Implementing API Security Posture Management

实施API安全态势管理

概述

API安全态势管理（API Security Posture Management，API-SPM）通过自动发现、分类和风险评分组织内所有API（包括内部、外部、合作伙伴和影子端点），持续监控API攻击面。与点时间测试工具不同，API-SPM持续运行，检测配置漂移（Configuration Drift）、策略违规、缺失安全控制、敏感数据暴露和合规差距。它从DAST、SAST、SCA和运行时监控工具聚合发现结果，提供整个组织API风险态势的统一视图。

前置条件

• 带流量日志的API网关（Kong、AWS API Gateway、Apigee、Envoy）
• 已记录API的OpenAPI规范
• SIEM或日志聚合平台（Splunk、Elastic）
• CI/CD流水线访问权限（安全左移集成）
• 用于基础设施发现的云提供商API
• Python 3.8+（自定义态势评估工具）

核心组件

1. API发现与清单管理

#!/usr/bin/env python3
"""API安全态势管理引擎

持续发现、分类和风险评分API，
以维护全面的安全态势清单。
"""

import json
import re
import hashlib
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum

class APIClassification(Enum):
    EXTERNAL = "external"     # 外部API
    INTERNAL = "internal"     # 内部API
    PARTNER = "partner"       # 合作伙伴API
    SHADOW = "shadow"         # 影子API
    DEPRECATED = "deprecated" # 已弃用API

class RiskLevel(Enum):
    CRITICAL = 4  # 严重
    HIGH = 3      # 高
    MEDIUM = 2    # 中
    LOW = 1       # 低
    INFO = 0      # 信息

@dataclass
class SecurityControl:
    name: str
    present: bool
    required: bool
    severity: RiskLevel
    details: str = ""

@dataclass
class APIEndpoint:
    api_id: str
    method: str
    path: str
    service_name: str
    classification: APIClassification
    owner: Optional[str] = None
    version: Optional[str] = None
    first_discovered: str = ""
    last_seen: str = ""
    documented: bool = False
    security_controls: List[SecurityControl] = field(default_factory=list)
    risk_score: float = 0.0
    sensitive_data_types: Set[str] = field(default_factory=set)
    compliance_tags: Set[str] = field(default_factory=set)
    traffic_volume_daily: int = 0

