02-ARCHITECTURE.md

02-ARCHITECTURE.md

System Architecture

This document breaks down how the system is designed and why certain architectural decisions were made.

High Level Architecture

┌─────────────────────────────────────────────────────────────────┐
│                          Client Browser                          │
└────────────────────────┬────────────────────────────────────────┘
                         │ HTTP/HTTPS
                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                      Nginx Reverse Proxy                         │
│  - Routes /api/* → Backend (FastAPI)                            │
│  - Routes /* → Frontend (Static Files)                          │
│  - Handles CORS, compression, caching                           │
└──────────────┬─────────────────────────┬────────────────────────┘
               │                         │
               ▼                         ▼
┌──────────────────────────┐   ┌─────────────────────────────────┐
│   Backend (FastAPI)      │   │   Frontend (React/Vite)         │
│  ┌────────────────────┐  │   │  ┌───────────────────────────┐  │
│  │     Routes         │  │   │  │    Components             │  │
│  │  /auth, /scans     │  │   │  │  Auth, Scan UI            │  │
│  └─────────┬──────────┘  │   │  └─────────┬─────────────────┘  │
│            ▼              │   │            │                    │
│  ┌────────────────────┐  │   │  ┌───────────────────────────┐  │
│  │    Services        │  │   │  │    TanStack Query         │  │
│  │  Business Logic    │  │   │  │  Server State Mgmt        │  │
│  └─────────┬──────────┘  │   │  └─────────┬─────────────────┘  │
│            ▼              │   │            │                    │
│  ┌────────────────────┐  │   │  ┌───────────────────────────┐  │
│  │   Repositories     │  │   │  │    Zustand Stores         │  │
│  │  Data Access       │  │   │  │  Local State Mgmt         │  │
│  └─────────┬──────────┘  │   │  └───────────────────────────┘  │
│            │              │   │                                 │
│  ┌─────────▼──────────┐  │   └─────────────────────────────────┘
│  │    Scanners        │  │
│  │  RateLimit, Auth,  │  │
│  │  SQLi, IDOR        │  │
│  └────────────────────┘  │
└──────────────┬────────────┘
               │
               ▼
┌─────────────────────────────────────────────────────────────────┐
│                    PostgreSQL Database                           │
│  Tables: users, scans, test_results                             │
└─────────────────────────────────────────────────────────────────┘

Component Breakdown

Nginx Reverse Proxy

• Purpose: Single entry point for all HTTP traffic
• Responsibilities: Route requests based on path, serve static files, handle SSL termination (production), compress responses, cache static assets
• Interfaces: Exposes port 80 (HTTP) and optionally 443 (HTTPS), proxies to backend on internal port 8000

FastAPI Backend

• Purpose: REST API server providing authentication and scanning services
• Responsibilities: Validate requests, enforce authentication, execute security scans, store results, return JSON responses
• Interfaces: Exposes HTTP endpoints at /auth/* and /scans/*, connects to PostgreSQL for data persistence

React Frontend

• Purpose: User interface for creating scans and viewing results
• Responsibilities: Form validation, API communication, state management, result visualization
• Interfaces: Communicates with backend via /api prefix, renders in browser

PostgreSQL Database

• Purpose: Persistent storage for users, scans, and test results
• Responsibilities: Data integrity, relationship enforcement, query optimization
• Interfaces: Accepts connections from backend on port 5432, enforces foreign key constraints

Scanner Modules

• Purpose: Execute security tests against target APIs
• Responsibilities: Send HTTP requests, analyze responses, detect vulnerabilities, collect evidence
• Interfaces: Inherit from BaseScanner, return TestResultCreate schemas

Data Flow

Primary Use Case: Creating and Running a Scan

Step by step walkthrough of what happens when a user submits a new scan:

1. User submits form → Frontend validates (Zod schema)
   Input: { targetUrl, authToken, testsToRun, maxRequests }
   Validation happens at frontend/src/lib/validation.ts:42-52

2. Frontend → POST /api/scans/ → Nginx
   Adds Authorization header with JWT from localStorage
   Request routed based on /api prefix

3. Nginx → Backend (routes/scans.py:23-37)
   Proxy passes to http://backend:8000/scans/
   Preserves headers including Authorization

4. Route handler → Dependencies check auth
   @limiter.limit("15/minute") - rate limits this endpoint
   get_current_user() extracts JWT, validates, loads user from DB
   Code at backend/core/dependencies.py:17-49

5. Route → Service layer (services/scan_service.py:23-65)
   ScanService.run_scan(db, user_id, scan_request)
   Creates Scan record in database via repository

6. Service → Scanner modules (scanners/*.py)
   Loops through requested tests (rate_limit, auth, sqli, idor)
   Instantiates appropriate scanner class for each test
   Each scanner inherits from BaseScanner

7. Scanner → Target API
   Makes HTTP requests using requests.Session
   Implements retry logic, rate limiting, timeout handling
   Base logic at backend/scanners/base_scanner.py:40-156

8. Scanner analyzes responses → Returns TestResultCreate
   Detects vulnerabilities based on:
   - Status codes (429 for rate limiting)
   - Response content (SQL errors)
   - Timing differences (blind SQLi)
   - Header patterns (JWT algorithms)

9. Service saves results → Repository → Database
   TestResultRepository.create_test_result() for each scanner output
   Foreign key links results to scan
   Code at backend/repositories/test_result_repository.py:19-50

10. Service → Route → JSON response
    ScanResponse includes full scan with nested test_results
    Frontend receives and redirects to /scans/{id}

11. Frontend fetches full scan → GET /api/scans/{id}
    TanStack Query caches result
    Renders TestResultCard for each result

Example with code references:

# Step 5: Route handler
@router.post("/", response_model=ScanResponse)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)  # "15/minute"
async def create_scan(
    request: Request,
    scan_request: ScanRequest,
    db: Session = Depends(get_db),
    current_user: UserResponse = Depends(get_current_user),  # Auth check
) -> ScanResponse:
    return ScanService.run_scan(db, current_user.id, scan_request)

# Step 6: Service orchestrates scanners
scanner_mapping: dict[TestType, type[BaseScanner]] = {
    TestType.RATE_LIMIT: RateLimitScanner,
    TestType.AUTH: AuthScanner,
    TestType.SQLI: SQLiScanner,
    TestType.IDOR: IDORScanner,
}

for test_type in scan_request.tests_to_run:
    scanner_class = scanner_mapping.get(test_type)
    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 test
    results.append(result)

# Step 9: Save to database
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,
    )

Secondary Use Case: User Registration Flow

1. User fills registration form → Frontend validates
   Zod schema checks: email format, password strength (8+ chars, uppercase, lowercase, number)
   frontend/src/lib/validation.ts:17-31

2. POST /api/auth/register → Rate limited to 15/minute
   Prevents automated account creation abuse
   backend/routes/auth.py:25-42

3. Service checks if email exists → Repository query
   UserRepository.get_by_email() checks for duplicates
   Returns 400 if email already registered

4. Service hashes password with bcrypt
   core/security.py:11-18 uses bcrypt.gensalt() and hashpw()
   Salt automatically generated per password

5. Repository creates user record
   UserRepository.create_user(email, hashed_password)
   Sets is_active=True, created_at=UTC timestamp

6. Response returns user data (NOT password)
   UserResponse schema excludes hashed_password field
   Frontend redirects to login page

Design Patterns

Repository Pattern

What it is: Abstraction layer between business logic and data access. All database queries go through repository classes that provide clean interfaces.

Where we use it:

• repositories/user_repository.py - User CRUD operations
• repositories/scan_repository.py - Scan queries with eager loading
• repositories/test_result_repository.py - Test result operations

Why we chose it: Keeps services clean and testable. Services call UserRepository.get_by_email(db, email) instead of writing raw queries. If we switch from SQLAlchemy to a different ORM or database entirely, we only change repository implementations, not service logic.

Trade-offs:

• Pros: Testable (mock repositories), maintainable (queries in one place), flexible (swap implementations)
• Cons: Extra layer of abstraction, more files to navigate, can feel like overkill for simple CRUD

Example implementation:

# repositories/user_repository.py:12-35
class UserRepository:
    """
    Repository for User database operations
    """
    @staticmethod
    def get_by_id(db: Session, user_id: int) -> User | None:
        return db.query(User).filter(User.id == user_id).first()
    
    @staticmethod
    def get_by_email(db: Session, email: str) -> User | None:
        return db.query(User).filter(User.email == email).first()
    
    @staticmethod
    def create_user(
        db: Session,
        email: str,
        hashed_password: str,
        commit: bool = True
    ) -> User:
        user = User(email=email, hashed_password=hashed_password)
        db.add(user)
        if commit:
            db.commit()
            db.refresh(user)
        return user

Services use it:

# services/auth_service.py:24-32
existing_user = UserRepository.get_by_email(db, user_data.email)
if existing_user:
    raise HTTPException(status_code=400, detail="Email already registered")

hashed_password = hash_password(user_data.password)

user = UserRepository.create_user(
    db=db,
    email=user_data.email,
    hashed_password=hashed_password,
)

Dependency Injection (FastAPI's Depends)

What it is: FastAPI's dependency injection system automatically provides values to route handler parameters. Used for database sessions, authentication, rate limiting.

Where we use it: Every route handler in routes/auth.py and routes/scans.py uses dependencies:

# routes/scans.py:23-37
@router.post("/", response_model=ScanResponse)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)
async def create_scan(
    request: Request,  # Injected by FastAPI
    scan_request: ScanRequest,  # Parsed and validated from request body
    db: Session = Depends(get_db),  # Database session injected
    current_user: UserResponse = Depends(get_current_user),  # Auth check injected
) -> ScanResponse:
    return ScanService.run_scan(db, current_user.id, scan_request)

Why we chose it: Clean separation of concerns. The route handler doesn't know how to:

• Get a database session (handled by get_db)
• Validate JWT tokens (handled by get_current_user)
• Parse request bodies (handled by Pydantic)

This makes testing easier - mock the dependencies, not the entire request cycle.

Trade-offs:

• Pros: Testable, reusable, explicit dependencies, automatic cleanup (session closing)
• Cons: "Magic" behavior for beginners, debugging can be tricky if dependency fails

The get_current_user dependency implementation (core/dependencies.py:17-49):

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:
        payload = decode_token(credentials.credentials)
        email: str | None = payload.get("sub")
        
        if email is None:
            raise HTTPException(status_code=401, detail="Invalid authentication credentials")
        
        user = UserRepository.get_by_email(db, email)
        
        if not user:
            raise HTTPException(status_code=401, detail="User not found")
        
        return UserResponse.model_validate(user)
    
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")

Template Method (BaseScanner Abstract Class)

What it is: Define the skeleton of an algorithm in a base class, let subclasses override specific steps. All scanners share common HTTP logic but implement their own scan() method.

Where we use it: scanners/base_scanner.py provides common functionality:

class BaseScanner(ABC):
    def __init__(self, target_url: str, auth_token: str | None = None, max_requests: int | None = None):
        self.target_url = target_url.rstrip("/")
        self.auth_token = auth_token
        self.max_requests = max_requests or settings.DEFAULT_MAX_REQUESTS
        self.session = self._create_session()
        self.last_request_time = 0.0
        self.request_count = 0
    
    def make_request(self, method: str, endpoint: str, **kwargs: Any) -> requests.Response:
        """Common HTTP request logic with retry and rate limiting"""
        self._wait_before_request()
        # ... retry logic, backoff, timeout handling
    
    def get_baseline_timing(self, endpoint: str, samples: int | None = None) -> tuple[float, float]:
        """Statistical baseline for time-based detection"""
        # ... takes samples, calculates mean and stdev
    
    @abstractmethod
    def scan(self) -> TestResultCreate:
        """Must be implemented by specific scanner classes"""

Subclasses implement scan():

# scanners/sqli_scanner.py:25-60
class SQLiScanner(BaseScanner):
    def scan(self) -> TestResultCreate:
        error_based_test = self._test_error_based_sqli()
        if error_based_test["vulnerable"]:
            return self._create_vulnerable_result(...)
        
        boolean_based_test = self._test_boolean_based_sqli()
        if boolean_based_test["vulnerable"]:
            return self._create_vulnerable_result(...)
        
        time_based_test = self._test_time_based_sqli()
        if time_based_test["vulnerable"]:
            return self._create_vulnerable_result(...)
        
        return TestResultCreate(status=ScanStatus.SAFE, ...)

Why we chose it: Eliminates code duplication. Request spacing, retry logic, session management - written once in BaseScanner, used by all four scanner types.

Trade-offs:

• Pros: DRY principle, consistent behavior, easy to add new scanners
• Cons: Tight coupling to base class, inheritance can be limiting

Layer Separation

The backend uses a three-layer architecture:

┌────────────────────────────────────┐
│    Layer 1: Routes                 │
│    - HTTP request/response         │
│    - Validation (Pydantic)         │
│    - Auth checks (dependencies)    │
│    - Rate limiting                 │
└────────────────┬───────────────────┘
                 ↓
┌────────────────────────────────────┐
│    Layer 2: Services               │
│    - Business logic                │
│    - Orchestration                 │
│    - Transaction management        │
│    - Error handling                │
└────────────────┬───────────────────┘
                 ↓
┌────────────────────────────────────┐
│    Layer 3: Repositories           │
│    - Database queries              │
│    - Data access only              │
│    - No business logic             │
└────────────────────────────────────┘

Why Layers?

Separation makes each layer testable in isolation:

• Test routes with mocked services
• Test services with mocked repositories
• Test repositories against a real test database

It also enforces single responsibility. Routes don't write SQL. Services don't parse HTTP headers. Repositories don't implement business rules.

What Lives Where

Layer 1: Routes (routes/auth.py, routes/scans.py)

• Files: Route handler functions decorated with @router.get/post/delete
• Imports: Can import from services, schemas, dependencies
• Forbidden: Direct database access, business logic, calling repositories directly

Example route:

# routes/scans.py:67-83
@router.get("/{scan_id}", response_model=ScanResponse)
@limiter.limit(settings.API_RATE_LIMIT_DEFAULT)
async def get_scan(
    request: Request,
    scan_id: int,
    db: Session = Depends(get_db),
    current_user: UserResponse = Depends(get_current_user),
) -> ScanResponse:
    """
    Get a specific scan by ID
    """
    return ScanService.get_scan_by_id(db, scan_id, current_user.id)

Layer 2: Services (services/auth_service.py, services/scan_service.py)

• Files: Service classes with static methods
• Imports: Repositories, models, schemas, utilities
• Forbidden: HTTP-specific code (requests, responses), direct SQL queries

Example service method:

# services/scan_service.py:23-65
@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 = scanner_mapping.get(test_type)
        scanner = scanner_class(...)
        result = scanner.scan()
        results.append(result)
    
    # Save all results
    for result in results:
        TestResultRepository.create_test_result(db=db, scan_id=scan.id, ...)
    
    db.refresh(scan)
    return ScanResponse.model_validate(scan)

Layer 3: Repositories (repositories/user_repository.py, repositories/scan_repository.py, repositories/test_result_repository.py)

• Files: Repository classes with static methods for database operations
• Imports: Models, SQLAlchemy, config
• Forbidden: Business logic, HTTP handling, calling other repositories

Example repository method:

# repositories/scan_repository.py:48-71
@staticmethod
def get_by_user(
    db: Session,
    user_id: int,
    skip: int = 0,
    limit: int | None = None
) -> list[Scan]:
    """
    Get all scans for a user with pagination
    """
    if limit is None:
        limit = settings.DEFAULT_PAGINATION_LIMIT
    
    return (
        db.query(Scan)
        .options(joinedload(Scan.test_results))  # Eager load relationships
        .filter(Scan.user_id == user_id)
        .order_by(Scan.scan_date.desc())
        .offset(skip)
        .limit(limit)
        .all()
    )

Data Models

User Model

# models/User.py:12-43
class User(BaseModel):
    """
    Stores authentication credentials and user information
    """
    __tablename__ = "users"
    
    email = Column(
        String(settings.EMAIL_MAX_LENGTH),  # 255 chars
        unique=True,
        nullable=False,
        index=True,  # Fast lookups by email
    )
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)

Fields explained:

• id: Auto-incrementing primary key (inherited from BaseModel)
• email: Unique identifier for login, indexed for fast authentication queries
• hashed_password: Bcrypt hash, never stored in plaintext, never returned in API responses
• is_active: Soft delete flag, allows disabling accounts without losing data
• created_at, updated_at: Timestamps inherited from BaseModel

Relationships:

• One-to-many with Scan: user.scans returns all scans created by this user
• Defined by relationship in Scan model: user = relationship("User", backref="scans")

Scan Model

# models/Scan.py:15-57
class Scan(BaseModel):
    """
    Stores metadata about scans performed on target URLs
    """
    __tablename__ = "scans"
    
    user_id = Column(
        Integer,
        ForeignKey("users.id", ondelete="CASCADE"),  # Delete scans when user deleted
        nullable=False,
        index=True,
    )
    target_url = Column(
        String(settings.URL_MAX_LENGTH),  # 2048 chars
        nullable=False,
    )
    scan_date = Column(
        DateTime(timezone=True),
        default=lambda: datetime.now(UTC),
        nullable=False,
    )
    
    user = relationship("User", backref="scans")
    test_results = relationship(
        "TestResult",
        back_populates="scan",
        cascade="all, delete-orphan",  # Delete results when scan deleted
    )

Fields explained:

• user_id: Foreign key to users table, indexed for filtering scans by user
• target_url: URL that was scanned, up to 2048 chars for long query strings
• scan_date: When scan was initiated, timezone-aware datetime in UTC
• CASCADE: When user is deleted, their scans are deleted. When scan is deleted, its test results are deleted

Relationships:

• Many-to-one with User: scan.user gets the user who created it
• One-to-many with TestResult: scan.test_results gets all vulnerability findings

Properties (computed, not stored):

@property
def has_vulnerabilities(self) -> bool:
    return any(result.status == "vulnerable" for result in self.test_results)

@property
def vulnerability_count(self) -> int:
    return sum(1 for result in self.test_results if result.status == "vulnerable")

TestResult Model

# models/TestResult.py:16-57
class TestResult(BaseModel):
    """
    Stores individual test results for each security scan
    """
    __tablename__ = "test_results"
    
    scan_id = Column(
        Integer,
        ForeignKey("scans.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    test_name = Column(
        Enum(TestType),  # rate_limit, auth, sqli, idor
        nullable=False,
        index=True,
    )
    status = Column(
        Enum(ScanStatus),  # vulnerable, safe, error
        nullable=False,
        index=True,
    )
    severity = Column(
        Enum(Severity),  # critical, high, medium, low, info
        nullable=False,
        index=True,
    )
    details = Column(Text, nullable=False)
    evidence_json = Column(JSON, nullable=False, default=dict)
    recommendations_json = Column(JSON, nullable=False, default=list)

Fields explained:

• scan_id: Foreign key to scans table, which scan this result belongs to
• test_name: Enum constraining values to valid test types, indexed for filtering by test
• status: Enum for vulnerable/safe/error, indexed for finding all vulnerabilities
• severity: Enum for CRITICAL/HIGH/MEDIUM/LOW/INFO, indexed for prioritization
• details: Text description of what was found
• evidence_json: JSON storing response codes, payloads, timings - varies by test type
• recommendations_json: Array of strings with remediation steps

Why JSON columns: Evidence varies by test type:

• Rate limit test: {"rate_limit_headers": {...}, "bypass_method": "IP spoofing"}
• SQLi test: {"database_type": "mysql", "payload": "' OR 1=1--", "response_time": "5.23s"}
• Auth test: {"algorithm_variant": "none", "status_code": 200}

JSON flexibility lets each scanner store relevant data without schema changes.

Security Architecture

Threat Model

What we're protecting against:

1. Unauthorized access to scan data - Users should only see their own scans. Attacker tries to view scan ID 123 when they only created scan ID 456. Defense: Authorization check in services/scan_service.py:77-80 verifies scan.user_id == user_id.

2. Token theft and replay - Attacker steals JWT from network traffic or XSS. Defense: HTTPS in production (enforced by nginx), short token lifetime (24 hours from config.py:24), httpOnly cookies (not implemented but recommended).

3. Brute force login attempts - Attacker tries common passwords against accounts. Defense: Rate limiting at routes/auth.py:49 limits login to 20/minute, bcrypt makes password verification slow.

4. SQL injection in scanner payloads - Malicious user creates scan with SQLi payload as target URL hoping to exploit our database. Defense: All database access uses parameterized queries via SQLAlchemy, never concatenation.

5. Resource exhaustion - Attacker submits scans with max_requests=50 repeatedly to consume backend resources. Defense: Rate limiting on scan creation (15/minute), timeout limits on scanners, max_requests capped at 50.

What we're NOT protecting against (out of scope):

• DDoS attacks - Application-level rate limiting can't stop volumetric network floods. Requires infrastructure defenses (CloudFlare, AWS Shield).
• Database compromise - If attacker gains direct database access, they can read all data. Requires infrastructure hardening, encrypted columns for sensitive data.
• Server-side request forgery (SSRF) - Scanners make requests to user-provided URLs. This is intentional functionality. Mitigation: scanners run with limited network access, not on internal network.

Defense Layers

Multiple layers of security create defense in depth:

Layer 1: Network (Nginx)
    ↓ HTTPS, CORS headers, rate limits
Layer 2: Application (FastAPI)
    ↓ JWT validation, endpoint rate limits, input validation
Layer 3: Business Logic (Services)
    ↓ Authorization checks, transaction management
Layer 4: Data Access (Repositories)
    ↓ Parameterized queries, row-level permissions

Why multiple layers? If one defense fails, others catch the attack. Example: Nginx rate limit bypassed via IP spoofing, but application-level rate limit (by user ID) still protects. JWT validation bypassed somehow, but service layer still checks scan.user_id before returning data.

Authentication Flow

Complete JWT authentication cycle:

1. Registration (services/auth_service.py:19-41):

# Hash password with bcrypt
hashed_password = hash_password(user_data.password)
# Bcrypt automatically generates salt, 10 rounds by default

# Store hashed password
user = UserRepository.create_user(
    db=db,
    email=user_data.email,
    hashed_password=hashed_password,
)

2. Login (services/auth_service.py:43-71):

# Verify password
if not verify_password(login_data.password, user.hashed_password):
    raise HTTPException(status_code=401, detail="Invalid email or password")

# Create JWT with expiration
access_token = create_access_token(
    data={"sub": user.email},  # Subject claim
    expires_delta=timedelta(minutes=1440)  # 24 hours
)

return TokenResponse(access_token=access_token, token_type="bearer")

3. Protected endpoint access (core/dependencies.py:17-49):

# Extract token from Authorization header
credentials: HTTPAuthorizationCredentials = Depends(security)
# security = HTTPBearer() from fastapi.security

# Decode and verify token
payload = decode_token(credentials.credentials)
email = payload.get("sub")

# Load user from database
user = UserRepository.get_by_email(db, email)
if not user:
    raise HTTPException(status_code=401)

return UserResponse.model_validate(user)

4. Route handler receives authenticated user:

@router.post("/scans/")
async def create_scan(
    current_user: UserResponse = Depends(get_current_user),  # Authenticated
    ...
):
    # current_user is guaranteed to be valid at this point
    return ScanService.run_scan(db, current_user.id, ...)

Rate Limiting Strategy

Multiple rate limit implementations:

1. Nginx level - Not implemented in dev, but production nginx can use limit_req:

limit_req_zone $binary_remote_addr zone=api_limit:10m rate=100r/m;
limit_req zone=api_limit burst=20 nodelay;

2. Application level - SlowAPI per-endpoint limits (backend/factory.py:34-36):

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

Applied to routes:

# routes/auth.py:25-28
@router.post("/register", ...)
@limiter.limit(settings.API_RATE_LIMIT_REGISTER)  # "15/minute"
async def register(...):

# routes/auth.py:47-50
@router.post("/login", ...)
@limiter.limit(settings.API_RATE_LIMIT_LOGIN)  # "20/minute"
async def login(...):

# routes/scans.py:23-26
@router.post("/", ...)
@limiter.limit(settings.API_RATE_LIMIT_SCAN)  # "15/minute"
async def create_scan(...):

3. Scanner-level - Outgoing requests to targets are spaced (base_scanner.py:64-90):

def _wait_before_request(self, jitter_ms: int | None = None) -> None:
    """
    Implement request spacing to avoid overwhelming target
    """
    required_delay = 1.0 / (self.max_requests / settings.SCANNER_RATE_LIMIT_WINDOW_SECONDS)
    # If max_requests=100 and window=60s, delay = 1.0 / (100/60) = 0.6s between requests
    
    jitter = random.uniform(0, jitter_ms / 1000.0)  # Random variation
    elapsed = time.time() - self.last_request_time
    
    if elapsed < required_delay:
        time.sleep(required_delay - elapsed + jitter)

This prevents scanners from hammering target APIs and getting IP banned.

Storage Strategy

PostgreSQL

What we store:

• User accounts (email, hashed password)
• Scan metadata (target URL, timestamp)
• Test results (findings, evidence, recommendations)

Why PostgreSQL:

• Relational data with foreign keys (scans → users, test_results → scans)
• JSON column support for flexible evidence storage
• ACID transactions for data integrity
• Mature, well-documented, widely deployed

Alternatives considered:

• MongoDB: Better for schema-less data, but we have clear relationships and benefit from foreign key constraints
• SQLite: Simpler setup, but doesn't handle concurrent writes well (multiple scans running simultaneously)

Schema design:

-- Automatically generated by SQLAlchemy from models
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    hashed_password VARCHAR NOT NULL,
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
);

CREATE INDEX ix_users_email ON users(email);

CREATE TABLE scans (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    target_url VARCHAR(2048) NOT NULL,
    scan_date TIMESTAMP WITH TIME ZONE NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
);

CREATE INDEX ix_scans_user_id ON scans(user_id);

CREATE TABLE test_results (
    id SERIAL PRIMARY KEY,
    scan_id INTEGER NOT NULL REFERENCES scans(id) ON DELETE CASCADE,
    test_name VARCHAR NOT NULL,  -- Enum: rate_limit, auth, sqli, idor
    status VARCHAR NOT NULL,      -- Enum: vulnerable, safe, error
    severity VARCHAR NOT NULL,    -- Enum: critical, high, medium, low, info
    details TEXT NOT NULL,
    evidence_json JSON NOT NULL,
    recommendations_json JSON NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL
);

CREATE INDEX ix_test_results_scan_id ON test_results(scan_id);
CREATE INDEX ix_test_results_test_name ON test_results(test_name);
CREATE INDEX ix_test_results_status ON test_results(status);
CREATE INDEX ix_test_results_severity ON test_results(severity);

Indexes explained:

• users.email: Fast login lookups (WHERE email = ?)
• scans.user_id: Fast filtering (WHERE user_id = ?)
• test_results.scan_id: Fast joins (JOIN scans WHERE scan_id = ?)
• test_results.status: Fast vulnerability queries (WHERE status = 'vulnerable')
• test_results.severity: Fast filtering by severity (WHERE severity = 'critical')

Connection Pooling

Database connections are expensive to create. SQLAlchemy maintains a pool:

# core/database.py:12-17
engine = create_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,  # Verify connections before use (handles DB restarts)
    echo=settings.DEBUG,  # Log all SQL when DEBUG=True
)

Default pool size: 5 connections, overflow: 10 (up to 15 total).

Session lifecycle managed by dependency:

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

Configuration

Environment Variables

All configuration lives in .env and config.py:

# config.py:14-69
class Settings(BaseSettings):
    # Application
    APP_NAME: str = "API Security Tester"
    VERSION: str = "1.0.0"
    DEBUG: bool = False
    
    # Database
    DATABASE_URL: str  # Required, no default
    
    # Security - JWT
    SECRET_KEY: str  # Required, MUST be random in production
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440
    
    # Rate limiting
    API_RATE_LIMIT_LOGIN: str = "20/minute"
    API_RATE_LIMIT_REGISTER: str = "15/minute"
    API_RATE_LIMIT_SCAN: str = "15/minute"
    
    # Scanner limits
    SCANNER_MAX_CONCURRENT_REQUESTS: int = 50
    SCANNER_CONNECTION_TIMEOUT: int = 180
    DEFAULT_MAX_REQUESTS: int = 100

Why centralized config:

• Single source of truth for all constants
• Type validation with Pydantic
• Easy to change without touching code
• Different values for dev/test/prod

Configuration strategy: Development uses .env file loaded by Docker Compose. Production uses environment variables set directly in Docker, Kubernetes, or cloud platform.

Critical settings that must be changed for production:

• SECRET_KEY - Generate with openssl rand -hex 32
• DEBUG - Must be false
• DATABASE_URL - Production database, not localhost
• CORS_ORIGINS - Actual frontend domain, not http://localhost

Performance Considerations

Bottlenecks

Where this system gets slow under load:

1. Scanner HTTP requests - Each scan makes 10-100 HTTP requests to external APIs. If target is slow (5s response time), scans take minutes. Can't parallelize much without overwhelming targets. Mitigated by timeout limits (180s from config.py:51).

2. Database queries with relationships - Loading scan.test_results triggers N+1 queries if not eager loaded. With 10 scans and 4 results each = 41 queries (1 for scans + 40 for results). Solved with joinedload() in repositories:

# repositories/scan_repository.py:50-59
return (
    db.query(Scan)
    .options(joinedload(Scan.test_results))  # Single query with JOIN
    .filter(Scan.user_id == user_id)
    .all()
)

3. Password hashing on login - Bcrypt is intentionally slow (prevents brute force). Each login takes ~100ms. Under load, authentication becomes bottleneck. Mitigated by rate limiting login attempts and using caching (not implemented, but could add Redis for session tokens).

Optimizations

What we did to make it faster:

• Request pooling - BaseScanner reuses requests.Session() which maintains connection pools to targets. Avoids TCP handshake overhead on each request.

• Database indexes - Foreign keys and commonly queried columns (email, user_id, status) are indexed. Queries that would do table scans become index lookups.

• Response pagination - Scan list queries use LIMIT/OFFSET to avoid loading thousands of records:

# repositories/scan_repository.py:50-71
@staticmethod
def get_by_user(
    db: Session,
    user_id: int,
    skip: int = 0,
    limit: int | None = None  # Default 100 from config
) -> list[Scan]:
    if limit is None:
        limit = settings.DEFAULT_PAGINATION_LIMIT
    
    return (
        db.query(Scan)
        .offset(skip)
        .limit(limit)  # Only load requested page
        .all()
    )

• Enum columns - test_name, status, severity use Postgres ENUMs, not strings. Smaller storage, faster comparisons, enforced validity.

Scalability

Vertical scaling (more CPU/RAM on single server):

• Database: Increase max_connections in Postgres config
• Backend: Run more gunicorn workers (4 workers in production Dockerfile)
• Frontend: Nginx already efficient, bottleneck is unlikely here

Current limits with single server:

• Database can handle ~100 concurrent connections
• Backend with 4 workers handles ~400 concurrent requests
• Scanners are the real limit - each scan is long-running (30-60s)

Horizontal scaling (more servers): Challenges:

• Scanners are stateless, can run on any backend instance ✓
• Database requires connection pooling strategy (PgBouncer)
• Shared session state needed (Redis) or stick to JWT (stateless) ✓
• Load balancer required (nginx, AWS ALB)

What needs to change:

• Add load balancer in front of backend
• Configure shared session store or rely solely on JWT
• Database connection pool management (PgBouncer)
• Consider async task queue (Celery, RQ) for long-running scans

Design Decisions

Decision 1: Synchronous scanners despite FastAPI async

What we chose: Scanners use synchronous requests library, not async httpx or aiohttp.

Alternatives considered:

• httpx async HTTP client - Could run all tests concurrently
• aiohttp - Similar benefits, different API

Trade-offs: Pros of sync:

• Simpler code, easier to reason about timing
• Time-based SQLi detection requires precise timing control
• Baseline timing calculation needs sequential requests
• Standard requests library is battle-tested

Cons of sync:

• Can't run multiple tests concurrently within a scan
• Blocks event loop (mitigated by running in thread pool)
• Slower for scans with many tests

Why we made this choice: Accuracy over speed. Time-based blind SQL injection detection (sqli_scanner.py:183-251) requires:

1. Establish baseline response time (multiple samples)
2. Send delay payload
3. Measure if response is slower by expected amount

Async concurrency would introduce timing noise. A delay of 5.1s vs 5.3s could be network jitter, not SQLi. Sequential requests with controlled spacing give cleaner signals.

Decision 2: Repository pattern over Active Record

What we chose: Repository pattern - UserRepository.get_by_email(db, email) instead of User.find_by_email(email).

Alternatives considered:

• Active Record (Django-style) - Models have class methods for queries
• Data Mapper (raw SQL) - Write SQL strings directly

Trade-offs: Pros of repository:

• Clear separation: models define structure, repositories define queries
• Testable: mock repositories in unit tests
• Flexible: swap ORM without changing service code

Cons of repository:

• More files, more navigation
• Extra abstraction layer
• Can feel like overkill for simple CRUD

Why we made this choice: Testability and maintainability. Services like AuthService.login_user() call UserRepository.get_by_email(). In tests, mock the repository to return a fake user without touching the database.

If we later migrate from SQLAlchemy to another ORM, we only change repository implementations. Services remain unchanged.

Decision 3: JWT without refresh tokens

What we chose: Single long-lived JWT (24 hours), no refresh token mechanism.

Alternatives considered:

• Short access tokens (15 min) + refresh tokens (30 days)
• Session-based auth with server-side storage

Trade-offs: Pros of current approach:

• Simpler implementation, no refresh endpoint
• Stateless - no session storage needed
• Works across multiple backend instances immediately

Cons of current approach:

• Can't invalidate tokens before expiration
• If token stolen, attacker has 24 hours of access
• No way to force logout on all devices

Why we made this choice: Simplicity for educational project. Adding refresh tokens requires:

• Refresh token storage (database or Redis)
• Refresh endpoint with rotation logic
• Token revocation tracking
• More complex frontend token management

For a production app, you'd implement refresh tokens. For learning how JWT works, this is clearer.

Decision 4: Docker Compose for local development

What we chose: Run everything in Docker containers with dev.compose.yml, even for local development.

Alternatives considered:

• Local Postgres + local Python + local Node (no Docker)
• Docker for services, local for development
• Kubernetes locally (minikube, kind)

Trade-offs: Pros of Docker Compose:

• Identical environment for all developers
• Spin up entire stack with one command
• Hot reload still works with volume mounts
• Production architecture matches dev (Docker in both)

Cons of Docker Compose:

• Slower file system on Mac (volume mounts)
• Extra resource usage (containers overhead)
• Learning curve for Docker debugging

Why we made this choice: "Works on my machine" is eliminated. Every developer gets Postgres 16, Python 3.11, Node 20 regardless of their host OS. New team member runs docker compose up and they're ready.

Volume mounts preserve hot reload:

# dev.compose.yml:33-35
volumes:
  - ./backend:/app  # Maps local backend/ to container /app
  # Changes to backend/*.py trigger uvicorn reload

Deployment Architecture

Production deployment uses optimized containers:

                    ┌──────────────────┐
Internet ────────>  │  Nginx (Port 80) │
                    └────────┬─────────┘
                             │
                    ┌────────▼─────────┐
                    │  Backend (8000)  │
                    │  Gunicorn        │
                    │  4 workers       │
                    └────────┬─────────┘
                             │
                    ┌────────▼─────────┐
                    │  PostgreSQL      │
                    │  (internal only) │
                    └──────────────────┘

Components:

Nginx container - Built from conf/docker/prod/vite.docker:

• Multi-stage build: compile React, then serve with nginx
• Serves static files from /usr/share/nginx/html
• Proxies /api to backend
• Gzip compression, caching headers

Backend container - Built from conf/docker/prod/fastapi.docker:

• Runs gunicorn with 4 uvicorn workers
• Non-root user for security
• No volume mounts (code baked into image)

Database container - Postgres 16 Alpine:

• Not exposed to host in production
• Data persists in Docker volume

Infrastructure: Minimal production setup:

• Single VPS (DigitalOcean Droplet, AWS EC2)
• Docker Compose orchestration
• SSL via Let's Encrypt (certbot) or Cloudflare proxy

Scaling beyond single server:

• Backend: Multiple instances behind load balancer
• Database: Read replicas, connection pooling
• Static files: CDN (CloudFront, Cloudflare)

Error Handling Strategy

Error Types

1. Validation errors (400) - Pydantic catches bad input:

# schemas/user_schemas.py:26-38
@field_validator("password")
@classmethod
def validate_password_strength(cls, v: str) -> str:
    if not re.search(r"[A-Z]", v):
        raise ValueError("Password must contain uppercase letter")
    # Pydantic converts to HTTP 422 automatically

2. Authentication errors (401) - JWT invalid or expired:

# core/dependencies.py:33-36
if email is None:
    raise HTTPException(
        status_code=401,
        detail="Invalid authentication credentials"
    )

3. Authorization errors (403) - Valid user, wrong resource:

# services/scan_service.py:77-80
if scan.user_id != user_id:
    raise HTTPException(
        status_code=403,
        detail="Not authorized to access this scan"
    )

4. Not found errors (404) - Resource doesn't exist:

# services/scan_service.py:73-74
if not scan:
    raise HTTPException(status_code=404, detail="Scan not found")

5. Scanner errors - Caught and returned as status="error":

# services/scan_service.py:52-65
try:
    scanner = scanner_class(...)
    result = scanner.scan()
    results.append(result)
except Exception as e:
    results.append(
        TestResultCreate(
            test_name=test_type,
            status="error",
            severity="info",
            details=f"Scanner error: {str(e)}",
            ...
        )
    )

Recovery Mechanisms

Database connection loss:

• Detection: pool_pre_ping=True tests connections before use
• Response: SQLAlchemy automatically reconnects
• Recovery: Failed transaction rolls back, next request gets new connection

Scanner timeout:

• Detection: requests.Timeout exception after SCANNER_CONNECTION_TIMEOUT seconds
• Response: Retry with exponential backoff (up to 3 times)
• Recovery: If all retries fail, return error result (scan continues with other tests)

Rate limit exceeded (429 from target):

• Detection: HTTP 429 status code in scanner response
• Response: Read Retry-After header, wait specified duration
• Recovery: Retry request after waiting

Code from base_scanner.py:92-156:

def make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
    retry_count = 0
    backoff_factor = 2.0
    
    while retry_count <= settings.DEFAULT_RETRY_COUNT:
        try:
            response = self.session.request(method, url, **kwargs)
            
            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
            
            if response.status_code >= 500:  # Server error
                wait_time = backoff_factor ** retry_count
                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 retries exhausted

Extensibility

Where to Add Features

Want to add a new security test (e.g., XSS detection)? Here's the process:

1. Create scanner in backend/scanners/xss_scanner.py:

from .base_scanner import BaseScanner
from core.enums import TestType, ScanStatus, Severity

class XSSScanner(BaseScanner):
    def scan(self) -> TestResultCreate:
        # Test for reflected XSS
        test_result = self._test_reflected_xss()
        
        if test_result["vulnerable"]:
            return TestResultCreate(
                test_name=TestType.XSS,  # Need to add to enum
                status=ScanStatus.VULNERABLE,
                severity=Severity.HIGH,
                details="Reflected XSS detected",
                evidence_json=test_result,
                recommendations_json=[...]
            )
        
        return TestResultCreate(test_name=TestType.XSS, status=ScanStatus.SAFE, ...)

2. Add to enum in backend/core/enums.py:19-25:

class TestType(str, Enum):
    RATE_LIMIT = "rate_limit"
    AUTH = "auth"
    SQLI = "sqli"
    IDOR = "idor"
    XSS = "xss"  # New test type

3. Register in service at backend/services/scan_service.py:32-37:

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. Update frontend constants in frontend/src/config/constants.ts:44-51:

export const SCAN_TEST_TYPES = {
  RATE_LIMIT: 'rate_limit',
  AUTH: 'auth',
  SQLI: 'sqli',
  IDOR: 'idor',
  XSS: 'xss',  // Add new test type
} as const;

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

No changes needed to:

• Database schema (test_name is enum, migrations auto-update)
• Routes (they just pass through test types)
• Repositories (they store whatever test types are sent)

Limitations

Current architectural limitations:

1. No scan queueing - Scans run synchronously in request handler. If 10 users submit scans simultaneously, they all block on scanner HTTP requests. Fix requires async task queue (Celery with Redis, or RQ).

2. No real-time scan progress - Frontend submits scan and waits for complete response. Long scans (2 minutes) show no progress. Fix requires WebSocket connection or polling for progress updates.

3. Single target per scan - Can't scan multiple URLs in one operation. Fix requires loop in service layer and UI for multiple target inputs.

4. No historical comparison - Can't compare scan results over time ("Was this vulnerable last week?"). Fix requires additional queries and UI for trend visualization.

5. Limited concurrency in scanners - Tests run sequentially within a scan. Could run all 4 tests simultaneously, but chose not to for timing accuracy. Trade-off between speed and precision.

These are not bugs - they're conscious trade-offs. Fixing them would require significant architectural changes.

Comparison to Similar Systems

Burp Suite

How we're different:

• Burp is a proxy, we're a standalone scanner
• Burp has GUI desktop app, we're web-based
• Burp is comprehensive (hundreds of tests), we focus on 4 core vulnerabilities

Why we made different choices: Educational focus. Burp is for professional pentesters. This project teaches how scanners work by implementing the core logic yourself.

OWASP ZAP

How we're different:

• ZAP is passive + active scanning, we're active only
• ZAP auto-discovers endpoints, we test provided URLs
• ZAP integrates with CI/CD, we're standalone

Why we made different choices: Simplicity. ZAP is powerful but complex. This project shows the fundamentals without overwhelming features.

Evolution

Version 1.0 Design (Current)

Initial design focused on:

• Four core vulnerability types
• Synchronous scanners for accuracy
• Repository pattern for clean separation
• Docker-first development

Future Improvements

Planned architectural changes:

1. Async task queue - Move scanning to background workers

   • Why: Non-blocking API, better scalability
   • What it enables: Real-time progress, scheduled scans

2. Plugin system - Load scanners dynamically

   • Why: Extensibility without modifying core
   • What it enables: Community contributions, custom tests

3. Report generation - PDF/HTML export of results

   • Why: Sharing findings with teams
   • What it enables: Professional documentation

4. Webhook notifications - Alert when scans complete

   • Why: Integration with other tools
   • What it enables: Slack/email notifications, CI/CD integration

Key Files Reference

Quick map of where to find things:

• backend/factory.py - Application factory, middleware setup, route registration
• backend/config.py - All environment variables and configuration
• backend/core/database.py - Database engine and session management
• backend/core/security.py - JWT creation, password hashing, token validation
• backend/core/dependencies.py - FastAPI dependencies (auth, database)
• backend/models/ - SQLAlchemy models (User, Scan, TestResult)
• backend/repositories/ - Database query functions
• backend/services/ - Business logic orchestration
• backend/routes/ - API endpoints
• backend/scanners/base_scanner.py - Common scanner functionality
• backend/scanners/*_scanner.py - Individual vulnerability tests
• frontend/src/hooks/ - React Query hooks for API calls
• frontend/src/services/ - API client functions
• frontend/src/store/ - Zustand state management
• conf/nginx/ - Nginx reverse proxy configuration
• compose.yml - Production Docker Compose
• dev.compose.yml - Development with volume mounts

Next Steps

Now that you understand the architecture:

1. Read 03-IMPLEMENTATION.md for code walkthrough - see how each scanner detects vulnerabilities, how authentication flows work, and how data moves through the layers
2. Try modifying scanners - change SQLi payloads, adjust timing thresholds, add new detection logic to understand the implementation details