fix(security): add username validation, constant-time login, default credential rejection, and tooling

This commit is contained in:
2026-02-18 19:11:14 +01:00
parent 1521005b62
commit f41b8fd8c2
8 changed files with 393 additions and 1 deletions

View File

@@ -0,0 +1,189 @@
---
description: Aegis backend Clean Architecture rules. Apply when working on any backend Python file under backend/app/ or backend/tests/.
globs: backend/**/*.py
---
# Aegis — Clean Modular Monolith Architecture
## Architecture Overview
Aegis follows a **Clean Architecture** pattern inside a modular monolith. The backend has four layers with strict dependency rules:
```
Presentation → Application → Domain ← Infrastructure
```
**The golden rule:** dependencies only point towards the Domain layer. Infrastructure implements the ports (interfaces) defined in Domain.
## Layer Structure and Rules
### Domain Layer (`backend/app/domain/`)
The innermost layer. **ZERO** imports from FastAPI, SQLAlchemy, Pydantic, or any framework.
| Directory | Purpose |
|-----------|---------|
| `domain/enums.py` | Canonical domain enums (TechniqueStatus, TestState, TeamSide, TestResult) |
| `domain/errors.py` | Exception hierarchy (DomainError → EntityNotFoundError, InvalidStateTransition, etc.) |
| `domain/exceptions.py` | Backward-compatible re-exports from errors.py |
| `domain/test_entity.py` | TestEntity — pure state machine with domain events |
| `domain/entities/` | Rich domain entities (TechniqueEntity, etc.) with business behavior |
| `domain/value_objects/` | Immutable value types (MitreId, ScoringWeights) |
| `domain/ports/repositories/` | Protocol interfaces defining data access contracts |
| `domain/ports/services/` | Protocol interfaces for external capabilities (storage, events) |
| `domain/unit_of_work.py` | UnitOfWork wrapping SQLAlchemy session |
**NEVER** import from `app.models`, `app.routers`, `app.infrastructure`, `fastapi`, or `sqlalchemy` inside `domain/`.
### Application Layer (`backend/app/application/` — future)
Use case orchestrators. Depends only on Domain.
| Directory | Purpose |
|-----------|---------|
| `application/use_cases/` | One class per business operation |
| `application/dto/` | Plain data containers for use case input/output |
| `application/interfaces/` | Application-level contracts (UnitOfWork protocol) |
### Infrastructure Layer (`backend/app/infrastructure/`)
Implements ports defined in Domain. Depends on Domain and Application.
| Directory | Purpose |
|-----------|---------|
| `infrastructure/redis_client.py` | Redis connection singleton |
| `infrastructure/persistence/repositories/` | SQLAlchemy implementations of repository ports |
| `infrastructure/persistence/mappers/` | ORM model ↔ domain entity converters |
### Presentation Layer (routers, schemas, dependencies)
HTTP boundary. Depends on Application and Domain (for exceptions).
| Directory | Purpose |
|-----------|---------|
| `routers/` | FastAPI routers — HTTP mapping only |
| `schemas/` | Pydantic request/response models |
| `dependencies/` | FastAPI `Depends()` wiring (auth, repositories) |
| `middleware/` | Error handler mapping domain exceptions → HTTP responses |
## Import Rules (Strict)
| From \ To | domain/ | application/ | infrastructure/ | presentation/ |
|-----------|---------|-------------|----------------|--------------|
| **domain/** | Self only | FORBIDDEN | FORBIDDEN | FORBIDDEN |
| **application/** | ALLOWED | Self only | FORBIDDEN | FORBIDDEN |
| **infrastructure/** | ALLOWED (ports) | ALLOWED (UoW) | Self only | FORBIDDEN |
| **presentation/** | ALLOWED (exceptions) | ALLOWED (use cases) | ALLOWED (wiring in dependencies/) | Self only |
## How to Add a New Feature
### 1. Start from the Domain
- Define or reuse domain entities in `domain/entities/`
- Add value objects if needed in `domain/value_objects/`
- Define repository port if a new aggregate root in `domain/ports/repositories/`
- Domain exceptions go in `domain/errors.py`
- Business rules live IN the entity, not in services or routers
### 2. Implement Infrastructure
- Create SQLAlchemy repository implementation in `infrastructure/persistence/repositories/`
- Create mapper if converting between ORM model and domain entity
- Repository does NOT call `commit()` — only `flush()`
- Transaction control belongs to the Unit of Work
### 3. Wire in Presentation
- Add FastAPI `Depends()` provider in `dependencies/repositories.py`
- Keep routers thin: parse request → call service/use case → return response
- Map domain exceptions to HTTP via the error handler middleware (automatic)
### 4. Tests (Mandatory)
Every change MUST include tests:
- **Domain entities/value objects**: pure unit tests, no DB, no mocking frameworks
- **Repositories**: integration tests using the `db` fixture from conftest
- **Routers**: API tests using the `client` fixture
- At least one success test + one failure/edge-case test per behavior
Before committing, run: `scripts/agent_validate_backend.sh`
## Existing Patterns to Follow
### Domain Entity Pattern (see `domain/test_entity.py`)
```python
@dataclass
class SomeEntity:
id: uuid.UUID
# fields...
_events: list[DomainEvent] = field(default_factory=list, repr=False)
@classmethod
def from_orm(cls, model: Any) -> "SomeEntity":
"""Build from SQLAlchemy model."""
...
def apply_to(self, model: Any) -> None:
"""Copy mutable fields back onto the ORM model."""
...
def some_business_method(self) -> None:
"""Business logic lives HERE, not in services."""
...
self._events.append(DomainEvent("something_happened"))
```
### Repository Port Pattern (Protocol)
```python
from typing import Protocol, runtime_checkable
@runtime_checkable
class SomeRepository(Protocol):
def find_by_id(self, id: uuid.UUID) -> SomeEntity | None: ...
def save(self, entity: SomeEntity) -> SomeEntity: ...
```
### Repository Implementation Pattern
```python
class SASomeRepository:
def __init__(self, session: Session) -> None:
self._session = session
def find_by_id(self, id: uuid.UUID) -> SomeEntity | None:
model = self._session.query(SomeModel).filter(SomeModel.id == id).first()
return SomeMapper.to_entity(model) if model else None
def save(self, entity: SomeEntity) -> SomeEntity:
model = SomeMapper.to_model(entity)
merged = self._session.merge(model)
self._session.flush() # NO commit — UoW does that
return SomeMapper.to_entity(merged)
```
### Error Handling (automatic via middleware)
Services raise domain exceptions → middleware maps to HTTP:
- `EntityNotFoundError` → 404
- `DuplicateEntityError` → 409
- `InvalidStateTransition` → 400
- `BusinessRuleViolation` → 400
- `PermissionViolation` → 403
### Coexistence Strategy
Old code (direct `db.query()` in routers) and new code (repositories) coexist. Migration is incremental:
1. New endpoints use repositories
2. Existing endpoints are migrated one at a time
3. Both access the same DB, same session, same tables
## Key Conventions
- **Enums**: canonical source is `domain/enums.py`, `models/enums.py` re-exports
- **Exceptions**: raise from `domain/errors.py`, never raise `HTTPException` from services
- **Commits**: only via `UnitOfWork.commit()` or at the router level, never inside services/repos
- **IDs**: UUID everywhere (primary keys, foreign keys)
- **Tests**: SQLite in-memory for unit/integration, PostgreSQL in CI
- **Validation**: Pydantic in schemas (presentation), domain rules in entities (domain)