Shadow APIs: The Hidden Cybersecurity Risk in 2026

# Install ffuf (if not installed)
# macOS: brew install ffuf
# Linux: Download from https://github.com/ffuf/ffuf

# Basic directory brute-forcing
ffuf -w /usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt \
     -u https://api.example.com/FUZZ \
     -mc 200,201,204 \
     -H "Authorization: Bearer YOUR_TOKEN" \
     -t 50

# API-specific wordlist
cat > api-wordlist.txt <<EOF
/api/v1/users
/api/v1/admin
/api/v2/users
/api/internal
/api/legacy
/api/test
/api/staging
/api/debug
/api/health
/api/metrics
/api/status
EOF

ffuf -w api-wordlist.txt -u https://api.example.com/FUZZ -mc 200,201,204

# Extract unique endpoints from access logs
cat access.log | grep -oE '(GET|POST|PUT|DELETE|PATCH) /[^ ]+' | sort -u > endpoints.txt

# Compare with documented endpoints
cat documented-endpoints.txt
# Manually compare or use diff

# Find endpoints not in documentation
comm -23 <(sort endpoints.txt) <(sort documented-endpoints.txt) > shadow-apis.txt

Complete Python Shadow API Discovery Tool:

#!/usr/bin/env python3
"""
Production-ready Shadow API Discovery Tool
Comprehensive tool for discovering undocumented/shadow APIs
"""

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

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


@dataclass
class DiscoveredEndpoint:
    """Discovered API endpoint."""
    url: str
    method: str
    status_code: int
    response_size: int
    content_type: Optional[str] = None
    is_documented: bool = False
    response_headers: Dict = None
    response_time: float = 0.0


@dataclass
class DiscoveryResult:
    """API discovery result."""
    endpoint: str
    method: str
    status_code: int
    is_shadow: bool
    confidence: float
    details: Dict


