# Implementation Guide

This document walks through the actual code. We'll build key features step by step and explain the decisions along the way.

## File Structure Walkthrough
```
backend/
├── main.py              # Application entry point
├── factory.py           # FastAPI app factory with middleware
├── config.py            # Environment variables and constants
├── core/
│   ├── database.py      # SQLAlchemy engine and session factory
│   ├── security.py      # Password hashing, JWT creation/validation
│   ├── dependencies.py  # FastAPI dependencies (auth, database)
│   └── enums.py         # Type-safe enums for status, severity, test types
├── models/
│   ├── Base.py          # Base model with common fields (id, timestamps)
│   ├── User.py          # User authentication model
│   ├── Scan.py          # Scan metadata model
│   └── TestResult.py    # Individual test results model
├── repositories/
│   ├── user_repository.py         # User database operations
│   ├── scan_repository.py         # Scan database operations
│   └── test_result_repository.py  # Test result database operations
├── routes/
│   ├── auth.py          # Registration and login endpoints
│   └── scans.py         # Scan CRUD endpoints
├── schemas/
│   ├── user_schemas.py        # Pydantic models for user data
│   ├── scan_schemas.py        # Pydantic models for scan data
│   └── test_result_schemas.py # Pydantic models for test results
├── services/
│   ├── auth_service.py  # Authentication business logic
│   └── scan_service.py  # Scan orchestration business logic
└── scanners/
    ├── base_scanner.py      # Common HTTP logic, retry handling
    ├── rate_limit_scanner.py # Rate limiting detection
    ├── auth_scanner.py      # Authentication vulnerability detection
    ├── sqli_scanner.py      # SQL injection detection
    ├── idor_scanner.py      # IDOR/BOLA detection
    └── payloads.py          # Attack payloads organized by type
```

## Building Authentication

### Step 1: Password Hashing with Bcrypt

What we're building: Secure password storage that protects user credentials even if the database is compromised.

The code lives in `backend/core/security.py:11-28`:
```python
import bcrypt
from jose import JWTError, jwt

def hash_password(password: str) -> str:
    """
    Hash a plain text password using bcrypt
    """
    password_bytes = password.encode("utf-8")  # Convert string to bytes
    salt = bcrypt.gensalt()  # Generate random salt (default 12 rounds)
    hashed = bcrypt.hashpw(password_bytes, salt)
    return hashed.decode("utf-8")  # Convert bytes back to string for storage
```

**Why this code works:**
- **Line 18-19**: Bcrypt requires byte input, not strings. The encoding/decoding dance converts between Python strings and byte arrays.
- **Line 19**: `bcrypt.gensalt()` generates a unique salt for each password. The salt is embedded in the output hash, so you don't store it separately.
- **Line 20**: `bcrypt.hashpw()` does the actual hashing. With default 12 rounds, this intentionally takes ~100ms. That's the point - makes brute forcing expensive.
- **Line 21**: Convert back to string because SQLAlchemy's `String` column type expects strings, not bytes.

The verification function (`security.py:21-31`):
```python
def verify_password(plain_password: str, hashed_password: str) -> bool:
    """
    Verify a plain text password against a hashed password
    """
    password_bytes = plain_password.encode("utf-8")
    hashed_bytes = hashed_password.encode("utf-8")
    return bcrypt.checkpw(password_bytes, hashed_bytes)
```

