API Security in 2026: The New API Threat Landscape Explained

Complete Python API Security Scanner:

#!/usr/bin/env python3
"""
Production-ready API Security Scanner
Discovers shadow APIs, validates security configurations, and monitors threats
"""

import requests
import json
import time
import hashlib
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, asdict
from datetime import datetime
import logging
from urllib.parse import urljoin, urlparse
import re

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class APIDiscoveryResult:
    """Result of API endpoint discovery."""
    endpoint: str
    method: str
    status_code: int
    requires_auth: bool
    content_type: Optional[str] = None
    response_size: int = 0
    is_documented: bool = False
    security_headers: Dict[str, str] = None


@dataclass
class SecurityFinding:
    """Security finding for an API endpoint."""
    endpoint: str
    severity: str  # "low", "medium", "high", "critical"
    issue: str
    recommendation: str
    details: Optional[Dict] = None


class APISecurityScanner:
    """Comprehensive API security scanner."""
    
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        """Initialize API security scanner.
        
        Args:
            base_url: Base URL of the API
            api_key: Optional API key for authenticated requests
        """
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        if api_key:
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})
        self.discovered_endpoints: Set[str] = set()
        self.security_findings: List[SecurityFinding] = []
        
    def discover_endpoints_from_openapi(self, openapi_path: str = '/.well-known/openapi.json') -> List[APIDiscoveryResult]:
        """Discover endpoints from OpenAPI specification.
        
        Args:
            openapi_path: Path to OpenAPI spec
            
        Returns:
            List of discovered endpoints
        """
        results = []
        try:
            url = urljoin(self.base_url, openapi_path)
            response = self.session.get(url, timeout=10)
            
            if response.status_code != 200:
                logger.warning(f"OpenAPI spec not found at {url}")
                return results
            
            spec = response.json()
            paths = spec.get('paths', {})
            
            for path, methods in paths.items():
                for method, details in methods.items():
                    if method.lower() in ['get', 'post', 'put', 'delete', 'patch']:
                        full_url = urljoin(self.base_url, path)
                        self.discovered_endpoints.add(f"{method.upper()} {full_url}")
                        results.append(APIDiscoveryResult(
                            endpoint=full_url,
                            method=method.upper(),
                            status_code=200,  # Assumed documented
                            requires_auth='security' in details or '401' in str(details),
                            is_documented=True
                        ))
            
            logger.info(f"Discovered {len(results)} documented endpoints from OpenAPI")
        except Exception as e:
            logger.error(f"Error discovering endpoints from OpenAPI: {e}")
        
        return results
    
    def test_endpoint_security(self, endpoint: str, method: str = 'GET') -> SecurityFinding:
        """Test security of a single endpoint.
        
        Args:
            endpoint: Endpoint URL
            method: HTTP method
            
        Returns:
            SecurityFinding object
        """
        findings = []
        
        # Test 1: Check authentication
        test_session = requests.Session()
        try:
            response = test_session.request(
                method,
                endpoint,
                timeout=10,
                allow_redirects=False
            )
            
            if response.status_code == 200:
                findings.append(SecurityFinding(
                    endpoint=endpoint,
                    severity='critical',
                    issue='Endpoint accessible without authentication',
                    recommendation='Require authentication for this endpoint',
                    details={'status_code': response.status_code}
                ))
        except Exception as e:
            logger.error(f"Error testing endpoint {endpoint}: {e}")
        
        # Test 2: Check security headers
        try:
            response = self.session.request(method, endpoint, timeout=10)
            security_headers = {
                'strict-transport-security': response.headers.get('Strict-Transport-Security'),
                'x-content-type-options': response.headers.get('X-Content-Type-Options'),
                'x-frame-options': response.headers.get('X-Frame-Options'),
                'content-security-policy': response.headers.get('Content-Security-Policy'),
            }
            
            missing_headers = [k for k, v in security_headers.items() if not v]
            if missing_headers:
                findings.append(SecurityFinding(
                    endpoint=endpoint,
                    severity='medium',
                    issue=f'Missing security headers: {", ".join(missing_headers)}',
                    recommendation='Add recommended security headers',
                    details={'missing_headers': missing_headers}
                ))
        except Exception as e:
            logger.error(f"Error checking headers for {endpoint}: {e}")
        
        # Test 3: Check for sensitive data exposure
        try:
            response = self.session.request(method, endpoint, timeout=10)
            if response.status_code == 200:
                content = response.text.lower()
                sensitive_patterns = [
                    r'password["\s]*[:=]["\s]*[^,}\s]+',
                    r'api[_-]?key["\s]*[:=]["\s]*[^,}\s]+',
                    r'token["\s]*[:=]["\s]*[^,}\s]+',
                    r'secret["\s]*[:=]["\s]*[^,}\s]+',
                ]
                
                for pattern in sensitive_patterns:
                    if re.search(pattern, content):
                        findings.append(SecurityFinding(
                            endpoint=endpoint,
                            severity='high',
                            issue='Potential sensitive data exposure in response',
                            recommendation='Review response content and remove sensitive data',
                            details={'pattern_matched': pattern}
                        ))
                        break
        except Exception as e:
            logger.error(f"Error checking for sensitive data: {e}")
        
        return findings[0] if findings else None
    
    def scan_shadow_apis(self, common_paths: List[str] = None) -> List[SecurityFinding]:
        """Scan for shadow/undocumented APIs.
        
        Args:
            common_paths: List of common API paths to test
            
        Returns:
            List of security findings
        """
        if common_paths is None:
            common_paths = [
                '/api/v1', '/api/v2', '/v1', '/v2',
                '/internal', '/admin', '/dev', '/test',
                '/graphql', '/graphiql', '/swagger', '/docs'
            ]
        
        findings = []
        for path in common_paths:
            full_url = urljoin(self.base_url, path)
            endpoint_key = f"GET {full_url}"
            
            if endpoint_key not in self.discovered_endpoints:
                try:
                    response = self.session.get(full_url, timeout=10, allow_redirects=False)
                    
                    # If endpoint responds, it might be a shadow API
                    if response.status_code in [200, 401, 403]:
                        finding = SecurityFinding(
                            endpoint=full_url,
                            severity='high',
                            issue='Potential shadow/undocumented API endpoint discovered',
                            recommendation='Document endpoint or remove if not needed',
                            details={
                                'status_code': response.status_code,
                                'content_length': len(response.content)
                            }
                        )
                        findings.append(finding)
                        logger.warning(f"Shadow API discovered: {full_url}")
                except Exception as e:
                    logger.debug(f"Endpoint {full_url} not accessible: {e}")
        
        return findings
    
    def generate_report(self) -> Dict:
        """Generate comprehensive security report.
        
        Returns:
            Security report dictionary
        """
        return {
            'scan_timestamp': datetime.utcnow().isoformat(),
            'base_url': self.base_url,
            'endpoints_discovered': len(self.discovered_endpoints),
            'findings': {
                'critical': len([f for f in self.security_findings if f.severity == 'critical']),
                'high': len([f for f in self.security_findings if f.severity == 'high']),
                'medium': len([f for f in self.security_findings if f.severity == 'medium']),
                'low': len([f for f in self.security_findings if f.severity == 'low']),
            },
            'detailed_findings': [asdict(f) for f in self.security_findings]
        }


