Sequence Diagram 15: Security Incident Response

Overview

This diagram illustrates the complete security incident detection, response, and remediation workflow for the CODITECT License Management Platform. The system employs multiple layers of security monitoring and automated response mechanisms.

Purpose: Demonstrate how security incidents are detected, escalated, and resolved

Actors:

• Attacker - Malicious actor attempting unauthorized access
• Cloud Armor - DDoS protection and WAF
• License API - Django REST Framework backend
• Security Monitoring - Cloud Logging + Security Command Center
• Incident Response - Automated and manual response systems
• Security Team - Human security operations team
• PagerDuty - Incident alerting and escalation

---

Sequence Diagram

---

Incident Severity Levels

Severity	Response Time	Escalation	Examples
P0 - CRITICAL	Immediate (< 5 min)	Page entire security team	Data breach attempt, tenant isolation violation
P1 - HIGH	< 15 minutes	Page on-call engineer	DDoS attack, compromised credentials
P2 - MEDIUM	< 1 hour	Slack notification	SQL injection attempt, brute force attack
P3 - LOW	< 4 hours	Email notification	Single failed login, rate limit triggered

---

Security Monitoring Components

1. Cloud Armor (WAF + DDoS Protection)

Configuration:

# opentofu/modules/cloud-armor/main.tf

resource "google_compute_security_policy" "license_api_policy" {
  name = "license-api-security-policy"

  # Rate limiting rule
  rule {
    action   = "rate_based_ban"
    priority = 1000
    match {
      versioned_expr = "SRC_IPS_V1"
      config {
        src_ip_ranges = ["*"]
      }
    }
    rate_limit_options {
      conform_action = "allow"
      exceed_action  = "deny(429)"
      enforce_on_key = "IP"

      rate_limit_threshold {
        count        = 100
        interval_sec = 60
      }

      ban_duration_sec = 600  # 10-minute ban
    }
  }

  # SQL injection detection
  rule {
    action   = "deny(403)"
    priority = 2000
    match {
      expr {
        expression = "evaluatePreconfiguredExpr('sqli-stable')"
      }
    }
  }

  # XSS detection
  rule {
    action   = "deny(403)"
    priority = 3000
    match {
      expr {
        expression = "evaluatePreconfiguredExpr('xss-stable')"
      }
    }
  }

  # Geo-blocking (optional)
  rule {
    action   = "deny(403)"
    priority = 4000
    match {
      expr {
        expression = "origin.region_code in ['CN', 'RU', 'KP']"  # Block high-risk regions
      }
    }
    description = "Block traffic from high-risk regions"
  }

  # Default allow rule
  rule {
    action   = "allow"
    priority = 2147483647
    match {
      versioned_expr = "SRC_IPS_V1"
      config {
        src_ip_ranges = ["*"]
      }
    }
  }
}

2. Django Security Middleware

Backend Configuration:

# backend/coditect_license/settings.py

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'corsheaders.middleware.CorsMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',

    # Custom security middleware
    'core.middleware.SecurityLoggingMiddleware',
    'core.middleware.TenantIsolationMiddleware',
    'core.middleware.AnomalyDetectionMiddleware',
]

# Security headers
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
X_FRAME_OPTIONS = 'DENY'
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_HSTS_SECONDS = 31536000
SECURE_HSTS_INCLUDE_SUBDOMAINS = True
SECURE_HSTS_PRELOAD = True

Security Logging Middleware:

# backend/core/middleware.py

import logging
from django.utils.deprecation import MiddlewareMixin

logger = logging.getLogger('security')

class SecurityLoggingMiddleware(MiddlewareMixin):
    """Log all security-relevant events."""

    def process_request(self, request):
        # Log authentication failures
        if request.path.startswith('/api/v1/auth/'):
            logger.info(f"Auth request: {request.method} {request.path}", extra={
                'ip': request.META.get('REMOTE_ADDR'),
                'user_agent': request.META.get('HTTP_USER_AGENT'),
                'path': request.path,
            })

    def process_exception(self, request, exception):
        # Log security exceptions
        if isinstance(exception, (PermissionDenied, SuspiciousOperation)):
            logger.warning(f"Security exception: {exception}", extra={
                'ip': request.META.get('REMOTE_ADDR'),
                'path': request.path,
                'exception': str(exception),
            })

3. Anomaly Detection Middleware

Behavioral Analysis:

# backend/core/middleware.py

