fix(security): address review findings for audit logging

benchmarks/security_audit_bench.py

@@ -16,6 +16,9 @@ from cleveragents.config.settings import Settings
 from cleveragents.infrastructure.database.models import AuditLogModel, Base
 
 
+from cleveragents.application.services.audit_service import _TIMESTAMP_FMT
+
+
 def _make_service() -> AuditService:
     """Create a service backed by in-memory SQLite."""
     Settings._instance = None
@@ -109,16 +112,22 @@ class TimeQueryEntries:
 
 
 class TimePrune:
-    """Benchmark pruning old audit entries."""
+    """Benchmark pruning old audit entries.
+
+    Uses ``setup`` to re-seed 200 backdated entries before *each*
+    benchmark iteration so that every call to ``prune()`` actually
+    deletes rows (ASV calls ``time_*`` multiple times per ``setup``
+    by default).
+    """
 
     timeout = 30
+    # ASV calls setup() before each iteration when number=1.
+    number = 1
 
     def setup(self) -> None:
         self.service = _make_service()
         # Insert entries backdated to 60 days ago
-        old_time = (datetime.now(tz=UTC) - timedelta(days=60)).strftime(
-            "%Y-%m-%dT%H:%M:%S.%f"
-        )
+        old_time = (datetime.now(tz=UTC) - timedelta(days=60)).strftime(_TIMESTAMP_FMT)
         for i in range(200):
             row = AuditLogModel(
                 event_type="plan_applied",

features/security_audit.feature

@@ -43,6 +43,13 @@ Feature: SEC7 - Audit logging for apply
     Then the entry details key "entity_type" should be "resource"
     And the entry details key "entity_name" should be "old-res"
 
+  Scenario: Record a session_created event
+    Given a fresh audit service
+    When I record a "session_created" event with session_id "sess-001"
+    Then the audit log should contain 1 entry
+    And the entry event_type should be "session_created"
+    And the entry details key "session_id" should be "sess-001"
+
   Scenario: Record event with empty details
     Given a fresh audit service
     When I record a "plan_applied" event with no details
@@ -176,6 +183,12 @@ Feature: SEC7 - Audit logging for apply
     Then the audit log should contain 1 entry
     And the entry details should contain key "timestamp"
 
+  Scenario: Recording with invalid event_type raises ValueError
+    Given a fresh audit service
+    When I record an event with invalid event_type "plan_appleid"
+    Then a ValueError should be raised for audit service
+    And the audit error message should contain "plan_appleid"
+
   Scenario: Service requires Settings instance
     When I create an AuditService with invalid settings
     Then a TypeError should be raised for audit service

features/steps/security_audit_steps.py

@@ -13,7 +13,10 @@ from behave.runner import Context
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
-from cleveragents.application.services.audit_service import AuditService
+from cleveragents.application.services.audit_service import (
+    _TIMESTAMP_FMT,
+    AuditService,
+)
 from cleveragents.config.settings import Settings
 from cleveragents.infrastructure.database.models import AuditLogModel, Base
 
@@ -44,9 +47,17 @@ def _insert_old_entry(
     days_ago: int,
     event_type: str = "plan_applied",
 ) -> None:
-    """Insert an entry with a manually backdated created_at."""
+    """Insert an entry with a manually backdated created_at.
+
+    Uses the same timestamp format as the service (``_TIMESTAMP_FMT``)
+    so that lexicographic comparisons in ``prune()`` and ``since``
+    filtering remain correct.
+
+    Directly accesses ``service._session`` because there is no public
+    API for inserting entries with an arbitrary ``created_at``.
+    """
     old_time = (datetime.now(tz=UTC) - timedelta(days=days_ago)).strftime(
-        "%Y-%m-%dT%H:%M:%S.%f"
+        _TIMESTAMP_FMT
     )
     row = AuditLogModel(
         event_type=event_type,
@@ -280,6 +291,25 @@ def step_record_structured_details(
     )
 
 
+@when('I record a "{event_type}" event with session_id "{session_id}"')
+def step_record_session_created(
+    context: Context, event_type: str, session_id: str
+) -> None:
+    context.entry = context.service.record(
+        event_type=event_type,
+        details={"session_id": session_id},
+    )
+
+
+@when('I record an event with invalid event_type "{event_type}"')
+def step_record_invalid_type(context: Context, event_type: str) -> None:
+    try:
+        context.service.record(event_type=event_type)
+        context.error = None
+    except ValueError as exc:
+        context.error = exc
+
+
 @when("I create an AuditService with invalid settings")
 def step_create_invalid(context: Context) -> None:
     try:
@@ -457,3 +487,13 @@ def step_user_none(context: Context) -> None:
 @then("a TypeError should be raised for audit service")
 def step_type_error(context: Context) -> None:
     assert isinstance(context.error, TypeError)
+
+
+@then("a ValueError should be raised for audit service")
+def step_value_error(context: Context) -> None:
+    assert isinstance(context.error, ValueError)
+
+
+@then('the audit error message should contain "{text}"')
+def step_audit_error_message_contains(context: Context, text: str) -> None:
+    assert text in str(context.error), f"Expected {text!r} in {context.error!s}"

robot/security_audit.robot

@@ -33,6 +33,7 @@ Audit Service Has Record Method
     ${content}=    Get File    ${AUDIT_SERVICE}
     Should Contain    ${content}    def record(
     Should Contain    ${content}    event_type
+    Should Contain    ${content}    VALID_EVENT_TYPES
 
 Audit Service Has List Method
     [Documentation]    Verify the audit service has a list_entries method
@@ -48,6 +49,13 @@ Audit Service Has Prune Method
     Should Contain    ${content}    def prune(
     Should Contain    ${content}    retention_days
 
+Audit Service Has Close And Context Manager
+    [Documentation]    Verify the audit service supports close() and context manager
+    ${content}=    Get File    ${AUDIT_SERVICE}
+    Should Contain    ${content}    def close(
+    Should Contain    ${content}    def __enter__(
+    Should Contain    ${content}    def __exit__(
+
 Audit CLI Has List Command
     [Documentation]    Verify the audit CLI has a list command
     ${content}=    Get File    ${AUDIT_CLI}

src/cleveragents/application/services/audit_service.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import json
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
-from typing import Any
+from typing import Any, Self
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
@@ -20,10 +20,32 @@ from cleveragents.config.settings import Settings
 from cleveragents.infrastructure.database.models import AuditLogModel, Base
 
 __all__ = [
+    "VALID_EVENT_TYPES",
     "AuditLogEntry",
     "AuditService",
 ]
 
+# Spec-defined event types (specification.md ~L39764).
+VALID_EVENT_TYPES: frozenset[str] = frozenset(
+    {
+        "plan_applied",
+        "plan_cancelled",
+        "resource_modified",
+        "correction_applied",
+        "config_changed",
+        "entity_deleted",
+        "session_created",
+    }
+)
+
+# Timestamp format matching the spec: ``strftime('%Y-%m-%dT%H:%M:%f')``
+# which produces ``YYYY-MM-DDTHH:MM:ffffff`` (no seconds field, microseconds
+# immediately after minutes).  All timestamps must use this format so that
+# lexicographic ``>=`` / ``<`` comparisons on the ``created_at`` TEXT column
+# remain correct.
+_TIMESTAMP_FMT = "%Y-%m-%dT%H:%M:%f"
+
+
 # ── Domain data class ────────────────────────────────────────────
 
 
@@ -60,6 +82,11 @@ class AuditLogEntry:
 class AuditService:
     """Records and queries audit log entries.
 
+    Implements the context-manager protocol so callers can use::
+
+        with AuditService(settings=get_settings()) as svc:
+            svc.record(event_type="plan_applied", ...)
+
     Args:
         settings: Application settings (for DB URL and retention).
         session: Optional pre-built SQLAlchemy session (for testing).
@@ -73,6 +100,7 @@ class AuditService:
         if not isinstance(settings, Settings):
             raise TypeError("settings must be a Settings instance")
         self._settings = settings
+        self._owns_session = session is None
         if session is not None:
             self._session = session
         else:
@@ -81,6 +109,23 @@ class AuditService:
             factory = sessionmaker(bind=engine)
             self._session = factory()
 
+    # ── Context manager ──────────────────────────────────────────
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, *_exc: object) -> None:
+        self.close()
+
+    def close(self) -> None:
+        """Close the underlying SQLAlchemy session.
+
+        Only closes sessions that were created by this service (i.e. not
+        externally-provided test sessions).  Safe to call multiple times.
+        """
+        if self._owns_session:
+            self._session.close()
+
     # ── Record ───────────────────────────────────────────────────
 
     def record(
@@ -95,10 +140,19 @@ class AuditService:
     ) -> AuditLogEntry:
         """Write an audit event to the database.
 
+        Raises:
+            ValueError: If *event_type* is not one of the spec-defined
+                event types in :data:`VALID_EVENT_TYPES`.
+
         Returns:
             The created :class:`AuditLogEntry` with its assigned ``id``.
         """
-        now = datetime.now(tz=UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")
+        if event_type not in VALID_EVENT_TYPES:
+            raise ValueError(
+                f"Unknown event_type {event_type!r}; "
+                f"expected one of {sorted(VALID_EVENT_TYPES)}"
+            )
+        now = datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT)
         details_json = json.dumps(details or {}, default=str)
 
         row = AuditLogModel(
@@ -133,7 +187,10 @@ class AuditService:
             plan_id: Filter by exact plan_id.
             project_name: Filter by exact project name.
             event_type: Filter by event type.
-            since: ISO-8601 timestamp; only entries after this time.
+            since: ISO-8601 timestamp (``YYYY-MM-DDTHH:MM:ffffff``
+                format recommended to match the stored format); only
+                entries created at or after this time are returned.
+                The comparison is lexicographic on the TEXT column.
             limit: Maximum number of entries to return.
 
         Returns:
@@ -193,9 +250,7 @@ class AuditService:
         )
         if days <= 0:
             return 0
-        cutoff = (datetime.now(tz=UTC) - timedelta(days=days)).strftime(
-            "%Y-%m-%dT%H:%M:%S.%f"
-        )
+        cutoff = (datetime.now(tz=UTC) - timedelta(days=days)).strftime(_TIMESTAMP_FMT)
         result = (
             self._session.query(AuditLogModel)
             .filter(AuditLogModel.created_at < cutoff)

src/cleveragents/cli/commands/audit.py

@@ -6,6 +6,7 @@ and pruning audit log entries.
 
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING, Annotated
 
 import typer
@@ -13,7 +14,10 @@ import typer
 from cleveragents.cli.main import get_console, get_err_console
 
 if TYPE_CHECKING:
-    from cleveragents.application.services.audit_service import AuditService
+    from cleveragents.application.services.audit_service import (
+        AuditLogEntry,
+        AuditService,
+    )
 
 app = typer.Typer(
     help="View and manage the audit log for security-relevant operations.",
@@ -56,15 +60,15 @@ def list_entries(
 ) -> None:
     """List audit log entries with optional filters."""
     console = get_console()
-    service = _get_audit_service()
 
-    entries = service.list_entries(
-        plan_id=plan,
-        project_name=project,
-        event_type=event_type,
-        since=since,
-        limit=limit,
-    )
+    with _get_audit_service() as service:
+        entries = service.list_entries(
+            plan_id=plan,
+            project_name=project,
+            event_type=event_type,
+            since=since,
+            limit=limit,
+        )
 
     if not entries:
         console.print("[dim]No audit entries found.[/dim]")
@@ -86,9 +90,10 @@ def show_entry(
     """Show a single audit entry with full metadata."""
     console = get_console()
     err_console = get_err_console()
-    service = _get_audit_service()
 
-    entry = service.get_entry(audit_id)
+    with _get_audit_service() as service:
+        entry = service.get_entry(audit_id)
+
     if entry is None:
         err_console.print(f"[red]Audit entry {audit_id} not found.[/red]")
         raise typer.Exit(code=1)
@@ -102,8 +107,6 @@ def show_entry(
     console.print(f"  User:           {entry.user_identity or '-'}")
     console.print("  Details:")
     if entry.details:
-        import json
-
         for line in json.dumps(entry.details, indent=2).splitlines():
             console.print(f"    {line}")
     else:
@@ -128,7 +131,6 @@ def prune(
     """Delete audit entries older than the retention period."""
     console = get_console()
     err_console = get_err_console()
-    service = _get_audit_service()
 
     if not yes:
         from cleveragents.config.settings import get_settings
@@ -148,7 +150,8 @@ def prune(
             err_console.print("[yellow]Aborted.[/yellow]")
             raise typer.Abort()
 
-    deleted = service.prune(retention_days=days)
+    with _get_audit_service() as service:
+        deleted = service.prune(retention_days=days)
     console.print(f"[green]Pruned {deleted} audit log entries.[/green]")
 
 
@@ -156,20 +159,16 @@ def prune(
 def count_entries() -> None:
     """Show the total number of audit log entries."""
     console = get_console()
-    service = _get_audit_service()
-    total = service.count()
+    with _get_audit_service() as service:
+        total = service.count()
     console.print(f"Total audit log entries: {total}")
 
 
 # ── Helpers ───────────────────────────────────────────────────────
 
 
-def _print_entry_summary(entry: object) -> None:
+def _print_entry_summary(entry: AuditLogEntry) -> None:
     """Print a one-line summary of an audit entry."""
-    from cleveragents.application.services.audit_service import AuditLogEntry
-
-    if not isinstance(entry, AuditLogEntry):
-        return
     console = get_console()
     plan_info = f" plan={entry.plan_id}" if entry.plan_id else ""
     project_info = f" project={entry.project_name}" if entry.project_name else ""