ipam_control/app/models.py

import enum
import uuid
from datetime import datetime

from sqlalchemy import (
    BigInteger, Boolean, CheckConstraint, DateTime, Enum, ForeignKey, ForeignKeyConstraint, Index, Integer, String, Text,
    UniqueConstraint, func, text,
)
from sqlalchemy.dialects.postgresql import CIDR, INET, JSONB, UUID
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class PrefixStatus(str, enum.Enum):
    active = "active"
    reserved = "reserved"
    deprecated = "deprecated"


class AddressStatus(str, enum.Enum):
    assigned = "assigned"
    reserved = "reserved"
    deprecated = "deprecated"


class Role(str, enum.Enum):
    superadmin = "superadmin"
    admin = "admin"
    viewer = "viewer"


class Organization(Base):
    __tablename__ = "organizations"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255), unique=True)
    short_name: Mapped[str] = mapped_column(String(100), default="")
    inn: Mapped[str] = mapped_column(String(12), unique=True)
    address: Mapped[str] = mapped_column(String(500), default="")
    contact_person: Mapped[str] = mapped_column(String(255), default="")
    phone: Mapped[str] = mapped_column(String(50), default="")
    email: Mapped[str] = mapped_column(String(255), default="")
    note: Mapped[str] = mapped_column(Text, default="")


class Vrf(Base):
    __tablename__ = "vrfs"
    # (id, organization_id) — цель составного FK префиксов: VRF префикса всегда из его организации
    __table_args__ = (UniqueConstraint("id", "organization_id", name="uq_vrfs_id_organization_id"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"))
    name: Mapped[str] = mapped_column(String(100))
    route_target: Mapped[str] = mapped_column(String(50), default="")
    note: Mapped[str] = mapped_column(Text, default="")


Index("uq_vrfs_org_lower_name", Vrf.organization_id, func.lower(Vrf.name), unique=True)  # имя VRF уникально в организации без учёта регистра


class Prefix(Base):
    __tablename__ = "prefixes"
    __table_args__ = (
        UniqueConstraint("vrf_id", "prefix"),
        UniqueConstraint("id", "vrf_id", name="uq_prefixes_id_vrf_id"),  # цель составного FK адресов
        ForeignKeyConstraint(["vrf_id", "organization_id"], ["vrfs.id", "vrfs.organization_id"], name="fk_prefixes_vrf_org"),
    )
    id: Mapped[int] = mapped_column(primary_key=True)
    organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True)
    vrf_id: Mapped[int] = mapped_column(ForeignKey("vrfs.id"), index=True)
    prefix: Mapped[str] = mapped_column(CIDR)
    description: Mapped[str] = mapped_column(String(500), default="")
    status: Mapped[PrefixStatus] = mapped_column(Enum(PrefixStatus, name="prefix_status"), default=PrefixStatus.active)
    parent_id: Mapped[int | None] = mapped_column(ForeignKey("prefixes.id", ondelete="SET NULL"))
    is_pool: Mapped[bool] = mapped_column(Boolean, default=False)
    note: Mapped[str] = mapped_column(Text, default="")

    vrf: Mapped[Vrf] = relationship(primaryjoin="Prefix.vrf_id == Vrf.id", foreign_keys=[vrf_id])


class Isp(Base):
    __tablename__ = "isps"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True)
    contract_number: Mapped[str] = mapped_column(String(100), default="")
    hotline: Mapped[str] = mapped_column(String(50), default="")
    note: Mapped[str] = mapped_column(Text, default="")

    networks: Mapped[list["IspNetwork"]] = relationship(cascade="all, delete-orphan", order_by="IspNetwork.id")


class IspNetwork(Base):
    __tablename__ = "isp_networks"
    id: Mapped[int] = mapped_column(primary_key=True)
    isp_id: Mapped[int] = mapped_column(ForeignKey("isps.id", ondelete="CASCADE"), index=True)
    cidr: Mapped[str] = mapped_column(CIDR)


class DeviceType(Base):
    __tablename__ = "device_types"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    is_default: Mapped[bool] = mapped_column(Boolean, default=False)


class Device(Base):
    __tablename__ = "devices"
    __table_args__ = (UniqueConstraint("organization_id", "name"),)
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255))
    device_type_id: Mapped[int] = mapped_column(ForeignKey("device_types.id"))
    organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True)
    mac: Mapped[str] = mapped_column(String(17), default="")
    note: Mapped[str] = mapped_column(Text, default="")


class Address(Base):
    __tablename__ = "addresses"
    # vrf_id дублирует VRF префикса (составной FK, обновляется каскадом при переносе): так БД гарантирует уникальность IP в VRF
    __table_args__ = (
        UniqueConstraint("prefix_id", "address"),
        UniqueConstraint("vrf_id", "address", name="uq_addresses_vrf_address"),
        ForeignKeyConstraint(["prefix_id", "vrf_id"], ["prefixes.id", "prefixes.vrf_id"], name="fk_addresses_prefix_vrf", ondelete="CASCADE", onupdate="CASCADE"),
    )
    id: Mapped[int] = mapped_column(primary_key=True)
    prefix_id: Mapped[int] = mapped_column(index=True)
    vrf_id: Mapped[int] = mapped_column()
    address: Mapped[str] = mapped_column(INET)
    status: Mapped[AddressStatus] = mapped_column(Enum(AddressStatus, name="address_status"), default=AddressStatus.assigned)
    dns_name: Mapped[str] = mapped_column(String(255), default="")
    description: Mapped[str] = mapped_column(String(500), default="")
    device_id: Mapped[int | None] = mapped_column(ForeignKey("devices.id", ondelete="SET NULL"), index=True)
    note: Mapped[str] = mapped_column(Text, default="")
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())


