Create Post Login
Menu

Cybersecurity Topic

Team 760 views

API Security

1. API FUNDAMENTALS

REST vs GraphQL Security Concerns

REST API Security:

text

GET /api/users/123
POST /api/users
PUT /api/users/123
DELETE /api/users/123

Security Concerns:

  • Verbose Error Messages: Stack traces in responses

  • Information Disclosure: Via OPTIONS, HEAD methods

  • Insecure Direct Object References: Predictable IDs

  • Mass Assignment: User can update any field

  • Missing Rate Limiting: Per endpoint basis

GraphQL Security:

graphql

# Query
{
  user(id: "123") {
    name
    email
    creditCard {
      number
      expiry
    }
  }
}

# Mutation
mutation {
  updateUser(id: "123", input: {
    email: "attacker@evil.com",
    role: "admin"
  }) {
    id
    role
  }
}

Security Concerns:

  1. Over-querying/DoS:

graphql

# Query batching attack
query {
  user1: user(id: "1") { posts { comments { user { posts { ... } } } } }
  user2: user(id: "2") { posts { comments { user { posts { ... } } } } }
  # ... repeat 1000 times
}

  1. Introspection Abuse:

graphql

# Exposes schema (often enabled in production)
{
  __schema {
    types { name fields { name type { name } } }
  }
}

  1. Field-Level Authorization: Missing per-field checks

  2. Circular Queries: Infinite nested queries

  3. Query Complexity: No query depth/size limits

Comparison Table:

Aspect REST GraphQL
Rate Limiting Per endpoint Query complexity based
Authorization Resource-level Field/resource-level
Data Exposure Fixed responses Client-defined responses
Error Handling HTTP status codes Generic "errors" array
Caching HTTP caching Complex, query-specific

API Authentication & Authorization Models

Authentication Methods:

  1. API Keys:

http

GET /api/data HTTP/1.1
X-API-Key: abc123def456

  • Pros: Simple, easy to implement

  • Cons: No expiration, easily leaked, no user context

  1. JWT (JSON Web Tokens):

json

{
  "alg": "HS256",
  "typ": "JWT"
}
{
  "sub": "1234567890",
  "name": "John Doe",
  "admin": true,
  "iat": 1516239022,
  "exp": 1516242622
}

  • Security Concerns:

    • None algorithm attacks

    • Weak HMAC secrets

    • Storing sensitive data in payload

    • Lack of token revocation

  1. OAuth 2.0 Flows:

Lecturre learning image

Authorization Models:

  1. RBAC (Role-Based Access Control):

json

{
  "roles": ["user", "moderator", "admin"],
  "permissions": {
    "user": ["read:own_profile", "write:own_posts"],
    "admin": ["*"]
  }
}

  1. ABAC (Attribute-Based Access Control):

json

{
  "rules": [
    {
      "effect": "allow",
      "action": "read",
      "resource": "document",
      "condition": {
        "user.department": "==",
        "document.department": true
      }
    }
  ]
}

  1. ReBAC (Relationship-Based Access Control):

  • Check user-resource relationships

  • Example: Can user edit document if they're the owner or shared collaborator?

API Gateways & Rate Limiting

API Gateway Features:

text

Client → API Gateway → Microservices
      ↓               ↓
  Authentication   Rate Limiting
      ↓               ↓
  Authorization    Logging
      ↓               ↓
    TLS Termination   Caching

Common Gateway Implementations:

  • Kong: Open-source, plugin-based

  • AWS API Gateway: Serverless, integrated with AWS ecosystem

  • Apigee: Google Cloud, enterprise features

  • Tyk: Open-source, Go-based

Rate Limiting Strategies:

  1. Token Bucket Algorithm:

python

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity  # Max tokens
        self.tokens = capacity
        self.refill_rate = refill_rate  # tokens/second
        self.last_refill = time.time()
    
    def consume(self, tokens=1):
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

  1. Fixed Window Counter:

python

# Redis implementation
import redis
r = redis.Redis()

def check_rate_limit(user_id, limit=100, window=3600):
    key = f"rate_limit:{user_id}:{int(time.time() // window)}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current <= limit

  1. Sliding Window Log:

python

def sliding_window_rate_limit(user_id, limit=100, window=60):
    key = f"rate_limit:{user_id}"
    now = time.time()
    window_start = now - window
    
    # Remove old entries
    r.zremrangebyscore(key, 0, window_start)
    
    # Count requests in window
    request_count = r.zcard(key)
    
    if request_count < limit:
        r.zadd(key, {str(now): now})
        r.expire(key, window)
        return True
    
    return False

Rate Limiting Headers (RFC 6585):

http

HTTP/1.1 429 Too Many Requests
Retry-After: 3600
X-RateLimit-Limit: 100
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 1640995200

2. COMMON API VULNERABILITIES

Broken Object Level Authorization (BOLA/IDOR)

Vulnerable Code:

python

