1512 lines
55 KiB
Markdown
1512 lines
55 KiB
Markdown
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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`):
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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()`:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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
|
|
|
|
```python
|
|
# 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):
|
|
|
|
```python
|
|
@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
|
|
|
|
```python
|
|
# 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`):
|
|
```python
|
|
# 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`):
|
|
```python
|
|
# 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`):
|
|
```python
|
|
# 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**:
|
|
```python
|
|
@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:
|
|
```nginx
|
|
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`):
|
|
```python
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
app.state.limiter = limiter
|
|
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
|
|
```
|
|
|
|
Applied to routes:
|
|
```python
|
|
# 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`):
|
|
```python
|
|
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:**
|
|
|
|
```sql
|
|
-- 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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`:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
# 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:
|
|
```yaml
|
|
# 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:
|
|
```python
|
|
# 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:
|
|
```python
|
|
# 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:
|
|
```python
|
|
# 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:
|
|
```python
|
|
# 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"`:
|
|
```python
|
|
# 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`:
|
|
```python
|
|
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`:
|
|
```python
|
|
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`:
|
|
```python
|
|
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`:
|
|
```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. Update frontend constants** in `frontend/src/config/constants.ts:44-51`:
|
|
```typescript
|
|
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](./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
|