Cybersecurity-Projects/PROJECTS/advanced/bug-bounty-platform/learn/PATTERNS.md

Design Patterns Explained

This document explains the key design patterns used in the Bug Bounty Platform, why they were chosen, and how they improve code quality.

---

Table of Contents

1. Dependency Injection
2. Repository Pattern
3. Layered Architecture
4. Factory Pattern
5. Strategy Pattern
6. Singleton Pattern
7. Mixin Pattern
8. Observer Pattern (Pub/Sub)

---

Dependency Injection

What it is: Instead of creating dependencies inside a class, they are "injected" from outside.

Without Dependency Injection:

class UserService:
    def __init__(self):
        self.db = Database()  # Hardcoded dependency
        self.cache = Redis()  # Hardcoded dependency

    def get_user(self, user_id: str):
        cached = self.cache.get(f"user:{user_id}")
        if cached:
            return cached
        user = self.db.query(f"SELECT * FROM users WHERE id = '{user_id}'")
        self.cache.set(f"user:{user_id}", user)
        return user

Problems:

• Hard to test (can't mock Database or Redis)
• Tight coupling (UserService depends on specific implementations)
• Can't swap implementations (e.g., use a different cache)

With Dependency Injection:

class UserService:
    def __init__(self, db: Database, cache: Cache):
        self.db = db      # Injected from outside
        self.cache = cache # Injected from outside

    def get_user(self, user_id: str):
        cached = self.cache.get(f"user:{user_id}")
        if cached:
            return cached
        user = self.db.query(f"SELECT * FROM users WHERE id = '{user_id}'")
        self.cache.set(f"user:{user_id}", user)
        return user

# Usage
db = Database()
cache = Redis()
user_service = UserService(db=db, cache=cache)

Benefits:

• Easy to test (inject mocks)
• Loose coupling (depends on interfaces, not implementations)
• Flexible (can inject different implementations)

Dependency Injection in FastAPI

FastAPI has built-in DI using Depends():

from fastapi import Depends
from typing import Annotated

# Dependency function
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with DatabaseSessionManager().session() as session:
        yield session

# Type alias for convenience
DatabaseSession = Annotated[AsyncSession, Depends(get_db_session)]

# Route using dependency
@router.get("/users/{user_id}")
async def get_user(
    user_id: UUID,
    session: DatabaseSession,  # ← Automatically injected
) -> UserResponse:
    repo = UserRepository(session)
    user = await repo.get_by_id(user_id)
    return user

How it works:

1. FastAPI sees session: DatabaseSession parameter
2. Recognizes DatabaseSession is an Annotated type with Depends()
3. Calls get_db_session() to get a session
4. Injects the session into the route handler
5. Automatically cleans up when request finishes

Real example from the codebase:

# backend/app/core/dependencies.py

async def get_current_user(
    token: str = Header(...),
    session: DatabaseSession,
) -> User:
    payload = decode_access_token(token)
    user_id = payload["sub"]
    repo = UserRepository(session)
    user = await repo.get_by_id(user_id)
    if not user:
        raise UnauthorizedError()
    return user

# Type alias
CurrentUser = Annotated[User, Depends(get_current_user)]

# Usage in routes
@router.get("/me")
async def get_me(user: CurrentUser) -> UserResponse:
    return user  # User is automatically injected and authenticated!

Why this is powerful:

• Route handlers don't need to worry about authentication
• DRY principle (don't repeat auth logic in every route)
• Easy to test (mock get_current_user)
• Composable (dependencies can depend on other dependencies)

---

Repository Pattern

What it is: An abstraction layer between business logic and data storage.

The problem:

# Business logic mixed with database queries
@router.post("/users")
async def create_user(data: UserCreate, session: DatabaseSession):
    # Database query in route handler
    stmt = select(User).where(User.email == data.email)
    result = await session.execute(stmt)
    existing_user = result.scalar_one_or_none()

    if existing_user:
        raise HTTPException(status_code=400, detail="Email already exists")

    user = User(
        email=data.email,
        password_hash=hash_password(data.password),
    )
    session.add(user)
    await session.commit()
    return user

Problems:

• Route handler knows about database schema
• Hard to test (need to set up database)
• Can't switch databases easily
• Duplicate query logic across multiple routes

The Repository Pattern solution:

# 1. Create a repository class
class UserRepository(BaseRepository[User]):
    async def find_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, data: UserCreate) -> User:
        user = User(
            email=data.email,
            password_hash=hash_password(data.password),
        )
        self.session.add(user)
        await self.session.flush()
        await self.session.refresh(user)
        return user

# 2. Use repository in route handler
@router.post("/users")
async def create_user(data: UserCreate, session: DatabaseSession):
    repo = UserRepository(session)

    existing_user = await repo.find_by_email(data.email)
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already exists")

    user = await repo.create(data)
    await session.commit()
    return user

Benefits:

• Route handler doesn't know about database schema
• Query logic is reusable
• Easy to test (mock repository)
• Can swap database implementation

Generic Base Repository

To avoid repeating CRUD operations, we use a generic base repository:

from typing import Generic, TypeVar

ModelT = TypeVar("ModelT", bound=Base)

class BaseRepository(Generic[ModelT]):
    def __init__(self, session: AsyncSession, model: type[ModelT]):
        self.session = session
        self.model = model

    async def get_by_id(self, id: UUID) -> ModelT | None:
        stmt = select(self.model).where(self.model.id == id)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(self, limit: int = 100, offset: int = 0) -> list[ModelT]:
        stmt = select(self.model).limit(limit).offset(offset)
        result = await self.session.execute(stmt)
        return list(result.scalars().all())

    async def create(self, **kwargs) -> ModelT:
        instance = self.model(**kwargs)
        self.session.add(instance)
        await self.session.flush()
        await self.session.refresh(instance)
        return instance

    async def delete(self, id: UUID) -> None:
        stmt = delete(self.model).where(self.model.id == id)
        await self.session.execute(stmt)

Usage:

# Specific repository extends base
class UserRepository(BaseRepository[User]):
    def __init__(self, session: AsyncSession):
        super().__init__(session, User)

    # Add domain-specific methods
    async def find_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()

# Usage
repo = UserRepository(session)
user = await repo.get_by_id(user_id)      # From BaseRepository
user = await repo.find_by_email(email)    # From UserRepository

Why generics?

• Type safety: repo.get_by_id() returns User, not Any
• Reusable: All models get CRUD for free
• DRY: Write common operations once

---

Layered Architecture

What it is: Organizing code into layers, where each layer has a specific responsibility.

┌─────────────────────────────────────┐
│         Presentation Layer          │  ← Routes (HTTP)
│  - Handle HTTP requests/responses   │
│  - Input validation                 │
│  - Serialize data to JSON           │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│          Business Layer             │  ← Services
│  - Core business logic              │
│  - Orchestrate repositories         │
│  - Transaction management           │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│         Data Access Layer           │  ← Repositories
│  - Database queries                 │
│  - CRUD operations                  │
│  - No business logic                │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│           Database                  │  ← PostgreSQL
└─────────────────────────────────────┘

Rules:

1. Top-down dependency: Upper layers depend on lower layers, never the reverse
2. Single responsibility: Each layer has one job
3. No layer skipping: Routes can't directly query the database

Example: Creating a report

# 1. Presentation Layer (routes.py)
@router.post("/reports")
async def create_report(
    data: ReportCreate,
    user: CurrentUser,
    session: DatabaseSession,
) -> ReportResponse:
    report = await submit_vulnerability_report(
        session=session,
        author_id=user.id,
        data=data,
    )
    return ReportResponse.from_orm(report)

The route:

• Validates input (ReportCreate Pydantic model)
• Authenticates user (CurrentUser dependency)
• Calls service function
• Serializes output (ReportResponse)

# 2. Business Layer (service.py)
async def submit_vulnerability_report(
    session: AsyncSession,
    author_id: UUID,
    data: ReportCreate,
) -> Report:
    # Validate program exists
    program_repo = ProgramRepository(session)
    program = await program_repo.get_by_slug(data.program_slug)
    if not program:
        raise ProgramNotFoundError()

    # Check if program accepts reports
    if program.status != ProgramStatus.ACTIVE:
        raise ProgramNotActiveError()

    # Create report
    report_repo = ReportRepository(session)
    report = await report_repo.create(
        author_id=author_id,
        program_id=program.id,
        title=data.title,
        description=data.description,
        severity=data.severity,
    )

    # Send notification (in real app)
    # await notify_program_owner(program.owner_id, report.id)

    await session.commit()
    return report

The service:

• Implements business rules (program must be active)
• Coordinates multiple repositories (Program + Report)
• Manages transactions (commit)
• Throws domain exceptions

# 3. Data Access Layer (repository.py)
class ReportRepository(BaseRepository[Report]):
    async def create(
        self,
        author_id: UUID,
        program_id: UUID,
        title: str,
        description: str,
        severity: Severity,
    ) -> Report:
        report = Report(
            author_id=author_id,
            program_id=program_id,
            title=title,
            description=description,
            severity=severity,
            status=ReportStatus.NEW,
        )
        self.session.add(report)
        await self.session.flush()
        await self.session.refresh(report)
        return report

The repository:

• Builds the Report model
• Adds to session
• Returns the instance
• No business logic

Benefits:

• Testability: Mock each layer independently
• Maintainability: Change one layer without affecting others
• Scalability: Extract layers to separate services if needed
• Clarity: Easy to find where logic lives

---

Factory Pattern

What it is: A function/class that creates objects without specifying their exact class.

Real example from the codebase:

# backend/app/factory.py

def create_app() -> FastAPI:
    app = FastAPI(
        title="Bug Bounty Platform",
        version="1.0.0",
        docs_url="/api/docs",
        redoc_url="/api/redoc",
    )

    # Add middleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.CORS_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Add rate limiting
    app.state.limiter = limiter
    app.add_exception_handler(RateLimitExceeded, rate_limit_handler)

    # Register routers
    app.include_router(auth_router, prefix="/api/v1/auth", tags=["auth"])
    app.include_router(user_router, prefix="/api/v1/users", tags=["users"])
    app.include_router(program_router, prefix="/api/v1/programs", tags=["programs"])
    app.include_router(report_router, prefix="/api/v1/reports", tags=["reports"])
    app.include_router(admin_router, prefix="/api/v1/admin", tags=["admin"])

    return app

# Usage
app = create_app()

Benefits:

• Centralized configuration
• Easy to create multiple app instances (testing, different environments)
• Follows Single Responsibility Principle

Another example: Database session factory

class DatabaseSessionManager:
    def __init__(self):
        self._engine: AsyncEngine | None = None
        self._sessionmaker: async_sessionmaker | None = None

    def init(self, database_url: str):
        self._engine = create_async_engine(
            database_url,
            pool_size=settings.DB_POOL_SIZE,
            max_overflow=settings.DB_MAX_OVERFLOW,
            pool_timeout=settings.DB_POOL_TIMEOUT,
            pool_recycle=settings.DB_POOL_RECYCLE,
        )
        self._sessionmaker = async_sessionmaker(
            self._engine,
            expire_on_commit=False,
        )

    async def session(self) -> AsyncGenerator[AsyncSession, None]:
        if self._sessionmaker is None:
            raise DatabaseNotInitializedError()

        async with self._sessionmaker() as session:
            try:
                yield session
            except Exception:
                await session.rollback()
                raise
            finally:
                await session.close()

# Usage
db_manager = DatabaseSessionManager()
db_manager.init(settings.DATABASE_URL)

async with db_manager.session() as session:
    # Use session
    pass

---

Strategy Pattern

What it is: Define a family of algorithms, encapsulate each one, and make them interchangeable.

Example: Password hashing

from abc import ABC, abstractmethod

# Strategy interface
class PasswordHasher(ABC):
    @abstractmethod
    def hash(self, password: str) -> str:
        pass

    @abstractmethod
    def verify(self, password: str, hash: str) -> bool:
        pass

# Concrete strategies
class Argon2Hasher(PasswordHasher):
    def hash(self, password: str) -> str:
        return argon2.hash(password)

    def verify(self, password: str, hash: str) -> bool:
        try:
            return argon2.verify(hash, password)
        except Exception:
            return False

class BcryptHasher(PasswordHasher):
    def hash(self, password: str) -> str:
        return bcrypt.hashpw(password.encode(), bcrypt.gensalt())

    def verify(self, password: str, hash: str) -> bool:
        return bcrypt.checkpw(password.encode(), hash.encode())

# Context
class PasswordService:
    def __init__(self, hasher: PasswordHasher):
        self.hasher = hasher

    def hash_password(self, password: str) -> str:
        return self.hasher.hash(password)

    def verify_password(self, password: str, hash: str) -> bool:
        return self.hasher.verify(password, hash)

# Usage
service = PasswordService(Argon2Hasher())  # Can swap to BcryptHasher()
hash = service.hash_password("secret123")
is_valid = service.verify_password("secret123", hash)

Benefits:

• Easy to add new algorithms (e.g., PBKDF2)
• Can switch algorithms without changing code
• Each strategy is independently testable

---

Singleton Pattern

What it is: Ensure a class has only one instance and provide a global access point.

Example: Database session manager

class DatabaseSessionManager:
    _instance: "DatabaseSessionManager | None" = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._engine = None
            cls._instance._sessionmaker = None
        return cls._instance

    def init(self, database_url: str):
        if self._engine is not None:
            return  # Already initialized

        self._engine = create_async_engine(database_url)
        self._sessionmaker = async_sessionmaker(self._engine)

    async def session(self):
        async with self._sessionmaker() as session:
            yield session

# Usage - always returns the same instance
db1 = DatabaseSessionManager()
db2 = DatabaseSessionManager()
assert db1 is db2  # True

When to use:

• Database connections (expensive to create)
• Loggers (single logging configuration)
• Configuration (load once, use everywhere)

When NOT to use:

• Most other cases (singletons are global state = hard to test)

---

Mixin Pattern

What it is: A class that provides methods to other classes but isn't meant to be instantiated itself.

Real example from the codebase:

# backend/app/core/Base.py

class UUIDMixin:
    id: Mapped[UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        default=uuid.uuid7,
    )

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=func.now(),
        onupdate=func.now(),
    )

