Performing API Inventory And Discovery

执行 API 资产清点与发现

适用场景

• 在安全评估前测绘组织的完整 API 攻击面
• 识别由开发团队未经安全审查部署的影子 API（Shadow API）
• 发现仍可访问但无人维护的已废弃或僵尸 API 版本
• 发现通过移动应用、SPA 或微服务暴露的未记录 API 端点
• 为合规需求（PCI-DSS、SOC2、GDPR）建立 API 清单

不适用于 未经书面授权的场景。API 发现涉及扫描网络基础设施和分析流量。

前置条件

• 明确授权文件，指定目标域名和网络范围
• 被动流量捕获能力（网络分路器、代理或云流量镜像）
• 主动扫描工具：Amass、subfinder、httpx 和 nuclei
• JavaScript 分析工具：LinkFinder、JS-Miner 或自定义解析器
• 访问云控制台（AWS、Azure、GCP）进行 API 网关清点
• Burp Suite Professional 用于被动 API 端点发现

工作流程

步骤 1：通过流量分析进行被动 API 发现

import re
import json
from collections import defaultdict

# 从浏览器开发者工具或代理解析 HAR 文件
def analyze_har_for_apis(har_file_path):
    """从 HTTP 归档（HAR）文件中提取 API 端点。"""
    with open(har_file_path) as f:
        har = json.load(f)

    api_endpoints = defaultdict(lambda: {
        "methods": set(), "content_types": set(),
        "auth_types": set(), "count": 0
    })

    for entry in har["log"]["entries"]:
        url = entry["request"]["url"]
        method = entry["request"]["method"]

        # 识别 API 模式
        api_patterns = [
            r'/api/', r'/v\d+/', r'/graphql', r'/rest/',
            r'/ws/', r'/rpc/', r'/grpc', r'/json',
        ]

        if any(re.search(p, url) for p in api_patterns):
            # 规范化 URL（删除查询参数和 ID）
            normalized = re.sub(r'\?.*$', '', url)
            normalized = re.sub(r'/\d+(/|$)', '/{id}\\1', normalized)
            normalized = re.sub(
                r'/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
                '/{uuid}', normalized)

            ep = api_endpoints[normalized]
            ep["methods"].add(method)
            ep["count"] += 1

            # 检测认证类型
            for header in entry["request"]["headers"]:
                name = header["name"].lower()
                if name == "authorization":
                    if "bearer" in header["value"].lower():
                        ep["auth_types"].add("Bearer/JWT")
                    elif "basic" in header["value"].lower():
                        ep["auth_types"].add("Basic")
                elif name == "x-api-key":
                    ep["auth_types"].add("API Key")

            # 检测内容类型
            content_type = next(
                (h["value"] for h in entry["request"]["headers"]
                 if h["name"].lower() == "content-type"), None)
            if content_type:
                ep["content_types"].add(content_type.split(";")[0])

    print(f"发现 {len(api_endpoints)} 个唯一 API 端点：\n")
    for url, info in sorted(api_endpoints.items()):
        methods = ", ".join(sorted(info["methods"]))
        auth = ", ".join(info["auth_types"]) or "无"
        print(f"  [{methods}] {url}")
        print(f"    认证：{auth} | 请求次数：{info['count']}")

    return api_endpoints

步骤 2：主动 API 端点发现

# DNS 枚举发现 API 子域名
amass enum -d example.com -o amass_results.txt
subfinder -d example.com -o subfinder_results.txt

# 过滤与 API 相关的子域名
grep -iE '(api|rest|graphql|ws|gateway|backend|internal|staging|dev|v1|v2)' \
    amass_results.txt subfinder_results.txt | sort -u > api_subdomains.txt

# 检查哪些子域名处于活跃状态
cat api_subdomains.txt | httpx -status-code -content-length -title \
    -tech-detect -o live_apis.txt

# 在每个活跃子域名上探测常见 API 路径
cat api_subdomains.txt | while read domain; do
    for path in /api /api/v1 /api/v2 /graphql /swagger.json /openapi.json \
                /api-docs /docs /health /status /metrics /actuator; do
        curl -s -o /dev/null -w "%{http_code} %{url_effective}\n" \
            "https://${domain}${path}" 2>/dev/null | grep -v "^404"
    done
done

import requests
import concurrent.futures