# Vulnerable - No authorization check
@app.route('/api/users/<user_id>/documents')
def get_user_documents(user_id):
    documents = db.query("SELECT * FROM documents WHERE user_id = %s", user_id)
    return jsonify(documents)

# Attack
GET /api/users/123/documents  # Attacker changes to 124

Prevention:

python

# Secure implementation
@app.route('/api/users/<user_id>/documents')
@require_auth
def get_user_documents(user_id):
    # Verify requesting user owns the resource
    if current_user.id != int(user_id) and not current_user.is_admin:
        abort(403)
    
    documents = Document.query.filter_by(user_id=user_id).all()
    return jsonify([doc.to_dict() for doc in documents])

Advanced BOLA Patterns:

  1. Indirect IDOR:

http

GET /api/files?name=report.pdf  # Change to admin_report.pdf

  1. Mass Assignment + IDOR:

http

PUT /api/users/123
Content-Type: application/json

{"id": 456, "role": "admin"}  # Changes ID and role

Excessive Data Exposure

GraphQL Example:

graphql

# Backend returns full user object
type User {
  id: ID!
  username: String!
  email: String!
  passwordHash: String!
  ssn: String!
  creditCard: CreditCard!
}

# Client only needs username
query {
  user(id: "123") {
    username
  }
}
# But receives ALL fields!

REST Example:

json

{
  "user": {
    "id": 123,
    "username": "johndoe",
    "email": "john@example.com",
    "password_hash": "$2a$12$...",  // Should not be exposed!
    "ssn": "123-45-6789",
    "internal_notes": "VIP customer"
  }
}

Prevention Strategies:

  1. Response Filtering:

python

# Django Rest Framework
class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'username', 'email']  # Explicit allow-list
    
    def to_representation(self, instance):
        data = super().to_representation(instance)
        if not self.context['request'].user.is_staff:
            data.pop('internal_notes', None)
        return data

  1. GraphQL Field-Level Permissions:

python

# Graphene implementation
class UserType(DjangoObjectType):
    class Meta:
        model = User
        fields = ("id", "username", "email")
    
    @staticmethod
    def resolve_email(parent, info):
        # Only show email to the user themselves or admins
        if info.context.user.id == parent.id or info.context.user.is_staff:
            return parent.email
        return None

  1. API Schema Enforcement:

yaml

# OpenAPI response schema
components:
  schemas:
    PublicUser:
      type: object
      properties:
        id:
          type: integer
        username:
          type: string
        # NO sensitive fields!

Mass Assignment

Vulnerable Code:

javascript

// Express.js vulnerable endpoint
app.put('/api/users/:id', async (req, res) => {
  // Direct assignment - attacker can set any field!
  const updatedUser = await User.findByIdAndUpdate(
    req.params.id,
    req.body,  // Attacker sends {role: 'admin'}
    { new: true }
  );
  res.json(updatedUser);
});

Attack Payload:

json

{
  "username": "attacker",
  "email": "attacker@evil.com",
  "role": "admin",
  "is_active": true,
  "balance": 1000000
}

Prevention:

  1. Allow-List Approach:

python

# Django Rest Framework
class UserUpdateSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['username', 'email']  # Only updatable fields
    
    # Extra validation
    def validate(self, data):
        if 'role' in self.initial_data:
            raise serializers.ValidationError("Cannot update role")
        return data

  1. Block-List Approach (Less secure):

python

class UserUpdateSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        exclude = ['role', 'is_superuser', 'balance']

  1. Different Serializers for Different Actions:

python

class PublicUserSerializer(serializers.ModelSerializer):
    # For public views
    class Meta:
        model = User
        fields = ['id', 'username']

class AdminUserSerializer(serializers.ModelSerializer):
    # For admin actions
    class Meta:
        model = User
        fields = '__all__'

Improper Asset Management

Common Issues:

  1. Shadow/Orphaned APIs:

    • Old API versions still accessible

    • Internal/test APIs exposed publicly

    • Deprecated endpoints not disabled

  2. Inadequate Documentation:

    • No API inventory

    • Missing changelogs

    • Undocumented endpoints

API Inventory Management:

yaml

# API inventory YAML
apis:
  - name: user-service
    version: v2.1.0
    base_url: https://api.company.com/v2
    status: active
    owner: team-identity
    security_level: public
    documentation: https://docs.company.com/api/user
    
  - name: legacy-user-service
    version: v1.0.0
    base_url: https://api.company.com/v1
    status: deprecated
    sunset_date: 2024-12-31
    owner: team-identity
    security_level: private

Versioning Strategies:

  1. URI Versioning:

text

https://api.company.com/v1/users
https://api.company.com/v2/users

  1. Header Versioning:

http

GET /users HTTP/1.1
Accept: application/vnd.company.v2+json

  1. Query Parameter Versioning:

text

https://api.company.com/users?version=2

Deprecation Headers:

http

HTTP/1.1 200 OK
Deprecation: true
Sunset: Wed, 31 Dec 2025 23:59:59 GMT
Link: </v2/users>; rel="successor-version"

