Files
Knacky e995853f0d feat(m1): DB schema, migrations, diag visibility
23 tables + alembic_version covering the v1 data model:
- Auth/RBAC (8): users, groups, permissions, user_groups, group_permissions,
  invitations, invitation_groups, refresh_tokens.
- MITRE (4): mitre_tactics, mitre_techniques, mitre_subtechniques + the
  technique↔tactic many-to-many.
- Templates (4): test_templates, test_template_mitre_tags (3 nullable FKs +
  CHECK exactly_one_mitre_fk), scenario_templates, scenario_template_tests
  (UUID PK + UNIQUE(scenario_id, position) so a test can appear at multiple
  positions).
- Missions (6): missions, mission_members, mission_scenarios, mission_tests,
  mission_test_mitre_tags (deliberately denormalised — copies external_id +
  name + url, no FK to mitre_* — so a re-sync of the catalogue can't purge
  historical tags), mission_categories.
- Evidence/settings/notifications (5): evidence_files, settings (JSONB
  value), detection_levels, notifications.

SQLAlchemy 2.x with Mapped[]/mapped_column(), pk_/fk_/ck_/uq_/ix_ naming
convention. Reusable mixins (UuidPkMixin, TimestampMixin, SoftDeleteMixin —
no auto __table_args__ since classes silently clobber the mixin's).

Soft delete: deleted_at + partial indexes ix_<table>_active WHERE deleted_at
IS NULL on 9 tables (users, groups, test_templates, scenario_templates,
missions, mission_scenarios, mission_tests, mission_categories,
evidence_files). Notifications gets ix_..._unread WHERE read_at IS NULL.

CHECK constraints for status / state / opsec_level / mitre_kind enums.

New API endpoint GET /api/v1/diag/db: returns alembic_revision (short hash)
and the public-schema table_count. 503 with {"reachable": false} on a DB
outage. Database card on the SPA home consumes it.

Test stage in backend/Dockerfile (--target test): runtime + dev extras +
tests/. New make test-api spins an ephemeral pytest container against the
live DB on the compose network. backend/tests/test_schema.py: 8 integration
tests (tables, FK pairs, CHECK constraints, partial indexes, alembic-at-head,
negative INSERT proving the exactly_one_mitre_fk CHECK fires).

e2e/tests/m1-db.spec.ts: 4 Playwright tests covering the diag endpoint
contract + the Database card + footer/roadmap labels.

DoD: make clean && make up && make migrate → 23 tables, 32 FKs, 9 CHECKs,
make test-api → 9 passed, make e2e → 12 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 06:16:24 +02:00

189 lines
6.8 KiB
Python

"""Auth + RBAC: users, groups, permissions, invitations, refresh tokens."""
from __future__ import annotations
import uuid
from datetime import datetime
from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey, Index, String, Text, UniqueConstraint, Uuid
from sqlalchemy import DateTime as SADateTime
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db.base import Base
from app.db.mixins import SoftDeleteMixin, TimestampMixin, UuidPkMixin
if TYPE_CHECKING:
from app.models.evidence import EvidenceFile
from app.models.notification import Notification
class User(Base, UuidPkMixin, TimestampMixin, SoftDeleteMixin):
__tablename__ = "users"
email: Mapped[str] = mapped_column(String(254), nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
display_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
locale: Mapped[str] = mapped_column(String(8), default="fr", nullable=False)
is_active: Mapped[bool] = mapped_column(default=True, nullable=False)
groups: Mapped[list["Group"]] = relationship(
secondary="user_groups",
back_populates="users",
lazy="selectin",
)
refresh_tokens: Mapped[list["RefreshToken"]] = relationship(
back_populates="user",
cascade="all, delete-orphan",
)
notifications: Mapped[list["Notification"]] = relationship(
back_populates="user",
cascade="all, delete-orphan",
)
uploaded_evidence: Mapped[list["EvidenceFile"]] = relationship(
back_populates="uploaded_by",
)
__table_args__ = (
# Email uniqueness scoped to non-deleted rows so an admin can re-invite
# a previously-soft-deleted user.
Index(
"uq_users_email_active",
"email",
unique=True,
postgresql_where="deleted_at IS NULL",
),
Index("ix_users_active", "deleted_at", postgresql_where="deleted_at IS NULL"),
)
class Group(Base, UuidPkMixin, TimestampMixin, SoftDeleteMixin):
__tablename__ = "groups"
name: Mapped[str] = mapped_column(String(80), nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
# Built-in groups (admin/redteam/blueteam) are protected from deletion.
is_system: Mapped[bool] = mapped_column(default=False, nullable=False)
users: Mapped[list[User]] = relationship(
secondary="user_groups",
back_populates="groups",
)
permissions: Mapped[list["Permission"]] = relationship(
secondary="group_permissions",
back_populates="groups",
lazy="selectin",
)
__table_args__ = (
Index(
"uq_groups_name_active",
"name",
unique=True,
postgresql_where="deleted_at IS NULL",
),
Index("ix_groups_active", "deleted_at", postgresql_where="deleted_at IS NULL"),
)
class Permission(Base, UuidPkMixin, TimestampMixin):
"""Atomic permission. Code follows the `<entity>.<action>` convention."""
__tablename__ = "permissions"
code: Mapped[str] = mapped_column(String(80), unique=True, nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
groups: Mapped[list[Group]] = relationship(
secondary="group_permissions",
back_populates="permissions",
)
class UserGroup(Base):
"""User ↔ Group join — no soft delete, just attach/detach."""
__tablename__ = "user_groups"
user_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
group_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True
)
class GroupPermission(Base):
"""Group ↔ Permission join."""
__tablename__ = "group_permissions"
group_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True
)
permission_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("permissions.id", ondelete="CASCADE"), primary_key=True
)
class Invitation(Base, UuidPkMixin, TimestampMixin):
__tablename__ = "invitations"
# Hash of the URL token, never the token itself.
token_hash: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
email_hint: Mapped[str | None] = mapped_column(String(254), nullable=True)
created_by_user_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False
)
expires_at: Mapped[datetime] = mapped_column(SADateTime(timezone=True), nullable=False)
consumed_at: Mapped[datetime | None] = mapped_column(SADateTime(timezone=True), nullable=True)
revoked_at: Mapped[datetime | None] = mapped_column(SADateTime(timezone=True), nullable=True)
consumed_by_user_id: Mapped[uuid.UUID | None] = mapped_column(
Uuid(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True
)
pre_assigned_groups: Mapped[list[Group]] = relationship(
secondary="invitation_groups",
lazy="selectin",
)
__table_args__ = (Index("ix_invitations_expires_at", "expires_at"),)
class InvitationGroup(Base):
"""Pre-assigned groups attached to an invitation; applied at acceptance."""
__tablename__ = "invitation_groups"
invitation_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("invitations.id", ondelete="CASCADE"), primary_key=True
)
group_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True
)
class RefreshToken(Base, UuidPkMixin, TimestampMixin):
"""Long-lived refresh tokens. The hash, never the token, is stored."""
__tablename__ = "refresh_tokens"
user_id: Mapped[uuid.UUID] = mapped_column(
Uuid(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
)
jti: Mapped[str] = mapped_column(String(64), nullable=False)
token_hash: Mapped[str] = mapped_column(String(128), nullable=False)
issued_at: Mapped[datetime] = mapped_column(SADateTime(timezone=True), nullable=False)
expires_at: Mapped[datetime] = mapped_column(SADateTime(timezone=True), nullable=False)
revoked_at: Mapped[datetime | None] = mapped_column(SADateTime(timezone=True), nullable=True)
replaced_by_id: Mapped[uuid.UUID | None] = mapped_column(
Uuid(as_uuid=True), ForeignKey("refresh_tokens.id", ondelete="SET NULL"), nullable=True
)
user: Mapped[User] = relationship(back_populates="refresh_tokens")
__table_args__ = (
UniqueConstraint("jti", name="uq_refresh_tokens_jti"),
Index("ix_refresh_tokens_user_id_expires_at", "user_id", "expires_at"),
)