class ShadowAPIDiscoverer:
    """Comprehensive shadow API discovery tool."""
    
    def __init__(
        self,
        base_url: str,
        documented_endpoints: Optional[List[str]] = None,
        auth_token: Optional[str] = None,
        max_workers: int = 10
    ):
        """Initialize shadow API discoverer.
        
        Args:
            base_url: Base URL of the API
            documented_endpoints: List of documented endpoint paths
            auth_token: Optional authentication token
            max_workers: Maximum concurrent workers
        """
        self.base_url = base_url.rstrip('/')
        self.documented_endpoints = set(documented_endpoints or [])
        self.auth_token = auth_token
        self.max_workers = max_workers
        self.session = requests.Session()
        
        if auth_token:
            self.session.headers.update({
                'Authorization': f'Bearer {auth_token}'
            })
        
        self.discovered_endpoints: Set[str] = set()
        self.shadow_apis: List[DiscoveryResult] = []
        
        # Common API wordlist
        self.common_paths = [
            '/api/v1', '/api/v2', '/api/v3',
            '/v1', '/v2', '/v3',
            '/internal', '/admin', '/dev', '/test', '/staging',
            '/debug', '/health', '/status', '/metrics', '/ping',
            '/graphql', '/graphiql', '/playground',
            '/swagger', '/swagger.json', '/openapi.json',
            '/docs', '/documentation',
            '/users', '/user', '/accounts', '/account',
            '/auth', '/login', '/logout', '/token',
            '/data', '/export', '/import',
            '/config', '/settings', '/preferences'
        ]
        
        # HTTP methods to test
        self.methods = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS']
    
    def discover_from_openapi(self, openapi_path: str = '/.well-known/openapi.json') -> List[str]:
        """Discover endpoints from OpenAPI specification.
        
        Args:
            openapi_path: Path to OpenAPI spec
            
        Returns:
            List of discovered endpoint paths
        """
        documented = []
        try:
            url = urljoin(self.base_url, openapi_path)
            response = self.session.get(url, timeout=10)
            
            if response.status_code == 200:
                spec = response.json()
                paths = spec.get('paths', {})
                
                for path, methods in paths.items():
                    for method in methods.keys():
                        if method.upper() in self.methods:
                            endpoint_key = f"{method.upper()} {path}"
                            documented.append(endpoint_key)
                            self.documented_endpoints.add(endpoint_key)
                
                logger.info(f"Discovered {len(documented)} documented endpoints from OpenAPI")
        except Exception as e:
            logger.error(f"Error discovering from OpenAPI: {e}")
        
        return documented
    
    def discover_from_source_code(self, code_paths: List[str]) -> List[str]:
        """Discover endpoints from source code analysis.
        
        Args:
            code_paths: List of file paths to analyze
            
        Returns:
            List of discovered endpoint patterns
        """
        discovered = []
        
        # Common route patterns
        route_patterns = [
            r'@app\.route\([\'"]([^\'"]+)[\'"]',
            r'router\.(get|post|put|delete|patch)\([\'"]([^\'"]+)[\'"]',
            r'\.(get|post|put|delete|patch)\([\'"]([^\'"]+)[\'"]',
            r'route\([\'"]([^\'"]+)[\'"]',
        ]
        
        for code_path in code_paths:
            try:
                with open(code_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                    
                    for pattern in route_patterns:
                        matches = re.finditer(pattern, content, re.IGNORECASE)
                        for match in matches:
                            endpoint = match.group(1) or match.group(2)
                            if endpoint:
                                discovered.append(endpoint)
                                endpoint_key = f"GET {endpoint}"  # Default to GET
                                self.documented_endpoints.add(endpoint_key)
            except Exception as e:
                logger.error(f"Error reading {code_path}: {e}")
        
        logger.info(f"Discovered {len(discovered)} endpoints from source code")
        return discovered
    
    def discover_from_traffic_logs(self, log_file: str) -> List[str]:
        """Discover endpoints from traffic logs.
        
        Args:
            log_file: Path to log file
            
        Returns:
            List of discovered endpoint paths
        """
        discovered = []
        endpoint_pattern = re.compile(r'(GET|POST|PUT|DELETE|PATCH|HEAD|OPTIONS)\s+([^\s]+)')
        
        try:
            with open(log_file, 'r') as f:
                for line in f:
                    match = endpoint_pattern.search(line)
                    if match:
                        method = match.group(1)
                        path = match.group(2).split('?')[0]  # Remove query params
                        endpoint_key = f"{method} {path}"
                        discovered.append(endpoint_key)
                        self.discovered_endpoints.add(endpoint_key)
        except Exception as e:
            logger.error(f"Error reading log file: {log_file}: {e}")
        
        logger.info(f"Discovered {len(set(discovered))} unique endpoints from logs")
        return list(set(discovered))
    
    def brute_force_discovery(self, wordlist: Optional[List[str]] = None) -> List[DiscoveryResult]:
        """Brute force endpoint discovery.
        
        Args:
            wordlist: Optional custom wordlist
            
        Returns:
            List of discovery results
        """
        if wordlist is None:
            wordlist = self.common_paths
        
        results = []
        
        # Use thread pool for concurrent requests
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            
            for path in wordlist:
                for method in ['GET', 'POST', 'OPTIONS']:
                    future = executor.submit(self.test_endpoint, path, method)
                    futures.append(future)
            
            for future in concurrent.futures.as_completed(futures):
                try:
                    result = future.result()
                    if result:
                        results.append(result)
                except Exception as e:
                    logger.error(f"Error in brute force discovery: {e}")
        
        return results
    
    def test_endpoint(self, path: str, method: str) -> Optional[DiscoveryResult]:
        """Test a single endpoint.
        
        Args:
            path: Endpoint path
            method: HTTP method
            
        Returns:
            DiscoveryResult if endpoint exists, None otherwise
        """
        url = urljoin(self.base_url, path)
        endpoint_key = f"{method} {path}"
        
        # Skip if already documented
        if endpoint_key in self.documented_endpoints:
            return None
        
        try:
            start_time = time.time()
            response = self.session.request(
                method,
                url,
                timeout=10,
                allow_redirects=False
            )
            response_time = time.time() - start_time
            
            # Consider endpoint discovered if it returns non-404
            if response.status_code not in [404, 400]:
                is_shadow = endpoint_key not in self.documented_endpoints
                
                result = DiscoveryResult(
                    endpoint=url,
                    method=method,
                    status_code=response.status_code,
                    is_shadow=is_shadow,
                    confidence=self.calculate_confidence(response),
                    details={
                        'response_size': len(response.content),
                        'content_type': response.headers.get('Content-Type'),
                        'response_time': response_time,
                        'headers': dict(response.headers)
                    }
                )
                
                self.discovered_endpoints.add(endpoint_key)
                return result
        except requests.exceptions.RequestException as e:
            logger.debug(f"Error testing {url}: {e}")
        
        return None
    
    def calculate_confidence(self, response: requests.Response) -> float:
        """Calculate confidence that endpoint is a shadow API.
        
        Args:
            response: HTTP response
            
        Returns:
            Confidence score (0.0-1.0)
        """
        confidence = 0.5
        
        # Higher confidence for successful responses
        if 200 <= response.status_code < 300:
            confidence += 0.3
        
        # Higher confidence for authentication errors (endpoint exists but needs auth)
        if response.status_code in [401, 403]:
            confidence += 0.2
        
        # Lower confidence for server errors
        if response.status_code >= 500:
            confidence -= 0.2
        
        # Higher confidence if response has API-like content
        content_type = response.headers.get('Content-Type', '').lower()
        if 'application/json' in content_type:
            confidence += 0.2
        
        # Check for API-like response
        try:
            data = response.json()
            if isinstance(data, dict) and any(key in data for key in ['data', 'result', 'error', 'message']):
                confidence += 0.1
        except:
            pass
        
        return min(1.0, max(0.0, confidence))
    
    def generate_report(self) -> Dict:
        """Generate comprehensive discovery report.
        
        Returns:
            Discovery report dictionary
        """
        shadow_count = len([r for r in self.shadow_apis if r.is_shadow])
        
        return {
            'timestamp': datetime.utcnow().isoformat(),
            'base_url': self.base_url,
            'total_endpoints_discovered': len(self.discovered_endpoints),
            'documented_endpoints': len(self.documented_endpoints),
            'shadow_apis_count': shadow_count,
            'shadow_apis': [asdict(r) for r in self.shadow_apis],
            'summary': {
                'high_confidence_shadow': len([r for r in self.shadow_apis if r.is_shadow and r.confidence > 0.7]),
                'medium_confidence_shadow': len([r for r in self.shadow_apis if r.is_shadow and 0.4 < r.confidence <= 0.7]),
                'low_confidence_shadow': len([r for r in self.shadow_apis if r.is_shadow and r.confidence <= 0.4])
            }
        }


# Example usage
if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Shadow API Discovery Tool')
    parser.add_argument('--url', required=True, help='Base URL to scan')
    parser.add_argument('--token', help='Authentication token')
    parser.add_argument('--openapi', help='Path to OpenAPI spec')
    parser.add_argument('--logs', help='Path to traffic log file')
    parser.add_argument('--output', help='Output file for report')
    
    args = parser.parse_args()
    
    discoverer = ShadowAPIDiscoverer(
        base_url=args.url,
        auth_token=args.token,
        max_workers=10
    )
    
    # Discover documented endpoints
    if args.openapi:
        discoverer.discover_from_openapi(args.openapi)
    
    if args.logs:
        discoverer.discover_from_traffic_logs(args.logs)
    
    # Brute force discovery
    logger.info("Starting brute force discovery...")
    results = discoverer.brute_force_discovery()
    discoverer.shadow_apis = [r for r in results if r.is_shadow]
    
    # Generate report
    report = discoverer.generate_report()
    
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(report, f, indent=2)
    else:
        print(json.dumps(report, indent=2))
    
    logger.info(f"Discovery complete. Found {len(discoverer.shadow_apis)} shadow APIs.")

# Install nuclei
# macOS: brew install nuclei
# Or: go install -v github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest

# Discover APIs with nuclei
nuclei -u https://api.example.com -t ~/nuclei-templates/http/discoveries/

# Custom nuclei template for API discovery
cat > api-discovery.yaml <<EOF
id: api-discovery
info:
  name: API Endpoint Discovery
  author: YourName
  severity: info

http:
  - method: GET
    path:
      - "{{BaseURL}}/api/v1/users"
      - "{{BaseURL}}/api/v2/users"
      - "{{BaseURL}}/api/internal/users"
    matchers:
      - type: status
        status:
          - 200
          - 201
          - 204
EOF

nuclei -u https://api.example.com -t api-discovery.yaml

# Find route definitions (example for Express.js)
grep -r "app\.\(get\|post\|put\|delete\|patch\)" src/ > routes.txt

# Extract endpoint paths
cat routes.txt | grep -oE "'/[^']+'" | sort -u > code-endpoints.txt

# Compare with documented endpoints
comm -23 <(sort code-endpoints.txt) <(sort documented-endpoints.txt) > shadow-from-code.txt

// Express.js example with express-rate-limit
const rateLimit = require('express-rate-limit');

// Global rate limiter (applies to all routes)
const globalLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100, // Limit each IP to 100 requests per windowMs
  message: 'Too many requests from this IP',
  standardHeaders: true,
  legacyHeaders: false,
});