Injection Attacks in APIs

REST API Injection:

http

# SQL Injection
GET /api/users?search=admin' OR '1'='1

# NoSQL Injection
POST /api/search
Content-Type: application/json

{
  "query": {
    "$where": "function() { return this.role == 'admin' }"
  }
}

# Command Injection in file upload
POST /api/upload
Content-Type: multipart/form-data

file=`cat /etc/passwd`&name=;ls -la;

GraphQL Injection:

graphql

# Field injection
query {
  users(search: "admin\")) { UNION SELECT * FROM secret_table --"}) {
    id
    username
  }
}

# Directive injection
query {
  __schema {
    directives {
      name
      # Malicious nested query
      ... on __Directive {
        locations
      }
    }
  }
}

Prevention:

  1. Input Validation with JSON Schema:

python

from jsonschema import validate

user_schema = {
    "type": "object",
    "properties": {
        "username": {"type": "string", "pattern": "^[a-zA-Z0-9_]{3,20}$"},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 0, "maximum": 150}
    },
    "required": ["username", "email"],
    "additionalProperties": False  # Prevent mass assignment
}

# Validate before processing
validate(instance=request.json, schema=user_schema)

  1. GraphQL Query Analysis:

python

from graphql import validate, parse
from graphql.validation import validate_sdl

# Query depth limiting
MAX_DEPTH = 10

class DepthLimitingVisitor:
    def __init__(self, max_depth):
        self.max_depth = max_depth
        self.current_depth = 0
    
    def enter(self, node, key, parent, path, ancestors):
        if isinstance(node, Field):
            self.current_depth += 1
            if self.current_depth > self.max_depth:
                raise Exception(f"Query too deep: {self.current_depth}")

# Query cost/complexity limiting
QUERY_COST_RULES = {
    'User': 1,
    'Post': 2,
    'Comment': 1,
    'friends': {'multiplier': 5},
    'posts': {'multiplier': 10}
}

3. SECURITY CONTROLS

OAuth 2.0 & OpenID Connect Deep Dive

OAuth 2.0 Flows Comparison:

Flow Use Case Security Considerations
Authorization Code Web apps (server-side) Most secure, uses PKCE for mobile/native
Implicit Deprecated Tokens in URL fragments, not recommended
Client Credentials Server-to-server No user context, machine-to-machine
Resource Owner Password Legacy migration Store credentials, avoid if possible
Device Code IoT/CLI devices Limited input devices

Authorization Code Flow with PKCE:

Lecturre learning image

OAuth Security Best Practices:

yaml

# OAuth configuration example
oauth_config:
  client_registration:
    require_https: true
    redirect_uri_validation:
      exact_match: true
      allow_wildcard: false
    
  token_management:
    access_token_lifetime: 3600  # 1 hour
    refresh_token_lifetime: 2592000  # 30 days
    refresh_token_rotation: true
    token_binding: required  # mTLS or DPoP
    
  security:
    require_pkce: true
    allowed_grant_types: ["authorization_code"]
    allowed_response_types: ["code"]
    enable_token_revocation: true

OpenID Connect Extensions:

json

{
  "id_token": {
    "iss": "https://auth.company.com",
    "sub": "123456",
    "aud": "client_id",
    "exp": 1640995200,
    "iat": 1640991600,
    "auth_time": 1640991000,
    "nonce": "abc123",
    "acr": "urn:mace:incommon:iap:silver",
    "amr": ["pwd", "mfa"],
    "claims": {
      "email": "user@example.com",
      "email_verified": true,
      "profile": "https://profile.url"
    }
  }
}

API Keys vs JWT

API Key Implementation:

python

class APIKeyAuth:
    def __init__(self):
        self.storage = RedisStorage()
    
    def create_api_key(self, user_id, permissions):
        # Generate cryptographically secure key
        key = secrets.token_urlsafe(32)
        key_hash = self._hash_key(key)
        
        # Store with metadata
        self.storage.set(f"apikey:{key_hash}", {
            "user_id": user_id,
            "permissions": permissions,
            "created_at": datetime.now(),
            "last_used": None,
            "usage_count": 0
        })
        
        return f"sk_live_{key}"
    
    def validate_api_key(self, api_key):
        # Extract key from format
        if not api_key.startswith("sk_live_"):
            return None
        
        key = api_key[8:]
        key_hash = self._hash_key(key)
        
        data = self.storage.get(f"apikey:{key_hash}")
        if not data:
            return None
        
        # Update usage stats
        data["last_used"] = datetime.now()
        data["usage_count"] += 1
        self.storage.set(f"apikey:{key_hash}", data)
        
        return data

JWT Implementation with Best Practices:

python

import jwt
from cryptography.hazmat.primitives import serialization
from datetime import datetime, timedelta

