fix: сканер audit_event ловит псевдонимы импорта, __table__.update()/delete() и Query API

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api/tests/test_audit_service.py

@@ -60,44 +60,79 @@ def _call_name(node: ast.Call) -> str | None:
     return None
 
 
-def _mentions_audit_event(node: ast.AST) -> bool:
+def _audit_event_aliases(tree: ast.AST) -> set[str]:
+    """Имена AuditEvent в модуле: сам `AuditEvent`, `from ... import AuditEvent as AE`, `AE2 = AE` (до фикспойнта)."""
+    aliases = {"AuditEvent"}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom):
+            aliases.update(alias.asname or alias.name for alias in node.names if alias.name == "AuditEvent")
+    changed = True
+    while changed:
+        changed = False
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Assign) and _mentions(node.value, aliases) and not isinstance(node.value, ast.Call):
+                for target in node.targets:
+                    if isinstance(target, ast.Name) and target.id not in aliases:
+                        aliases.add(target.id)
+                        changed = True
+    return aliases
+
+
+def _mutating_names(tree: ast.AST) -> set[str]:
+    """`update`/`delete` и их импортные псевдонимы (`from sqlalchemy import update as upd`)."""
+    names = set(MUTATING_CALLS)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom):
+            names.update(alias.asname for alias in node.names if alias.name in MUTATING_CALLS and alias.asname)
+    return names
+
+
+def _mentions(node: ast.AST, aliases: set[str]) -> bool:
     return any(
-        (isinstance(sub, ast.Name) and sub.id == "AuditEvent")
+        (isinstance(sub, ast.Name) and sub.id in aliases) or (isinstance(sub, ast.Attribute) and sub.attr in aliases)
-        or (isinstance(sub, ast.Attribute) and sub.attr == "AuditEvent")
         for sub in ast.walk(node)
     )
 
 
-def _is_audit_event_expr(node: ast.AST | None) -> bool:
+def _is_audit_event_expr(node: ast.AST | None, aliases: set[str]) -> bool:
-    """Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)`."""
+    """Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)` / `AE(...)`."""
-    return isinstance(node, ast.Call) and _mentions_audit_event(node.func)
+    return isinstance(node, ast.Call) and _mentions(node.func, aliases)
 
 
-def _tainted_names(tree: ast.AST) -> set[str]:
+def _tainted_names(tree: ast.AST, aliases: set[str]) -> set[str]:
     """Имена, связанные с экземплярами AuditEvent: присваивание `AuditEvent(...)`, аннотация `: AuditEvent`."""
     names: set[str] = set()
     for node in ast.walk(tree):
-        if isinstance(node, ast.Assign) and _is_audit_event_expr(node.value):
+        if isinstance(node, ast.Assign) and _is_audit_event_expr(node.value, aliases):
             names.update(t.id for t in node.targets if isinstance(t, ast.Name))
         elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
-            if _mentions_audit_event(node.annotation) or _is_audit_event_expr(node.value):
+            if _mentions(node.annotation, aliases) or _is_audit_event_expr(node.value, aliases):
                 names.add(node.target.id)
-        elif isinstance(node, ast.arg) and node.annotation is not None and _mentions_audit_event(node.annotation):
+        elif isinstance(node, ast.arg) and node.annotation is not None and _mentions(node.annotation, aliases):
             names.add(node.arg)
-        elif isinstance(node, ast.For | ast.comprehension) and _is_audit_event_expr(getattr(node, "iter", None)):
-            pass
     return names
 
 
 def find_audit_event_mutations(source: str, filename: str = "<src>") -> list[str]:
     """Находки вида `file:line: описание` — любое изменение/удаление audit_event в исходнике.
 
-    Ловит: `update(AuditEvent...)` / `delete(AuditEvent...)` (в т.ч. `sa.update`, `AuditEvent.__table__`);
+    Ловит (best-effort, статически, в пределах одного модуля):
-    `.delete(x)`/`delete(x)`, где x — экземпляр AuditEvent (внутримодульное отслеживание имён); присваивание/удаление
+    - `update(AuditEvent...)` / `delete(AuditEvent...)`, в т.ч. `sa.update`, `update as upd`, `AuditEvent.__table__`
-    атрибутов таких экземпляров и `AuditEvent.<поле> = ...`; сырой SQL UPDATE/DELETE/TRUNCATE по audit_event.
+      как аргумент; псевдонимы `from ... import AuditEvent as AE` и `AE = AuditEvent`;
+    - вызов `.update()`/`.delete()`, где приёмник упоминает AuditEvent: `AuditEvent.__table__.update()`,
+      `session.query(AuditEvent).update(...)/.delete()`;
+    - `.delete(x)`/`delete(x)`, где x — экземпляр AuditEvent (имя связано с `AuditEvent(...)` или аннотацией);
+    - присваивание/удаление атрибутов таких экземпляров и `AuditEvent.<поле> = ...`;
+    - сырой SQL UPDATE/DELETE/TRUNCATE по audit_event в строковой константе.
+
+    НЕ ловит: объекты, загруженные из БД без аннотации (`obj = (await s.execute(select(...))).scalar_one()`),
+    `setattr`, `session.merge()`, SQL, собранный конкатенацией/f-строкой, межмодульный поток типов. Защита на уровне
+    БД (триггер/REVOKE UPDATE, DELETE) в Срез 0 не входит.
     """
     tree = ast.parse(source, filename=filename)
