feat(backend): add pagination + user/member/audit DTOs (D-016)

Adds the `Page[T]` envelope `{items, total, page, page_size}` documented in
D-016, the matching `PageQuery` for `?page=&page_size=` parsing (default 50,
max 200), and `parse_page_query()` helper for blueprints.

DTOs:
- `UserRead` / `UserCreate` / `UserUpdate` (sprint 2). `UserRead` never
  exposes `local_password_hash`. `UserCreate` validates email via
  pydantic-email-validator and pins password to 8..128 chars.
- `EngagementMemberRead` / `EngagementMemberCreate`. `role` is a free-form
  string ≤ 40 chars (D-017), defaulting to `"member"`.
- `AuditLogEntry` for the upcoming audit viewer.
This commit is contained in:
knacky
2026-05-23 15:52:56 +02:00
parent 48a1c756bf
commit feda5d1485
6 changed files with 158 additions and 0 deletions

View File

@@ -15,6 +15,7 @@ from mimic.auth.identity import AuthUser
from mimic.extensions import db from mimic.extensions import db
from mimic.rbac.matrix import GroupName from mimic.rbac.matrix import GroupName
from mimic.schemas import CurrentUser from mimic.schemas import CurrentUser
from mimic.schemas.pagination import PageQuery
def parse_body[T: BaseModel](model: type[T]) -> T: def parse_body[T: BaseModel](model: type[T]) -> T:
@@ -69,6 +70,15 @@ def api_error(code: str, message: str, status: int) -> tuple[Response, int]:
return jsonify({"error": code, "message": message}), status return jsonify({"error": code, "message": message}), status
def parse_page_query() -> PageQuery:
"""Read `?page=` / `?page_size=` from the current request (D-016)."""
raw = {key: value for key, value in request.args.items() if key in {"page", "page_size"}}
try:
return PageQuery.model_validate(raw)
except ValidationError as exc:
abort(422, description=exc.errors())
def audit_write( def audit_write(
*, *,
action: str, action: str,

View File

@@ -1,12 +1,15 @@
"""Pydantic 2 request/response DTOs.""" """Pydantic 2 request/response DTOs."""
from mimic.schemas.audit import AuditLogEntry
from mimic.schemas.auth import CurrentUser, LoginRequest from mimic.schemas.auth import CurrentUser, LoginRequest
from mimic.schemas.engagement import ( from mimic.schemas.engagement import (
EngagementCreate, EngagementCreate,
EngagementRead, EngagementRead,
EngagementUpdate, EngagementUpdate,
) )
from mimic.schemas.engagement_member import EngagementMemberCreate, EngagementMemberRead
from mimic.schemas.host import HostCreate, HostRead, HostUpdate from mimic.schemas.host import HostCreate, HostRead, HostUpdate
from mimic.schemas.pagination import Page, PageQuery
from mimic.schemas.scenario import ( from mimic.schemas.scenario import (
ScenarioCreate, ScenarioCreate,
ScenarioRead, ScenarioRead,
@@ -15,16 +18,22 @@ from mimic.schemas.scenario import (
ScenarioUpdate, ScenarioUpdate,
) )
from mimic.schemas.ttp import TtpCreate, TtpRead, TtpUpdate from mimic.schemas.ttp import TtpCreate, TtpRead, TtpUpdate
from mimic.schemas.user import UserCreate, UserRead, UserUpdate
__all__ = [ __all__ = [
"AuditLogEntry",
"CurrentUser", "CurrentUser",
"EngagementCreate", "EngagementCreate",
"EngagementMemberCreate",
"EngagementMemberRead",
"EngagementRead", "EngagementRead",
"EngagementUpdate", "EngagementUpdate",
"HostCreate", "HostCreate",
"HostRead", "HostRead",
"HostUpdate", "HostUpdate",
"LoginRequest", "LoginRequest",
"Page",
"PageQuery",
"ScenarioCreate", "ScenarioCreate",
"ScenarioRead", "ScenarioRead",
"ScenarioStepCreate", "ScenarioStepCreate",
@@ -33,4 +42,7 @@ __all__ = [
"TtpCreate", "TtpCreate",
"TtpRead", "TtpRead",
"TtpUpdate", "TtpUpdate",
"UserCreate",
"UserRead",
"UserUpdate",
] ]

View File

@@ -0,0 +1,28 @@
"""Audit log viewer DTO (sprint 2)."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from uuid import UUID
from pydantic import BaseModel, ConfigDict
class AuditLogEntry(BaseModel):
"""Single audit log row as exposed by `GET /api/v1/audit/log`."""
model_config = ConfigDict(from_attributes=True)
id: UUID
ts: datetime
actor_id: UUID | None
action: str
resource_type: str
resource_id: str | None
metadata_json: dict[str, Any]
prev_hash: str | None
row_hash: str
source_ip: str | None
user_agent: str | None
comment: str | None

View File

@@ -0,0 +1,27 @@
"""Engagement membership DTOs (sprint 2).
`role` is a free-form label per D-017 — not a permission gate. Application-
level RBAC stays the responsibility of the F11 `group` membership; per-
engagement role is informational (e.g. "lead", "shadow", "binôme A").
"""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
class EngagementMemberRead(BaseModel):
model_config = ConfigDict(from_attributes=True)
engagement_id: UUID
user_id: UUID
role: str
added_at: datetime
class EngagementMemberCreate(BaseModel):
user_id: UUID
role: str = Field(default="member", min_length=1, max_length=40)

View File

@@ -0,0 +1,40 @@
"""Generic pagination envelope (D-016).
Frontend reads `{items, total, page, page_size}`. Sprint 2 uses this on
`/users` and `/audit/log`; existing endpoints (`/engagements`) stay
non-paginated for backwards-compatibility and will migrate together in a
later opt-in (`?paginate=true` or `/api/v2/`).
"""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
DEFAULT_PAGE_SIZE = 50
MAX_PAGE_SIZE = 200
class Page[T](BaseModel):
"""Paginated response envelope."""
model_config = ConfigDict(arbitrary_types_allowed=True)
items: list[T]
total: int = Field(ge=0)
page: int = Field(ge=1)
page_size: int = Field(ge=1, le=MAX_PAGE_SIZE)
class PageQuery(BaseModel):
"""Parsed `?page=` / `?page_size=` query string (always normalised)."""
page: int = Field(default=1, ge=1)
page_size: int = Field(default=DEFAULT_PAGE_SIZE, ge=1, le=MAX_PAGE_SIZE)
@property
def offset(self) -> int:
return (self.page - 1) * self.page_size
@property
def limit(self) -> int:
return self.page_size

View File

@@ -0,0 +1,41 @@
"""User management DTOs (sprint 2)."""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from mimic.db.types import UserType
class UserRead(BaseModel):
"""Public user representation. Never exposes `local_password_hash`."""
model_config = ConfigDict(from_attributes=True)
id: UUID
email: str
display_name: str | None
type: UserType
disabled_at: datetime | None
last_login_at: datetime | None
created_at: datetime
class UserCreate(BaseModel):
"""`POST /api/v1/users` body."""
email: EmailStr
display_name: str | None = Field(default=None, max_length=120)
password: str = Field(min_length=8, max_length=128)
type: UserType
class UserUpdate(BaseModel):
"""`PATCH /api/v1/users/<uid>` body (all fields optional)."""
display_name: str | None = Field(default=None, max_length=120)
password: str | None = Field(default=None, min_length=8, max_length=128)
type: UserType | None = None