app.use(globalLimiter); // Apply to all routes

// Per-endpoint rate limiter (stricter for sensitive endpoints)
const strictLimiter = rateLimit({
  windowMs: 15 * 60 * 1000,
  max: 10,
});

app.post('/api/v1/admin/users', strictLimiter, adminHandler);

// Middleware to validate all requests
app.use((req, res, next) => {
  // Log all requests (including shadow APIs)
  console.log(`${req.method} ${req.path} - ${req.ip}`);
  
  // Check if endpoint is documented
  const isDocumented = documentedEndpoints.includes(`${req.method} ${req.path}`);
  
  if (!isDocumented) {
    // Alert on undocumented endpoint access
    logger.warn(`Shadow API accessed: ${req.method} ${req.path}`, {
      ip: req.ip,
      userAgent: req.headers['user-agent']
    });
  }
  
  // Apply security controls regardless
  // (auth, validation, rate limiting, etc.)
  next();
});

// Production-ready 404 logging and brute-force detection
class ShadowAPIDetection {
  constructor() {
    // Store recent 404s per IP (in production, use Redis or similar)
    this.recent404s = new Map();
    
    // Cleanup old entries every 5 minutes
    setInterval(() => this.cleanupOldEntries(), 5 * 60 * 1000);
  }
  