class JWTManager:
    def __init__(self):
        # Load asymmetric keys (RS256)
        with open("private.pem", "rb") as f:
            self.private_key = serialization.load_pem_private_key(
                f.read(),
                password=None
            )
        
        with open("public.pem", "rb") as f:
            self.public_key = serialization.load_pem_public_key(
                f.read()
            )
    
    def create_token(self, user_id, claims=None):
        payload = {
            "sub": str(user_id),
            "iss": "https://api.company.com",
            "aud": "https://api.company.com",
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(hours=1),
            "jti": secrets.token_urlsafe(16),  # Unique token ID
            "type": "access"
        }
        
        if claims:
            payload.update(claims)
        
        # Sign with private key
        token = jwt.encode(
            payload,
            self.private_key,
            algorithm="RS256",
            headers={"kid": "2024-01"}  # Key ID for rotation
        )
        
        return token
    
    def validate_token(self, token):
        try:
            # Decode with public key
            payload = jwt.decode(
                token,
                self.public_key,
                algorithms=["RS256"],
                issuer="https://api.company.com",
                audience="https://api.company.com",
                options={
                    "require": ["exp", "iat", "sub", "jti"],
                    "verify_signature": True,
                    "verify_exp": True,
                    "verify_iat": True,
                    "verify_aud": True,
                    "verify_iss": True
                }
            )
            
            # Check token revocation
            if self.is_token_revoked(payload["jti"]):
                return None
                
            return payload
        except jwt.InvalidTokenError as e:
            logger.warning(f"Invalid token: {e}")
            return None
    
    def refresh_token_flow(self, refresh_token):
        # Validate refresh token
        payload = self.validate_token(refresh_token)
        if not payload or payload.get("type") != "refresh":
            return None
        
        # Check if refresh token is revoked
        if self.is_refresh_token_revoked(payload["jti"]):
            return None
        
        # Create new access token
        new_access_token = self.create_token(payload["sub"])
        
        # Rotate refresh token (optional)
        new_refresh_token = self.create_refresh_token(payload["sub"])
        
        # Revoke old refresh token
        self.revoke_token(payload["jti"])
        
        return {
            "access_token": new_access_token,
            "refresh_token": new_refresh_token,
            "token_type": "Bearer",
            "expires_in": 3600
        }

Input Validation & Schema Validation

Layered Validation Approach:

python

# Layer 1: Request Schema Validation
from pydantic import BaseModel, validator, Field
from typing import List, Optional
import re

class UserCreateRequest(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, regex=r'^[a-zA-Z0-9_]+$')
    email: str = Field(..., regex=r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
    age: Optional[int] = Field(None, ge=0, le=150)
    roles: List[str] = Field(default_factory=list, max_items=5)
    
    @validator('roles')
    def validate_roles(cls, v):
        valid_roles = {'user', 'editor', 'admin'}
        for role in v:
            if role not in valid_roles:
                raise ValueError(f'Invalid role: {role}')
        return v

# Layer 2: Business Logic Validation
class UserValidator:
    def validate_create(self, data: dict):
        # Check username uniqueness
        if User.objects.filter(username=data['username']).exists():
            raise ValidationError("Username already exists")
        
        # Check rate limiting
        if self._is_creation_rate_limited(data['email']):
            raise ValidationError("Too many accounts created")
        
        # Check against blacklist
        if self._is_email_blacklisted(data['email']):
            raise ValidationError("Email not allowed")
        
        return True

# Layer 3: Database-Level Constraints
# SQLAlchemy model
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(20), unique=True, nullable=False, 
                      index=True, check_constraint='LENGTH(username) >= 3')
    email = Column(String(255), unique=True, nullable=False, 
                   check_constraint="email ~* '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$'")
    age = Column(Integer, CheckConstraint('age >= 0 AND age <= 150'))

OpenAPI Schema Validation:

yaml

openapi: 3.0.0
components:
  schemas:
    User:
      type: object
      required:
        - username
        - email
      properties:
        id:
          type: integer
          readOnly: true
        username:
          type: string
          minLength: 3
          maxLength: 20
          pattern: '^[a-zA-Z0-9_]+$'
        email:
          type: string
          format: email
        age:
          type: integer
          minimum: 0
          maximum: 150
        roles:
          type: array
          maxItems: 5
          items:
            type: string
            enum: [user, editor, admin]
      additionalProperties: false  # Prevent mass assignment

API Versioning Security

Secure Versioning Strategy:

python

from flask import Flask, request, jsonify
import semver

app = Flask(__name__)

