detecting-shadow-api-endpoints

name	detecting-shadow-api-endpoints
description	通过流量分析、代码扫描和API发现平台，发现和清点在已记录规范之外运行的影子API（Shadow API）端点。
domain	cybersecurity
subdomain	api-security
tags	["api-security","shadow-apis","api-discovery","undocumented-apis","zombie-apis","api-inventory","attack-surface-management","api-governance"]
version	1.0
author	mahipal
license	Apache-2.0

检测影子API端点

概述

影子API（Shadow API）是在组织环境中运行但未被追踪、记录或保护的API端点。它们来源于快速开发周期、被遗忘的测试环境、仍在运行的已废弃API版本、第三方集成，或未经治理部署的开发人员个人项目。影子API绕过认证和监控控制，为攻击者创造隐藏入口。研究表明，大型组织中多达30%的API端点未被记录，使得影子API检测成为API安全态势管理（API Security Posture Management）的关键组成部分。

前置条件

• 带流量日志的API网关或反向代理（Kong、AWS API Gateway、Envoy）
• 网络流量捕获能力（分组代理、端口镜像）
• 可访问源代码仓库和CI/CD管道配置
• 云提供商访问权限用于配置扫描（AWS、GCP、Azure）
• API文档清单（OpenAPI规范、Swagger文档）
• Python 3.8+用于自定义发现工具

检测方法

1. 流量分析与对比

将实时API流量与已记录的OpenAPI规范进行对比，识别未记录的端点：

#!/usr/bin/env python3
"""影子API端点检测器

将观察到的API流量模式与已记录的OpenAPI规范进行对比，
以识别未记录的（影子）端点。
"""

import json
import re
import yaml
import sys
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Set, Tuple, Optional
from dataclasses import dataclass, field

@dataclass
class DiscoveredEndpoint:
    method: str
    path_pattern: str
    first_seen: str
    last_seen: str
    request_count: int
    source_ips: Set[str] = field(default_factory=set)
    status_codes: Set[int] = field(default_factory=set)
    has_auth_header: bool = False
    documented: bool = False

