Newer
Older
gnexus-creds / gnexus_creds / schemas.py
@Eugene Sukhodolskiy Eugene Sukhodolskiy 10 hours ago 4 KB Add in-app extension download page
"""Pydantic schemas for REST and services."""

from datetime import datetime
from enum import StrEnum
from typing import Generic, TypeVar
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field, field_validator

T = TypeVar("T")


class SecretStatus(StrEnum):
    actual = "actual"
    outdated = "outdated"


class Channel(StrEnum):
    ui = "ui"
    rest = "rest"
    mcp = "mcp"


class Scope(StrEnum):
    read = "read"
    reveal = "reveal"
    write = "write"
    admin = "admin"
    mcp = "mcp"


class SecretFieldIn(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    value: str = Field(max_length=65536)
    encrypted: bool = False
    masked: bool = False
    position: int = 0


class SecretFieldOut(BaseModel):
    name: str
    value: str | None = None
    encrypted: bool
    masked: bool
    position: int


class SecretCreate(BaseModel):
    title: str = Field(min_length=1, max_length=255)
    purpose: str | None = Field(default=None, max_length=255)
    category: str | None = Field(default=None, max_length=120)
    source: str | None = Field(default=None, max_length=255)
    notes: str | None = Field(default=None, max_length=140)
    tags: list[str] = Field(default_factory=list)
    status: SecretStatus = SecretStatus.actual
    archived: bool = False
    allow_ui: bool = True
    allow_rest_api: bool = True
    allow_mcp: bool = False
    fields: list[SecretFieldIn] = Field(default_factory=list)

    @field_validator("tags")
    @classmethod
    def normalize_tags(cls, tags: list[str]) -> list[str]:
        result = []
        for tag in tags:
            value = tag.strip().lower()
            if value and value not in result:
                result.append(value[:80])
        return result


class SecretUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=255)
    purpose: str | None = Field(default=None, max_length=255)
    category: str | None = Field(default=None, max_length=120)
    source: str | None = Field(default=None, max_length=255)
    notes: str | None = Field(default=None, max_length=140)
    tags: list[str] | None = None
    status: SecretStatus | None = None
    archived: bool | None = None
    allow_ui: bool | None = None
    allow_rest_api: bool | None = None
    allow_mcp: bool | None = None
    fields: list[SecretFieldIn] | None = None


class SecretRead(BaseModel):
    id: UUID
    title: str
    purpose: str | None
    category: str | None
    source: str | None
    notes: str | None
    tags: list[str]
    status: SecretStatus
    archived: bool
    allow_ui: bool
    allow_rest_api: bool
    allow_mcp: bool
    created_at: datetime
    updated_at: datetime
    fields: list[SecretFieldOut]


class SecretReveal(SecretRead):
    version_id: UUID
    version_number: int


class SecretVersionRead(BaseModel):
    id: UUID
    version_number: int
    created_at: datetime
    fields: list[SecretFieldOut]


class Page(BaseModel, Generic[T]):
    items: list[T]
    total: int
    offset: int
    limit: int


class ApiTokenCreate(BaseModel):
    name: str = Field(min_length=1, max_length=120)
    scopes: list[Scope]


class ApiTokenRead(BaseModel):
    id: UUID
    public_id: str
    name: str
    scopes: list[str]
    created_at: datetime
    revoked_at: datetime | None
    last_used_at: datetime | None


class ApiTokenCreated(ApiTokenRead):
    token: str


class AuditEventRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    user_id: UUID | None
    actor_user_id: UUID | None
    api_token_id: UUID | None
    secret_id: UUID | None
    channel: str
    action: str
    ip_address: str | None
    user_agent: str | None
    audit_metadata: dict = Field(serialization_alias="metadata")
    created_at: datetime


class UserRead(BaseModel):
    id: UUID
    email: str
    display_name: str | None
    locale: str | None
    role: str
    status: str
    avatar_url: str | None = None
    auth_profile_url: str | None = None


class UserUpdate(BaseModel):
    display_name: str | None = Field(default=None, max_length=120)
    locale: str | None = Field(default=None, max_length=10)


class ExportResponse(BaseModel):
    format: str
    version: int
    exported_at: datetime
    secrets: list[dict]


class ImportPayload(BaseModel):
    format: str
    version: int
    exported_at: datetime
    secrets: list[SecretCreate]


class StatsRead(BaseModel):
    total_secrets: int
    active_secrets: int
    mcp_enabled_secrets: int


class ExtensionBuildRead(BaseModel):
    browser: str
    filename: str
    sha256: str
    size: int
    download_url: str
    released_at: datetime | None = None
    notes: str | None = None


class ExtensionRead(BaseModel):
    version: str
    released_at: datetime | None = None
    notes: str | None = None
    builds: list[ExtensionBuildRead]