# Example usage
if __name__ == '__main__':
    scanner = APISecurityScanner(base_url='https://api.example.com')
    
    # Discover documented endpoints
    documented = scanner.discover_endpoints_from_openapi()
    
    # Scan for shadow APIs
    shadow_findings = scanner.scan_shadow_apis()
    scanner.security_findings.extend(shadow_findings)
    
    # Test security of discovered endpoints
    for endpoint_result in documented[:5]:  # Test first 5
        finding = scanner.test_endpoint_security(
            endpoint_result.endpoint,
            endpoint_result.method
        )
        if finding:
            scanner.security_findings.append(finding)
    
    # Generate report
    report = scanner.generate_report()
    print(json.dumps(report, indent=2))

Complete Node.js API Security Middleware:

// requirements: npm install express express-rate-limit jsonwebtoken express-validator helmet

const express = require('express');
const jwt = require('jsonwebtoken');
const rateLimit = require('express-rate-limit');
const { body, validationResult } = require('express-validator');
const helmet = require('helmet');

const app = express();

// Security headers
app.use(helmet());

// JWT authentication middleware
function authenticateToken(req, res, next) {
    const authHeader = req.headers['authorization'];
    const token = authHeader && authHeader.split(' ')[1];
    
    if (!token) {
        return res.status(401).json({ error: 'Authentication required' });
    }
    
    jwt.verify(token, process.env.JWT_SECRET, (err, user) => {
        if (err) {
            return res.status(403).json({ error: 'Invalid token' });
        }
        req.user = user;
        next();
    });
}

// Rate limiting per endpoint
const apiLimiter = rateLimit({
    windowMs: 15 * 60 * 1000, // 15 minutes
    max: 100, // 100 requests per window
    message: 'Too many requests from this IP'
});

// Schema validation middleware
function validateSchema(schema) {
    return [
        ...schema.map(field => body(field.name).isLength({ min: field.min, max: field.max })),
        (req, res, next) => {
            const errors = validationResult(req);
            if (!errors.isEmpty()) {
                return res.status(400).json({ errors: errors.array() });
            }
            next();
        }
    ];
}

