"""Pydantic request/response schemas shared across routers.""" from __future__ import annotations import uuid from datetime import datetime from typing import Annotated, Any, Literal from pydantic import BaseModel, Field, StringConstraints from app.models import ManagementRole UsernameStr = Annotated[str, StringConstraints(pattern=r"^[a-zA-Z0-9_-]{1,100}$")] TopicFilterStr = Annotated[str, StringConstraints(min_length=1, max_length=200)] ACL_TYPES = Literal[ "publishClientSend", "publishClientReceive", "subscribeLiteral", "subscribePattern", "unsubscribeLiteral", "unsubscribePattern", ] # --- Auth ------------------------------------------------------------------ class LoginRequest(BaseModel): username: str password: str class SessionInfo(BaseModel): username: str role: ManagementRole must_change_password: bool class PasswordChangeRequest(BaseModel): current_password: str new_password: str = Field(min_length=12) # --- Management users ------------------------------------------------------- class ManagementUserOut(BaseModel): id: uuid.UUID username: str display_name: str | None role: ManagementRole enabled: bool must_change_password: bool created_at: datetime model_config = {"from_attributes": True} class ManagementUserCreate(BaseModel): username: UsernameStr password: str = Field(min_length=12) display_name: str | None = None role: ManagementRole = ManagementRole.OPERATOR class ManagementUserUpdate(BaseModel): display_name: str | None = None role: ManagementRole | None = None enabled: bool | None = None # --- MQTT clients ------------------------------------------------------------ class AclRuleIn(BaseModel): """One ACL rule as edited in the UI (spec section 14).""" acltype: ACL_TYPES topic: TopicFilterStr allow: bool = True priority: int = Field(default=0, ge=0, le=100000) class RoleIn(BaseModel): rolename: UsernameStr text_name: str | None = None text_description: str | None = None acls: list[AclRuleIn] = Field(default_factory=list) class RoleUpdate(BaseModel): """PUT /roles/{name} payload: rolename comes from the URL path.""" acls: list[AclRuleIn] class GroupRoleIn(BaseModel): rolename: str priority: int = 1 class GroupIn(BaseModel): groupname: UsernameStr text_name: str | None = None text_description: str | None = None roles: list[GroupRoleIn] = Field(default_factory=list) class ClientRoleIn(BaseModel): rolename: str priority: int = 1 class ClientGroupIn(BaseModel): groupname: str priority: int = 1 class ClientRoleAssign(BaseModel): """POST /clients/{username}/roles body — add one role incrementally.""" rolename: str = Field(min_length=1) priority: int = Field(default=-1, ge=-1, le=100000) class GroupRoleAssign(BaseModel): """POST /groups/{groupname}/roles body — add one role incrementally.""" rolename: str = Field(min_length=1) priority: int = Field(default=-1, ge=-1, le=100000) class GroupMemberAssign(BaseModel): """POST /groups/{groupname}/clients body — add one client to a group.""" username: str = Field(min_length=1) priority: int = Field(default=-1, ge=-1, le=100000) class MqttClientCreate(BaseModel): username: UsernameStr password: str | None = Field( default=None, description="If omitted a random password is generated and returned once.", ) clientid: str | None = None text_name: str | None = None text_description: str | None = None groups: list[ClientGroupIn] = Field(default_factory=list) roles: list[ClientRoleIn] = Field(default_factory=list) # PostgreSQL-side metadata (spec section 11) owner: str | None = None purpose: str | None = None tags: list[str] = Field(default_factory=list) class MqttClientOut(BaseModel): username: str clientid: str | None text_name: str | None text_description: str | None enabled: bool groups: list[dict[str, Any]] roles: list[dict[str, Any]] owner: str | None purpose: str | None tags: list[str] last_known_connection: datetime | None class ClientIdUpdate(BaseModel): """PUT /clients/{username}/clientid — bind or clear the expected client ID (spec section 11: "bind an expected client ID"). An empty string clears the binding: any client id may then connect with these credentials. A bound ID restricts connections to exactly that client id (anti-credential-sharing). """ clientid: str = Field(default="", max_length=200) class MqttPasswordReset(BaseModel): """Response for create/reset — plaintext shown exactly once (spec 11).""" username: str password: str # --- Broker configuration ---------------------------------------------------- class BrokerSettingsIn(BaseModel): key: str = Field(min_length=1, max_length=100) value: str value_type: Literal["string", "int", "bool"] = "string" class BrokerSettingsOut(BrokerSettingsIn): """GET /broker/settings rows — same fields, from ORM attributes.""" model_config = {"from_attributes": True} class ListenerIn(BaseModel): name: str = Field(min_length=1, max_length=100) enabled: bool = True port: int = Field(ge=1, le=65535) bind_address: str | None = None protocol: Literal["mqtt", "websockets"] = "mqtt" allow_anonymous: bool = False max_connections: int = -1 tls_enabled: bool = False tls_min_version: Literal["tlsv1.2", "tlsv1.3"] | None = None require_certificate: bool = False use_identity_as_username: bool = False # Bound server certificate (PKI): emitted as certfile/keyfile/cafile. cert_id: uuid.UUID | None = None class ListenerOut(ListenerIn): id: uuid.UUID model_config = {"from_attributes": True} class ConfigPreviewOut(BaseModel): config_text: str warnings: list[str] class DeployRequest(BaseModel): comment: str | None = None class RevisionOut(BaseModel): id: uuid.UUID revision_number: int created_by: str | None comment: str | None deployed: bool deployed_at: datetime | None created_at: datetime model_config = {"from_attributes": True} class RevisionDetailOut(RevisionOut): config_text: str validation_result: dict[str, Any] # --- PKI: certificate authorities + server certificates (spec 16-19) --------- class CaCreate(BaseModel): common_name: str = Field(min_length=1, max_length=200) validity_days: int = Field(gt=0, le=36500) key_algorithm: Literal["rsa3072", "rsa4096", "ecdsa-p256", "ecdsa-p384"] organization: str | None = None organizational_unit: str | None = None country: str | None = Field(default=None, max_length=2) class ServerCertCreate(BaseModel): """Issue a server certificate signed by an existing CA (spec 19).""" ca_id: uuid.UUID common_name: str = Field(min_length=1, max_length=200) validity_days: int = Field(gt=0, le=36500) dns_sans: list[str] = Field(default_factory=list) ip_sans: list[str] = Field(default_factory=list) organization: str | None = None organizational_unit: str | None = None country: str | None = Field(default=None, max_length=2) class CertificateOut(BaseModel): """Inventory view — NEVER includes the private key (spec section 21).""" id: uuid.UUID common_name: str is_ca: bool parent_id: uuid.UUID | None cert_type: str # MQTT client this certificate authenticates (client certs only; # the CN equals the username — spec section 24). username: str | None = None key_algorithm: str serial_number: str dns_sans: list[str] ip_sans: list[str] fingerprint: str status: str expiry_date: datetime created_at: datetime model_config = {"from_attributes": True} class ClientCertCreate(BaseModel): """Issue a client certificate bound to an MQTT client (spec section 20). The certificate CN is the MQTT username — with use_identity_as_username the broker maps CN -> dynsec identity (spec section 24), so the dynsec client must already exist. """ username: str = Field(min_length=1, max_length=100) ca_id: uuid.UUID validity_days: int = Field(gt=0, le=36500) class CertificateBundle(BaseModel): """One-time download payload (spec sections 20, 44). Contains EVERYTHING including the plaintext private key — shown exactly once, never retrievable afterwards from the inventory. """ certificate_id: uuid.UUID username: str certificate_pem: str private_key_pem: str ca_chain_pem: str download_hint: str = ( "Store these files securely now — the private key cannot be " "retrieved again." )