def discover_api_endpoints(base_domains):
    """主动探测已发现域名上的 API 端点。"""

    # 待测试的常见 API 路径
    API_PATHS = [
        "/api", "/api/v1", "/api/v2", "/api/v3",
        "/graphql", "/gql", "/query",
        "/rest", "/json", "/rpc",
        "/swagger.json", "/swagger/v1/swagger.json",
        "/openapi.json", "/openapi.yaml", "/api-docs",
        "/docs", "/redoc", "/explorer",
        "/.well-known/openid-configuration",
        "/health", "/healthz", "/ready",
        "/status", "/info", "/version",
        "/metrics", "/prometheus",
        "/actuator", "/actuator/health", "/actuator/info",
        "/admin", "/admin/api", "/internal",
        "/debug", "/debug/vars", "/debug/pprof",
        "/ws", "/websocket", "/socket.io",
        "/grpc", "/twirp",
    ]

    discovered = []

    def check_endpoint(domain, path):
        for scheme in ["https", "http"]:
            url = f"{scheme}://{domain}{path}"
            try:
                resp = requests.get(url, timeout=5, allow_redirects=False,
                                  verify=False)
                if resp.status_code not in (404, 502, 503):
                    return {
                        "url": url,
                        "status": resp.status_code,
                        "content_type": resp.headers.get("Content-Type", ""),
                        "server": resp.headers.get("Server", ""),
                        "size": len(resp.content),
                    }
            except requests.exceptions.RequestException:
                pass
        return None

    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        futures = {}
        for domain in base_domains:
            for path in API_PATHS:
                future = executor.submit(check_endpoint, domain, path)
                futures[future] = (domain, path)

        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result:
                discovered.append(result)
                print(f"  [发现] {result['url']} -> {result['status']} ({result['content_type']})")

    return discovered

步骤 3：从 JavaScript 源码分析 API 端点

import re
import requests

def extract_apis_from_javascript(js_urls):
    """从 JavaScript 源文件中提取 API 端点。"""
    api_pattern = re.compile(
        r'''(?:['"`])((?:/api/|/v[0-9]+/|/graphql|/rest/)[^'"`\s<>{}]+)(?:['"`])''',
        re.IGNORECASE
    )
    url_pattern = re.compile(
        r'''(?:['"`])(https?://[a-zA-Z0-9._-]+(?:\.[a-zA-Z]{2,})+(?:/[^'"`\s<>{}]*)?)(?:['"`])'''
    )
    fetch_pattern = re.compile(
        r'''(?:fetch|axios|ajax|XMLHttpRequest|\.get|\.post|\.put|\.delete|\.patch)\s*\(\s*(?:['"`])([^'"`]+)'''
    )

    all_endpoints = set()

    for js_url in js_urls:
        try:
            resp = requests.get(js_url, timeout=10)
            content = resp.text

            # 提取相对 API 路径
            for match in api_pattern.findall(content):
                all_endpoints.add(("relative", match))

            # 提取绝对 URL
            for match in url_pattern.findall(content):
                if any(kw in match.lower() for kw in ["/api", "/v1", "/v2", "graphql"]):
                    all_endpoints.add(("absolute", match))

            # 从 fetch/axios 调用中提取
            for match in fetch_pattern.findall(content):
                all_endpoints.add(("fetch", match))

        except requests.exceptions.RequestException:
            pass

    print(f"\n从 JavaScript 中发现的 API 端点（{len(all_endpoints)} 个）：")
    for source, endpoint in sorted(all_endpoints):
        print(f"  [{source}] {endpoint}")

    return all_endpoints

