Update audit logging and dashboard

packages/sshecret-admin/src/sshecret_admin/frontend/views/audit.py

@@ -3,10 +3,12 @@
 # pyright: reportUnusedFunction=false
 import logging
 import math
-from typing import Annotated
+from typing import Annotated, cast
 from fastapi import APIRouter, Depends, Request, Response
 from pydantic import BaseModel
 
+from sshecret.backend import AuditFilter, Operation
+
 from sshecret_admin.auth import User
 from sshecret_admin.services import AdminBackend
 
@@ -45,28 +47,33 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
     templates = dependencies.templates
 
     async def resolve_audit_entries(
-        request: Request, current_user: User, admin: AdminBackend, page: int
+        request: Request,
+        current_user: User,
+        admin: AdminBackend,
+        page: int,
+        filters: AuditFilter,
     ) -> Response:
         """Resolve audit entries."""
         LOG.info("Page: %r", page)
-        total_messages = await admin.get_audit_log_count()
         per_page = 20
         offset = 0
         if page > 1:
             offset = (page - 1) * per_page
 
-        entries = await admin.get_audit_log(offset=offset, limit=per_page)
-        LOG.info("Entries: %r", entries)
+        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=total_messages, offset=offset
+            page=page, limit=per_page, total=audit_log.total, offset=offset
         )
+        operations = list(Operation)
         if request.headers.get("HX-Request"):
             return templates.TemplateResponse(
                 request,
                 "audit/inner.html.j2",
                 {
-                    "entries": entries,
+                    "entries": audit_log.results,
                     "page_info": page_info,
+                    "operations": operations,
                 },
             )
         return templates.TemplateResponse(
@@ -74,9 +81,10 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
             "audit/index.html.j2",
             {
                 "page_title": "Audit",
-                "entries": entries,
+                "entries": audit_log.results,
                 "user": current_user.username,
                 "page_info": page_info,
+                "operations": operations,
             },
         )
 
@@ -85,19 +93,21 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
         request: Request,
         current_user: Annotated[User, Depends(dependencies.get_user_from_access_token)],
         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)
+        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[User, Depends(dependencies.get_user_from_access_token)],
         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)
+        return await resolve_audit_entries(request, current_user, admin, page, filters)
 
     return app

packages/sshecret-admin/src/sshecret_admin/frontend/views/auth.py

@@ -8,6 +8,7 @@ from fastapi import APIRouter, Depends, Query, Request, Response, status
 from fastapi.responses import RedirectResponse
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlmodel import Session
+from sshecret_admin.services import AdminBackend
 from starlette.datastructures import URL
 
 from sshecret_admin.auth import (
@@ -17,6 +18,8 @@ from sshecret_admin.auth import (
     create_refresh_token,
 )
 
+from sshecret.backend.models import Operation
+
 from ..dependencies import FrontendDependencies
 from ..exceptions import RedirectException
 
@@ -30,6 +33,19 @@ class LoginError(BaseModel):
     message: str
 
 
+async def audit_login_failure(admin: AdminBackend, username: str, request: Request) -> None:
+    """Write login failure to audit log."""
+    origin: str | None = None
+    if request.client:
+        origin = request.client.host
+    await admin.write_audit_message(
+        operation=Operation.DENY,
+        message="Login failed",
+        origin=origin or "UNKNOWN",
+        username=username,
+    )
+
+
 def create_router(dependencies: FrontendDependencies) -> APIRouter:
     """Create auth router."""
 
@@ -64,6 +80,7 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
         request: Request,
         response: Response,
         session: Annotated[Session, Depends(dependencies.get_db_session)],
+        admin: Annotated[AdminBackend, Depends(dependencies.get_admin_backend)],
         form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
         next: Annotated[str, Query()] = "/dashboard",
         error_title: str | None = None,
@@ -89,6 +106,7 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
             )
         )
         if not user:
+            await audit_login_failure(admin, form_data.username, request)
             raise login_failed
         token_data: dict[str, str] = {"sub": user.username}
         access_token = create_access_token(dependencies.settings, data=token_data)
@@ -108,6 +126,17 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
             secure=False,
             samesite="strict",
         )
+        origin = "UNKNOWN"
+        if request.client:
+            origin = request.client.host
+
+        await admin.write_audit_message(
+            operation=Operation.LOGIN,
+            message="Logged in to admin frontend",
+            origin=origin,
+            username=form_data.username,
+        )
+
         return response
 
     @app.get("/refresh")

packages/sshecret-admin/src/sshecret_admin/frontend/views/index.py

@@ -56,6 +56,9 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
     ):
         """Dashboard for mocking up the dashboard."""
         stats = await get_stats(admin)
+        last_login_events = await admin.get_audit_log_detailed(limit=5, operation="login")
+        last_audit_events = await admin.get_audit_log_detailed(limit=10)
+
 
         return templates.TemplateResponse(
             request,
@@ -64,6 +67,9 @@ def create_router(dependencies: FrontendDependencies) -> APIRouter:
                 "page_title": "sshecret",
                 "user": current_user.username,
                 "stats": stats,
+                "last_login_events": last_login_events,
+                "last_audit_events": last_audit_events,
+
             },
         )