Задачи 025-030: ёмкость префиксов, политика входа, дерево префиксов

Повторный анализ кодовой базы (docs/reviews/2026-09-26-codebase-review-2.md) и доработки:
025 Ёмкость префикса — размер его подсети (а не сумма листьев); «Обзор» считает ёмкость
    по корневым активным IPv4-префиксам и адреса внутри них.
026 Политика блокировки входа: 5 неудач на логин+IP, 20 на IP, 50 на логин со всех IP,
    кроме известных IP (known_logins, миграция 0009) — владельца нельзя заблокировать анонимно.
027 Сериализация попыток входа по IP (advisory-lock после блокировки логина).
028 UI «Префиксы»: загрузка всех страниц (до 20 000), счётчики по total, предупреждение об усечении.
029 Advisory-lock по VRF для операций, меняющих дерево префиксов и раскладку адресов.
030 Исправление замечаний ревью 025-029: _lock_prefix (VRF блокируется до чтения префикса,
    409 при одновременном переносе), константы политики входа перенесены в app/services.py.

Тесты: 14 passed (проверка ёмкости родителя приведена к семантике 025); сквозные сценарии
и гонки — docs/reviews/2026-09-26-changes-025-029-review.md, 2026-09-27-changes-030-review.md.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/api/v1/auth.py

@@ -1,42 +1,61 @@
-"""Вход в UI. Перебор паролей ограничен: {MAX_PER_LOGIN} неудач на логин и {MAX_PER_IP} на IP за {WINDOW_MIN} минут → 429 (изменение 012)."""
+"""Вход в UI. Перебор паролей ограничен тремя областями (окно {WINDOW_MIN} минут → 429, изменения 012, 026, 027):
+{MAX_PER_LOGIN_IP} неудач на пару логин+IP, {MAX_PER_IP} на IP (любые логины), {MAX_PER_LOGIN} на логин со всех IP, кроме «известных»
+(IP, с которого этот пользователь успешно входил за последние {KNOWN_IP_DAYS} дн.) — иначе анонимный клиент, знающий логин,
+мог бы держать чужую учётную запись заблокированной с любого IP (находка №2 ревью 2026-09-26)."""
 from datetime import datetime, timedelta, timezone
 
 from fastapi import APIRouter, Depends, HTTPException
-from sqlalchemy import delete, func, select
+from sqlalchemy import and_, delete, func, select
+from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.orm import Session
 
 from app.db import get_db
-from app.models import LoginAttempt, User
+from app.models import KnownLogin, LoginAttempt, User
 from app.request_context import request_meta
 from app.schemas import LoginIn, TokenOut, UserOut
 from app.security import create_token, current_user, hash_password, verify_password
-from app.services import ANONYMOUS, audit
+from app.services import ANONYMOUS, KNOWN_IP_DAYS, LOGIN_WINDOW, MAX_PER_IP, MAX_PER_LOGIN, MAX_PER_LOGIN_IP, audit
 
 router = APIRouter(prefix="/auth", tags=["auth"])