# 从目标域名中查找 JavaScript 文件
def find_js_files(domain):
    """从 Web 应用中发现 JavaScript 文件。"""
    resp = requests.get(f"https://{domain}", timeout=10)
    js_files = re.findall(r'src=["']([^"']+\.js[^"']*)', resp.text)
    full_urls = []
    for js in js_files:
        if js.startswith("http"):
            full_urls.append(js)
        elif js.startswith("//"):
            full_urls.append(f"https:{js}")
        elif js.startswith("/"):
            full_urls.append(f"https://{domain}{js}")
    return full_urls

步骤 4：云 API 网关清点

import boto3

def inventory_aws_apis():
    """清点 AWS API Gateway 中的所有 API。"""
    apigw = boto3.client('apigateway')
    apigwv2 = boto3.client('apigatewayv2')

    apis = []

    # REST API（API Gateway v1）
    rest_apis = apigw.get_rest_apis()
    for api in rest_apis['items']:
        resources = apigw.get_resources(restApiId=api['id'])
        stages = apigw.get_stages(restApiId=api['id'])

        for stage in stages['item']:
            for resource in resources['items']:
                for method in resource.get('resourceMethods', {}).keys():
                    apis.append({
                        "type": "REST",
                        "name": api['name'],
                        "stage": stage['stageName'],
                        "path": resource['path'],
                        "method": method,
                        "url": f"https://{api['id']}.execute-api.{boto3.session.Session().region_name}.amazonaws.com/{stage['stageName']}{resource['path']}",
                        "created": str(api.get('createdDate', '')),
                    })

    # HTTP API（API Gateway v2）
    http_apis = apigwv2.get_apis()
    for api in http_apis['Items']:
        routes = apigwv2.get_routes(ApiId=api['ApiId'])
        stages = apigwv2.get_stages(ApiId=api['ApiId'])

        for route in routes['Items']:
            apis.append({
                "type": "HTTP",
                "name": api['Name'],
                "route": route['RouteKey'],
                "api_id": api['ApiId'],
                "protocol": api['ProtocolType'],
            })

    print(f"\nAWS API 清单（{len(apis)} 个端点）：")
    for api in apis:
        print(f"  [{api['type']}] {api.get('name')} - {api.get('method', '')} {api.get('path', api.get('route', ''))}")

    return apis

步骤 5：API 版本与影子 API 检测

def detect_shadow_and_zombie_apis(discovered_endpoints, documented_endpoints):
    """将发现的 API 与已记录的清单进行比对。"""

    # 规范化端点以便比对
    def normalize(ep):
        ep = re.sub(r'/v\d+/', '/vX/', ep)
        ep = re.sub(r'/\d+', '/{id}', ep)
        return ep.lower().rstrip('/')

    documented_normalized = {normalize(ep) for ep in documented_endpoints}

    shadow_apis = []  # 已发现但未记录
    zombie_apis = []  # 旧版本仍可访问

    for ep in discovered_endpoints:
        normalized = normalize(ep["url"])

        if normalized not in documented_normalized:
            # 检查是否为已记录 API 的旧版本
            if re.search(r'/v[0-9]+/', ep["url"]):
                zombie_apis.append(ep)
            else:
                shadow_apis.append(ep)

    print(f"\n影子 API（未记录）：{len(shadow_apis)} 个")
    for api in shadow_apis:
        print(f"  [影子] {api['url']} -> {api['status']}")

    print(f"\n僵尸 API（已废弃版本）：{len(zombie_apis)} 个")
    for api in zombie_apis:
        print(f"  [僵尸] {api['url']} -> {api['status']}")

    # 检查僵尸 API 是否缺少安全控制
    for api in zombie_apis:
        resp = requests.get(api["url"], timeout=5)
        if resp.status_code not in (401, 403):
            print(f"  [严重] 僵尸 API 无需认证即可访问：{api['url']}")

    return shadow_apis, zombie_apis

核心概念

工具与系统

• Amass：通过 DNS 枚举、网页爬取和 API 发现进行攻击面测绘的 OWASP 工具
• httpx：用于验证已发现域名并识别活跃 API 端点的快速 HTTP 探测工具
• nuclei：基于模板的扫描器，用于检测暴露的 API 文档、调试端点和配置错误的服务
• Swagger UI Detector：用于在组织内查找暴露的 Swagger/OpenAPI 文档端点的工具
• Akto：通过流量分析发现 API 并维护自动化清单的 API 安全平台

常见场景

场景：企业 API 攻击面评估

场景背景：一家大型企业有 200 多个开发团队使用微服务架构。安全团队怀疑有许多未记录的 API 暴露在互联网上，需要为安全审计进行全面的 API 清点。

方法：

1. DNS 枚举发现 340 个子域名，其中 45 个包含 API 相关关键词（api、rest、gateway、backend）
2. 使用 API 路径词典主动探测所有子域名，发现 127 个活跃 API 端点
3. 主 Web 应用的 JavaScript 分析揭示 34 个 API 端点，其中 8 个指向未记录的内部服务
4. AWS API Gateway 清点显示跨 12 个账户有 67 个 REST API 和 23 个 HTTP API
5. 与官方 API 目录交叉对比：31 个影子 API（未记录），14 个僵尸 API（已废弃版本）
6. 3 个僵尸 API 无认证，通过本应下线的端点暴露客户数据
7. 2 个影子 API 无需授权即可向互联网暴露内部管理功能

常见陷阱：

• 只检查已记录的 API 端点，遗漏在 API 网关之外部署的影子 API
• 未扫描前端应用程序硬编码 API 端点 URL 的 JavaScript 包
• 遗漏非标准端口或子路径后的 API
• 未检查多个 API 版本，其中旧版本可能缺乏安全控制
• 假设所有 API 都通过 API 网关，而实际上部分 API 可能直接暴露

输出格式

## API 清单与发现报告

**组织**：示例公司
**评估日期**：2024-12-15
**扫描域名**：340 个

### 摘要

| 类别 | 数量 |
|----------|-------|
| 发现的总 API 数 | 127 |
| 已记录的 API | 82 |
| 影子 API（未记录） | 31 |
| 僵尸 API（已废弃） | 14 |
| 无认证的 API | 8 |
| 暴露敏感数据的 API | 5 |

### 关键发现

1. **僵尸 API**：api-v1.example.com/api/v1/users - 2022 年已废弃，
   仍可访问，无需认证，返回完整用户数据
2. **影子 API**：internal-tools.example.com/api/admin - 管理功能
   无需授权暴露至互联网
3. **暴露的文档**：12 个 Swagger UI 实例可公开访问，
   泄露了完整的 API 模式和端点详情