inkwell/src/inkwell/audit.py

"""The audit log: what happened to accounts, kept where an admin can read it.

Sign-ins, failed sign-ins, password changes and resets, devices linked and
unlinked, invites made and revoked. Each is still written to the app log as it
happens; this keeps a copy in `audit_events` for `audit_retention_days`, which the
app log can't promise past the container's log rotation.

What is NOT recorded here: a throttled attempt (429). That stays in the app log
only. A throttle trip is the answer to a flood, and a row per refused request
would make each one of the flood cost a database write.

`record` opens its own session and commits on its own, so an event outlives the
caller's transaction. A refused registration rolls back, but its refusal is still
recorded. A failure to record is logged and swallowed: the audit log is never the
reason a sign-in fails.
"""
from __future__ import annotations

import logging
import uuid
from datetime import datetime, timezone

from sqlalchemy import delete, select

from .common import expired_before, iso
from .db import session_scope
from .models.audit_event import AuditEvent
from .proxy import client_address
from .settings import get_setting

logger = logging.getLogger(__name__)

# The events, by name. The web Activity list words each one; a new event needs a
# line there too (`frontend/src/components/ActivityList.vue`).
SIGN_IN = "sign_in"
SIGN_IN_FAILED = "sign_in_failed"
REGISTERED = "registered"
REGISTRATION_REFUSED = "registration_refused"
RESET_REQUESTED = "reset_requested"
RESET_LINK_MADE = "reset_link_made"
PASSWORD_RESET = "password_reset"
RESET_REFUSED = "reset_refused"
PASSWORD_CHANGED = "password_changed"
PASSWORD_CHANGE_REFUSED = "password_change_refused"
SIGNED_OUT_ELSEWHERE = "signed_out_elsewhere"
DEVICE_LINKED = "device_linked"
DEVICE_UNLINKED = "device_unlinked"
INVITE_CREATED = "invite_created"
INVITE_REVOKED = "invite_revoked"

# One page of the Activity list. Older events are reached with `before`.
PAGE = 100


async def record(
    event: str,
    *,
    user_id: uuid.UUID | None = None,
    email: str | None = None,
    detail: str | None = None,
) -> None:
    """Keep one event. Call it from inside a request: the address is read from it."""
    try:
        async with session_scope() as db:
            db.add(
                AuditEvent(
                    event=event,
                    user_id=user_id,
                    email=email or None,
                    address=client_address(),
                    detail=detail,
                )
            )
            await db.commit()
    except Exception:
        logger.exception("audit event %s was not recorded", event)


def _serialize(e: AuditEvent) -> dict:
    return {
        "id": str(e.id),
        "at": iso(e.at),
        "event": e.event,
        "email": e.email,
        "address": e.address,
        "detail": e.detail,
    }


async def recent(db, before: datetime | None = None) -> list[dict]:
    """The newest page of events, or the page older than `before`."""
    query = select(AuditEvent).order_by(AuditEvent.at.desc(), AuditEvent.id.desc()).limit(PAGE)
    if before is not None:
        query = query.where(AuditEvent.at < before)
    return [_serialize(e) for e in (await db.scalars(query)).all()]


async def sweep_once(*, now: datetime | None = None) -> int:
    """Delete events older than `audit_retention_days`; 0 keeps them forever."""
    async with session_scope() as db:
        days = int(await get_setting(db, "audit_retention_days"))
        cutoff = expired_before(now or datetime.now(timezone.utc), days)
        if cutoff is None:
            return 0
        deleted = (await db.execute(delete(AuditEvent).where(AuditEvent.at < cutoff))).rowcount
        await db.commit()
        return deleted