class APIVersionManager:
    def __init__(self):
        self.supported_versions = {
            "1.0.0": {"status": "deprecated", "sunset": "2024-06-30"},
            "1.1.0": {"status": "active", "deprecation_warning": True},
            "2.0.0": {"status": "active", "latest": True}
        }
    
    def validate_version(self, requested_version):
        # Parse version from header
        version = request.headers.get('X-API-Version', '1.0.0')
        
        # Validate format
        try:
            semver.VersionInfo.parse(version)
        except ValueError:
            return None, "Invalid version format"
        
        # Check if supported
        if version not in self.supported_versions:
            # Find compatible version
            compatible = self.find_compatible_version(version)
            if not compatible:
                return None, "Unsupported version"
            version = compatible
        
        # Check deprecation status
        version_info = self.supported_versions[version]
        return version, version_info
    
    def add_version_headers(self, response, version):
        version_info = self.supported_versions.get(version, {})
        
        response.headers['X-API-Version'] = version
        response.headers['X-API-Version-Status'] = version_info.get('status', 'unknown')
        
        if version_info.get('deprecation_warning'):
            response.headers['Deprecation'] = 'true'
            response.headers['Sunset'] = version_info.get('sunset', '')
            response.headers['Link'] = f'</api/v2>; rel="successor-version"'
        
        if version_info.get('latest'):
            response.headers['X-API-Latest-Version'] = 'true'
        
        return response

@app.route('/api/users')
def get_users():
    version_manager = APIVersionManager()
    version, error = version_manager.validate_version(request)
    
    if error:
        return jsonify({"error": error}), 400
    
    # Route to appropriate version handler
    if version.startswith('1.'):
        return handle_v1_users()
    elif version.startswith('2.'):
        return handle_v2_users()
    
    response = jsonify({"error": "Version not implemented"})
    return version_manager.add_version_headers(response, version), 501

4. TESTING & DEFENSE

API Penetration Testing Methodology

Testing Framework:

python

# Custom API testing tool
class APIPentestFramework:
    def __init__(self, target_url):
        self.target = target_url
        self.session = requests.Session()
        self.findings = []
    
    def test_authentication(self):
        """Test authentication endpoints"""
        tests = [
            self._test_rate_limiting,
            self._test_credential_stuffing,
            self._test_jwt_weaknesses,
            self._test_oauth_flows
        ]
        
        for test in tests:
            test()
    
    def test_authorization(self):
        """Test BOLA/IDOR vulnerabilities"""
        # Horizontal privilege escalation
        self._test_horizontal_escalation()
        
        # Vertical privilege escalation
        self._test_vertical_escalation()
        
        # Mass assignment
        self._test_mass_assignment()
    
    def test_injection(self):
        """Test various injection attacks"""
        injection_points = self._find_injection_points()
        
        for point in injection_points:
            self._test_sql_injection(point)
            self._test_nosql_injection(point)
            self._test_command_injection(point)
            self._test_xxe(point)
    
    def test_business_logic(self):
        """Test business logic flaws"""
        # Price manipulation
        self._test_price_manipulation()
        
        # Workflow bypass
        self._test_workflow_bypass()
        
        # Excessive data exposure
        self._test_data_exposure()
    
    def _test_rate_limiting(self):
        """Test rate limiting bypass"""
        endpoints = ['/api/login', '/api/register', '/api/password-reset']
        
        for endpoint in endpoints:
            print(f"Testing rate limiting on {endpoint}")
            
            # Test different bypass techniques
            techniques = [
                self._rotate_ip_addresses,
                self._use_different_user_agents,
                self._add_random_headers,
                self._parameter_pollution
            ]
            
            for technique in techniques:
                if self._bypass_rate_limit(endpoint, technique):
                    self.findings.append({
                        "type": "Rate Limiting Bypass",
                        "endpoint": endpoint,
                        "technique": technique.__name__,
                        "severity": "Medium"
                    })
    
    def generate_report(self):
        """Generate penetration test report"""
        report = {
            "target": self.target,
            "date": datetime.now().isoformat(),
            "findings": self.findings,
            "summary": {
                "critical": len([f for f in self.findings if f["severity"] == "Critical"]),
                "high": len([f for f in self.findings if f["severity"] == "High"]),
                "medium": len([f for f in self.findings if f["severity"] == "Medium"]),
                "low": len([f for f in self.findings if f["severity"] == "Low"])
            },
            "recommendations": self._generate_recommendations()
        }
        
        return json.dumps(report, indent=2)

Testing Tools Arsenal:

  1. Automated Scanners:

    • OWASP ZAP (with API add-on)

    • Burp Suite Professional

    • Nessus

    • Nuclei (with API templates)

  2. Manual Testing Tools:

    • Postman/Insomnia with fuzzing collections

    • Custom Python scripts with requests

    • sqlmap for API endpoints

    • nosqlmap for NoSQL injection

  3. GraphQL Specific:

    • InQL (Burp extension)

    • GraphQL Raider

    • GraphQLmap

Securing Public vs Private APIs

Public API Security Controls:

yaml

public_api_security:
  authentication:
    required: true
    methods:
      - api_key
      - oauth2_client_credentials
      - jwt_bearer
    
  authorization:
    required: true
    model: role_based
    default_role: anonymous
    rate_limits:
      anonymous: 10/hour
      authenticated: 1000/hour
      partner: 10000/hour
    
  network:
    require_tls_1.2+: true
    enable_waf: true
    ddos_protection: cloudflare
    ip_whitelisting: false  # Not feasible for public
    
  monitoring:
    enable_behavioral_analysis: true
    alert_on:
      - suspicious_geo_location
      - rapid_key_rotation
      - abnormal_request_patterns
      - data_exfiltration_patterns
    
  data_protection:
    response_filtering: enabled
    pii_detection: enabled
    encryption_at_rest: required
    data_retention: 30_days

