92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""Audit view factory."""
|
|
|
|
# pyright: reportUnusedFunction=false
|
|
import logging
|
|
from typing import Annotated, cast
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
|
|
from sshecret.backend import AuditFilter, Operation
|
|
|
|
from sshecret_admin.auth import LocalUserInfo
|
|
from sshecret_admin.services import AdminBackend
|
|
|
|
from .common import PagingInfo
|
|
from ..dependencies import FrontendDependencies
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
def create_router(dependencies: FrontendDependencies) -> APIRouter:
|
|
"""Create clients router."""
|
|
|
|
app = APIRouter()
|
|
templates = dependencies.templates
|
|
|
|
async def resolve_audit_entries(
|
|
request: Request,
|
|
current_user: LocalUserInfo,
|
|
admin: AdminBackend,
|
|
page: int,
|
|
filters: AuditFilter,
|
|
) -> Response:
|
|
"""Resolve audit entries."""
|
|
LOG.info("Page: %r", page)
|
|
per_page = 20
|
|
offset = 0
|
|
if page > 1:
|
|
offset = (page - 1) * per_page
|
|
|
|
filter_args = cast(dict[str, str], filters.model_dump(exclude_none=True))
|
|
audit_log = await admin.get_audit_log_detailed(offset, per_page, **filter_args)
|
|
page_info = PagingInfo(
|
|
page=page, limit=per_page, total=audit_log.total, offset=offset
|
|
)
|
|
operations = list(Operation)
|
|
breadcrumbs = [("Audit", "/audit/")]
|
|
if request.headers.get("HX-Request"):
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"audit/inner.html.j2",
|
|
{
|
|
"entries": audit_log.results,
|
|
"page_info": page_info,
|
|
"operations": operations,
|
|
},
|
|
)
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"audit/index.html.j2",
|
|
{
|
|
"page_title": "Audit Log",
|
|
"breadcrumbs": breadcrumbs,
|
|
"entries": audit_log.results,
|
|
"user": current_user,
|
|
"page_info": page_info,
|
|
"operations": operations,
|
|
},
|
|
)
|
|
|
|
@app.get("/audit/")
|
|
async def get_audit_entries(
|
|
request: Request,
|
|
current_user: Annotated[LocalUserInfo, Depends(dependencies.get_user_info)],
|
|
admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)],
|
|
filters: Annotated[AuditFilter, Depends()],
|
|
) -> Response:
|
|
"""Get audit entries."""
|
|
return await resolve_audit_entries(request, current_user, admin, 1, filters)
|
|
|
|
@app.get("/audit/page/{page}")
|
|
async def get_audit_entries_page(
|
|
request: Request,
|
|
current_user: Annotated[LocalUserInfo, Depends(dependencies.get_user_info)],
|
|
admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)],
|
|
filters: Annotated[AuditFilter, Depends()],
|
|
page: int,
|
|
) -> Response:
|
|
"""Get audit entries."""
|
|
LOG.info("Get audit entries page: %r", page)
|
|
return await resolve_audit_entries(request, current_user, admin, page, filters)
|
|
|
|
return app
|