class ShadowAPIDetector:
    # 用于参数化路径段的常见模式
    PARAM_PATTERNS = [
        (re.compile(r'/\d+'), '/{id}'),
        (re.compile(r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'), '/{uuid}'),
        (re.compile(r'/[a-zA-Z0-9]{20,40}'), ),
    ]

     ():
        .documented_endpoints: [[, ]] = ()
        .discovered_endpoints: [[, ], DiscoveredEndpoint] = {}

     ():
        
         (spec_path, )  f:
             spec_path.endswith():
                spec = json.load(f)
            :
                spec = yaml.safe_load(f)

        paths = spec.get(, {})
         path, methods  paths.items():
            
            normalized_path = re.sub(, , path)
             method  methods:
                 method.upper()  (, , , , , , ):
                    .documented_endpoints.add((method.upper(), normalized_path))

        ()

     () -> :
        
        
        path = path.split()[]

         pattern, replacement  .PARAM_PATTERNS:
            path = pattern.sub(replacement, path)

         path

     ():
        
        patterns = {
            : re.(
                
                
            ),
            :   
        }

         (log_file, )  f:
             line  f:
                 log_format == :
                    :
                        entry = json.loads(line)
                        method = entry.get(, entry.get(, ))
                        path = entry.get(, entry.get(, ))
                        status = (entry.get(, entry.get(, )))
                        ip = entry.get(, entry.get(, ))
                        timestamp = entry.get(, entry.get(, ))
                        has_auth = (entry.get(, entry.get(, )))
                     json.JSONDecodeError:
                        
                :
                     = patterns[log_format].(line)
                      :
                        
                    method = .group()
                    path = .group()
                    status = (.group())
                    ip = .group()
                    timestamp = .group()
                    has_auth =   line

                
                  path.startswith()   path.startswith():
                    

                normalized = .normalize_path(path)
                key = (method.upper(), normalized)

                 key   .discovered_endpoints:
                    .discovered_endpoints[key] = DiscoveredEndpoint(
                        method=method.upper(),
                        path_pattern=normalized,
                        first_seen=timestamp,
                        last_seen=timestamp,
                        request_count=,
                        documented=(key  .documented_endpoints)
                    )

                endpoint = .discovered_endpoints[key]
                endpoint.request_count += 
                endpoint.last_seen = timestamp
                endpoint.source_ips.add(ip)
                endpoint.status_codes.add(status)
                 has_auth:
                    endpoint.has_auth_header = 

     () -> [DiscoveredEndpoint]:
        
        shadows = []
         key, endpoint  .discovered_endpoints.items():
              endpoint.documented:
                shadows.append(endpoint)

        
        shadows.sort(key= e: e.request_count, reverse=)
         shadows

     () -> :
        
        risk_score = 

        
          endpoint.has_auth_header:
            risk_score += 

        
         endpoint.request_count > :
            risk_score += 
         endpoint.request_count > :
            risk_score += 

        
         (endpoint.source_ips) > :
            risk_score += 

        
           endpoint.status_codes    endpoint.status_codes:
            risk_score += 

        
         endpoint.method  (, , , ):
            risk_score += 

        
        sensitive_patterns = [, , , , ,
                            , , , , ]
         pattern  sensitive_patterns:
             pattern  endpoint.path_pattern.lower():
                risk_score += 
                

         risk_score >= :
             
         risk_score >= :
             
         risk_score >= :
             
         

     () -> :
        
        shadows = .identify_shadow_apis()
        total_documented = (.documented_endpoints)
        total_discovered = (.discovered_endpoints)

        report = {
            : datetime.now().isoformat(),
            : {
                : total_documented,
                : total_discovered,
                : (shadows),
                : ,
            },
            : []
        }

         endpoint  shadows:
            risk = .classify_risk(endpoint)
            report[].append({
                : endpoint.method,
                : endpoint.path_pattern,
                : risk,
                : endpoint.request_count,
                : (endpoint.source_ips),
                : endpoint.has_auth_header,
                : (endpoint.status_codes),
                : endpoint.first_seen,
                : endpoint.last_seen,
            })

         report


 ():
    detector = ShadowAPIDetector()

    
    spec_files = sys.argv[:]  (sys.argv) >   []
     spec  spec_files:
         spec.endswith((, , )):
            detector.load_openapi_spec(spec)

    
    detector.process_access_log()

    report = detector.generate_report()

    ()
    ()
    ()
    ()
    ()
    ()
    ()

     ep  report[]:
        risk_marker = {: , : , : , : }
        ()
        ()

    
     (, )  f:
        json.dump(report, f, indent=, default=)
    ()


 __name__ == :
    main()

2. 云配置扫描

# AWS：发现不在文档中的API Gateway端点
aws apigateway get-rest-apis --query 'items[*].[name,id]' --output table

# 列出每个API的所有路由
aws apigatewayv2 get-apis --query 'Items[*].[Name,ApiId,ProtocolType]' --output table

# AWS Lambda函数URL（潜在的影子API）
aws lambda list-function-url-configs --function-name "*" 2>/dev/null

# 查找路由到未记录后端的ALB监听器规则
aws elbv2 describe-rules --listener-arn $LISTENER_ARN \
  --query 'Rules[*].[Priority,Conditions[0].Values[0],Actions[0].TargetGroupArn]'

3. 源代码仓库挖掘

# 在源代码中搜索未记录的路由定义
# Express.js路由
grep -rn "app\.\(get\|post\|put\|delete\|patch\)" --include="*.js" --include="*.ts" src/

# Flask/Django路由
grep -rn "@app\.route\|@api\.route\|path(" --include="*.py" src/

# Spring Boot端点
grep -rn "@\(Get\|Post\|Put\|Delete\|Patch\)Mapping\|@RequestMapping" --include="*.java" src/

# 将找到的路由与OpenAPI规范对比
diff <(grep -roh "'/api/[^']*'" src/ | sort -u) \
     <(yq '.paths | keys[]' openapi.yaml | sort -u)

防护与治理

API注册网关策略

# Kong插件配置 - 拒绝未注册的路由
plugins:
  - name: request-validator
    config:
      allowed_content_types:
        - application/json
      body_schema: null
  - name: pre-function
    config:
      access:
        - |
          -- 阻止对未注册端点的请求
          local registered = kong.cache:get("registered_endpoints")
          local path = kong.request.get_path()
          local method = kong.request.get_method()
          local key = method .. ":" .. path
          if not registered[key] then
            kong.log.warn("Shadow API access attempt: ", key)
            return kong.response.exit(404, {error = "Endpoint not registered"})
          end

参考资料

• APIsec Shadow API Best Practices: https://www.apisec.ai/blog/secure-your-shadow-apis-best-practices-for-api-discovery
• Wiz Shadow API Guide: https://www.wiz.io/academy/api-security/shadow-api
• Checkmarx Shadow and Zombie APIs: https://checkmarx.com/learn/api-security/shadow-zombie-apis-undocumented-api-vulnerabilities-threaten-security-posture/
• Treblle Shadow API Tools: https://treblle.com/blog/top-tools-for-detecting-shadow-apis-and-how-treblle-differs
• SecureLayer7 Shadow APIs: https://blog.securelayer7.net/shadow-apis-explained-risks-detection-and-prevention/