-WINDOW = timedelta(minutes=10)
-MAX_PER_LOGIN = 5
-MAX_PER_IP = 20
-__doc__ = __doc__.format(MAX_PER_LOGIN=MAX_PER_LOGIN, MAX_PER_IP=MAX_PER_IP, WINDOW_MIN=int(WINDOW.total_seconds() // 60))
+# пороги перебора (LOGIN_WINDOW, MAX_PER_*, KNOWN_IP_DAYS) — в app.services (изменение 030, ревью 025-029 находка №2)
+# пространства ключей двухаргументного advisory-lock (не пересекаются с одноаргументными: LOCK_KEY/USERS_ADMIN_LOCK)
+LOGIN_LOCK_NS = 7031
+IP_LOCK_NS = 7032
+__doc__ = __doc__.format(MAX_PER_LOGIN_IP=MAX_PER_LOGIN_IP, MAX_PER_IP=MAX_PER_IP, MAX_PER_LOGIN=MAX_PER_LOGIN,
+                          WINDOW_MIN=int(LOGIN_WINDOW.total_seconds() // 60), KNOWN_IP_DAYS=KNOWN_IP_DAYS)
 _DUMMY_HASH = hash_password("dummy-password-for-timing")  # выравнивает время ответа для несуществующего логина
 
 
 def _condition(login: str, ip: str | None, scope: str):
-    return (LoginAttempt.username == login) if scope == "login" else (LoginAttempt.client_ip == ip)
+    if scope == "login_ip":
+        return and_(LoginAttempt.username == login, LoginAttempt.client_ip == ip)
+    if scope == "ip":
+        return LoginAttempt.client_ip == ip
+    return LoginAttempt.username == login  # "login" — по всем IP
 
 
-def _retry_after(db: Session, login: str, ip: str | None) -> tuple[int, str | None, dict[str, int]]:
-    """(секунд до конца блокировки, причина 'login'|'ip', {"login": n, "ip": m} — число неудач в каждой области, до лимита).
-    Блокировка длится окно после последней неудачи. Счётчики по областям отдельно (изменение 024, находка №4):
-    решение о записи в журнал принимается по конкретному логину, а не по максимуму среди логина и IP."""
+def _is_known(db: Session, login: str, ip: str | None) -> bool:
+    """IP считается известным для логина, если вход с него был успешен не более KNOWN_IP_DAYS дней назад."""
+    if ip is None:
+        return False
+    cutoff = datetime.now(timezone.utc) - timedelta(days=KNOWN_IP_DAYS)
+    return db.scalar(select(KnownLogin.username).where(KnownLogin.username == login, KnownLogin.client_ip == ip, KnownLogin.last_seen > cutoff)) is not None
+
+
+def _retry_after(db: Session, login: str, ip: str | None, known: bool) -> tuple[int, str | None, dict[str, int]]:
+    """(секунд до конца блокировки, область 'login_ip'|'ip'|'login', счётчики по каждой области, до лимита).
+    Блокировка длится окно после последней неудачи. Область 'login' не блокирует, если IP запроса известен
+    (изменение 026, находка №2); 'login_ip' и 'ip' действуют всегда. Счётчики считаются отдельно (изменение 024,
+    находка №4): решение о записи в журнал принимается по конкретной области, а не по максимуму среди них."""
     now = datetime.now(timezone.utc)
     worst, why, counts = 0, None, {}
-    for scope, limit in (("login", MAX_PER_LOGIN), ("ip", MAX_PER_IP)):
-        if scope == "ip" and ip is None:
+    for scope, limit in (("login_ip", MAX_PER_LOGIN_IP), ("ip", MAX_PER_IP), ("login", MAX_PER_LOGIN)):
+        if scope in ("login_ip", "ip") and ip is None:
             continue
-        rows = db.scalars(select(LoginAttempt.ts).where(_condition(login, ip, scope), LoginAttempt.ts > now - WINDOW).order_by(LoginAttempt.ts.desc()).limit(limit)).all()
+        rows = db.scalars(select(LoginAttempt.ts).where(_condition(login, ip, scope), LoginAttempt.ts > now - LOGIN_WINDOW).order_by(LoginAttempt.ts.desc()).limit(limit)).all()
         counts[scope] = len(rows)
-        if len(rows) >= limit:
-            left = int((rows[0] + WINDOW - now).total_seconds()) + 1
+        if len(rows) >= limit and not (scope == "login" and known):
+            left = int((rows[0] + LOGIN_WINDOW - now).total_seconds()) + 1
             if left > worst:
                 worst, why = left, scope
     return worst, why, counts
@@ -45,7 +64,15 @@ def _retry_after(db: Session, login: str, ip: str | None) -> tuple[int, str | No
 def _distinct_logins(db: Session, ip: str) -> int:
     """Число различных логинов, для которых была неудачная попытка с этого IP в окне (для diff записи session.locked)."""
     now = datetime.now(timezone.utc)
-    return db.scalar(select(func.count(func.distinct(LoginAttempt.username))).where(LoginAttempt.client_ip == ip, LoginAttempt.ts > now - WINDOW)) or 0
+    return db.scalar(select(func.count(func.distinct(LoginAttempt.username))).where(LoginAttempt.client_ip == ip, LoginAttempt.ts > now - LOGIN_WINDOW)) or 0
+
+
+def _remember_login(db: Session, login: str, ip: str | None) -> None:
+    """Отмечает IP как известный для логина (upsert last_seen); вызывается при успешном входе."""
+    if ip is None:
+        return
+    stmt = insert(KnownLogin).values(username=login, client_ip=ip, last_seen=func.now())
+    db.execute(stmt.on_conflict_do_update(index_elements=[KnownLogin.username, KnownLogin.client_ip], set_={"last_seen": func.now()}))
 
 
 def _locked(retry: int) -> HTTPException:
@@ -58,27 +85,34 @@ def login(body: LoginIn, db: Session = Depends(get_db)):
     ctx = request_meta.get()
     ip = ctx["client_ip"] if ctx else None
     name = body.username.strip().lower()
-    # сериализация попыток одного логина (изменение 024, находка №1): без этого параллельные запросы
-    # проходят проверку блокировки одновременно, и лимит «N за окно» превращается в «N + степень параллелизма»
-    db.execute(select(func.pg_advisory_xact_lock(func.hashtext(name))))
-    retry, _, _ = _retry_after(db, name, ip)
+    # сериализация попыток одного логина (изменение 024, находка №1), затем — этого IP (изменение 027, находка №7):
+    # порядок всегда «логин, затем IP» исключает взаимную блокировку; без второй блокировки параллельный перебор
+    # разных логинов с одного IP проходит проверку лимита по IP одновременно, и лимит превышается на степень
+    # параллелизма — ценой служит то, что попытки с одного IP (в том числе за NAT) обрабатываются по одной,
+    # и при массовом входе время ответа растёт на время проверки argon2.
+    db.execute(select(func.pg_advisory_xact_lock(LOGIN_LOCK_NS, func.hashtext(name))))
+    if ip is not None:
+        db.execute(select(func.pg_advisory_xact_lock(IP_LOCK_NS, func.hashtext(ip))))
+    known = _is_known(db, name, ip)
+    retry, _, _ = _retry_after(db, name, ip, known)
     if retry:  # блокировка: пароль не проверяем, в журнал не пишем (запись о блокировке уже есть)
         raise _locked(retry)
     user = db.scalar(select(User).where(User.username == body.username, User.is_active))
     if user is None:
         verify_password(body.password, _DUMMY_HASH)
     if user is None or not verify_password(body.password, user.password_hash):
-        _, _, before = _retry_after(db, name, ip)
+        _, _, before = _retry_after(db, name, ip, known)
         db.add(LoginAttempt(client_ip=ip, username=name))
         db.flush()
-        retry, why, after = _retry_after(db, name, ip)
+        retry, why, after = _retry_after(db, name, ip, known)
         if retry and before[why] < after[why]:  # именно этот запрос впервые пересёк лимит — запись пишем один раз
             attempts = after[why]
             diff = {"scope": why, "attempts": attempts, "retry_after_seconds": retry}
             if why == "ip":
                 diff["distinct_logins"] = _distinct_logins(db, ip)
+            scope_ru = {"login_ip": "по логину и IP", "ip": "с IP", "login": "по логину"}[why]
             audit(db, ANONYMOUS, "session", None, "locked", body.username[:100], diff,
-                  message=f"Вход заблокирован на {max(1, -(-retry // 60))} мин.: {attempts} неудачных попыток ({'по логину' if why == 'login' else 'с IP'})")
+                  message=f"Вход заблокирован на {max(1, -(-retry // 60))} мин.: {attempts} неудачных попыток ({scope_ru})")
         elif before["login"] == 0:  # в журнал — только первая неудача по этому логину в окне (не по IP: иначе перебор логинов с одного IP её не оставит)
             audit(db, ANONYMOUS, "session", None, "failed", body.username[:100], message="Неудачная попытка входа в UI")
         db.commit()
@@ -86,6 +120,7 @@ def login(body: LoginIn, db: Session = Depends(get_db)):
             raise _locked(retry)
         raise HTTPException(401, "Неверный логин или пароль")
     db.execute(delete(LoginAttempt).where(LoginAttempt.username == name))
+    _remember_login(db, name, ip)
     audit(db, user, "session", None, "login", user.username, message=f"Вход в UI: {user.username}")
     db.commit()
     return TokenOut(access_token=create_token(user))

app/api/v1/overview.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Depends, Query
 from pydantic import BaseModel
-from sqlalchemy import func, select
+from sqlalchemy import and_, func, select
 from sqlalchemy.orm import Session
 
 from app import schemas as s
@@ -24,18 +24,40 @@ class Overview(BaseModel):
     recent_changes: list[s.AuditOut]
 
 
+def _ipv4_roots(db: Session):
+    """Активные IPv4-префиксы, не вложенные ни в один другой активный IPv4-префикс того же VRF (по CIDR, не по
+    parent_id — дерево может включать неактивные ветки). Основа ёмкости и загрузки «Обзора» (изменение 025, находка №1):
+    раньше ёмкость считалась суммой листьев, а назначенные адреса — по всем IPv4-адресам, включая нелистовые префиксы."""
+    q = Prefix.__table__.alias("q")
+    contained = (
+        select(1).select_from(q)
+        .where(q.c.vrf_id == Prefix.vrf_id, q.c.id != Prefix.id, q.c.status == PrefixStatus.active,
+               func.family(q.c.prefix) == 4, q.c.prefix.op(">>")(Prefix.prefix))
+        .exists()
+    )
+    return select(Prefix.vrf_id, Prefix.prefix).where(Prefix.status == PrefixStatus.active, func.family(Prefix.prefix) == 4, ~contained)
+
+
 @router.get("/overview", response_model=Overview, tags=["overview"])
 def overview(db: Session = Depends(get_db)):
-    prefixes = db.scalars(select(Prefix).where(Prefix.status == PrefixStatus.active)).all()
-    outs = _prefix_outs(db, list(prefixes))
+    active = db.scalars(select(Prefix).where(Prefix.status == PrefixStatus.active)).all()
+    outs = _prefix_outs(db, list(active))
     parents = {p.parent_id for p in outs if p.parent_id}
-    leaves = [p for p in outs if p.id not in parents]  # ёмкость считаем по листьям, чтобы не удваивать
-    leaves4 = [p for p in leaves if p.family == 4]  # IPv6-пространство несопоставимо по размеру — в метрики использования не входит
-    cap = sum(p.capacity for p in leaves4)
-    v4 = func.family(Address.address) == 4
-    assigned = db.scalar(select(func.count()).select_from(Address).where(Address.status == AddressStatus.assigned, v4)) or 0
-    reserved = db.scalar(select(func.count()).select_from(Address).where(Address.status == AddressStatus.reserved, v4)) or 0
+    leaves4 = [p for p in outs if p.id not in parents and p.family == 4]  # top_prefixes — как раньше, по листьям
     top = sorted((p for p in leaves4 if p.capacity > 1), key=lambda p: p.utilization, reverse=True)[:4]
+
+    roots_stmt = _ipv4_roots(db)
+    roots = db.execute(roots_stmt).all()
+    cap = sum(capacity(str(r.prefix)) for r in roots)
+    roots_cte = roots_stmt.cte("overview_roots")
+    counts = dict(db.execute(
+        select(Address.status, func.count(func.distinct(Address.id)))
+        .select_from(Address.__table__.join(roots_cte, and_(roots_cte.c.vrf_id == Address.vrf_id, Address.address.op("<<=")(roots_cte.c.prefix))))
+        .where(func.family(Address.address) == 4, Address.status.in_((AddressStatus.assigned, AddressStatus.reserved)))
+        .group_by(Address.status)
+    ).all())
+    assigned = counts.get(AddressStatus.assigned, 0)
+    reserved = counts.get(AddressStatus.reserved, 0)
     recent = db.scalars(select(AuditLog).order_by(AuditLog.id.desc()).limit(4)).all()
     return Overview(
         prefixes=db.scalar(select(func.count()).select_from(Prefix)) or 0,

app/api/v1/prefixes.py

@@ -9,12 +9,43 @@ from app.db import get_db
 from app.models import Address, AddressStatus, Device, Organization, Prefix, PrefixStatus, User, Vrf
 from app.security import admin_user, current_user
 from app.services import (
-    MAX_CAPACITY, MAX_OFFSET, apply_update, audit, blockers, capacity, commit, count, flush, contains, free_page, get_or_404, network_role, next_free_address, next_free_subnet, refuse_delete,
+    MAX_OFFSET, apply_update, audit, blockers, capacity, commit, count, flush, contains, free_page, get_or_404, network_role, next_free_address, next_free_subnet, refuse_delete,
     utilization,
 )
 
 router = APIRouter(dependencies=[Depends(current_user)], tags=["prefixes"])
 FREE_LISTING_LIMIT = 4096  # «свободные» строки показываем, только если подсеть не больше /20
+VRF_LOCK_NS = 7033  # advisory-lock пространство: изменения дерева/раскладки адресов одного VRF выполняются по одному (изменение 029, находка №5)
+
+
+def _lock_vrf(db: Session, *vrf_ids: int) -> None:
+    """Сериализует операции над деревом префиксов и раскладкой адресов одного VRF: без этого параллельные запросы
+    (например, ручное создание и автовыделение в одном родителе) читают дерево по разным снимкам транзакции и могут
+    оставить неверный parent_id. Несколько VRF (перенос между ними) блокируются в порядке возрастания id — исключает
+    взаимную блокировку. Вызывается до любых чтений дерева этого VRF; читатели (GET) не блокируются."""
+    for vrf_id in sorted(set(vrf_ids)):
+        db.execute(select(func.pg_advisory_xact_lock(VRF_LOCK_NS, vrf_id)))
+
+
+def _lock_prefix(db: Session, id: int, *extra_vrf_ids: int) -> Prefix:
+    """Читает и блокирует префикс вместе с деревом его VRF, устраняя гонку «vrf_id прочитан до блокировки»
+    (изменение 030, ревью 025-029 находка №1): `vrf_id` читается отдельным SELECT, затем блокируется через
+    `_lock_vrf` (плюс `extra_vrf_ids` — например, целевой VRF при переносе, одним вызовом), и только потом
+    префикс перечитывается под `FOR UPDATE` с `populate_existing`, чтобы не взять устаревший объект из identity map.
+    Если под блокировкой `vrf_id` оказался другим (префикс успели перенести в другой VRF) — 409, без повтора:
+    повтор накопил бы в транзакции лишнюю блокировку устаревшего VRF, и порядок захвата между попытками перестал
+    бы быть строго возрастающим (риск взаимной блокировки с зеркальной операцией). Транзакция откатится при
+    закрытии сессии, и клиент повторяет запрос с нуля, без унаследованных блокировок."""
+    vrf_id = db.scalar(select(Prefix.vrf_id).where(Prefix.id == id))
+    if vrf_id is None:
+        raise HTTPException(404, "Префикс не найден")
+    _lock_vrf(db, vrf_id, *extra_vrf_ids)
+    p = db.scalar(select(Prefix).where(Prefix.id == id).with_for_update().execution_options(populate_existing=True))
+    if p is None:
+        raise HTTPException(404, "Префикс не найден")
+    if p.vrf_id != vrf_id:
+        raise HTTPException(409, "Префикс одновременно изменяется, повторите запрос")
+    return p
 
 
 # -------------------------------------------------------------------- prefixes
@@ -50,36 +81,19 @@ def _depths(db: Session, organization_id: int) -> dict[int, int]:
     return {i: depth(i) for i in parents}
 
 
-def _capacities(db: Session, organization_id: int) -> dict[int, int]:
-    """Ёмкость листа — размер подсети; ёмкость родителя — сумма ёмкостей вложенных листьев."""
-    rows = db.execute(select(Prefix.id, Prefix.parent_id, Prefix.prefix).where(Prefix.organization_id == organization_id)).all()
-    kids: dict[int, list[int]] = {}
-    for i, parent, _ in rows:
-        if parent is not None:
-            kids.setdefault(parent, []).append(i)
-    own = {i: capacity(str(cidr)) for i, _, cidr in rows}
-    memo: dict[int, int] = {}
-
-    def cap(i: int) -> int:
-        if i not in memo:
-            memo[i] = sum(cap(k) for k in kids[i]) if i in kids else own[i]
-        return memo[i]
-
-    return {i: min(cap(i), MAX_CAPACITY) for i in own}
-
-
 def _prefix_outs(db: Session, rows: list[Prefix]) -> list[s.PrefixOut]:
+    """Ёмкость префикса — размер его собственной подсети (изменение 025, находка №1): раньше ёмкость родителя
+    считалась суммой ёмкостей вложенных листьев, и после частичного разбиения (изменение 010) расходилась
+    с экраном адресов того же префикса."""
     usage = _usage(db, [r.id for r in rows])
     depths: dict[int, int] = {}
-    caps: dict[int, int] = {}
     vrf_names = dict(db.execute(select(Vrf.id, Vrf.name).where(Vrf.id.in_(list({r.vrf_id for r in rows})))).all()) if rows else {}
     for org_id in {r.organization_id for r in rows}:
         depths.update(_depths(db, org_id))
-        caps.update(_capacities(db, org_id))
     out = []
     for r in rows:
         used, stored = usage.get(r.id, (0, 0))
-        cap = caps.get(r.id, capacity(str(r.prefix)))
+        cap = capacity(str(r.prefix))
         out.append(s.PrefixOut(
             id=r.id, organization_id=r.organization_id, vrf_id=r.vrf_id, vrf_name=vrf_names[r.vrf_id],
             prefix=str(r.prefix), family=ipaddress.ip_network(str(r.prefix)).version,
@@ -216,6 +230,7 @@ def get_prefix(id: int, db: Session = Depends(get_db)):
 
 @router.post("/prefixes", response_model=s.PrefixOut, status_code=201)
 def create_prefix(body: s.PrefixIn, db: Session = Depends(get_db), user: User = Depends(admin_user)):
+    _lock_vrf(db, body.vrf_id)  # до любых чтений дерева VRF (изменение 029)
     vrf = get_or_404(db, Vrf, body.vrf_id, "VRF")
     if vrf.organization_id != body.organization_id:
         raise HTTPException(422, "VRF принадлежит другой организации")
@@ -272,9 +287,7 @@ def preview_subnet(id: int, length: int = Query(ge=1, le=128), db: Session = Dep
 @router.post("/prefixes/{id}/subnets/next", response_model=s.PrefixOut, status_code=201)
 def allocate_subnet(id: int, body: s.SubnetNextIn, db: Session = Depends(get_db), user: User = Depends(admin_user)):
     """Создаёт вложенный префикс заданного размера в первом свободном выровненном блоке родителя."""
-    parent = db.scalar(select(Prefix).where(Prefix.id == id).with_for_update())  # сериализуем параллельные выделения из одного родителя
-    if parent is None:
-        raise HTTPException(404, "Префикс не найден")
+    parent = _lock_prefix(db, id)  # до любых чтений дерева (изменение 029/030) — сериализует параллельные выделения из одного родителя
     found, _, _ = _find_subnet(db, parent, body.length)
     if found is None:
         raise HTTPException(409, f"В префиксе {parent.prefix} нет свободного блока /{body.length}")
@@ -291,9 +304,12 @@ def allocate_subnet(id: int, body: s.SubnetNextIn, db: Session = Depends(get_db)
 
 @router.patch("/prefixes/{id}", response_model=s.PrefixOut)
 def update_prefix(id: int, body: s.PrefixUpdate, db: Session = Depends(get_db), user: User = Depends(admin_user)):
-    p = get_or_404(db, Prefix, id, "Префикс")
     data = body.model_dump(exclude_unset=True, exclude_none=True)
     new_vrf = data.pop("vrf_id", None)
+    # целевой VRF известен из тела запроса до чтения префикса — блокируем исходный и целевой VRF одним вызовом,
+    # раньше чем прочитан текущий p.vrf_id (изменение 030, ревью 025-029 находка №1); если new_vrf совпадёт
+    # с текущим VRF, лишняя блокировка того же VRF безвредна
+    p = _lock_prefix(db, id, new_vrf) if new_vrf is not None else get_or_404(db, Prefix, id, "Префикс")
     changed = {k: str(v) for k, v in apply_update(p, data).items()}
     if new_vrf is not None and new_vrf != p.vrf_id:
         changed.update(_move_to_vrf(db, p, new_vrf)[0])
@@ -304,7 +320,7 @@ def update_prefix(id: int, body: s.PrefixUpdate, db: Session = Depends(get_db),
 
 @router.delete("/prefixes/{id}", status_code=204)
 def delete_prefix(id: int, force: bool = False, db: Session = Depends(get_db), user: User = Depends(admin_user)):
-    p = get_or_404(db, Prefix, id, "Префикс")
+    p = _lock_prefix(db, id)  # до переподвешивания детей: устаревший p.parent_id иначе достанется всем детям (изменение 029/030)
     used = None if force else blockers(db, select(func.host(Address.address)).where(Address.prefix_id == id).order_by(Address.address))
     if used:
         refuse_delete(db, user, "prefix", p, str(p.prefix), "В префиксе есть адреса; удалите их или используйте force=true", {"addresses": used})
@@ -377,7 +393,7 @@ def list_addresses(
 
 @router.post("/prefixes/{id}/addresses", response_model=s.AddressOut, status_code=201)
 def create_address(id: int, body: s.AddressIn, db: Session = Depends(get_db), user: User = Depends(admin_user)):
-    p = get_or_404(db, Prefix, id, "Префикс")
+    p = _lock_prefix(db, id)  # выбор самого узкого префикса должен видеть согласованное дерево VRF (изменение 029/030)
     net, ip = ipaddress.ip_network(str(p.prefix)), ipaddress.ip_address(body.address)
     if ip not in net:
         raise HTTPException(422, f"Адрес {body.address} не принадлежит префиксу {p.prefix}")
@@ -402,9 +418,7 @@ def allocate_next(
     id: int, body: s.AddressUpdate | None = None, db: Session = Depends(get_db), user: User = Depends(admin_user)
 ):
     """Автоназначение первого свободного адреса; только для префиксов с флагом is_pool."""
-    p = db.scalar(select(Prefix).where(Prefix.id == id).with_for_update())  # параллельные запросы получают разные адреса
-    if p is None:
-        raise HTTPException(404, "Префикс не найден")
+    p = _lock_prefix(db, id)  # занятые диапазоны (вложенные префиксы) должны читаться по согласованному дереву (изменение 029/030)
     if not p.is_pool:
         raise HTTPException(422, "Префикс не является пулом для автоназначения")
     ip = next_free_address(ipaddress.ip_network(str(p.prefix)), _busy_ranges(db, p))  # вложенные префиксы и адрес сети/broadcast пропускаются