-    tainted = _tainted_names(tree)
+    aliases = _audit_event_aliases(tree)
+    tainted = _tainted_names(tree, aliases)
+    mutating = _mutating_names(tree)
     findings: list[str] = []
 
     def add(node: ast.AST, message: str) -> None:
@@ -107,9 +142,12 @@ def find_audit_event_mutations(source: str, filename: str = "<src>") -> list[str
         if isinstance(node, ast.Call):
             name = _call_name(node)
             args: list[ast.AST] = [*node.args, *(kw.value for kw in node.keywords)]
-            if name in MUTATING_CALLS:
+            if name in mutating:
-                if any(_mentions_audit_event(arg) for arg in args):
+                receiver = node.func.value if isinstance(node.func, ast.Attribute) else None
+                if any(_mentions(arg, aliases) for arg in args):
                     add(node, f"{name}() по AuditEvent")
+                elif receiver is not None and _mentions(receiver, aliases):
+                    add(node, f".{name}() на выражении с AuditEvent")
                 elif any(isinstance(arg, ast.Name) and arg.id in tainted for arg in args):
                     add(node, f"{name}() экземпляра AuditEvent")
         elif isinstance(node, ast.Assign | ast.AugAssign | ast.AnnAssign | ast.Delete):
@@ -118,8 +156,8 @@ def find_audit_event_mutations(source: str, filename: str = "<src>") -> list[str
                 if isinstance(target, ast.Attribute) and target.attr in AUDIT_FIELDS:
                     base = target.value
                     if (
-                        isinstance(base, ast.Name) and (base.id in tainted or base.id == "AuditEvent")
+                        isinstance(base, ast.Name) and (base.id in tainted or base.id in aliases)
-                    ) or _is_audit_event_expr(base):
+                    ) or _is_audit_event_expr(base, aliases):
                         add(node, f"изменение поля {target.attr} у AuditEvent")
         elif isinstance(node, ast.Constant) and isinstance(node.value, str) and RAW_SQL_MUTATION.search(node.value):
             add(node, "сырой SQL, изменяющий audit_event")
@@ -154,6 +192,20 @@ def test_src_has_no_audit_event_update_or_delete() -> None:
         "stmt = text('UPDATE dkb.audit_event SET comment = 1')",
         "stmt = text('delete from audit_event')",
         "stmt = text('TRUNCATE TABLE dkb.audit_event')",
+        # обходы через псевдонимы
+        "from src.models.audit_ex import AuditEvent as AE\nstmt = delete(AE)",
+        "from src.models.audit_ex import AuditEvent as AE\nstmt = sa.update(AE).values(result='x')",
+        "from src.models.audit_ex import AuditEvent as AE\nAE.comment = 'x'",
+        "from src.models.audit_ex import AuditEvent as AE\ndef f(s):\n    e = AE(object_type='x')\n    s.delete(e)",
+        "AE = AuditEvent\nstmt = update(AE)",
+        "from sqlalchemy import update as upd\nstmt = upd(AuditEvent)",
+        # __table__ как приёмник
+        "stmt = AuditEvent.__table__.update().values(result='x')",
+        "stmt = AuditEvent.__table__.delete()",
+        # устаревший Query API
+        "def f(s):\n    s.query(AuditEvent).update({'result': 'x'})",
+        "def f(s):\n    s.query(AuditEvent).filter(AuditEvent.id == 1).delete()",
+        "def f(s):\n    from src.models.audit_ex import AuditEvent as AE\n    s.query(AE).delete()",
     ],
 )
 def test_scanner_detects_violation(source: str) -> None:
@@ -169,6 +221,13 @@ def test_scanner_detects_violation(source: str) -> None:
         "async def f(session, office):\n    await session.delete(office)",
         "def f(office):\n    office.comment = 'x'",
         "stmt = text('select * from audit_event')",
+        "from src.models.audit_ex import AuditEvent as AE\nstmt = select(AE).where(AE.id == 1)",
+        "stmt = select(AuditEvent.__table__)",
+        "def f(s):\n    return s.query(AuditEvent).filter(AuditEvent.id == 1).all()",
+        "def f(s):\n    s.query(RegionalOffice).update({'name': 'x'})\n    s.query(RegionalOffice).delete()",
+        "async def f(session):\n    await AuditService(session).record(action=A, section=S, object_type='x')",
+        "from src.services.audit import AuditService\nservice = AuditService(session)\nservice.update_cache()",
+        "def f(d):\n    d.update({'a': 1})\n    d.get('x').update({})",
     ],
 )
 def test_scanner_passes_allowed_code(source: str) -> None:
@@ -282,6 +341,8 @@ async def _count_offices(session: AsyncSession, name: str) -> int:
 
 @pytest.mark.integration
 async def test_system_event_without_user(db_session: AsyncSession) -> None:
+    # action/section выбраны произвольно: тест проверяет user_id=NULL и jsonb, а не предметный сценарий (по R-7
+    # обмен recognition_job audit_event в Срезе 0 не пишет)
     before = datetime.datetime.now(datetime.UTC)
     event = await AuditService(db_session).record(
         action=AuditAction.RECOGNIZED_DATA_EDIT,