class APIPostureManager:
    SENSITIVE_PATTERNS = {
        "ssn": re.compile(r'\b\d{3}-\d{2}-\d{4}\b'),
        "credit_card": re.compile(r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b'),
        "email": re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
        "api_key": re.compile(r'\b[A-Za-z0-9]{32,}\b'),
        "jwt": re.compile(r'eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+'),
        "phone": re.compile(r'\b\+?1?\d{10,15}\b'),
    }

    def __init__(self):
        self.inventory: Dict[str, APIEndpoint] = {}
        self.policy_rules: List[dict] = []

    def generate_api_id(self, method: str, path: str, service: str) -> str:
        raw = f"{service}:{method}:{path}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

    def register_api(self, method: str, path: str, service_name: str,
                     classification: APIClassification,
                     documented: bool = False, owner: str = None) -> APIEndpoint:
        api_id = self.generate_api_id(method, path, service_name)
        now = datetime.now().isoformat()

        if api_id in self.inventory:
            endpoint = self.inventory[api_id]
            endpoint.last_seen = now
            return endpoint

        endpoint = APIEndpoint(
            api_id=api_id,
            method=method,
            path=path,
            service_name=service_name,
            classification=classification,
            owner=owner,
            first_discovered=now,
            last_seen=now,
            documented=documented
        )
        self.inventory[api_id] = endpoint
        return endpoint

    def assess_security_controls(self, endpoint: APIEndpoint,
                                  traffic_sample: dict) -> List[SecurityControl]:
        """评估API端点上存在的安全控制措施。"""
        controls = []

        # 认证检查
        has_auth = any(h in traffic_sample.get('request_headers', {})
                      for h in ['Authorization', 'X-API-Key', 'Cookie'])
        controls.append(SecurityControl(
            name="authentication",
            present=has_auth,
            required=True,
            severity=RiskLevel.CRITICAL,
            details="未检测到认证机制" if not has_auth else "认证已存在"
        ))

        # TLS/HTTPS检查
        is_https = traffic_sample.get('scheme', '').lower() == 'https'
        controls.append(SecurityControl(
            name="transport_encryption",
            present=is_https,
            required=True,
            severity=RiskLevel.CRITICAL,
            details="API通过无TLS的HTTP访问" if not is_https else "HTTPS已强制执行"
        ))

        # 速率限制检查
        has_rate_limit = any(h.startswith('X-RateLimit') or h == 'Retry-After'
                           for h in traffic_sample.get('response_headers', {}).keys())
        controls.append(SecurityControl(
            name="rate_limiting",
            present=has_rate_limit,
            required=True,
            severity=RiskLevel.HIGH,
            details="未检测到速率限制头" if not has_rate_limit else "速率限制已启用"
        ))

        # CORS策略检查
        cors_origin = traffic_sample.get('response_headers', {}).get('Access-Control-Allow-Origin', '')
        has_strict_cors = cors_origin and cors_origin != '*'
        controls.append(SecurityControl(
            name="cors_policy",
            present=has_strict_cors,
            required=endpoint.classification == APIClassification.EXTERNAL,
            severity=RiskLevel.HIGH if cors_origin == '*' else RiskLevel.MEDIUM,
            details=f"CORS来源: {cors_origin}" if cors_origin else "无CORS头"
        ))

        # 安全响应头检查
        sec_headers = traffic_sample.get('response_headers', {})
        required_headers = {
            'X-Content-Type-Options': 'nosniff',
            'Strict-Transport-Security': None,
            'X-Frame-Options': None,
            'Cache-Control': 'no-store',
        }
        missing = [h for h in required_headers if h not in sec_headers]
        controls.append(SecurityControl(
            name="security_headers",
            present=len(missing) == 0,
            required=True,
            severity=RiskLevel.MEDIUM,
            details=f"缺少响应头: {', '.join(missing)}" if missing else "所有安全响应头均已存在"
        ))

        # 输入验证检查（从日志检测Schema验证错误）
        has_validation = traffic_sample.get('has_schema_validation', False)
        controls.append(SecurityControl(
            name="input_validation",
            present=has_validation,
            required=True,
            severity=RiskLevel.HIGH,
            details="未检测到Schema验证" if not has_validation else "输入验证已启用"
        ))

        endpoint.security_controls = controls
        return controls

    def calculate_risk_score(self, endpoint: APIEndpoint) -> float:
        """计算API端点的综合风险分数（0-100）。"""
        score = 0.0
        max_score = 0.0

        # 安全控制评分
        for control in endpoint.security_controls:
            weight = control.severity.value * 5
            max_score += weight
            if not control.present and control.required:
                score += weight

        # 分类风险乘数
        classification_weights = {
            APIClassification.EXTERNAL: 1.5,
            APIClassification.PARTNER: 1.3,
            APIClassification.SHADOW: 2.0,
            APIClassification.DEPRECATED: 1.8,
            APIClassification.INTERNAL: 1.0,
        }
        multiplier = classification_weights.get(endpoint.classification, 1.0)

        # 未文档化惩罚
        if not endpoint.documented:
            score += 10

        # 敏感数据惩罚
        score += len(endpoint.sensitive_data_types) * 5

        # 归一化到0-100
        if max_score > 0:
            normalized = min(100, (score / max_score) * 100 * multiplier)
        else:
            normalized = 0

        endpoint.risk_score = round(normalized, 1)
        return endpoint.risk_score

    def generate_posture_report(self) -> dict:
        """生成整个组织的API安全态势报告。"""
        total = len(self.inventory)
        if total == 0:
            return {"error": "清单中无API"}

        risk_distribution = {level.name: 0 for level in RiskLevel}
        classification_counts = {c.value: 0 for c in APIClassification}
        undocumented = 0
        missing_auth = 0
        missing_tls = 0

        for endpoint in self.inventory.values():
            self.calculate_risk_score(endpoint)

            if endpoint.risk_score >= 75:
                risk_distribution["CRITICAL"] += 1
            elif endpoint.risk_score >= 50:
                risk_distribution["HIGH"] += 1
            elif endpoint.risk_score >= 25:
                risk_distribution["MEDIUM"] += 1
            else:
                risk_distribution["LOW"] += 1

            classification_counts[endpoint.classification.value] += 1

            if not endpoint.documented:
                undocumented += 1

            for control in endpoint.security_controls:
                if control.name == "authentication" and not control.present:
                    missing_auth += 1
                if control.name == "transport_encryption" and not control.present:
                    missing_tls += 1

        avg_risk = sum(e.risk_score for e in self.inventory.values()) / total

        return {
            "report_date": datetime.now().isoformat(),
            "total_apis": total,
            "average_risk_score": round(avg_risk, 1),
            "risk_distribution": risk_distribution,
            "classification": classification_counts,
            "undocumented_apis": undocumented,
            "missing_authentication": missing_auth,
            "missing_tls": missing_tls,
            "top_risks": sorted(
                [{"api_id": e.api_id, "method": e.method, "path": e.path,
                  "service": e.service_name, "risk_score": e.risk_score,
                  "classification": e.classification.value}
                 for e in self.inventory.values()],
                key=lambda x: x["risk_score"],
                reverse=True
            )[:20]
        }

2. 策略执行

在所有API中定义并强制执行安全策略：

# api-security-policies.yaml
policies:
  - name: require-authentication
    description: 所有外部API必须要求认证
    scope:
      classification: [external, partner]
    rule:
      control: authentication
      required: true
    severity: critical
    remediation: "添加OAuth2、API密钥或JWT认证"

  - name: enforce-tls
    description: 所有API必须使用HTTPS
    scope:
      classification: [external, internal, partner]
    rule:
      control: transport_encryption
      required: true
    severity: critical
    remediation: "配置TLS证书并将HTTP重定向到HTTPS"

  - name: require-rate-limiting
    description: 外部API必须实施速率限制
    scope:
      classification: [external]
    rule:
      control: rate_limiting
      required: true
    severity: high
    remediation: "在API网关层面配置速率限制"

  - name: no-wildcard-cors
    description: API不得使用通配符CORS来源
    scope:
      classification: [external]
    rule:
      control: cors_policy
      condition: "origin != '*'"
    severity: high
    remediation: "在CORS配置中指定明确允许的来源"

  - name: documentation-required
    description: 所有API必须有OpenAPI文档
    scope:
      classification: [external, partner]
    rule:
      documented: true
    severity: medium
    remediation: "创建并发布OpenAPI规范"

  - name: deprecation-sunset
    description: 已弃用API必须包含Sunset响应头
    scope:
      classification: [deprecated]
    rule:
      header_present: "Sunset"
    severity: medium
    remediation: "添加包含计划下线日期的Sunset响应头"

持续监控仪表盘指标

参考资料

• OX Security ASPM指南2025: https://www.ox.security/blog/application-security-posture-management-aspm/
• IBM ASPM概述: https://www.ibm.com/think/topics/aspm
• StackHawk最佳ASPM工具: https://www.stackhawk.com/blog/best-aspm-tools/
• AppSentinels API安全态势管理: https://appsentinels.ai/blog/api-security-posture-management-from-reactive-protection-to-continuous-governance/
• Palo Alto Networks ASPM: https://www.paloaltonetworks.com/cyberpedia/aspm-application-security-posture-management