ipam_control/app/schemas.py

import ipaddress
import re
from datetime import datetime
from typing import Annotated, Generic, TypeVar

from pydantic import AfterValidator, BaseModel, ConfigDict, EmailStr, Field, field_validator

from app.models import AddressStatus, PrefixStatus

T = TypeVar("T")


def _cidr(v: str) -> str:
    try:
        return str(ipaddress.ip_network(v.strip(), strict=True))
    except ValueError:
        raise ValueError("Некорректный CIDR (пример: 10.30.0.0/24, биты хоста должны быть нулевыми)")


def _ip(v: str) -> str:
    try:
        return str(ipaddress.ip_address(v.strip()))
    except ValueError:
        raise ValueError("Некорректный IP-адрес")


Cidr = Annotated[str, AfterValidator(_cidr)]
IpAddr = Annotated[str, AfterValidator(_ip)]
_FQDN = re.compile(r"^(?=.{1,253}$)([A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9_])?)(\.[A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9_])?)*$")
_MAC = re.compile(r"^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$")


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


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


# --- auth
class LoginIn(BaseModel):
    username: str
    password: str


class TokenOut(BaseModel):
    access_token: str
    token_type: str = "bearer"


class UserOut(ORM):
    username: str
    role: str


# --- organizations
class OrgIn(BaseModel):
    name: str = Field(min_length=1, max_length=255)
    short_name: str = Field("", max_length=100)
    inn: str = Field(pattern=r"^(\d{10}|\d{12})$")
    address: str = Field("", max_length=500)
    contact_person: str = ""
    phone: str = ""
    email: EmailStr | str = ""
    note: str = ""

    @field_validator("email")
    @classmethod
    def _email(cls, v):
        return str(v)


class OrgOut(ORM, OrgIn):
    id: int
    prefixes_count: int = 0
    addresses_count: int = 0


# --- vrf
class VrfIn(BaseModel):
    organization_id: int
    name: str = Field(min_length=1, max_length=100)
    route_target: str = Field("", max_length=50, pattern=r"^(\d+:\d+)?$")
    note: str = ""


class VrfUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    route_target: str | None = Field(None, max_length=50, pattern=r"^(\d+:\d+)?$")
    note: str | None = None


class VrfOut(ORM):
    id: int
    organization_id: int
    name: str
    route_target: str
    note: str
    prefixes_count: int = 0


# --- device types / devices
class DeviceTypeIn(BaseModel):
    name: str = Field(min_length=1, max_length=100)


class DeviceTypeOut(ORM):
    id: int
    name: str
    is_default: bool
    devices_count: int = 0


class DeviceIn(BaseModel):
    name: str = Field(min_length=1, max_length=255)
    device_type_id: int
    organization_id: int
    mac: str = ""
    note: str = ""

    @field_validator("name")
    @classmethod
    def _name(cls, v):
        if not _FQDN.match(v):
            raise ValueError("Некорректное имя устройства (hostname)")
        return v

    @field_validator("mac")
    @classmethod
    def _mac(cls, v):
        if v and not _MAC.match(v):
            raise ValueError("Некорректный MAC-адрес (AA:BB:CC:DD:EE:FF)")
        return v.upper().replace("-", ":")


class DeviceUpdate(BaseModel):
    name: str | None = None
    device_type_id: int | None = None
    mac: str | None = None
    note: str | None = None


class DeviceOut(ORM):
    id: int
    name: str
    device_type_id: int
    device_type_name: str
    organization_id: int
    mac: str
    note: str
    ip_addresses: list[str] = []
    first_prefix_id: int | None = None
    all_deprecated: bool = False


# --- isp
class IspIn(BaseModel):
    name: str = Field(min_length=1, max_length=255)
    organization_id: int
    networks: list[Cidr] = []
    hotline: str = Field("", max_length=50)
    contract_number: str = Field("", max_length=100)
    note: str = ""


class IspOut(ORM):
    id: int
    name: str
    organization_id: int
    organization_name: str
    networks: list[str]
    hotline: str
    contract_number: str
    note: str


# --- prefixes
class PrefixIn(BaseModel):
    organization_id: int
    vrf_id: int
    prefix: Cidr
    description: str = Field("", max_length=500)
    status: PrefixStatus = PrefixStatus.active
    parent_id: int | None = None
    is_pool: bool = False
    note: str = ""


class PrefixUpdate(BaseModel):
    vrf_id: int | None = None
    description: str | None = Field(None, max_length=500)
    status: PrefixStatus | None = None
    is_pool: bool | None = None
    note: str | None = None


class PrefixOut(ORM):
    id: int
    organization_id: int
    vrf_id: int
    vrf_name: str
    prefix: str
    family: int
    description: str
    status: PrefixStatus
    parent_id: int | None
    depth: int
    is_pool: bool
    note: str
    used: int
    capacity: int
    utilization: int
    addresses_count: int


# --- addresses
class AddressIn(BaseModel):
    address: IpAddr
    status: AddressStatus = AddressStatus.assigned
    dns_name: str = ""
    description: str = Field("", max_length=500)
    device_id: int | None = None
    note: str = ""

    @field_validator("dns_name")
    @classmethod
    def _dns(cls, v):
        if v and not _FQDN.match(v):
            raise ValueError("Некорректное DNS-имя (FQDN)")
        return v


class AddressUpdate(BaseModel):
    status: AddressStatus | None = None
    dns_name: str | None = None
    description: str | None = Field(None, max_length=500)
    device_id: int | None = None
    note: str | None = None


class AddressOut(BaseModel):
    id: int | None
    prefix_id: int
    address: str
    status: str  # assigned | reserved | deprecated | free
    dns_name: str = ""
    description: str = ""
    device_id: int | None = None
    device_name: str | None = None
    note: str = ""
    updated_at: datetime | None = None


class AddressSummary(BaseModel):
    assigned: int
    reserved: int
    deprecated: int
    free: int
    capacity: int


class AddressPage(Page[AddressOut]):
    summary: AddressSummary


class AuditOut(ORM):
    id: int
    uid: str
    short_id: str
    ts: datetime
    username: str
    actor: str
    entity_type: str
    entity_id: int | None
    entity_label: str
    action: str
    event_type: str
    message: str
    diff: dict | None
    client_ip: str | None = None
    meta: dict | None = None

    @classmethod
    def from_row(cls, r) -> "AuditOut":
        return cls(
            id=r.id, uid=str(r.uid), short_id=str(r.uid)[:8], ts=r.ts, username=r.username,
            actor=r.username if r.username in ("system", "anonymous") else f"ui:{r.username}",
            entity_type=r.entity_type, entity_id=r.entity_id, entity_label=r.entity_label, action=r.action,
            event_type=f"{r.entity_type}.{r.action}", message=r.message, diff=r.diff,
            client_ip=ip_text(r.client_ip) if r.client_ip is not None else None, meta=r.meta,
        )


def ip_text(v) -> str:
    """psycopg отдаёт inet/cidr объектами ipaddress; для хоста убираем /32 и /128."""
    text = str(v)
    if text.endswith("/32") or text.endswith("/128"):
        return text.rsplit("/", 1)[0]
    return text