from django.core.cache import cache
from django.contrib.gis.geoip2 import GeoIP2

class AnomalyDetectionMiddleware(MiddlewareMixin):
    """Detect anomalous user behavior."""

    def __init__(self, get_response):
        super().__init__(get_response)
        self.geoip = GeoIP2()

    def process_request(self, request):
        if not request.user.is_authenticated:
            return

        user_id = request.user.id
        ip_address = request.META.get('REMOTE_ADDR')

        # Get current location
        try:
            current_country = self.geoip.country(ip_address)
        except:
            current_country = None

        # Get last known location
        cache_key = f"user:{user_id}:last_location"
        last_location = cache.get(cache_key)

        if last_location and current_country:
            # Detect impossible travel
            if last_location != current_country:
                # Different country in < 1 hour = suspicious
                logger.warning(f"Anomalous access: Impossible travel detected", extra={
                    'user_id': user_id,
                    'last_location': last_location,
                    'current_location': current_country,
                    'ip': ip_address,
                })

                # Trigger security alert
                self.trigger_security_alert(
                    user_id=user_id,
                    alert_type='IMPOSSIBLE_TRAVEL',
                    severity='HIGH',
                )

        # Update last known location
        if current_country:
            cache.set(cache_key, current_country, timeout=3600)

    def trigger_security_alert(self, user_id, alert_type, severity):
        """Send security alert to SIEM."""
        from google.cloud import logging

        client = logging.Client()
        logger = client.logger('security-alerts')

        logger.log_struct({
            'user_id': str(user_id),
            'alert_type': alert_type,
            'severity': severity,
            'timestamp': timezone.now().isoformat(),
        }, severity='WARNING')

---

Automated Incident Response

Response Actions by Severity

# backend/core/incident_response.py

class IncidentResponse:
    """Automated incident response actions."""

    @staticmethod
    def respond_to_ddos(source_ip: str):
        """Respond to DDoS attack."""
        # 1. Add to Cloud Armor blocklist
        add_to_cloud_armor_blocklist(source_ip, duration=3600)

        # 2. Log incident
        logger.critical(f"DDoS attack detected from {source_ip}")

        # 3. Create PagerDuty incident
        create_pagerduty_incident(
            title=f"DDoS Attack from {source_ip}",
            severity="P1",
            description="Automated DDoS mitigation triggered",
        )

    @staticmethod
    def respond_to_sql_injection(user_id: str, request_data: dict):
        """Respond to SQL injection attempt."""
        # 1. Log detailed attack info
        logger.warning(f"SQL injection attempt by user {user_id}", extra={
            'user_id': user_id,
            'request_data': request_data,
        })

        # 2. Rate limit user
        cache.set(f"rate_limit:{user_id}", True, timeout=3600)

        # 3. Create incident
        create_pagerduty_incident(
            title=f"SQL Injection Attempt by User {user_id}",
            severity="P2",
            description="Automated blocking applied",
        )

    @staticmethod
    def respond_to_tenant_isolation_breach(user_id: str, attempted_tenant_id: str):
        """Respond to tenant isolation breach."""
        # 1. Suspend user account immediately
        user = User.objects.get(id=user_id)
        user.is_active = False
        user.save()

        # 2. Force logout all sessions
        Session.objects.filter(user_id=user_id).update(status='released')

        # 3. Log critical incident
        logger.critical(f"Tenant isolation breach by user {user_id}", extra={
            'user_id': user_id,
            'attempted_tenant': attempted_tenant_id,
            'user_tenant': user.tenant_id.id,
        })

        # 4. Create P0 incident
        create_pagerduty_incident(
            title=f"CRITICAL: Tenant Isolation Breach by User {user_id}",
            severity="P0",
            description="User account suspended. Manual forensic review required.",
            urgency="high",
        )

        # 5. Notify security team immediately
        send_security_alert_email(
            subject="URGENT: Tenant Isolation Breach Detected",
            body=f"User {user_id} attempted to access tenant {attempted_tenant_id}",
        )

---

Security Monitoring Dashboards

Grafana Dashboard Configuration

# grafana/dashboards/security-monitoring.json

