ayurishchevandClaude Opus 5.5
d3495fb077
Пользователь сам включает TOTP (QR, приложение-аутентификатор); 10 одноразовых кодов восстановления; superadmin сбрасывает 2FA другому пользователю. Вход в два шага: /auth/login → mfa_token, /auth/login/2fa — под теми же лимитами и advisory-lock, что пароль. Секрет шифруется Fernet-ключом TOTP_ENC_KEY, защита от повторного использования кода (totp_last_step), лимиты перебора кода при отключении. Миграция 0015, зависимости pyotp, segno, cryptography. UI: второй шаг входа, диалоги 2FA в меню логина, бейдж и сброс в «Пользователях»; submitDialog не закрывает окно, открытое следующим шагом цепочки. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
229 lines
13 KiB
Python
229 lines
13 KiB
Python
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))
|