| 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规范进行对比，识别未记录的端点：
"""影子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 apigateway get-rest-apis --query 'items[*].[name,id]' --output table
aws apigatewayv2 get-apis --query 'Items[*].[Name,ApiId,ProtocolType]' --output table
aws lambda list-function-url-configs --function-name "*" 2>/dev/null
aws elbv2 describe-rules --listener-arn $LISTENER_ARN \
--query 'Rules[*].[Priority,Conditions[0].Values[0],Actions[0].TargetGroupArn]'
3. 源代码仓库挖掘
grep -rn "app\.\(get\|post\|put\|delete\|patch\)" --include="*.js" --include="*.ts" src/
grep -rn "@app\.route\|@api\.route\|path(" --include="*.py" src/
grep -rn "@\(Get\|Post\|Put\|Delete\|Patch\)Mapping\|@RequestMapping" --include="*.java" src/
diff <(grep -roh "'/api/[^']*'" src/ | sort -u) \
<(yq '.paths | keys[]' openapi.yaml | sort -u)
防护与治理
API注册网关策略
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
参考资料