  // Get recent 404s for an IP within time window
  getRecent404s(ip, timeWindow) {
    const now = Date.now();
    const windowMs = this.parseTimeWindow(timeWindow);
    const cutoff = now - windowMs;
    
    if (!this.recent404s.has(ip)) {
      return [];
    }
    
    const entries = this.recent404s.get(ip);
    return entries.filter(entry => entry.timestamp > cutoff);
  }
  
  // Parse time window string to milliseconds
  parseTimeWindow(window) {
    const match = window.match(/(\d+)([smhd])/);
    if (!match) return 5 * 60 * 1000; // Default 5 minutes
    
    const value = parseInt(match[1]);
    const unit = match[2];
    
    const multipliers = {
      's': 1000,
      'm': 60 * 1000,
      'h': 60 * 60 * 1000,
      'd': 24 * 60 * 60 * 1000
    };
    
    return value * (multipliers[unit] || 1000);
  }
  
  // Record 404 event
  record404(ip, path, method, userAgent) {
    if (!this.recent404s.has(ip)) {
      this.recent404s.set(ip, []);
    }
    
    const entries = this.recent404s.get(ip);
    entries.push({
      timestamp: Date.now(),
      path,
      method,
      userAgent
    });
    
    // Keep only last 1000 entries per IP
    if (entries.length > 1000) {
      entries.shift();
    }
  }
  
  // Cleanup old entries
  cleanupOldEntries() {
    const now = Date.now();
    const maxAge = 60 * 60 * 1000; // 1 hour
    
    for (const [ip, entries] of this.recent404s.entries()) {
      const filtered = entries.filter(entry => entry.timestamp > now - maxAge);
      if (filtered.length === 0) {
        this.recent404s.delete(ip);
      } else {
        this.recent404s.set(ip, filtered);
      }
    }
  }
  
  // Check for brute-forcing pattern
  checkBruteForcing(ip) {
    const recent404s = this.getRecent404s(ip, '5m');
    
    // Check for high volume of 404s
    if (recent404s.length > 50) {
      return {
        isBruteForcing: true,
        reason: 'High volume of 404s',
        count: recent404s.length
      };
    }
    
    // Check for directory brute-forcing patterns
    const paths = recent404s.map(e => e.path);
    const uniquePaths = new Set(paths);
    const pathVariations = paths.length / uniquePaths.size;
    
    if (pathVariations > 10 && recent404s.length > 20) {
      return {
        isBruteForcing: true,
        reason: 'Directory brute-forcing pattern',
        uniquePaths: uniquePaths.size,
        totalRequests: recent404s.length
      };
    }
    
    return { isBruteForcing: false };
  }
}

// Initialize detector
const shadowAPIDetector = new ShadowAPIDetection();

// Express middleware for 404 logging and detection
app.use((req, res, next) => {
  const originalSend = res.send;
  res.send = function(data) {
    if (res.statusCode === 404) {
      const ip = req.ip || req.connection.remoteAddress;
      
      // Record 404 event
      shadowAPIDetector.record404(
        ip,
        req.path,
        req.method,
        req.headers['user-agent']
      );
      
      // Log warning
      logger.warn('404 - Potential API discovery attempt', {
        method: req.method,
        path: req.path,
        ip: ip,
        userAgent: req.headers['user-agent'],
        timestamp: new Date().toISOString()
      });
      
      // Check for brute-forcing
      const bruteForceCheck = shadowAPIDetector.checkBruteForcing(ip);
      if (bruteForceCheck.isBruteForcing) {
        logger.alert('Possible directory brute-forcing detected', {
          ip: ip,
          reason: bruteForceCheck.reason,
          details: bruteForceCheck
        });
        
        // Optionally block IP (in production, use rate limiting middleware)
        // res.status(429).json({ error: 'Too many requests' });
        // return;
      }
    }
    originalSend.call(this, data);
  };
  next();
});