class SoftDeleteMixin:
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True),
        nullable=True,
        default=None,
    )

# Usage: Compose mixins
class User(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "users"

    email: Mapped[str] = mapped_column(String(255))
    password_hash: Mapped[str] = mapped_column(String(255))

class Report(Base, UUIDMixin, TimestampMixin, SoftDeleteMixin):
    __tablename__ = "reports"

    title: Mapped[str] = mapped_column(String(255))
    description: Mapped[str] = mapped_column(Text)

Benefits:

• DRY: Don't repeat id, created_at, updated_at in every model
• Composable: Mix and match (User doesn't need SoftDeleteMixin)
• Maintainable: Change UUIDMixin → all models updated

---

Observer Pattern (Pub/Sub)

What it is: When an object changes state, all its dependents are notified automatically.

Example: Event system

from typing import Callable

class EventBus:
    def __init__(self):
        self._subscribers: dict[str, list[Callable]] = {}

    def subscribe(self, event: str, handler: Callable):
        if event not in self._subscribers:
            self._subscribers[event] = []
        self._subscribers[event].append(handler)

    async def publish(self, event: str, data: dict):
        if event not in self._subscribers:
            return

        for handler in self._subscribers[event]:
            await handler(data)

# Event handlers
async def send_email_notification(data: dict):
    print(f"Sending email to {data['email']}")

async def log_event(data: dict):
    print(f"Logging event: {data}")

# Setup
bus = EventBus()
bus.subscribe("user.registered", send_email_notification)
bus.subscribe("user.registered", log_event)

# Trigger event
await bus.publish("user.registered", {
    "email": "user@example.com",
    "user_id": "123",
})

Real-world use case:

# When a report is submitted
async def submit_report(...):
    report = await report_repo.create(...)

    # Publish event
    await event_bus.publish("report.submitted", {
        "report_id": report.id,
        "program_id": report.program_id,
        "author_id": report.author_id,
    })

    return report

# Multiple handlers react to the event
bus.subscribe("report.submitted", notify_program_owner)
bus.subscribe("report.submitted", update_analytics)
bus.subscribe("report.submitted", log_submission)

Benefits:

• Decoupled: Report submission doesn't know about notifications
• Extensible: Add new handlers without modifying code
• Testable: Can test each handler independently

---

Combining Patterns

Real-world code often combines multiple patterns:

# Dependency Injection + Repository + Layered Architecture

# 1. Route (Presentation Layer)
@router.post("/users")
async def create_user(
    data: UserCreate,
    session: DatabaseSession,  # Dependency Injection
) -> UserResponse:
    user = await register_user(session, data)  # Service Layer
    return UserResponse.from_orm(user)

# 2. Service (Business Layer)
async def register_user(
    session: AsyncSession,
    data: UserCreate,
) -> User:
    repo = UserRepository(session)  # Repository Pattern

    existing_user = await repo.find_by_email(data.email)
    if existing_user:
        raise UserAlreadyExistsError()

    password_hash = hash_password(data.password)  # Strategy Pattern
    user = await repo.create(
        email=data.email,
        password_hash=password_hash,
    )

    await session.commit()

    await event_bus.publish("user.registered", {  # Observer Pattern
        "user_id": user.id,
        "email": user.email,
    })

    return user

# 3. Repository (Data Access Layer)
class UserRepository(BaseRepository[User]):  # Generic Pattern
    async def find_by_email(self, email: str) -> User | None:
        ...

    async def create(self, **kwargs) -> User:
        ...

This combines:

• Dependency Injection (session injected)
• Repository Pattern (data access abstraction)
• Layered Architecture (route → service → repository)
• Strategy Pattern (hash_password can use different algorithms)
• Observer Pattern (event_bus.publish)
• Generic Pattern (BaseRepository[User])

---

Conclusion

Design patterns are not about memorization, but about recognizing common problems and applying proven solutions.

Key takeaways:

1. Dependency Injection - Inject dependencies, don't hardcode them
2. Repository Pattern - Abstract data access from business logic
3. Layered Architecture - Separate presentation, business, and data layers
4. Factory Pattern - Centralize object creation
5. Strategy Pattern - Make algorithms interchangeable
6. Singleton Pattern - One instance for expensive resources
7. Mixin Pattern - Compose behavior from reusable pieces
8. Observer Pattern - Decouple event producers from consumers

For more information:

• System architecture: ARCHITECTURE.md
• Database design: DATABASE.md
• Security implementation: SECURITY.md
• Hands-on tutorial: GETTING-STARTED.md