**What's happening:**
1. Convert both inputs to bytes
2. `bcrypt.checkpw()` extracts the salt from `hashed_bytes` (it's embedded in the hash)
3. Hashes `password_bytes` with that same salt
4. Compares the result - if they match, password is correct

**Common mistakes here:**
```python
# Wrong - storing plaintext
user.password = password  # Database breach = everyone's password leaked

# Wrong - using weak hashing
import hashlib
user.password = hashlib.md5(password.encode()).hexdigest()  # Fast = easy to brute force

# Wrong - forgetting to salt
user.password = hashlib.sha256(password.encode()).hexdigest()  # Rainbow tables break this

# Right - bcrypt with automatic salting
user.hashed_password = hash_password(password)  # Secure
```

### Step 2: JWT Token Creation

Now we need to create authentication tokens after successful login.

In `core/security.py:34-56`:
```python
from datetime import datetime, timedelta
from config import settings

def create_access_token(
    data: dict[str, str],
    expires_delta: timedelta | None = None
) -> str:
    """
    Create a JWT access token
    """
    to_encode = data.copy()  # Don't modify the original dict
    
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES  # 1440 = 24 hours
        )
    
    to_encode.update({"exp": expire})  # Add expiration claim
    
    encoded_jwt = jwt.encode(
        to_encode,
        settings.SECRET_KEY,  # MUST be random, 256+ bits
        algorithm=settings.ALGORITHM  # "HS256"
    )
    return encoded_jwt
```

**Key parts explained:**

**Line 43** - Copy the data dict because we're about to modify it. If we mutated the original, the caller would see `{"sub": "user@example.com", "exp": 1234567890}` instead of just the email.

**Line 45-50** - Calculate when the token expires. If no `expires_delta` is passed, default to 24 hours from `config.py`. The `exp` claim is part of the JWT standard - libraries automatically check it.

**Line 52** - This is where the magic happens. The `jwt.encode()` function:
1. Converts `to_encode` dict to JSON
2. Base64 encodes it as the payload
3. Creates a header: `{"alg": "HS256", "typ": "JWT"}`
4. Computes HMAC-SHA256 signature using `SECRET_KEY`
5. Joins header.payload.signature with periods

The result looks like:
```
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyQGV4YW1wbGUuY29tIiwiZXhwIjoxNzM4MzY4MDAwfQ.signature_here
```

**Why we do it this way:**
JWTs are stateless. The server doesn't need to store session data. When a request comes in with a JWT, the server verifies the signature and trusts the payload. This means you can scale horizontally - any backend instance can validate any token without coordinating with other instances or a shared session store.

**Alternative approaches:**
- **Session-based auth**: Store session ID in cookie, look up user data on each request. Simpler but requires session storage (Redis, database). Doesn't scale as easily.
- **OAuth 2.0 with refresh tokens**: Short-lived access tokens (15 min) + long-lived refresh tokens. More secure (can revoke refresh tokens) but more complex to implement.

### Step 3: Token Validation

When a protected endpoint receives a request, we need to validate the JWT and load the user.

The dependency injection function in `core/dependencies.py:17-49`:
```python
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from .security import decode_token
from .database import get_db
from repositories.user_repository import UserRepository
from schemas.user_schemas import UserResponse

security = HTTPBearer()  # Extracts "Authorization: Bearer <token>" header

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db),
) -> UserResponse:
    """
    FastAPI dependency to extract and verify the current authenticated user
    """
    try:
        # Decode and verify the JWT
        payload = decode_token(credentials.credentials)
        email: str | None = payload.get("sub")
        
        if email is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication credentials",
                headers={"WWW-Authenticate": "Bearer"},
            )
        
        # Load user from database
        user = UserRepository.get_by_email(db, email)
        
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="User not found",
                headers={"WWW-Authenticate": "Bearer"},
            )
        
        return UserResponse.model_validate(user)
    
    except ValueError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        ) from None
```

**What's happening:**

**Line 27** - `HTTPBearer()` is FastAPI's built-in extractor for `Authorization: Bearer <token>` headers. It parses the header and gives us the token string in `credentials.credentials`.

**Line 35** - Call `decode_token()` which lives in `security.py:48-62`:
```python
def decode_token(token: str) -> dict[str, str]:
    """
    Decode and verify a JWT token
    """
    try:
        payload = jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=[settings.ALGORITHM]  # Only accept HS256
        )
        return payload
    except JWTError as e:
        raise ValueError(f"Invalid token: {str(e)}") from e
```

This verifies:
- Signature is valid (token wasn't tampered with)
- Algorithm matches expected (prevents "none" algorithm attack)
- Token hasn't expired (checks `exp` claim automatically)

**Line 36** - Extract the `sub` (subject) claim. This is the user identifier we put in the token during login.

**Line 38-43** - If `sub` is missing, the token is malformed. Return 401 with `WWW-Authenticate` header per HTTP standards.

**Line 46** - Load the full user record from the database. We could skip this and just use the email from the token, but loading from DB ensures:
- User still exists (wasn't deleted after token was issued)
- User is still active (wasn't deactivated)
- We get the full user object with all fields

**Line 48-52** - If user doesn't exist in database, token is invalid. This catches deleted users with valid tokens.

**Line 54** - Convert SQLAlchemy model to Pydantic schema. This excludes `hashed_password` from the response (Pydantic schema doesn't include it).

**Line 56-61** - Catch `ValueError` from `decode_token()` and convert to HTTP 401. The `from None` suppresses the chained exception traceback - cleaner error messages for clients.

### Testing Authentication

How to verify this works:
```bash
# Register a user
curl -X POST http://localhost:8000/auth/register \
  -H "Content-Type: application/json" \
  -d '{
    "email": "test@example.com",
    "password": "SecurePass123"
  }'

# Response: {"id": 1, "email": "test@example.com", ...}

# Login
curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "email": "test@example.com",
    "password": "SecurePass123"
  }'

# Response: {"access_token": "eyJ...", "token_type": "bearer"}

# Use token to access protected endpoint
curl http://localhost:8000/scans/ \
  -H "Authorization: Bearer eyJ..."

# Response: [list of scans]
```

Expected output: First request creates user, second returns JWT, third returns scans array (empty initially).

If you see `401 Unauthorized`, check:
- Token is being sent in header (not query param, not body)
- Format is exactly `Bearer <token>` with capital B
- Token hasn't expired (24 hours from login)
- User still exists in database

## Building SQL Injection Detection

### The Problem

We need to detect three types of SQL injection:
1. **Error-based** - Database errors leak information
2. **Boolean-based blind** - Responses differ for true/false conditions
3. **Time-based blind** - Database delays reveal injection

### The Solution

Use payload libraries and response analysis to detect each type. The scanner lives in `backend/scanners/sqli_scanner.py`.

### Implementation

Starting with error-based detection (`sqli_scanner.py:51-94`):
```python
def _test_error_based_sqli(self) -> dict[str, Any]:
    """
    Test for error based SQL injection
    
    Detects database errors in responses indicating SQLi vulnerability
    """
    error_signatures = SQLiPayloads.get_error_signatures()
    
    basic_payloads = SQLiPayloads.BASIC_AUTHENTICATION_BYPASS
    
    for payload in basic_payloads:
        try:
            response = self.make_request("GET", f"/?id={payload}")
            
            response_text_lower = response.text.lower()
            
            # Look for database error signatures
            for db_type, signatures in error_signatures.items():
                for signature in signatures:
                    if signature in response_text_lower:
                        return {
                            "vulnerable": True,
                            "database_type": db_type,
                            "payload": payload,
                            "status_code": response.status_code,
                            "error_signature": signature,
                            "response_excerpt": response.text[:500],
                        }
        
        except Exception:
            continue  # Network errors don't indicate SQLi
    
    return {
        "vulnerable": False,
        "payloads_tested": len(basic_payloads),
        "description": "No database errors detected",
    }
```

**Key parts explained:**

**Line 57** - Load error signatures from `scanners/payloads.py:9-30`. These are database-specific error messages:
```python
ERROR_SIGNATURES = {
    "mysql": [
        "sql syntax",
        "mysql_fetch",
        "mysql error",
    ],
    "postgres": [
        "postgresql",
        "pg_query",
        "syntax error",
    ],
    "mssql": [
        "sqlserver jdbc driver",
        "sqlexception",
    ],
}
```

**Line 59** - Basic payloads like `' OR '1'='1`, `' OR 1=1--`, `admin'--` from `payloads.py:33-48`.

**Line 63** - Make request with payload in query parameter. If the backend does:
```python
query = f"SELECT * FROM users WHERE id = {request.args['id']}"
```

And we send `id=' OR 1=1--`, it becomes:
```sql
SELECT * FROM users WHERE id = ' OR 1=1--'
```

That's invalid SQL. Database returns error: `"You have an error in your SQL syntax"`.

**Line 65** - Convert response to lowercase for case-insensitive matching. Error messages vary: "SQL syntax", "Sql Syntax", "sql syntax".

**Line 68-72** - Check each signature against response. If `"sql syntax"` appears in the HTML, we found SQLi.

**Line 73-79** - Return evidence: which payload worked, what database type was detected, the actual error message (first 500 chars).

**Line 81-82** - Network errors (timeout, connection refused) don't mean SQLi. Skip to next payload.

**Line 84-88** - No errors found across all payloads = not vulnerable (to error-based SQLi at least).

### Boolean-Based Blind Detection

When errors are suppressed, check if responses differ for true vs false conditions.

Code at `sqli_scanner.py:96-164`:
```python
def _test_boolean_based_sqli(self) -> dict[str, Any]:
    """
    Test for boolean based blind SQL injection
    
    Compares responses from true vs false conditions to detect SQLi
    """
    try:
        # Establish baseline
        baseline_response = self.make_request("GET", "/?id=1")
        baseline_length = len(baseline_response.text)
        baseline_status = baseline_response.status_code
        
        if baseline_status != 200:
            return {
                "vulnerable": False,
                "description": "Baseline request failed",
                "baseline_status": baseline_status,
            }
        
        boolean_payloads = SQLiPayloads.BOOLEAN_BASED_BLIND
        
        # Test true conditions: ' AND 1=1--
        true_payloads = [
            p for p in boolean_payloads
            if "AND '1'='1" in p or "AND 1=1" in p
        ]
        false_payloads = [
            p for p in boolean_payloads
            if "AND '1'='2" in p or "AND 1=2" in p or "AND 1=0" in p
        ]
        
        true_lengths = []
        for payload in true_payloads:
            response = self.make_request("GET", f"/?id={payload}")
            true_lengths.append(len(response.text))
        
        false_lengths = []
        for payload in false_payloads:
            response = self.make_request("GET", f"/?id={payload}")
            false_lengths.append(len(response.text))
        
        # Calculate averages
        avg_true = statistics.mean(true_lengths)
        avg_false = statistics.mean(false_lengths)
        
        length_diff = abs(avg_true - avg_false)
        
        # Significant difference indicates SQLi
        if length_diff > 100 and avg_true != avg_false:
            return {
                "vulnerable": True,
                "baseline_length": baseline_length,
                "true_condition_avg_length": avg_true,
                "false_condition_avg_length": avg_false,
                "length_difference": length_diff,
                "confidence": "HIGH" if length_diff > 500 else "MEDIUM",
            }
        
        return {
            "vulnerable": False,
            "description": "No boolean-based SQLi detected",
            "length_difference": length_diff,
        }
```

**What's happening:**

**Line 105-106** - Get baseline response with normal input (`id=1`). We need this to compare against.

**Line 121-128** - Split payloads into "always true" and "always false" conditions. 

True condition example: `1' AND 1=1--`
```sql
SELECT * FROM users WHERE id = 1' AND 1=1--'
```
The `1=1` is always true, so if vulnerable, this returns data.

False condition example: `1' AND 1=2--`
```sql
SELECT * FROM users WHERE id = 1' AND 1=2--'
```
The `1=2` is never true, so if vulnerable, this returns empty result.

**Line 130-141** - Send multiple true and false payloads, record response lengths.

**Line 144-147** - Calculate average length for true vs false. Statistical approach reduces false positives from network variance.

**Line 151** - If difference is significant (>100 bytes) and not zero, likely vulnerable. A difference of 500+ bytes is high confidence - probably seeing full records vs empty results.

**Why this works:**
If SQLi exists, true conditions return data (longer response), false conditions return nothing (shorter response). If no SQLi, both are treated as invalid input and return the same error page.

### Time-Based Blind Detection

Most sophisticated technique. Use database sleep functions to measure injection.

Code at `sqli_scanner.py:183-251`:
```python
def _test_time_based_sqli(self, delay_seconds: int = 5) -> dict[str, Any]:
    """
    Test for time based blind SQL injection
    
    Uses baseline timing comparison with statistical analysis
    for false positive reduction
    """
    try:
        # Establish baseline timing
        baseline_mean, baseline_stdev = self.get_baseline_timing("/")
        
        threshold = baseline_mean + (3 * baseline_stdev)
        expected_delay_time = baseline_mean + delay_seconds
        
        all_time_payloads = SQLiPayloads.TIME_BASED_BLIND
        
        # Group by database type
        delay_payloads = {
            "mysql": [p for p in all_time_payloads if "SLEEP" in p],
            "postgres": [p for p in all_time_payloads if "pg_sleep" in p],
            "mssql": [p for p in all_time_payloads if "WAITFOR" in p],
        }
        
        for db_type, payloads in delay_payloads.items():
            for payload in payloads:
                delay_times = []
                
                # Take multiple samples
                for _ in range(3):
                    try:
                        response = self.make_request(
                            "GET",
                            f"/?id={payload}",
                            timeout=delay_seconds + 10,
                        )
                        elapsed = getattr(response, "request_time", 0.0)
                        delay_times.append(elapsed)
                    
                    except Exception:
                        delay_times.append(delay_seconds + 10)  # Assume timeout = worked
                    
                    time.sleep(1)  # Space out requests
                
                avg_delay = statistics.mean(delay_times)
                
                # Check if delay matches expected
                if avg_delay >= expected_delay_time - 1:
                    confidence = "HIGH" if avg_delay >= expected_delay_time else "MEDIUM"
                    
                    return {
                        "vulnerable": True,
                        "database_type": db_type,
                        "payload": payload,
                        "baseline_time": f"{baseline_mean:.3f}s",
                        "response_time": f"{avg_delay:.3f}s",
                        "expected_delay": f"{expected_delay_time:.3f}s",
                        "confidence": confidence,
                        "individual_times": [f"{t:.3f}s" for t in delay_times],
                    }
```

**Important details:**

**Line 192** - Get baseline timing from `base_scanner.py:158-177`:
```python
def get_baseline_timing(
    self,
    endpoint: str,
    samples: int | None = None
) -> tuple[float, float]:
    """
    Establish baseline response time for an endpoint
    
    Takes multiple samples and calculates mean and standard deviation
    """
    if samples is None:
        samples = settings.DEFAULT_BASELINE_SAMPLES  # 10
    
    times = []
    
    for _ in range(samples):
        response = self.make_request("GET", endpoint)
        times.append(getattr(response, "request_time", 0.0))
        time.sleep(0.5)  # Space out samples
    
    return statistics.mean(times), statistics.stdev(times)
```

This makes 10 normal requests and calculates average and standard deviation. Example results:
- Mean: 0.15s
- Stdev: 0.02s

**Line 194** - Calculate threshold: `mean + 3*stdev`. With example above: `0.15 + 3*0.02 = 0.21s`. Any response over 0.21s is "unusually slow" (99.7% confidence if normally distributed).

**Line 195** - Expected delay: `baseline + 5s`. If baseline is 0.15s and we inject `SLEEP(5)`, we expect ~5.15s response.

**Line 200-204** - Group payloads by database. MySQL uses `SLEEP(5)`, Postgres uses `pg_sleep(5)`, MSSQL uses `WAITFOR DELAY '0:0:5'`.

**Line 211-225** - Take 3 samples of each payload. Network jitter can cause ±0.5s variance. Averaging 3 samples gives cleaner signal.

**Line 230** - If average delay is within 1 second of expected (5.15s ± 1s), that's SQLi. The 1 second tolerance accounts for network overhead.

**Why this works:**
Payloads like `1'; SELECT SLEEP(5)--` execute the sleep if SQLi exists. The response takes 5 extra seconds. Without SQLi, the payload is just treated as invalid input and returns immediately.

## Security Implementation

### JWT Signature Validation

Critical to prevent token forgery. The none algorithm attack test in `auth_scanner.py:168-213`:
```python
def _test_none_algorithm(self) -> dict[str, Any]:
    """
    Test if server accepts JWT with 'none' algorithm
    
    Critical vulnerability: allows unsigned tokens to be accepted
    """
    try:
        header, payload, signature = self.auth_token.split(".")
        
        none_variants = AuthPayloads.get_jwt_none_variants()
        # ["none", "None", "NONE", "nOnE", "NoNe", "NOne"]
        
        for variant in none_variants:
            # Create malicious header
            malicious_header = self._base64url_encode(
                json.dumps({"alg": variant, "typ": "JWT"})
            )
            
            # Remove signature (trailing period means "no signature")
            malicious_token = f"{malicious_header}.{payload}."
            
            response = self.make_request(
                "GET",
                "/",
                headers={"Authorization": f"Bearer {malicious_token}"}
            )
            
            if response.status_code == 200:
                return {
                    "vulnerable": True,
                    "vulnerability_type": "JWT None Algorithm",
                    "algorithm_variant": variant,
                    "status_code": response.status_code,
                    "recommendations": [
                        "Reject tokens with 'none' algorithm (all case variations)",
                        "Explicitly verify signature before accepting tokens",
                        "Use allowlist of accepted algorithms",
                    ],
                }
        
        return {
            "vulnerable": False,
            "description": "None algorithm properly rejected",
        }
```

**What this prevents:**

Normal JWT: `header.payload.signature`
```
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbiJ9.signature
```

Malicious JWT: `header.payload.` (no signature)
```
eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJzdWIiOiJhZG1pbiJ9.
```

Decoded malicious header:
```json
{"alg": "none", "typ": "JWT"}
```

Decoded malicious payload:
```json
{"sub": "admin"}
```

If the server accepts this, attacker can forge tokens with any user identity.

**How to defend:**

The project's JWT validation (`core/security.py:54-57`) explicitly specifies algorithms:
```python
payload = jwt.decode(
    token,
    settings.SECRET_KEY,
    algorithms=[settings.ALGORITHM]  # ["HS256"] - none not in this list
)
```

The `algorithms` parameter is an allowlist. The library will reject tokens with `alg: none` because it's not in `["HS256"]`.

**What NOT to do:**
```python
# BAD - accepts any algorithm
payload = jwt.decode(token, settings.SECRET_KEY)

# BAD - doesn't verify signature
payload = jwt.decode(token, options={"verify_signature": False})

# BAD - allows none
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256", "none"])
```

### IDOR Prevention Pattern

Authorization checks in `services/scan_service.py:67-84`:
```python
@staticmethod
def get_scan_by_id(db: Session, scan_id: int, user_id: int) -> ScanResponse:
    """
    Get scan by ID with authorization check
    """
    scan = ScanRepository.get_by_id(db, scan_id)
    
    if not scan:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Scan not found",
        )
    
    # CRITICAL: Check if user owns this scan
    if scan.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not authorized to access this scan",
        )
    
    return ScanResponse.model_validate(scan)
```

**Why 404 before 403:**
Line 75-78 returns 404 if scan doesn't exist, BEFORE checking ownership. This prevents information disclosure.

If we checked ownership first:
- Request scan 999 (doesn't exist) → 404 "Scan not found"
- Request scan 123 (exists, belongs to someone else) → 403 "Not authorized"

Now attacker knows scan 123 exists. They can enumerate all scan IDs and find which ones are valid.

Better approach:
- Request scan 999 → 404
- Request scan 123 → 404

Can't tell difference between "doesn't exist" and "exists but you can't access it". Leak less information.

**The authorization check:**
Line 81-84 compares `scan.user_id` (owner) with `user_id` (requester). If they don't match, return 403.

This happens at the service layer, NOT the route layer. Every code path that retrieves scans goes through the service, so we can't forget the check.

## Data Flow Example

Let's trace a complete request through the system: User creates a scan that tests for SQLi.

### Request Comes In

Entry point: `routes/scans.py:23-37`
```python
@router.post("/", response_model=ScanResponse, status_code=status.HTTP_201_CREATED)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)  # "15/minute"
async def create_scan(
    request: Request,
    scan_request: ScanRequest,  # Pydantic validates this
    db: Session = Depends(get_db),  # Injects database session
    current_user: UserResponse = Depends(get_current_user),  # Validates JWT
) -> ScanResponse:
    """
    Create and execute a new security scan
    """
    return ScanService.run_scan(db, current_user.id, scan_request)
```

At this point:
- `scan_request` is validated by Pydantic (correct URL format, valid test types)
- `current_user` is authenticated (JWT was valid, user exists in database)
- `db` is a fresh SQLAlchemy session
- Rate limiting checked - if user exceeded 15 scans/minute, request was rejected before this function runs

What happens next: Call service layer to orchestrate the scan.

### Processing Layer

Service orchestrates scanners: `services/scan_service.py:23-65`
```python
@staticmethod
def run_scan(db: Session, user_id: int, scan_request: ScanRequest) -> ScanResponse:
    # Create scan record
    scan = ScanRepository.create_scan(
        db=db,
        user_id=user_id,
        target_url=str(scan_request.target_url),
    )
    
    # Map test types to scanner classes
    scanner_mapping: dict[TestType, type[BaseScanner]] = {
        TestType.RATE_LIMIT: RateLimitScanner,
        TestType.AUTH: AuthScanner,
        TestType.SQLI: SQLiScanner,
        TestType.IDOR: IDORScanner,
    }
    
    results: list[TestResultCreate] = []
    
    # Execute each requested test
    for test_type in scan_request.tests_to_run:
        scanner_class: type[BaseScanner] | None = scanner_mapping.get(test_type)
        
        if not scanner_class:
            continue  # Skip unknown test types
        
        try:
            scanner = scanner_class(
                target_url=str(scan_request.target_url),
                auth_token=scan_request.auth_token,
                max_requests=scan_request.max_requests,
            )
            
            result = scanner.scan()  # Execute the actual test
            results.append(result)
        
        except Exception as e:
            # Scanner crashed - return error result instead of failing entire scan
            results.append(
                TestResultCreate(
                    test_name=test_type,
                    status="error",
                    severity="info",
                    details=f"Scanner error: {str(e)}",
                    evidence_json={"error": str(e)},
                    recommendations_json=[
                        "Check target URL is accessible",
                        "Verify authentication token if provided",
                    ],
                )
            )
```

This code:
- Creates the scan record immediately (line 26-30) so it has an ID
- Maps test type enums to actual scanner classes (line 33-37)
- Loops through requested tests (line 42-60)
- Instantiates each scanner with target URL and auth token
- Calls `scanner.scan()` which returns `TestResultCreate` with findings
- Catches exceptions so one failing scanner doesn't kill the entire scan

### Storage/Output

Save results to database: `services/scan_service.py:62-65` continues:
```python
    # Save all results
    for result in results:
        TestResultRepository.create_test_result(
            db=db,
            scan_id=scan.id,
            test_name=result.test_name,
            status=result.status,
            severity=result.severity,
            details=result.details,
            evidence_json=result.evidence_json,
            recommendations_json=result.recommendations_json,
        )
    
    db.refresh(scan)  # Reload scan with test_results relationship populated
    
    return ScanResponse.model_validate(scan)
```

The result is a JSON response containing the scan with nested test results:
```json
{
  "id": 42,
  "user_id": 1,
  "target_url": "https://api.example.com/users",
  "scan_date": "2026-02-04T10:30:00Z",
  "created_at": "2026-02-04T10:30:00Z",
  "test_results": [
    {
      "id": 101,
      "scan_id": 42,
      "test_name": "sqli",
      "status": "vulnerable",
      "severity": "critical",
      "details": "Error-based SQL injection detected: mysql",
      "evidence_json": {
        "database_type": "mysql",
        "payload": "' OR 1=1--",
        "error_signature": "sql syntax"
      },
      "recommendations_json": [
        "Use parameterized queries (prepared statements)",
        "Never concatenate user input into SQL queries"
      ]
    }
  ]
}
```

## Error Handling Patterns

### Database Errors with Automatic Rollback

The `get_db()` dependency handles transaction management:
```python
# core/database.py:28-36
def get_db() -> Generator[Session, None, None]:
    """
    FastAPI dependency for database sessions
    """
    db = SessionLocal()
    try:
        yield db  # Provide session to route handler
    finally:
        db.close()  # Always close, even if exception raised
```

If an exception occurs during request processing:
1. FastAPI catches it
2. Control returns to finally block
3. `db.close()` runs
4. SQLAlchemy automatically rolls back uncommitted transaction
5. Connection returned to pool

Example error scenario:
```python
@router.post("/scans/")
async def create_scan(db: Session = Depends(get_db), ...):
    scan = ScanRepository.create_scan(db, ...)  # INSERT INTO scans
    
    # Something goes wrong here
    raise ValueError("Oops")
    
    # This never runs
    TestResultRepository.create_test_result(db, ...)
```

Without explicit transaction handling:
- Scan record would be inserted
- Test result would not be inserted
- Database left in inconsistent state

With `get_db()` cleanup:
- ValueError propagates to FastAPI
- `db.close()` runs
- Scan INSERT is rolled back
- Database remains consistent

**What NOT to do:**
```python
# Bad - manual session management
db = SessionLocal()
try:
    scan = ScanRepository.create_scan(db, ...)
    db.commit()  # Committed before error could happen
    raise ValueError("Oops")
finally:
    db.close()

# Scan is committed, error occurs after - inconsistent state
```

### Scanner Timeout Recovery

Scanners use retry logic with exponential backoff (`base_scanner.py:92-156`):
```python
def make_request(
    self,
    method: str,
    endpoint: str,
    **kwargs: Any,
) -> requests.Response:
    """
    Make HTTP request with retry logic and rate limit handling
    """
    self._wait_before_request()  # Implement request spacing
    
    url = urljoin(self.target_url, endpoint)
    retry_count = 0
    backoff_factor = 2.0
    
    kwargs.setdefault("timeout", settings.SCANNER_CONNECTION_TIMEOUT)
    
    while retry_count <= settings.DEFAULT_RETRY_COUNT:  # 3 retries
        try:
            start_time = time.time()
            response = self.session.request(method, url, **kwargs)
            
            # Track timing for time-based detection
            setattr(response, "request_time", time.time() - start_time)
            
            self.request_count += 1
            
            # Handle 429 Too Many Requests
            if response.status_code == 429:
                retry_after = response.headers.get("Retry-After", "60")
                wait_time = int(retry_after) if retry_after.isdigit() else 60
                time.sleep(wait_time)
                retry_count += 1
                continue  # Try again after waiting
            
            # Handle server errors with backoff
            if response.status_code >= 500 and retry_count < settings.DEFAULT_RETRY_COUNT:
                wait_time = backoff_factor ** retry_count  # 1s, 2s, 4s
                time.sleep(wait_time)
                retry_count += 1
                continue
            
            return response  # Success
        
        except (requests.Timeout, requests.ConnectionError):
            if retry_count < settings.DEFAULT_RETRY_COUNT:
                wait_time = backoff_factor ** retry_count
                time.sleep(wait_time)
                retry_count += 1
            else:
                raise  # Give up after 3 retries
    
    return response  # Return last response if loop completes
```

**Retry scenarios:**

1. **Timeout**: Wait 1s, retry. Still timeout? Wait 2s, retry. Still timeout? Wait 4s, retry. Still timeout? Raise exception.

2. **Connection refused**: Same exponential backoff strategy.

3. **429 Rate Limited**: Read `Retry-After` header, wait that long, retry. No exponential backoff needed - server told us exactly how long to wait.

4. **500 Server Error**: Exponential backoff, but only retry if `retry_count < 3`. After 3 attempts, return the 500 response (don't raise exception). Let caller decide how to handle.

## Performance Optimizations

### Before: Naive Database Queries (N+1 Problem)
```python
# Bad - triggers N+1 queries
@router.get("/scans/")
async def get_scans(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
    scans = db.query(Scan).filter(Scan.user_id == user.id).all()
    # SQL: SELECT * FROM scans WHERE user_id = 1
    
    for scan in scans:
        print(scan.test_results)  # Each access triggers new query!
        # SQL: SELECT * FROM test_results WHERE scan_id = 42
        # SQL: SELECT * FROM test_results WHERE scan_id = 43
        # ... repeated for each scan
    
    return scans
```

With 10 scans, 4 results each = 41 queries (1 + 10*4).

### After: Eager Loading with joinedload
```python
# Good - single query with JOIN
def get_by_user(db: Session, user_id: int) -> list[Scan]:
    return (
        db.query(Scan)
        .options(joinedload(Scan.test_results))  # Load relationship in same query
        .filter(Scan.user_id == user_id)
        .all()
    )
    # SQL: SELECT scans.*, test_results.* 
    #      FROM scans 
    #      LEFT JOIN test_results ON scans.id = test_results.scan_id
    #      WHERE scans.user_id = 1
```

Single query loads everything. Accessing `scan.test_results` uses already-loaded data, no additional query.

**Benchmarks:**
- Before: 41 queries, ~205ms (5ms per query)
- After: 1 query, ~15ms
- Improvement: **13x faster**

### Request Connection Pooling

Base scanner reuses HTTP session (`base_scanner.py:40-62`):
```python
def __init__(self, target_url: str, auth_token: str | None = None, ...):
    self.target_url = target_url.rstrip("/")
    self.auth_token = auth_token
    self.session = self._create_session()  # Created once
    
def _create_session(self) -> requests.Session:
    """
    Create persistent HTTP session with proper headers
    """
    session = requests.Session()
    
    session.headers.update({
        "User-Agent": f"{settings.APP_NAME}/{settings.VERSION}",
        "Accept": "application/json",
    })
    
    if self.auth_token:
        session.headers.update({"Authorization": f"Bearer {self.auth_token}"})
    
    return session
```

**Why this matters:**

Without session (making new request each time):
```python
# Each call creates new TCP connection
requests.get("https://api.example.com/endpoint1")  # Connect, TLS handshake, request, close
requests.get("https://api.example.com/endpoint2")  # Connect, TLS handshake, request, close
# 2 connections, 2 TLS handshakes
```

With session:
```python
session = requests.Session()
session.get("https://api.example.com/endpoint1")  # Connect, TLS handshake, request, keep-alive
session.get("https://api.example.com/endpoint2")  # Reuse connection, request
# 1 connection, 1 TLS handshake
```

TLS handshake takes 50-100ms. Over 100 requests, that's 5-10 seconds saved.

## Configuration Management

### Loading Config

All settings loaded from environment via Pydantic:
```python
# config.py:14-69
from functools import lru_cache
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file="../.env",  # Load from .env file
        env_file_encoding="utf-8",
        case_sensitive=True  # DATABASE_URL != database_url
    )
    
    # Required fields (no default)
    DATABASE_URL: str
    SECRET_KEY: str
    
    # Optional fields (have defaults)
    APP_NAME: str = "API Security Tester"
    DEBUG: bool = False
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440
    
    @property
    def cors_origins_list(self) -> list[str]:
        """Convert comma-separated string to list"""
        return [origin.strip() for origin in self.CORS_ORIGINS.split(",")]

@lru_cache
def get_settings() -> Settings:
    """
    Get cached settings instance
    
    @lru_cache ensures settings are loaded only once
    """
    return Settings()

settings = get_settings()
```

**Validation:**

Pydantic validates types at startup:
```python
# .env contains:
# DEBUG=not_a_boolean

# When Settings() instantiates:
# ValidationError: field required to be bool, got str 'not_a_boolean'
```

Application crashes immediately with clear error instead of mysterious runtime failures.

**Why cache with `@lru_cache`:**

Without cache:
```python
# Every import creates new Settings instance
from config import settings  # Loads .env
from config import settings  # Loads .env again
```

With cache:
```python
# First import loads .env
from config import settings  # Loads .env, caches result

# Subsequent imports reuse cached instance
from config import settings  # Returns cached Settings
```

Faster startup, consistent state across application.

## Database/Storage Operations

### Creating Records with Transactions

Repository method with explicit commit control:
```python
# repositories/scan_repository.py:13-32
@staticmethod
def create_scan(
    db: Session,
    user_id: int,
    target_url: str,
    commit: bool = True  # Allow caller to control commit
) -> Scan:
    """
    Create a new scan
    """
    scan = Scan(
        user_id=user_id,
        target_url=target_url,
        scan_date=datetime.now(UTC),
    )
    db.add(scan)
    
    if commit:
        db.commit()  # Flush to database
        db.refresh(scan)  # Reload to get auto-generated ID
    
    return scan
```

**Important details:**

- **Transaction management**: The `commit` parameter lets callers batch operations. Create scan + create results in single transaction.
- **Refresh after commit**: Line 30 reloads the object from database. This populates auto-generated fields like `id`, `created_at`, `updated_at`.
- **Foreign key constraints**: PostgreSQL enforces `test_results.scan_id` references `scans.id`. If we create test results before committing scan, foreign key check fails.

### Bulk Inserts for Performance

Creating test results in batch:
```python
# repositories/test_result_repository.py:52-67
@staticmethod
def bulk_create(
    db: Session,
    test_results: list[TestResult],
    commit: bool = True
) -> list[TestResult]:
    """
    Create multiple test results in bulk
    """
    db.add_all(test_results)  # Add all at once
    
    if commit:
        db.commit()
        for result in test_results:
            db.refresh(result)  # Refresh each to get IDs
    
    return test_results
```

**Why bulk insert:**

Individual inserts:
```python
for result in results:
    db.add(result)
    db.commit()  # Commit after each - 4 commits for 4 results
# 4 round trips to database
```

Bulk insert:
```python
db.add_all(results)
db.commit()  # Single commit for all 4 results
# 1 round trip to database
```

With 4 test results, bulk insert is 4x faster.

## Common Implementation Pitfalls

### Pitfall 1: Forgetting to Validate JWT Algorithm

**Symptom:**
Attacker sends token with `"alg": "none"` and gains admin access.

**Cause:**
```python
# Problematic code - accepts any algorithm
payload = jwt.decode(token, settings.SECRET_KEY)

# Attacker sends:
# eyJhbGciOiJub25lIn0.eyJzdWIiOiJhZG1pbiJ9.
# (header: {"alg": "none"}, payload: {"sub": "admin"}, no signature)

# Library accepts it because no algorithm restriction
```

**Fix:**
```python
# Correct - explicitly allow only HS256
payload = jwt.decode(
    token,
    settings.SECRET_KEY,
    algorithms=["HS256"]  # Reject "none" and other algorithms
)
```

**Why this matters:**
Without algorithm validation, JWT security is completely bypassed. Attacker can forge tokens with any claims.

### Pitfall 2: SQL Injection in Raw Queries

**Symptom:**
Attacker sends `email=admin'--` and gets admin access.

**Cause:**
```python
# Vulnerable - concatenating user input
email = request.get("email")
query = f"SELECT * FROM users WHERE email = '{email}'"
db.execute(text(query))

# Becomes: SELECT * FROM users WHERE email = 'admin'--'
# Comment (--) removes password check
```

**Fix:**
```python
# Correct - parameterized query
email = request.get("email")
db.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": email}
)

# SQLAlchemy escapes the email value safely
```

Or better, use ORM:
```python
# Best - ORM automatically parameterizes
db.query(User).filter(User.email == email).first()
```

**Why this matters:**
String concatenation treats user input as SQL code. Parameterization treats it as data.

### Pitfall 3: Missing Authorization Check

**Symptom:**
User can view other users' scans by changing scan ID in URL.

**Cause:**
```python
# Vulnerable - no ownership check
@router.get("/scans/{scan_id}")
async def get_scan(scan_id: int, db: Session = Depends(get_db)):
    scan = ScanRepository.get_by_id(db, scan_id)
    if not scan:
        raise HTTPException(404)
    return scan  # Returns any scan, regardless of ownership
```

**Fix:**
```python
# Correct - verify ownership
@router.get("/scans/{scan_id}")
async def get_scan(
    scan_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    scan = ScanRepository.get_by_id(db, scan_id)
    if not scan:
        raise HTTPException(404)
    
    if scan.user_id != current_user.id:
        raise HTTPException(403, detail="Not authorized")
    
    return scan
```

**Why this matters:**
Authentication proves who you are. Authorization proves what you can access. Both are required.

## Debugging Tips

### Issue Type 1: JWT Token Appears Invalid But Format Looks Correct

**Problem:** Getting 401 errors when using a token that decodes properly on jwt.io.

**How to debug:**

1. Check token expiration:
```python
import jwt
import datetime

token = "eyJ..."
decoded = jwt.decode(token, options={"verify_signature": False})  # Skip verification for debugging
print(decoded)
# {"sub": "user@example.com", "exp": 1704067200}

exp_time = datetime.datetime.fromtimestamp(decoded["exp"])
now = datetime.datetime.now()
print(f"Expires: {exp_time}, Now: {now}, Valid: {exp_time > now}")
```

2. Verify secret key matches:
```python
# In Python shell with access to settings
from config import settings
print(f"SECRET_KEY: {settings.SECRET_KEY}")
# Make sure this matches the key used to create the token
```

3. Check algorithm matches:
```python
# Decode header without verification
import base64
import json

token = "eyJ..."
header = token.split(".")[0]
# Add padding if needed
padding = 4 - (len(header) % 4)
if padding != 4:
    header += "=" * padding

decoded_header = json.loads(base64.urlsafe_b64decode(header))
print(decoded_header)
# {"alg": "HS256", "typ": "JWT"}

# Make sure "alg" matches settings.ALGORITHM
```

**Common causes:**
- Token expired (check `exp` claim)
- Wrong secret key (dev vs prod environments)
- Algorithm mismatch (created with HS256, verifying with RS256)

### Issue Type 2: Scanner Times Out on All Tests

**Problem:** All test results return `status="error"` with timeout messages.

**How to debug:**

1. Test target is accessible:
```bash
# From inside backend container
docker exec -it apisec_backend_dev bash
curl -v https://target-api.com/endpoint

# Check:
# - Does connection succeed?
# - What's the response time?
# - Are there redirects?
```

2. Check timeout settings:
```python
# config.py:51
SCANNER_CONNECTION_TIMEOUT: int = 180  # 3 minutes

# If target is slower, increase this
```

3. Look at actual error:
```python
# services/scan_service.py exception block shows error
except Exception as e:
    details=f"Scanner error: {str(e)}"
    
# Check what the exception says:
# - "Connection timeout" = target slow or unreachable
# - "Name resolution failed" = DNS issue
# - "SSL certificate verify failed" = TLS problem
```

4. Try with a known-good target:
```bash
# Test with httpbin (always responsive)
curl -X POST http://localhost:8000/scans/ \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "target_url": "https://httpbin.org/get",
    "tests_to_run": ["rate_limit"],
    "max_requests": 10
  }'
```

**Common causes:**
- Target requires VPN or is behind firewall
- Target rate limiting scanner (ironically)
- DNS resolution failing in Docker network
- TLS certificate issues (self-signed, expired)

### Issue Type 3: Database Deadlock Errors

**Problem:** Intermittent `DeadlockDetected` errors under concurrent load.

**How to debug:**

1. Check transaction order:
```sql
-- PostgreSQL query to see locks
SELECT
  pid,
  state,
  query_start,
  state_change,
  query
FROM pg_stat_activity
WHERE state = 'active';
```

2. Look for transaction patterns:
```python
# Problematic pattern - different order
# Thread 1:
UPDATE scans SET ... WHERE id = 1;
UPDATE test_results SET ... WHERE scan_id = 1;

# Thread 2:
UPDATE test_results SET ... WHERE scan_id = 2;
UPDATE scans SET ... WHERE id = 2;

# Thread 1 locks scans.id=1, waits for test_results
# Thread 2 locks test_results.scan_id=2, waits for scans
# = Deadlock
```

3. Fix by consistent ordering:
```python
# Both threads update in same order = no deadlock
# Always update parent (scans) before child (test_results)
UPDATE scans SET ... WHERE id = ?;
UPDATE test_results SET ... WHERE scan_id = ?;
```

**Common causes:**
- Multiple transactions updating same rows in different orders
- Long-running transactions holding locks
- Missing indexes causing table scans that lock many rows

## Code Organization Principles

### Why Routes Stay Thin

Routes in `routes/scans.py` are intentionally simple:
```python
@router.post("/", response_model=ScanResponse)
async def create_scan(
    request: Request,
    scan_request: ScanRequest,
    db: Session = Depends(get_db),
    current_user: UserResponse = Depends(get_current_user),
) -> ScanResponse:
    return ScanService.run_scan(db, current_user.id, scan_request)
```

Just 3 lines:
1. Dependency injection provides db and current_user
2. Call service layer
3. Return result

**All business logic lives in services.** Routes handle HTTP concerns:
- Parsing request
- Validating with Pydantic
- Checking authentication
- Serializing response

This makes routes easy to test:
```python
# Test route with mocked service
def test_create_scan(mock_service):
    mock_service.run_scan.return_value = fake_scan
    
    response = client.post("/scans/", json={...})
    
    assert response.status_code == 201
    assert mock_service.run_scan.called
```

No need to mock database, scanners, or any complex logic. Service is mocked, route just handles HTTP.

### Naming Conventions

- `*Repository` = Data access classes (UserRepository, ScanRepository)
- `*Service` = Business logic classes (AuthService, ScanService)
- `*Scanner` = Security test implementations (SQLiScanner, AuthScanner)
- `*Schema` = Pydantic validation models (UserCreate, ScanResponse)
- `get_*` functions = FastAPI dependencies (get_db, get_current_user)
- `_private_method` = Internal helper (not part of public interface)

Following these patterns makes it easier to find code. Need to add a database query? Look in repositories. Need to change business logic? Look in services.

## Extending the Code

### Adding a New Security Test (XSS Example)

Want to add XSS (Cross-Site Scripting) detection? Here's the complete process:

**1. Create scanner** `backend/scanners/xss_scanner.py`:
```python
from .base_scanner import BaseScanner
from .payloads import XSSPayloads
from core.enums import TestType, ScanStatus, Severity
from schemas.test_result_schemas import TestResultCreate

class XSSScanner(BaseScanner):
    """
    Tests for Cross-Site Scripting vulnerabilities
    """
    
    def scan(self) -> TestResultCreate:
        reflected_test = self._test_reflected_xss()
        
        if reflected_test["vulnerable"]:
            return TestResultCreate(
                test_name=TestType.XSS,
                status=ScanStatus.VULNERABLE,
                severity=Severity.HIGH,
                details=f"Reflected XSS detected: {reflected_test['payload']}",
                evidence_json=reflected_test,
                recommendations_json=[
                    "Encode user input before rendering in HTML",
                    "Use Content-Security-Policy headers",
                    "Validate input on server side",
                ],
            )
        
        return TestResultCreate(
            test_name=TestType.XSS,
            status=ScanStatus.SAFE,
            severity=Severity.INFO,
            details="No XSS vulnerabilities detected",
            evidence_json=reflected_test,
            recommendations_json=[],
        )
    
    def _test_reflected_xss(self) -> dict[str, Any]:
        payloads = XSSPayloads.get_basic_payloads()
        
        for payload in payloads:
            response = self.make_request("GET", f"/?q={payload}")
            
            # Check if payload appears unencoded in response
            if payload in response.text:
                return {
                    "vulnerable": True,
                    "payload": payload,
                    "status_code": response.status_code,
                }
        
        return {"vulnerable": False}
```

**2. Add enum value** in `backend/core/enums.py:19-26`:
```python
class TestType(str, Enum):
    RATE_LIMIT = "rate_limit"
    AUTH = "auth"
    SQLI = "sqli"
    IDOR = "idor"
    XSS = "xss"  # New test type
```

**3. Register scanner** in `backend/services/scan_service.py:32-38`:
```python
scanner_mapping: dict[TestType, type[BaseScanner]] = {
    TestType.RATE_LIMIT: RateLimitScanner,
    TestType.AUTH: AuthScanner,
    TestType.SQLI: SQLiScanner,
    TestType.IDOR: IDORScanner,
    TestType.XSS: XSSScanner,  # Register new scanner
}
```

**4. Add payloads** in `backend/scanners/payloads.py:250-280`:
```python
class XSSPayloads:
    BASIC_XSS = [
        "<script>alert('XSS')</script>",
        "<img src=x onerror=alert('XSS')>",
        "<svg/onload=alert('XSS')>",
    ]
    
    @classmethod
    def get_basic_payloads(cls) -> list[str]:
        return cls.BASIC_XSS
```

**5. Update frontend** in `frontend/src/config/constants.ts`:
```typescript
export const SCAN_TEST_TYPES = {
  // ... existing
  XSS: 'xss',
} as const;

export const TEST_TYPE_LABELS: Record<ScanTestType, string> = {
  // ... existing
  [SCAN_TEST_TYPES.XSS]: 'Cross-Site Scripting',
};
```

That's it. No database changes needed (TestType enum automatically updates). No route changes (they pass through test types). Just scanner implementation and registration.

## Next Steps

You've seen how the code works. Now:

1. **Try the challenges** - [04-CHALLENGES.md](./04-CHALLENGES.md) has extension ideas from adding stored XSS detection to implementing custom scanner plugins
2. **Modify scanners** - Change SQLi payload timing thresholds, add new auth bypass techniques, implement DOM-based XSS detection
3. **Read related projects** - The docker-security-audit project builds on container scanning concepts, network-traffic-analyzer goes deeper into packet analysis