// Protected endpoint example
app.post('/api/data', 
    authenticateToken,
    apiLimiter,
    validateSchema([
        { name: 'field1', min: 1, max: 100 },
        { name: 'field2', min: 1, max: 50 }
    ]),
    (req, res) => {
        res.json({ success: true, data: req.body });
    }
);

app.listen(3000, () => {
    console.log('Secure API server running on port 3000');
});

curl -i https://api.example.com/v1/data

curl -s -X POST https://api.example.com/graphql \
  -H "Content-Type: application/json" \
  -d '{"query":"{a{b{c{d{e}}}}}"}'

import json
import hashlib
from datetime import datetime
from typing import Dict, Any

def log_api_request(
    request_id: str,
    user: str,
    path: str,
    method: str,
    status: int,
    size: int,
    headers: Dict[str, str]
) -> Dict[str, Any]:
    """Log API request with privacy protection"""
    # Hash sensitive fields
    safe_headers = {}
    for key, value in headers.items():
        if key.lower() in ['authorization', 'cookie', 'x-api-key']:
            safe_headers[key] = hashlib.sha256(value.encode()).hexdigest()[:16]
        else:
            safe_headers[key] = value
    
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "request_id": request_id,
        "user": user,
        "path": path,
        "method": method,
        "status": status,
        "size": size,
        "headers": safe_headers
    }
    
    return log_entry

#!/usr/bin/env python3
"""
Advanced API Security Framework
Production-ready API security with comprehensive protection
"""

from typing import List, Dict, Optional, Set
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
from collections import defaultdict
import logging
import json
import hashlib
import hmac
import jwt
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SecurityLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class APISecurityEvent:
    """API security event record."""
    event_id: str
    timestamp: datetime
    endpoint: str
    method: str
    source_ip: str
    user_id: Optional[str] = None
    event_type: str = "access"
    severity: SecurityLevel = SecurityLevel.LOW
    details: Dict = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'severity': self.severity.value,
            'timestamp': self.timestamp.isoformat()
        }

class AdvancedAPISecurityFramework:
    """Production-ready API security framework."""
    
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
        self.rate_limits: Dict[str, Dict] = defaultdict(lambda: {'count': 0, 'reset_time': time.time()})
        self.security_events: List[APISecurityEvent] = []
        self.blocked_ips: Set[str] = set()
        self.allowed_endpoints: Set[str] = set()
    
    def validate_jwt(self, token: str) -> Optional[Dict]:
        """Validate JWT token.
        
        Args:
            token: JWT token string
            
        Returns:
            Decoded token payload if valid, None otherwise
        """
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=['HS256'],
                options={'verify_exp': True}
            )
            return payload
        except jwt.ExpiredSignatureError:
            logger.warning("JWT token expired")
            return None
        except jwt.InvalidTokenError as e:
            logger.warning(f"Invalid JWT token: {e}")
            return None
    
    def check_rate_limit(self, identifier: str, limit: int = 100, window: int = 60) -> bool:
        """Check rate limit for identifier.
        
        Args:
            identifier: Rate limit identifier (IP, user, etc.)
            limit: Request limit
            window: Time window in seconds
            
        Returns:
            True if within limit, False if exceeded
        """
        current_time = time.time()
        rate_data = self.rate_limits[identifier]
        
        # Reset if window expired
        if current_time > rate_data['reset_time']:
            rate_data['count'] = 0
            rate_data['reset_time'] = current_time + window
        
        # Check limit
        if rate_data['count'] >= limit:
            logger.warning(f"Rate limit exceeded for {identifier}")
            return False
        
        rate_data['count'] += 1
        return True
    
    def detect_anomaly(self, event: APISecurityEvent) -> bool:
        """Detect anomalous API access patterns.
        
        Args:
            event: API security event
            
        Returns:
            True if anomaly detected
        """
        # Check for rapid requests from same IP
        recent_events = [
            e for e in self.security_events
            if e.source_ip == event.source_ip and
            (event.timestamp - e.timestamp).total_seconds() < 10
        ]
        
        if len(recent_events) > 20:
            event.severity = SecurityLevel.HIGH
            event.details['anomaly'] = 'rapid_requests'
            return True
        
        # Check for access to blocked endpoints
        if event.endpoint not in self.allowed_endpoints and len(self.allowed_endpoints) > 0:
            event.severity = SecurityLevel.MEDIUM
            event.details['anomaly'] = 'unauthorized_endpoint'
            return True
        
        return False
    
    def log_security_event(self, event: APISecurityEvent):
        """Log security event.
        
        Args:
            event: Security event to log
        """
        self.security_events.append(event)
        
        # Detect anomalies
        if self.detect_anomaly(event):
            logger.warning(f"Anomaly detected: {event.event_id}")
        
        # Auto-block on critical events
        if event.severity == SecurityLevel.CRITICAL:
            self.blocked_ips.add(event.source_ip)
            logger.error(f"IP blocked: {event.source_ip}")
    
    def is_blocked(self, ip: str) -> bool:
        """Check if IP is blocked.
        
        Args:
            ip: IP address to check
            
        Returns:
            True if blocked
        """
        return ip in self.blocked_ips
    
    def get_statistics(self) -> Dict:
        """Get security statistics.
        
        Returns:
            Statistics dictionary
        """
        return {
            'total_events': len(self.security_events),
            'blocked_ips': len(self.blocked_ips),
            'by_severity': {
                level.value: len([e for e in self.security_events if e.severity == level])
                for level in SecurityLevel
            },
            'rate_limited': len([k for k, v in self.rate_limits.items() if v['count'] >= 100])
        }
    
    def cleanup(self):
        """Clean up resources."""
        logger.info("Cleaning up API security framework resources")