Private/Internal API Security Controls:

yaml

internal_api_security:
  authentication:
    required: true
    methods:
      - mutual_tls
      - service_account_jwt
      - kubernetes_service_account
    
  authorization:
    required: true
    model: service_mesh
    use_istio_rbac: true
    network_policies: calico
    
  network:
    service_mesh: istio
    mTLS_required: true
    network_segmentation: enabled
    zero_trust_network: true
    
  access_control:
    principle_of_least_privilege: enforced
    just_in_time_access: required
    audit_logging: comprehensive
    
  monitoring:
    enable_service_graph: true
    trace_all_requests: true
    anomaly_detection: machine_learning

Service Mesh Security (Istio Example):

yaml

apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: api-authz
  namespace: api-services
spec:
  selector:
    matchLabels:
      app: user-api
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/ingress/sa/istio-ingressgateway"]
    to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/api/v1/*"]
    when:
    - key: request.headers[X-API-Key]
      values: ["valid-keys-*"]
---
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: api-mtls
  namespace: api-services
spec:
  mtls:
    mode: STRICT

Monitoring API Abuse

Real-time Abuse Detection System:

python

import json
import redis
from datetime import datetime, timedelta
from collections import defaultdict
import re

class APIAbuseDetector:
    def __init__(self):
        self.redis = redis.Redis()
        self.rules = self._load_detection_rules()
        self.alerts = []
    
    def _load_detection_rules(self):
        return {
            "rate_limiting": {
                "window": 3600,
                "thresholds": {
                    "login": 10,
                    "password_reset": 5,
                    "api_call": 1000
                }
            },
            "anomaly_detection": {
                "unusual_geo": True,
                "unusual_timing": True,
                "data_exfiltration": True
            },
            "pattern_detection": {
                "sqli_patterns": [
                    r"('.+--|\b(UNION|SELECT|INSERT|UPDATE|DELETE|DROP)\b)",
                    r"(\b(OR|AND)\b\s+\d+\s*=\s*\d+)"
                ],
                "xss_patterns": [
                    r"(<script|javascript:|on\w+\s*=)",
                    r"(alert\(|document\.cookie|eval\()"
                ],
                "path_traversal": [
                    r"(\.\./|\.\.\\|%2e%2e%2f)",
                    r"(\b(etc/passwd|win.ini|boot.ini)\b)"
                ]
            }
        }
    
    def analyze_request(self, request_data):
        """Analyze incoming request for abuse patterns"""
        findings = []
        
        # Check rate limiting
        rate_finding = self._check_rate_limits(request_data)
        if rate_finding:
            findings.append(rate_finding)
        
        # Check for attack patterns
        pattern_finding = self._check_attack_patterns(request_data)
        if pattern_finding:
            findings.append(pattern_finding)
        
        # Check for anomalies
        anomaly_finding = self._check_anomalies(request_data)
        if anomaly_finding:
            findings.append(anomaly_finding)
        
        # Check for data exfiltration
        exfil_finding = self._check_data_exfiltration(request_data)
        if exfil_finding:
            findings.append(exfil_finding)
        
        # Take action based on severity
        if findings:
            self._take_action(findings, request_data)
        
        return findings
    
    def _check_rate_limits(self, request):
        client_id = request.get('client_ip') or request.get('api_key')
        endpoint = request['path']
        
        # Get thresholds for this endpoint
        endpoint_type = self._categorize_endpoint(endpoint)
        threshold = self.rules['rate_limiting']['thresholds'].get(
            endpoint_type, 
            100
        )
        
        # Use sliding window counter
        key = f"rate:{client_id}:{endpoint_type}"
        window = self.rules['rate_limiting']['window']
        
        # Add request to sorted set
        now = datetime.now().timestamp()
        self.redis.zadd(key, {str(now): now})
        
        # Remove old requests
        self.redis.zremrangebyscore(key, 0, now - window)
        
        # Count requests in window
        count = self.redis.zcard(key)
        
        if count > threshold:
            return {
                "type": "RATE_LIMIT_EXCEEDED",
                "client": client_id,
                "endpoint": endpoint,
                "count": count,
                "threshold": threshold,
                "severity": "MEDIUM"
            }
        
        return None
    
    def _check_attack_patterns(self, request):
        patterns = self.rules['pattern_detection']
        
        # Check request body
        body = json.dumps(request.get('body', {}))
        
        for category, pattern_list in patterns.items():
            for pattern in pattern_list:
                if re.search(pattern, body, re.IGNORECASE):
                    return {
                        "type": "ATTACK_PATTERN_DETECTED",
                        "category": category,
                        "pattern": pattern,
                        "severity": "HIGH",
                        "request_id": request.get('id')
                    }
        
        return None
    
    def _check_data_exfiltration(self, request):
        """Detect potential data exfiltration attempts"""
        
        # Check for large responses to the same client
        client_id = request.get('client_ip')
        response_size = request.get('response_size', 0)
        
        if response_size > 10 * 1024 * 1024:  # 10MB
            # Check if this client has been requesting large amounts of data
            key = f"exfil:{client_id}"
            total_data = self.redis.incrby(key, response_size)
            self.redis.expire(key, 3600)  # 1 hour window
            
            if total_data > 100 * 1024 * 1024:  # 100MB in 1 hour
                return {
                    "type": "DATA_EXFILTRATION_SUSPECTED",
                    "client": client_id,
                    "data_transferred": total_data,
                    "severity": "HIGH"
                }
        
        return None
    
    def _check_anomalies(self, request):
        """Machine learning based anomaly detection"""
        features = {
            "request_size": len(json.dumps(request.get('body', {}))),
            "response_size": request.get('response_size', 0),
            "response_time": request.get('response_time', 0),
            "unusual_hour": 1 if 2 <= datetime.now().hour <= 5 else 0,
            "error_rate": self._calculate_error_rate(request['path'])
        }
        
        # Simple rule-based anomaly detection
        anomalies = []
        
        if features['response_time'] > 5000:  # 5 seconds
            anomalies.append("HIGH_RESPONSE_TIME")
        
        if features['request_size'] > 100 * 1024:  # 100KB
            anomalies.append("LARGE_REQUEST")
        
        if features['error_rate'] > 0.5:  # 50% error rate
            anomalies.append("HIGH_ERROR_RATE")
        
        if anomalies:
            return {
                "type": "ANOMALY_DETECTED",
                "anomalies": anomalies,
                "features": features,
                "severity": "MEDIUM"
            }
        
        return None
    
    def _take_action(self, findings, request):
        """Take appropriate action based on findings"""
        
        for finding in findings:
            severity = finding.get('severity', 'LOW')
            client_id = finding.get('client')
            
            actions = {
                "LOW": ["LOG_ONLY"],
                "MEDIUM": ["LOG", "ALERT", "THROTTLE"],
                "HIGH": ["LOG", "ALERT", "BLOCK", "BAN_TEMPORARY"],
                "CRITICAL": ["LOG", "ALERT", "BLOCK", "BAN_PERMANENT"]
            }
            
            for action in actions.get(severity, ["LOG"]):
                if action == "BLOCK":
                    self._block_client(client_id, finding['type'])
                elif action == "THROTTLE":
                    self._throttle_client(client_id)
                elif action == "ALERT":
                    self._send_alert(finding)
    
    def generate_dashboard_metrics(self):
        """Generate metrics for security dashboard"""
        metrics = {
            "total_requests": self.redis.get("stats:total_requests") or 0,
            "blocked_requests": self.redis.get("stats:blocked_requests") or 0,
            "threat_categories": {
                "sqli": self.redis.get("threats:sqli") or 0,
                "xss": self.redis.get("threats:xss") or 0,
                "rate_limit": self.redis.get("threats:rate_limit") or 0,
                "anomaly": self.redis.get("threats:anomaly") or 0
            },
            "top_attackers": self._get_top_attackers(),
            "response_time_percentiles": self._get_response_time_stats()
        }
        
        return metrics

API Security Dashboard Metrics:

json

{
  "security_dashboard": {
    "traffic_analysis": {
      "total_requests": 1500000,
      "malicious_requests": 1250,
      "blocked_requests": 980,
      "false_positives": 12,
      "threat_prevention_rate": "99.92%"
    },
    "threat_intelligence": {
      "top_attack_types": [
        {"type": "SQLi", "count": 450, "trend": "↓ 15%"},
        {"type": "BOLA", "count": 320, "trend": "↑ 25%"},
        {"type": "Rate Limit Abuse", "count": 280, "trend": "→"},
        {"type": "Credential Stuffing", "count": 150, "trend": "↑ 40%"}
      ],
      "top_attacker_countries": ["US", "CN", "RU", "IN"],
      "top_targeted_endpoints": ["/api/login", "/api/users", "/api/payments"]
    },
    "performance_impact": {
      "avg_blocking_latency": "2.3ms",
      "avg_detection_time": "45ms",
      "system_overhead": "3.2%"
    },
    "compliance": {
      "pii_exposure_attempts": 45,
      "gdpr_compliance_score": "98.5%",
      "pci_dss_compliance": "100%",
      "hipaa_violations": 0
    }
  }
}

5. ADVANCED DEFENSE PATTERNS

Zero-Trust API Architecture

python

class ZeroTrustAPIGateway:
    def __init__(self):
        self.policies = ZeroTrustPolicies()
        self.context_engine = ContextEngine()
        self.decision_engine = DecisionEngine()
    
    def handle_request(self, request):
        # Step 1: Collect context
        context = self._collect_context(request)
        
        # Step 2: Authenticate
        auth_result = self._authenticate(request, context)
        if not auth_result.success:
            return self._deny_request(auth_result.reason)
        
        # Step 3: Authorize with context
        authz_result = self._authorize(request, context)
        if not authz_result.success:
            return self._deny_request(authz_result.reason)
        
        # Step 4: Apply data filtering
        filtered_request = self._apply_data_policies(request, context)
        
        # Step 5: Forward to backend
        response = self._forward_to_backend(filtered_request)
        
        # Step 6: Filter response data
        filtered_response = self._filter_response(response, context)
        
        # Step 7: Log and audit
        self._audit_request(request, response, context)
        
        return filtered_response
    
    def _collect_context(self, request):
        """Collect rich context for decision making"""
        return {
            "request": {
                "ip": request.remote_addr,
                "geo": self._get_geo_location(request.remote_addr),
                "user_agent": request.user_agent.string,
                "device_fingerprint": self._get_device_fingerprint(request),
                "time": datetime.now(),
                "tls_info": request.environ.get('SSL_CLIENT_CERT')
            },
            "user": {
                "id": self._extract_user_id(request),
                "role": self._extract_user_role(request),
                "behavior_baseline": self._get_user_behavior(request),
                "risk_score": self._calculate_risk_score(request)
            },
            "resource": {
                "sensitivity": self._get_resource_sensitivity(request.path),
                "data_classification": self._get_data_classification(request.path),
                "compliance_requirements": self._get_compliance_requirements(request.path)
            }
        }

API Security Automation

python

# Infrastructure as Code for API Security
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: api-security-policy
  namespace: api-services
spec:
  podSelector:
    matchLabels:
      app: api-gateway
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: ingress-namespace
    ports:
    - protocol: TCP
      port: 443
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: internal-services
    ports:
    - protocol: TCP
      port: 8080
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: api-rate-limit
  namespace: api-services
spec:
  selector:
    matchLabels:
      app: api-gateway
  rules:
  - to:
    - operation:
        paths: ["/api/*"]
    when:
    - key: request.headers[X-API-Key]
      values: ["*"]
    - key: request.headers[X-Forwarded-For]
      values: ["*"]
---
# GitOps security policy
apiVersion: templates.gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: apisecuritypolicies
spec:
  crd:
    spec:
      names:
        kind: APISecurityPolicy
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package apisecurity
        
        violation[{"msg": msg}] {
          input.request.kind.kind == "Deployment"
          deployment := input.request.object
          not deployment.spec.template.metadata.annotations["api-security-score"]
          msg := "All API deployments must have security score annotation"
        }
        
        violation[{"msg": msg}] {
          input.request.kind.kind == "Service"
          service := input.request.object
          not service.metadata.annotations["owasp-api-compliance"]
          msg := "API services must declare OWASP compliance level"
        }

6. CONTINUOUS API SECURITY PRACTICES

API Security Maturity Model

text

Level 0: Ad-hoc
├── No API security standards
├── Manual testing only
└── Reactive incident response

Level 1: Basic
├── API inventory exists
├── Basic authentication
├── Manual security testing
└── Basic monitoring

Level 2: Standardized
├── API security standards defined
├── Automated testing in CI/CD
├── OAuth 2.0 implemented
└── Basic threat detection

Level 3: Advanced
├── Zero-trust architecture
├── Real-time threat prevention
├── Machine learning anomaly detection
└── Comprehensive API governance

Level 4: Optimized
├── Predictive security analytics
├── Automated threat response
├── Continuous compliance validation
└── Business-aligned security metrics

API Security Checklist

markdown

# API Security Implementation Checklist

## Authentication & Authorization
- Implement OAuth 2.0 with PKCE
- Use asymmetric keys for JWT signing
- Implement token revocation
- Apply principle of least privilege
- Implement proper session management

## Input Validation
- Validate against JSON schemas
- Implement allow-list validation
- Sanitize all inputs
- Implement request size limits
- Validate content types

## Data Protection
- Implement field-level authorization
- Encrypt sensitive data at rest
- Use TLS 1.2+ everywhere
- Implement data masking
- Apply data retention policies

## Monitoring & Logging
- Log all security events
- Implement real-time monitoring
- Set up alerting for anomalies
- Implement audit trails
- Regular security dashboard reviews

## Infrastructure
- Implement API gateway
- Configure WAF rules
- Set up rate limiting
- Implement DDoS protection
- Regular penetration testing

## Development
- Security requirements in user stories
- Automated security testing in CI/CD
- Regular dependency updates
- Security code reviews
- Threat modeling for new features

This extensive API security guide provides both foundational knowledge and advanced implementation patterns. Remember that API security requires a defense-in-depth approach, combining proper authentication, authorization, input validation, monitoring, and continuous testing to protect against evolving threats.

Comments (0)

No comments yet. Be the first to comment!

Advertisements
Swapday Free Marketplace
Software Support
📢 Contact for Advertisement

LecturreAI Assistant

Hello! I'm LecturreAI

What would you like to learn today?