class User(Base):
    __tablename__ = "users"
    __table_args__ = (
        # изменение 032: superadmin без организации, admin/viewer с организацией или неактивны
        CheckConstraint("is_active = false OR (role = 'superadmin') = (organization_id IS NULL)", name="ck_users_role_org_scope"),  # изменение 033, находка №13
    )
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(100), unique=True)
    password_hash: Mapped[str] = mapped_column(String(255))
    role: Mapped[Role] = mapped_column(Enum(Role, name="user_role"), default=Role.viewer)
    organization_id: Mapped[int | None] = mapped_column(ForeignKey("organizations.id"), index=True)  # изменение 032: NK для admin/viewer, NULL для superadmin
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    password_changed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # токены с iat раньше — недействительны
    last_login_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # изменение 035: время последнего успешного входа
    last_login_ip: Mapped[str | None] = mapped_column(INET)  # изменение 035: IP последнего успешного входа
    # изменение 037: двухфакторная аутентификация TOTP, по выбору пользователя
    totp_secret_enc: Mapped[str | None] = mapped_column(Text)  # зашифрованный (Fernet, TOTP_ENC_KEY) подтверждённый секрет
    totp_pending_enc: Mapped[str | None] = mapped_column(Text)  # секрет на время настройки, до подтверждения кодом
    totp_enabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))  # NULL — 2FA выключена
    totp_last_step: Mapped[int | None] = mapped_column(BigInteger)  # защита от повторного использования кода: шаг должен строго возрастать

    @property
    def totp_enabled(self) -> bool:  # изменение 037: вычисляемое поле для UserOut (from_attributes читает его как обычный атрибут)
        return self.totp_enabled_at is not None


Index("uq_users_lower_username", func.lower(User.username), unique=True)  # логин уникален без учёта регистра


class AuditLog(Base):
    __tablename__ = "audit_log"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True)
    username: Mapped[str] = mapped_column(String(100))
    entity_type: Mapped[str] = mapped_column(String(50), index=True)
    entity_id: Mapped[int | None] = mapped_column(Integer)
    entity_label: Mapped[str] = mapped_column(String(255))
    action: Mapped[str] = mapped_column(String(20))
    diff: Mapped[dict | None] = mapped_column(JSONB)
    uid: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), server_default=text("gen_random_uuid()"), unique=True)
    message: Mapped[str] = mapped_column(Text, default="", server_default="")
    client_ip: Mapped[str | None] = mapped_column(INET, index=True)  # IP клиента запроса; NULL для системных событий
    meta: Mapped[dict | None] = mapped_column(JSONB)  # user_agent, method, path, request_id
    organization_id: Mapped[int | None] = mapped_column(ForeignKey("organizations.id", ondelete="SET NULL"), index=True)  # изменение 032: для событий по сущностям организации, NULL для системных; ondelete — изменение 033, находка №1

    __table_args__ = (Index("ix_audit_log_entity_type_action", "entity_type", "action"),)


class AppSetting(Base):
    __tablename__ = "app_settings"
    key: Mapped[str] = mapped_column(String(50), primary_key=True)
    value: Mapped[dict] = mapped_column(JSONB)


class LoginAttempt(Base):
    """Неудачные попытки входа (для ограничения перебора паролей); записи старше суток удаляет ротация."""
    __tablename__ = "login_attempts"
    id: Mapped[int] = mapped_column(primary_key=True)
    ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    client_ip: Mapped[str | None] = mapped_column(INET)
    username: Mapped[str] = mapped_column(String(100))  # в нижнем регистре
    __table_args__ = (Index("ix_login_attempts_ip_ts", "client_ip", "ts"), Index("ix_login_attempts_user_ts", "username", "ts"))


class KnownLogin(Base):
    """IP, с которых логин уже успешно входил (last_seen обновляется при каждом входе); запись не старше
    KNOWN_IP_DAYS исключает логин из общей блокировки «по логину со всех IP» (изменение 026, находка №2).
    Записи старше срока удаляет ротация."""
    __tablename__ = "known_logins"
    username: Mapped[str] = mapped_column(String(100), primary_key=True)  # в нижнем регистре, как в login_attempts
    client_ip: Mapped[str] = mapped_column(INET, primary_key=True)
    last_seen: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())


class ClearAttempt(Base):
    """Неудачные попытки подтверждения пароля при очистке журнала (для блокировки)."""
    __tablename__ = "clear_attempts"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
    ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())


class RecoveryCode(Base):
    """Коды восстановления 2FA (изменение 037): выдаются 10 штук при включении, каждый гасится один раз.
    Хэш — sha256 (не argon2): код высокоэнтропийный (10 символов base32), медленный хэш не нужен."""
    __tablename__ = "recovery_codes"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
    code_hash: Mapped[str] = mapped_column(String(64))
    used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))