# Example usage
if __name__ == "__main__":
    framework = AdvancedAPISecurityFramework(secret_key="your-secret-key")
    
    # Validate JWT
    token = jwt.encode({'user_id': '123', 'exp': time.time() + 3600}, framework.secret_key, algorithm='HS256')
    payload = framework.validate_jwt(token)
    print(f"Token valid: {payload is not None}")
    
    # Check rate limit
    allowed = framework.check_rate_limit("192.168.1.100", limit=10)
    print(f"Rate limit check: {allowed}")
    
    # Log event
    event = APISecurityEvent(
        event_id="EVT-001",
        timestamp=datetime.now(),
        endpoint="/api/users",
        method="GET",
        source_ip="192.168.1.100"
    )
    framework.log_security_event(event)
    
    stats = framework.get_statistics()
    print(f"Statistics: {json.dumps(stats, indent=2)}")

#!/usr/bin/env python3
"""
Unit tests for API Security Framework
"""

import pytest
import jwt
import time
from datetime import datetime
from api_security_framework import (
    AdvancedAPISecurityFramework, APISecurityEvent, SecurityLevel
)

class TestAPISecurityFramework:
    """Tests for AdvancedAPISecurityFramework."""
    
    @pytest.fixture
    def framework(self):
        return AdvancedAPISecurityFramework(secret_key="test-secret-key")
    
    def test_validate_jwt(self, framework):
        """Test JWT validation."""
        token = jwt.encode(
            {'user_id': '123', 'exp': time.time() + 3600},
            framework.secret_key,
            algorithm='HS256'
        )
        payload = framework.validate_jwt(token)
        assert payload is not None
        assert payload['user_id'] == '123'
    
    def test_rate_limit(self, framework):
        """Test rate limiting."""
        identifier = "test-ip"
        for i in range(10):
            allowed = framework.check_rate_limit(identifier, limit=10)
            if i < 10:
                assert allowed is True
            else:
                assert allowed is False
    
    def test_anomaly_detection(self, framework):
        """Test anomaly detection."""
        for i in range(25):
            event = APISecurityEvent(
                event_id=f"EVT-{i}",
                timestamp=datetime.now(),
                endpoint="/api/test",
                method="GET",
                source_ip="192.168.1.100"
            )
            framework.log_security_event(event)
        
        # Check if anomaly was detected
        events = [e for e in framework.security_events if e.severity == SecurityLevel.HIGH]
        assert len(events) > 0

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

#!/usr/bin/env python3
"""
API Security Framework Cleanup
Production-ready cleanup and resource management
"""

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class APISecurityFrameworkCleanup:
    """Handles cleanup operations."""
    
    def __init__(self, framework):
        self.framework = framework
    
    def cleanup_old_events(self, days: int = 90):
        """Remove events older than specified days."""
        cutoff_date = datetime.now() - timedelta(days=days)
        initial_count = len(self.framework.security_events)
        
        self.framework.security_events = [
            e for e in self.framework.security_events
            if e.timestamp >= cutoff_date
        ]
        
        removed = initial_count - len(self.framework.security_events)
        logger.info(f"Cleaned up {removed} old events")
        return removed
    
    def cleanup_rate_limits(self):
        """Clean up expired rate limit entries."""
        current_time = time.time()
        expired_keys = [
            k for k, v in self.framework.rate_limits.items()
            if current_time > v['reset_time']
        ]
        
        for key in expired_keys:
            del self.framework.rate_limits[key]
        
        logger.info(f"Cleaned up {len(expired_keys)} expired rate limit entries")
        return len(expired_keys)
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting API security framework cleanup")
        self.cleanup_old_events()
        self.cleanup_rate_limits()
        self.framework.cleanup()
        logger.info("API security framework cleanup complete")