{
  "dashboard": {
    "title": "Security Monitoring - License API",
    "panels": [
      {
        "title": "Failed Authentication Attempts (Last 24h)",
        "type": "graph",
        "targets": [{
          "expr": "sum(rate(django_auth_failed_total[5m])) by (reason)"
        }]
      },
      {
        "title": "Cloud Armor Blocked Requests",
        "type": "stat",
        "targets": [{
          "expr": "sum(cloudarmor_blocked_requests_total)"
        }]
      },
      {
        "title": "SQL Injection Attempts",
        "type": "table",
        "targets": [{
          "expr": "security_events{type='sql_injection'}"
        }]
      },
      {
        "title": "Tenant Isolation Violations",
        "type": "alert-list",
        "targets": [{
          "expr": "security_events{type='tenant_breach'}"
        }]
      },
      {
        "title": "Geographic Anomalies",
        "type": "worldmap",
        "targets": [{
          "expr": "security_events{type='geo_anomaly'}"
        }]
      }
    ]
  }
}

---

Incident Response Runbook

Response Checklist by Incident Type

DDoS Attack:

• Verify attack in Cloud Armor logs
• Check if auto-mitigation triggered
• Review rate limit rules effectiveness
• Add source IPs to permanent blocklist if needed
• Document attack pattern and mitigation

SQL Injection:

• Verify attack pattern in application logs
• Check if WAF rule triggered
• Review serializer validation
• Update validation rules if bypass discovered
• Document attack vector

Tenant Isolation Breach:

• URGENT: Suspend user account immediately
• Review all recent actions by user
• Check for data exfiltration
• Audit multi-tenant middleware
• Forensic analysis of breach attempt
• Update isolation rules if needed

Compromised Credentials:

• Revoke API keys/tokens immediately
• Force logout all sessions
• Review access logs for unauthorized actions
• Contact legitimate user
• Issue new credentials securely
• Update anomaly detection rules

---

Testing & Validation

Security Test Scenarios

# backend/licenses/tests/test_security.py

import pytest
from django.test import Client

@pytest.mark.security
def test_sql_injection_blocked():
    """Verify SQL injection attempts are blocked."""
    client = Client()

    # Attempt SQL injection
    response = client.post('/api/v1/license/acquire', {
        'license_key': "' OR 1=1--",
        'hardware_id': 'test123',
    })

    # Should be rejected
    assert response.status_code == 400
    assert 'Invalid license key format' in response.json()['detail']

@pytest.mark.security
def test_tenant_isolation_enforced():
    """Verify multi-tenant isolation is enforced."""
    # Create two tenants
    tenant1 = Tenant.objects.create(name="Tenant 1")
    tenant2 = Tenant.objects.create(name="Tenant 2")

    # Create user in tenant1
    user = User.objects.create(tenant_id=tenant1, email="user@tenant1.com")

    # Attempt to access tenant2 license
    client = Client()
    client.force_login(user)

    license_tenant2 = License.objects.create(tenant_id=tenant2)

    response = client.get(f'/api/v1/licenses/{license_tenant2.id}/')

    # Should be denied
    assert response.status_code == 403
    assert 'Tenant access denied' in response.json()['detail']

@pytest.mark.security
def test_rate_limiting_works():
    """Verify rate limiting prevents abuse."""
    client = Client()

    # Make 101 requests in rapid succession
    for i in range(101):
        response = client.post('/api/v1/license/acquire', {...})

    # 101st request should be rate limited
    assert response.status_code == 429

---

Compliance & Audit Trail

Security Event Logging Format

{
  "timestamp": "2025-11-30T12:00:00Z",
  "severity": "WARNING",
  "event_type": "SECURITY_INCIDENT",
  "incident_id": "INC-2025-11-30-001",
  "category": "SQL_INJECTION",
  "source_ip": "1.2.3.4",
  "user_id": "550e8400-e29b-41d4-a716-446655440000",
  "tenant_id": "660e8400-e29b-41d4-a716-446655440000",
  "request": {
    "method": "POST",
    "path": "/api/v1/license/acquire",
    "payload": {"license_key": "' OR 1=1--"}
  },
  "response": {
    "status_code": 400,
    "action": "BLOCKED"
  },
  "automated_response": {
    "action": "RATE_LIMIT_APPLIED",
    "duration": 3600,
    "pagerduty_incident": "PD-12345"
  },
  "tags": ["security", "attack", "sql_injection"]
}

---

Status: Specification Complete ✅ Implementation: Infrastructure ready, code implementation in Phase 2 Dependencies: Cloud Armor, Cloud Logging, Security Command Center, PagerDuty Estimated Implementation: 3-4 days

---

Last Updated: November 30, 2025 Owner: Security Team Reviewed By: DevOps Team, Compliance Team