add-ideas/govoplan-audit
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: add-ideas:v0.1.8
Branches Tags
add-ideas:main
add-ideas:v0.1.8 add-ideas:v0.1.7 add-ideas:v0.1.6
...
compare: add-ideas:main
Branches Tags
add-ideas:main
add-ideas:v0.1.8 add-ideas:v0.1.7 add-ideas:v0.1.6

1 Commits
v0.1.8 ... main

Author SHA1 Message Date
Albrecht Degering
5e4f84a789 intermittent commit 2026-07-14 13:22:10 +02:00
2 changed files with 148 additions and 131 deletions
Expand all files Collapse all files

277
src/govoplan_audit/backend/api/v1/routes.py
View File

@@ -1,6 +1,8 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import and_, false, func, or_
@@ -23,6 +25,20 @@ router = APIRouter(tags=["audit"])
AUDIT_ADMIN_CURSOR_SCOPE = "audit.admin"
@dataclass(slots=True)
class AuditAdminQueryContext:
query: Any
access_admin: AccessAdministration
effective_scope: str
resolved_tenant_id: str | None
sort_column: Any
order: Any
total: int
effective_page_size: int
pages: int
fingerprint: str
def _access_administration() -> AccessAdministration:
registry = get_registry()
if registry is None or not registry.has_capability(CAPABILITY_ACCESS_ADMINISTRATION):
@@ -286,26 +302,23 @@ def _audit_cursor_condition(sort_column, *, sort_by: str, sort_direction: str, c
return or_(primary_after, and_(sort_column == sort_value, AuditLog.id < cursor_id))
@router.get("/admin/audit", response_model=AuditAdminListResponse)
def list_admin_audit(
tenant_id: str | None = Query(default=None),
all_tenants: bool = Query(default=False),
audit_scope: str | None = Query(default=None, alias="scope"),
limit: int = Query(default=100, ge=1, le=500),
offset: int = Query(default=0, ge=0),
page: int | None = Query(default=None, ge=1),
page_size: int | None = Query(default=None, ge=1, le=500),
cursor: str | None = Query(default=None),
sort_by: str = Query(default="time"),
sort_direction: str = Query(default="desc"),
filter_time: str | None = Query(default=None),
filter_actor: str | None = Query(default=None),
filter_action: str | None = Query(default=None),
filter_object: str | None = Query(default=None),
filter_tenant: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
def _prepare_audit_admin_query(
session: Session,
principal: ApiPrincipal,
*,
tenant_id: str | None,
all_tenants: bool,
audit_scope: str | None,
limit: int,
page_size: int | None,
sort_by: str,
sort_direction: str,
filter_time: str | None,
filter_actor: str | None,
filter_action: str | None,
filter_object: str | None,
filter_tenant: str | None,
) -> AuditAdminQueryContext:
effective_scope = audit_scope or ("all" if all_tenants else "tenant")
if effective_scope not in {"tenant", "system", "all"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
@@ -333,14 +346,13 @@ def list_admin_audit(
object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
access_admin = _access_administration()
filters = [
for condition in (
_audit_time_filter(filter_time),
_audit_actor_filter(access_admin, session, filter_actor),
_audit_text_filter(AuditLog.action, filter_action),
_audit_text_filter(object_text, filter_object),
_audit_text_filter(AuditLog.tenant_id, filter_tenant),
]
for condition in filters:
):
if condition is not None:
query = query.filter(condition)
@@ -353,7 +365,6 @@ def list_admin_audit(
}
sort_column = sort_columns[sort_by]
order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
ordered_query = query.order_by(order, AuditLog.id.desc())
total = query.count()
effective_page_size = page_size or limit
pages = max(1, (total + effective_page_size - 1) // effective_page_size)
@@ -372,47 +383,100 @@ def list_admin_audit(
sort_direction=sort_direction,
filters=filters,
)
return AuditAdminQueryContext(
query=query,
access_admin=access_admin,
effective_scope=effective_scope,
resolved_tenant_id=resolved_tenant_id,
sort_column=sort_column,
order=order,
total=total,
effective_page_size=effective_page_size,
pages=pages,
fingerprint=fingerprint,
)
@router.get("/admin/audit", response_model=AuditAdminListResponse)
def list_admin_audit(
tenant_id: str | None = Query(default=None),
all_tenants: bool = Query(default=False),
audit_scope: str | None = Query(default=None, alias="scope"),
limit: int = Query(default=100, ge=1, le=500),
offset: int = Query(default=0, ge=0),
page: int | None = Query(default=None, ge=1),
page_size: int | None = Query(default=None, ge=1, le=500),
cursor: str | None = Query(default=None),
sort_by: str = Query(default="time"),
sort_direction: str = Query(default="desc"),
filter_time: str | None = Query(default=None),
filter_actor: str | None = Query(default=None),
filter_action: str | None = Query(default=None),
filter_object: str | None = Query(default=None),
filter_tenant: str | None = Query(default=None),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
context = _prepare_audit_admin_query(
session,
principal,
tenant_id=tenant_id,
all_tenants=all_tenants,
audit_scope=audit_scope,
limit=limit,
page_size=page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filter_time=filter_time,
filter_actor=filter_actor,
filter_action=filter_action,
filter_object=filter_object,
filter_tenant=filter_tenant,
)
ordered_query = context.query.order_by(context.order, AuditLog.id.desc())
start_cursor: str | None = None
if cursor:
try:
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint)
if cursor_values is None:
raise KeysetCursorError("Invalid pagination cursor")
page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
page_query = context.query.filter(
_audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values)
)
except KeysetCursorError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
effective_page = page or (offset // effective_page_size + 1)
effective_page = page or (offset // context.effective_page_size + 1)
effective_offset = 0
start_cursor = cursor
else:
if page is not None or page_size is not None:
effective_page = min(page or 1, pages)
effective_offset = (effective_page - 1) * effective_page_size
effective_page = min(page or 1, context.pages)
effective_offset = (effective_page - 1) * context.effective_page_size
else:
effective_page = offset // effective_page_size + 1
effective_page = offset // context.effective_page_size + 1
effective_offset = offset
page_query = query
page_query = context.query
if effective_offset > 0:
previous_row = ordered_query.offset(effective_offset - 1).limit(1).first()
if previous_row is not None:
start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
start_cursor = _audit_cursor_for_row(previous_row, sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).offset(effective_offset).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).offset(effective_offset).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminListResponse(
total=total,
total=context.total,
page=effective_page,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
)
@@ -435,150 +499,103 @@ def list_admin_audit_delta(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(require_any_scope("audit:read", "system:audit:read")),
):
effective_scope = audit_scope or ("all" if all_tenants else "tenant")
if effective_scope not in {"tenant", "system", "all"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit scope must be tenant, system or all.")
if sort_by not in {"time", "actor", "action", "object", "tenant"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Unsupported audit sort column.")
if sort_direction not in {"asc", "desc"}:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail="Audit sort direction must be asc or desc.")
query = session.query(AuditLog)
resolved_tenant_id: str | None = None
if effective_scope != "all":
query = query.filter(AuditLog.scope == effective_scope)
if effective_scope == "system":
if not has_scope(principal, "system:audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
elif effective_scope == "all" or all_tenants:
if not has_scope(principal, "system:audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:audit:read")
else:
if not has_scope(principal, "audit:read"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: audit:read")
tenant = _resolve_tenant(session, principal, tenant_id)
resolved_tenant_id = tenant.id
query = query.filter(AuditLog.tenant_id == tenant.id)
object_text = func.coalesce(AuditLog.object_type, "") + " " + func.coalesce(AuditLog.object_id, "")
access_admin = _access_administration()
filters = [
_audit_time_filter(filter_time),
_audit_actor_filter(access_admin, session, filter_actor),
_audit_text_filter(AuditLog.action, filter_action),
_audit_text_filter(object_text, filter_object),
_audit_text_filter(AuditLog.tenant_id, filter_tenant),
]
for condition in filters:
if condition is not None:
query = query.filter(condition)
sort_columns = {
"time": AuditLog.created_at,
"actor": func.coalesce(AuditLog.user_id, "System"),
"action": AuditLog.action,
"object": object_text,
"tenant": func.coalesce(AuditLog.tenant_id, ""),
}
sort_column = sort_columns[sort_by]
order = sort_column.asc() if sort_direction == "asc" else sort_column.desc()
total = query.count()
effective_page_size = page_size or limit
pages = max(1, (total + effective_page_size - 1) // effective_page_size)
filters = _audit_filter_params(
context = _prepare_audit_admin_query(
session,
principal,
tenant_id=tenant_id,
all_tenants=all_tenants,
audit_scope=audit_scope,
limit=limit,
page_size=page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filter_time=filter_time,
filter_actor=filter_actor,
filter_action=filter_action,
filter_object=filter_object,
filter_tenant=filter_tenant,
)
fingerprint = _audit_cursor_fingerprint(
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
page_size=effective_page_size,
sort_by=sort_by,
sort_direction=sort_direction,
filters=filters,
)
start_cursor: str | None = None
page_query = query
page_query = context.query
if cursor:
try:
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
cursor_values = decode_keyset_cursor(AUDIT_ADMIN_CURSOR_SCOPE, cursor, fingerprint=context.fingerprint)
if cursor_values is None:
raise KeysetCursorError("Invalid pagination cursor")
page_query = query.filter(_audit_cursor_condition(sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values))
page_query = context.query.filter(
_audit_cursor_condition(context.sort_column, sort_by=sort_by, sort_direction=sort_direction, cursor_values=cursor_values)
)
start_cursor = cursor
except KeysetCursorError as exc:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
if since is None:
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=total,
total=context.total,
page=1,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
)
entries, has_more = _audit_delta_entries(
session,
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
effective_scope=context.effective_scope,
tenant_id=context.resolved_tenant_id,
since=since,
limit=effective_page_size,
limit=context.effective_page_size,
)
if entries is None:
rows_plus_one = page_query.order_by(order, AuditLog.id.desc()).limit(effective_page_size + 1).all()
rows = rows_plus_one[:effective_page_size]
rows_plus_one = page_query.order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size + 1).all()
rows = rows_plus_one[:context.effective_page_size]
next_cursor = (
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=fingerprint)
if len(rows_plus_one) > effective_page_size and rows else None
_audit_cursor_for_row(rows[-1], sort_by=sort_by, sort_direction=sort_direction, fingerprint=context.fingerprint)
if len(rows_plus_one) > context.effective_page_size and rows else None
)
return AuditAdminDeltaResponse(
total=total,
total=context.total,
page=1,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=next_cursor,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_watermark(session, effective_scope=effective_scope, tenant_id=resolved_tenant_id),
watermark=_audit_delta_watermark(session, effective_scope=context.effective_scope, tenant_id=context.resolved_tenant_id),
has_more=False,
full=True,
)
changed_ids = [entry.resource_id for entry in entries if entry.resource_type == "audit_log"]
rows = (
page_query.filter(AuditLog.id.in_(changed_ids)).order_by(order, AuditLog.id.desc()).limit(effective_page_size).all()
page_query.filter(AuditLog.id.in_(changed_ids)).order_by(context.order, AuditLog.id.desc()).limit(context.effective_page_size).all()
if changed_ids else []
)
return AuditAdminDeltaResponse(
total=total,
total=context.total,
page=1,
page_size=effective_page_size,
pages=pages,
page_size=context.effective_page_size,
pages=context.pages,
cursor=start_cursor,
next_cursor=None,
items=_audit_items(session, rows, access_admin),
items=_audit_items(session, rows, context.access_admin),
deleted=[],
watermark=_audit_delta_response_watermark(
session,
effective_scope=effective_scope,
tenant_id=resolved_tenant_id,
effective_scope=context.effective_scope,
tenant_id=context.resolved_tenant_id,
entries=entries,
has_more=has_more,
),

2
tests/test_audit_module_contract.py
View File

@@ -13,7 +13,7 @@ class AuditModuleContractTests(unittest.TestCase):
project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
dependencies = tuple(project["dependencies"])
self.assertIn("govoplan-core>=0.1.6", dependencies)
self.assertIn("govoplan-core>=0.1.8", dependencies)
self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies))
def test_audit_source_does_not_import_access_implementation(self) -> None: