From dcbeea575655be3cf4c91cd199570f55cec28a38 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sat, 3 Oct 2026 23:30:27 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20=D1=81=D0=BA=D0=B0=D0=BD=D0=B5=D1=80=20a?= =?UTF-8?q?udit=5Fevent=20=D0=BB=D0=BE=D0=B2=D0=B8=D1=82=20=D0=BF=D1=81?= =?UTF-8?q?=D0=B5=D0=B2=D0=B4=D0=BE=D0=BD=D0=B8=D0=BC=D1=8B=20=D0=B8=D0=BC?= =?UTF-8?q?=D0=BF=D0=BE=D1=80=D1=82=D0=B0,=20=5F=5Ftable=5F=5F.update()/de?= =?UTF-8?q?lete()=20=D0=B8=20Query=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- api/tests/test_audit_service.py | 101 +++++++++++++++++++++++++------- 1 file changed, 81 insertions(+), 20 deletions(-) diff --git a/api/tests/test_audit_service.py b/api/tests/test_audit_service.py index 2378472..bfe9ac6 100644 --- a/api/tests/test_audit_service.py +++ b/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") - or (isinstance(sub, ast.Attribute) and sub.attr == "AuditEvent") + (isinstance(sub, ast.Name) and sub.id in aliases) or (isinstance(sub, ast.Attribute) and sub.attr in aliases) for sub in ast.walk(node) ) -def _is_audit_event_expr(node: ast.AST | None) -> bool: - """Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)`.""" - return isinstance(node, ast.Call) and _mentions_audit_event(node.func) +def _is_audit_event_expr(node: ast.AST | None, aliases: set[str]) -> bool: + """Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)` / `AE(...)`.""" + 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 = "") -> list[str]: """Находки вида `file:line: описание` — любое изменение/удаление audit_event в исходнике. - Ловит: `update(AuditEvent...)` / `delete(AuditEvent...)` (в т.ч. `sa.update`, `AuditEvent.__table__`); - `.delete(x)`/`delete(x)`, где x — экземпляр AuditEvent (внутримодульное отслеживание имён); присваивание/удаление - атрибутов таких экземпляров и `AuditEvent.<поле> = ...`; сырой SQL UPDATE/DELETE/TRUNCATE по audit_event. + Ловит (best-effort, статически, в пределах одного модуля): + - `update(AuditEvent...)` / `delete(AuditEvent...)`, в т.ч. `sa.update`, `update as upd`, `AuditEvent.__table__` + как аргумент; псевдонимы `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 = "") -> 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 any(_mentions_audit_event(arg) for arg in args): + if name in mutating: + 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 = "") -> 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") - ) or _is_audit_event_expr(base): + isinstance(base, ast.Name) and (base.id in tainted or base.id in aliases) + ) 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,