""" Pydantic-Schemas fuer alle API-Eingaben/-Ausgaben. Strikte Validierung ist Teil des Hardening-Konzepts (6.6): Laenge, Typ und erlaubte Zeichen werden hier durchgesetzt, bevor irgendein Wert die Business-Logik oder die Datenbank erreicht. """ from __future__ import annotations import re from typing import Literal from pydantic import BaseModel, Field, field_validator USERNAME_RE = re.compile(r"^[a-zA-Z0-9._-]{3,64}$") HOSTNAME_LABEL_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]{0,127}$") ROLE_NAME = Literal[ "ssh_connect", "rdp_connect", "file_transfer", "clipboard", "session_recording_view", "admin_hostgroup", # Hinzugefuegt fuer RBAC-Erweiterung 'Credentials ins RBAC-Modell' (Migration # 0008_credentials_roles.sql): erlaubt Nicht-Admins mit dieser Rolle auf # einer Hostgruppe gezielten Zugriff auf Zugangsdaten (SSH-Keys, RDP- # Passwoerter) der Hosts dieser Gruppe, siehe app/auth/deps.py:: # require_admin_scope_or_host_role und admin/routes.py (rdp-credentials, # ssh-keys-Zuordnung, GET /admin/hosts/{id}/credentials). "credentials_view", "credentials_manage", ] class LoginRequest(BaseModel): username: str = Field(min_length=3, max_length=64) password: str = Field(min_length=1, max_length=256) @field_validator("username") @classmethod def check_username(cls, v: str) -> str: if not USERNAME_RE.match(v): raise ValueError("Ungueltiger Benutzername") return v class TotpLoginRequest(BaseModel): pending_token: str code: str = Field(min_length=6, max_length=64) class TotpConfirmRequest(BaseModel): code: str = Field(min_length=6, max_length=6, pattern=r"^\d{6}$") class ChangePasswordRequest(BaseModel): current_password: str = Field(min_length=1, max_length=256) new_password: str = Field(min_length=12, max_length=256) class UserCreateRequest(BaseModel): username: str = Field(min_length=3, max_length=64) initial_password: str = Field(min_length=12, max_length=256) is_admin: bool = False # Nur relevant, wenn der anlegende Principal Mandanten-Admin ist: wird # serverseitig ohnehin auf den/die erlaubten Mandanten erzwungen (siehe # create_user() in admin/routes.py) -- ein Mandanten-Admin kann hierueber # KEINEN User einem fremden Mandanten zuordnen. Fuer Super-Admins optional. home_tenant_id: int | None = None @field_validator("username") @classmethod def check_username(cls, v: str) -> str: if not USERNAME_RE.match(v): raise ValueError("Ungueltiger Benutzername") return v class UserUpdateRequest(BaseModel): is_admin: bool | None = None is_active: bool | None = None new_password: str | None = Field(default=None, min_length=12, max_length=256) class TenantCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) class TenantUpdateRequest(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) is_active: bool | None = None class TenantAdminAssignRequest(BaseModel): user_id: int class HostGroupCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) # Nur fuer Super-Admins relevant (Mandanten-Admins werden serverseitig # ohnehin auf ihren eigenen Mandanten festgelegt, siehe admin/routes.py). tenant_id: int | None = None class HostGroupUpdateRequest(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) class HostCreateRequest(BaseModel): host_group_id: int hostname: str = Field(min_length=1, max_length=128) address: str = Field(min_length=1, max_length=255) protocol: Literal["ssh", "rdp"] port: int = Field(gt=0, le=65535) os_type: Literal["linux", "windows"] ssh_host_key_fingerprint: str | None = Field(default=None, max_length=512) # ---- Altlast (vor Migration 0010) ------------------------------------- # Benutzername und Domaene gehoeren zu den ZUGANGSDATEN (ssh_keys.username # bzw. rdp_credentials.username/.domain) und nicht zum Host. Die Felder # bleiben nur entgegennahmefaehig, damit bestehende Skripte/Tokens nicht # brechen; die Admin-Oberflaeche sendet sie nicht mehr, und der # Verbindungsaufbau liest sie nur noch als Fallback. ssh_username: str | None = Field(default=None, max_length=128, deprecated=True) rdp_username: str | None = Field(default=None, max_length=128, deprecated=True) rdp_domain: str | None = Field(default=None, max_length=128, deprecated=True) rdp_require_nla: bool = True clipboard_enabled: bool = True file_transfer_enabled: bool = True # Standard: selbstsigniertes RDP-Zertifikat des Ziels akzeptieren. Windows # praesentiert ohne interne PKI genau so eines; guacd/FreeRDP bricht sonst # vor dem ersten Bild ab. Wer eine PKI betreibt, setzt den Wert pro Host # auf False und erzwingt damit die Pruefung. rdp_ignore_cert: bool = True @field_validator("hostname") @classmethod def check_hostname(cls, v: str) -> str: if not HOSTNAME_LABEL_RE.match(v): raise ValueError("Ungueltiger Hostname") return v class HostUpdateRequest(BaseModel): host_group_id: int | None = None hostname: str | None = Field(default=None, min_length=1, max_length=128) address: str | None = Field(default=None, min_length=1, max_length=255) port: int | None = Field(default=None, gt=0, le=65535) # Siehe HostCreateRequest: Altlast, gehoert zu den Zugangsdaten. ssh_username: str | None = Field(default=None, deprecated=True) rdp_username: str | None = Field(default=None, deprecated=True) rdp_domain: str | None = Field(default=None, deprecated=True) rdp_require_nla: bool | None = None clipboard_enabled: bool | None = None file_transfer_enabled: bool | None = None rdp_ignore_cert: bool | None = None is_active: bool | None = None @field_validator("hostname") @classmethod def check_hostname(cls, v: str | None) -> str | None: if v is not None and not HOSTNAME_LABEL_RE.match(v): raise ValueError("Ungueltiger Hostname") return v class RoleGrantRequest(BaseModel): user_id: int host_group_id: int role_names: list[ROLE_NAME] = Field(min_length=1, max_length=8) expires_at: str | None = None class RoleRevokeRequest(BaseModel): user_id: int host_group_id: int role_name: ROLE_NAME class UserGroupCreateRequest(BaseModel): name: str = Field(min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) tenant_id: int | None = None class UserGroupUpdateRequest(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = Field(default=None, max_length=1024) class GroupMemberRequest(BaseModel): user_id: int class GroupRoleGrantRequest(BaseModel): user_group_id: int host_group_id: int role_names: list[ROLE_NAME] = Field(min_length=1, max_length=8) expires_at: str | None = None class GroupRoleRevokeRequest(BaseModel): user_group_id: int host_group_id: int role_name: ROLE_NAME class ApiTokenCreateRequest(BaseModel): label: str = Field(min_length=1, max_length=128) user_id: int scopes: list[str] = Field(min_length=1, max_length=32) expires_at: str | None = None # Nur fuer Super-Admins relevant -- ein Mandanten-Admin wird serverseitig # ohnehin auf seinen eigenen Mandanten festgelegt. tenant_id: int | None = None class SshKeyCreateRequest(BaseModel): label: str = Field(min_length=1, max_length=128) owner_user_id: int | None = None private_key_pem: str = Field(min_length=1, max_length=32_768) public_key: str = Field(min_length=1, max_length=8192) key_type: Literal["ed25519", "rsa-3072", "rsa-4096", "ca-cert"] tenant_id: int | None = None # Anmeldename am Zielsystem. Gehoert zum Schluessel, nicht zum Host # (Migration 0010): ein Schluessel ist immer "der Zugang von Benutzer X". # Fuer mehrere Konten wird je ein Schluesseleintrag angelegt. username: str | None = Field(default=None, max_length=128) # Passphrase eines verschluesselten Private Keys. Wird -- wie das # Schluesselmaterial selbst -- mit dem KEK verschluesselt gespeichert # (ssh_keys.passphrase_enc, Migration 0009) und beim Verbindungsaufbau an # asyncssh uebergeben. None/"" = Schluessel ist unverschluesselt. passphrase: str | None = Field(default=None, max_length=1024) class SshKeyUpdateRequest(BaseModel): label: str | None = Field(default=None, min_length=1, max_length=128) owner_user_id: int | None = None # Wenn gesetzt, wird der Schluessel rotiert (neues Schluesselmaterial, # rotated_at wird aktualisiert). Alle drei Felder muessen dann zusammen # angegeben werden (siehe rotate_ssh_key() in admin/routes.py). private_key_pem: str | None = Field(default=None, min_length=1, max_length=32_768) public_key: str | None = Field(default=None, min_length=1, max_length=8192) key_type: Literal["ed25519", "rsa-3072", "rsa-4096", "ca-cert"] | None = None # Wird -- wie passphrase -- nur beachtet, wenn das Feld im Request # vorkommt: Weglassen laesst den Benutzernamen unveraendert, explizites # null/"" entfernt ihn. username: str | None = Field(default=None, max_length=128) # Wird nur beachtet, wenn das Feld im Request vorkommt (siehe # update_ssh_key in app/admin/routes.py): explizites null/"" entfernt eine # hinterlegte Passphrase, Weglassen laesst sie unveraendert. Ohne # gleichzeitige Rotation wird die neue Passphrase gegen das bereits # gespeicherte Schluesselmaterial geprueft. passphrase: str | None = Field(default=None, max_length=1024) class ConnectRequest(BaseModel): host_id: int class RdpCredentialsRequest(BaseModel): password: str = Field(min_length=1, max_length=512) # Benutzername und (optional) Domaene der Windows-Anmeldung. Seit # Migration 0010 Teil der Zugangsdaten statt des Hosts. username: str | None = Field(default=None, max_length=128) domain: str | None = Field(default=None, max_length=128)