implementing-api-security-posture-management

What this skill does

# Implementing API Security Posture Management

## Overview

API Security Posture Management (API-SPM) provides continuous visibility into an organization's API attack surface by automatically discovering, classifying, and risk-scoring all APIs including internal, external, partner, and shadow endpoints. Unlike point-in-time testing tools, API-SPM operates continuously to detect configuration drift, policy violations, missing security controls, sensitive data exposure, and compliance gaps. It aggregates findings from DAST, SAST, SCA, and runtime monitoring tools to provide a unified view of API risk posture across the organization.


## When to Use

- When deploying or configuring implementing api security posture management capabilities in your environment
- When establishing security controls aligned to compliance requirements
- When building or improving security architecture for this domain
- When conducting security assessments that require this implementation

## Prerequisites

- API gateway with traffic logging (Kong, AWS API Gateway, Apigee, Envoy)
- OpenAPI specifications for documented APIs
- SIEM or log aggregation platform (Splunk, Elastic)
- CI/CD pipeline access for shift-left integration
- Cloud provider APIs for infrastructure discovery
- Python 3.8+ for custom posture assessment tooling

## Core Components

### 1. API Discovery and Inventory

```python
#!/usr/bin/env python3
"""API Security Posture Management Engine

Continuously discovers, classifies, and risk-scores APIs
to maintain a comprehensive security posture inventory.
"""

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"
    INTERNAL = "internal"
    PARTNER = "partner"
    SHADOW = "shadow"
    DEPRECATED = "deprecated"

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]:
        """Evaluate security controls present on an API endpoint."""
        controls = []

        # Authentication check
        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="No authentication mechanism detected" if not has_auth else "Authentication present"
        ))

        # TLS/HTTPS check
        is_https = traffic_sample.get('scheme', '').lower() == 'https'
        controls.append(SecurityControl(
            name="transport_encryption",
            present=is_https,
            required=True,
            severity=RiskLevel.CRITICAL,
            details="API accessible over HTTP without TLS" if not is_https else "HTTPS enforced"
        ))

        # Rate limiting check
        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="No rate limiting headers detected" if not has_rate_limit else "Rate limiting active"
        ))

        # CORS policy check
        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 origin: {cors_origin}" if cors_origin else "No CORS headers"
        ))

        # Security headers
        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"Missing headers: {', '.join(missing)}" if missing else "All security headers present"
        ))

        # Input validation (check for schema validation errors in logs)
        has_validation = traffic_sample.get('has_schema_validation', False)
        controls.append(SecurityControl(
            name="input_validation",
            present=has_validation,
            required=True,
            severity=RiskLevel.HIGH,
            details="No schema validation detected" if not has_validation else "Input validation active"
        ))

        endpoint.security_controls = controls
        return controls

    def calculate_risk_score(self, endpoint: APIEndpoint) -> float:
        """Calculate a composite risk score (0-100) for an API endpoint."""
        score = 0.0
        max_score = 0.0

        # Security controls scoring
        for control in endpoint.security_controls:
            weight = control.severity.value * 5
            max_score += we