fix(policy): API policy reloads survive a restart and are recorded (F21)

CHANGELOG.md

@@ -6,6 +6,13 @@ All notable changes to this project are documented here. Format follows
 
 ## [Unreleased]
 
+### Added
+- Every policy reload is recorded: a `PolicyChanged` span (applied, ignored,
+  rejected or failed, with the versions and the SHA-256 of the submitted text)
+  in a `policy-change` run owned by that tenant, exported to the SIEM at
+  severity `medium`, and a `POLICY` line in the security log. Earlier versions
+  cannot replay these runs, since they do not know the event type.
+
 ### Security
 - `versions.lock` moves past newly published advisories: certifi 2026.10.19
   (PYSEC-2026-4320), and for the pre-commit tooling filelock 4.2.1 (three CVEs)
@@ -16,6 +23,17 @@ All notable changes to this project are documented here. Format follows
   fresh install already picks up the fixed certifi.
 
 ### Fixed
+- A policy replaced through `POST /tenants/{tenant}/policy` survives a restart.
+  It lived only in memory, so a restart silently put the configured file back,
+  and nothing recorded that the policy had changed. An applied policy is now
+  saved under `${SENTINEL_DATA_DIR}/policies/` before it takes effect (a failed
+  save changes nothing and answers 500). At startup a saved policy replaces the
+  configured one only when its `policy_version` is higher. Reloads are
+  serialized, so two at once cannot save one version and enforce another.
+  Files are named by a SHA-256 of the tenant, so any tenant name works and none
+  can escape the directory or collide with another on a case-insensitive file
+  system. A saved policy that cannot be read or no longer loads stops
+  `sentinel serve` and `sentinel check` with an error naming the file.
 - Retention could keep a trace forever. The SQLite store filled in span
   timestamps only while adding the `ts` column, but 0.2.2 can still write to a
   migrated database and leaves `ts` empty, so after a rollback and a second

README.md

@@ -168,7 +168,7 @@ Read this before deploying anything. Each limitation changes what SENTINEL prote
 | **Only guarded tools are guarded** | A tool without rules can carry tainted data out. Add rules for every outbound tool; there is no general notion of an outbound "sink" yet |
 | **Quarantine is recorded, not enforced** (default) | The trust score never recovers, so agents making only allowed calls cross the threshold within tens of calls. `SENTINEL_ENFORCE_QUARANTINE=1` enforces it |
 | **Downstream servers: streamable HTTP only** | No stdio servers, no per-server credentials or OAuth, one shared session per server, and no reconnect. Most published MCP servers are stdio |
-| **State lives in one process** | Trust scores, MCP sessions, the event feed, the run index and API policy reloads are in memory. Two replicas share none of it |
+| **State lives in one process** | Trust scores, MCP sessions, the event feed and the run index are in memory. Two replicas share none of it, and a policy replaced through the API is saved only on the replica that received it |
 | **Cosmos persistence is partial** | Needs `pip install "sentinel-prox[cosmos]"`; run ownership, the run index after a restart and retention purging exist only for SQLite |
 | **Some state is visible across tenants** | `/events` ids are one global counter; `/downstream` and `/capabilities` describe the whole deployment |
 | **Calls the SDK rejects are not recorded** | Arguments that fail a tool's input schema are refused by the MCP SDK before SENTINEL sees them |

src/sentinel/authorization/registry.py

@@ -64,7 +64,10 @@ def reload_if_newer(self, tenant: str, text: str) -> ReloadResult:
         Raises ``PolicyLoadError`` (from :func:`load_policy`) on malformed input,
         BEFORE touching the active policy — a bad push cannot take a tenant down.
         """
-        candidate = load_policy(text)
+        return self.apply_if_newer(tenant, load_policy(text))
+
+    def apply_if_newer(self, tenant: str, candidate: CompiledPolicy) -> ReloadResult:
+        """Swap in an already-compiled ``candidate`` for ``tenant`` if it is newer."""
         current = self._policies.get(tenant)
         if current is None:
             self._policies[tenant] = candidate

src/sentinel/control/app.py

@@ -635,11 +635,16 @@ async def list_tenants(request: Request) -> dict[str, Any]:
 
     @app.post("/tenants/{tenant}/policy", dependencies=[Depends(require_admin)])
     async def reload_tenant_policy(tenant: str, body: PolicyReloadRequest) -> dict[str, Any]:
-        """Hot-reload a tenant's policy on version bump (rejects malformed)."""
+        """Replace a tenant's policy on version bump; saved, so it survives a restart."""
         try:
-            result = mgr.reload_policy(tenant, body.policy)
+            result = await mgr.reload_policy(tenant, body.policy)
         except PolicyLoadError as exc:
             raise HTTPException(status_code=400, detail=str(exc)) from exc
+        except OSError as exc:
+            raise HTTPException(
+                status_code=500,
+                detail="the policy could not be saved, so it was not applied; see the server log",
+            ) from exc
         return {
             "tenant": result.tenant,
             "reloaded": result.reloaded,

src/sentinel/control/manager.py

@@ -16,7 +16,9 @@
 
 import asyncio
 import functools
+import hashlib
 import importlib.util
+import json
 import logging
 import os
 import time
@@ -30,7 +32,12 @@
 
 from sentinel.authn import tenant_credentials, tenant_policy_paths
 from sentinel.authorization.engine import AuthorizationEngine
-from sentinel.authorization.policy import CompiledPolicy, load_default_policy, load_policy
+from sentinel.authorization.policy import (
+    CompiledPolicy,
+    PolicyLoadError,
+    load_default_policy,
+    load_policy,
+)
 from sentinel.authorization.registry import DEFAULT_TENANT, PolicyRegistry, ReloadResult
 from sentinel.classifier import AttackClassifier
 from sentinel.classifier.attack_classifier import BlockedAttempt
@@ -49,6 +56,8 @@
     AgentQuarantined,
     AuthorizationDecided,
     InjectionScanned,
+    PolicyChanged,
+    PolicyOutcome,
     ToolBlocked,
 )
 from sentinel.forensics.replay import TraceReplay, replay
@@ -59,6 +68,7 @@
 )
 from sentinel.mcp_proxy.proxy import SentinelProxy
 from sentinel.mcp_proxy.router import DownstreamConnection
+from sentinel.observability import log_policy_change
 from sentinel.shield import InputShield
 from sentinel.trust.config import load_default_trust_config
 from sentinel.trust.scorer import TrustScorer
@@ -104,6 +114,101 @@ def _load_policy_file(path_text: str, *, what: str) -> CompiledPolicy:
 _LOG = logging.getLogger("sentinel.control.manager")
 
 
+def _data_dir() -> Path:
+    return Path(os.environ.get("SENTINEL_DATA_DIR", "var"))
+
+
+def saved_policy_dir() -> Path:
+    """Where policies applied through ``POST /tenants/{tenant}/policy`` are kept."""
+    return _data_dir() / "policies"
+
+
+def _saved_policy_path(tenant: str) -> Path:
+    # Named by a digest of the tenant rather than the tenant itself: any tenant
+    # name is valid, none can reach outside the directory, and "acme" and
+    # "ACME" cannot share a file on a case-insensitive file system.
+    digest = hashlib.sha256(tenant.encode("utf-8")).hexdigest()
+    return saved_policy_dir() / f"{digest}.json"
+
+
+def _save_policy(tenant: str, text: str, version: int) -> None:
+    """Write ``tenant``'s policy so a restart keeps it: the old file or the new.
+
+    Staged, flushed to disk, then renamed over the old copy, so a crash leaves
+    one complete file or the other, never half of one.
+    """
+    path = _saved_policy_path(tenant)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    staging = path.with_name(f".{path.name}.tmp")
+    record = {
+        "tenant": tenant,
+        "policy_version": version,
+        "saved_at": datetime.now(UTC).isoformat(),
+        "policy": text,
+    }
+    with staging.open("w", encoding="utf-8") as handle:
+        json.dump(record, handle, ensure_ascii=False)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(staging, path)
+    if hasattr(os, "O_DIRECTORY"):  # POSIX: make the rename itself durable
+        fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+
+
+def _overlay_saved_policies(registry: PolicyRegistry) -> None:
+    """Apply the policies saved by earlier API reloads, where newer than configured.
+
+    A reload used to live only in memory, so a restart silently put the
+    configured file back. The rule that decides a reload decides here too: the
+    higher ``policy_version`` wins, so an operator who bumps the configured file
+    past a saved copy gets the file.
+
+    A saved policy that cannot be read or no longer loads stops startup, as a
+    broken configured file does: it is the operator's latest policy, and
+    quietly enforcing an older one instead would be the worse surprise.
+    """
+    directory = saved_policy_dir()
+    if not directory.is_dir():
+        return
+    for path in sorted(directory.glob("*.json")):
+        try:
+            record = json.loads(path.read_text(encoding="utf-8"))
+            tenant, text = record["tenant"], record["policy"]
+            if not isinstance(tenant, str) or not isinstance(text, str):
+                raise TypeError("'tenant' and 'policy' must be strings")
+        except (OSError, ValueError, KeyError, TypeError) as exc:
+            raise ValueError(
+                f"saved policy {path} cannot be read ({exc}); repair or delete it"
+            ) from exc
+        if path.name != _saved_policy_path(tenant).name:
+            raise ValueError(
+                f"saved policy {path} is for tenant {tenant!r} but is not named for it; "
+                "repair or delete it"
+            )
+        try:
+            saved = load_policy(text)
+        except PolicyLoadError as exc:
+            raise ValueError(
+                f"saved policy for tenant {tenant!r} ({path}) does not load: {exc}; "
+                "repair or delete it"
+            ) from exc
+        current = registry.get(tenant)
+        if current is None or saved.policy_version > current.policy_version:
+            registry.register(tenant, saved)
+            _LOG.warning(
+                "tenant %r: enforcing the policy saved by an API reload (v%d) %s",
+                tenant,
+                saved.policy_version,
+                "(no policy is configured for it)"
+                if current is None
+                else f"instead of the configured v{current.policy_version}",
+            )
+
+
 def build_policy_registry(settings: Settings) -> PolicyRegistry:
     """Give every provisioned tenant a policy, and nobody else.
 
@@ -116,6 +221,8 @@ def build_policy_registry(settings: Settings) -> PolicyRegistry:
       that authenticated callers and then could not execute a single action.
     * A tenant with neither a credential nor a file has no policy and is
       refused. That fail-closed behaviour is unchanged.
+    * A policy saved by an API reload replaces any of these that it is newer
+      than (see :func:`_overlay_saved_policies`).
 
     Failing loudly on a bad path is deliberate: silently running the example
     policy while the operator believes theirs is active would be a security
@@ -137,6 +244,7 @@ def build_policy_registry(settings: Settings) -> PolicyRegistry:
     for tenant in tenant_credentials():
         if tenant not in explicit and registry.get(tenant) is None:
             registry.register(tenant, deployment)
+    _overlay_saved_policies(registry)
     return registry
 
 
@@ -243,6 +351,10 @@ def __init__(
         # Persistent history is a startup import, not an index refill on every
         # GET /runs. Re-adding every evicted trace defeated max_retained_runs.
         self._persisted_runs_restored = False
+        # One policy reload at a time. Two interleaved reloads could otherwise
+        # save one version and activate another, and a restart would then
+        # bring back the policy that lost.
+        self._policy_lock = asyncio.Lock()
 
     @property
     def bus(self) -> EventBus:
@@ -684,9 +796,99 @@ def list_tenants(self) -> list[dict[str, Any]]:
             for t in self._registry.tenants()
         ]
 
-    def reload_policy(self, tenant: str, policy_text: str) -> ReloadResult:
-        """Hot-reload a tenant's policy on version bump (raises on malformed)."""
-        return self._registry.reload_if_newer(tenant, policy_text)
+    async def reload_policy(self, tenant: str, policy_text: str) -> ReloadResult:
+        """Replace ``tenant``'s policy if the submitted one is newer; save it; record it.
+
+        A reload used to live only in memory, so a restart silently put the
+        configured file back, and nothing recorded that the policy had changed.
+        Now an applied policy is saved under the data directory BEFORE it takes
+        effect, so a failed save changes nothing, and every attempt becomes a
+        ``PolicyChanged`` record in the tenant's history: applied, ignored (not
+        newer), rejected (did not load) or failed (could not be saved).
+
+        Raises :class:`PolicyLoadError` for a policy that does not load and
+        :class:`OSError` for one that could not be saved, after recording either.
+        """
+        digest = hashlib.sha256(policy_text.encode("utf-8")).hexdigest()
+        async with self._policy_lock:
+            previous = self._registry.version(tenant)
+            try:
+                candidate = load_policy(policy_text)
+            except PolicyLoadError as exc:
+                await self._record_policy_change(
+                    tenant, "rejected", None, previous, digest, str(exc)
+                )
+                raise
+            if previous is None or candidate.policy_version > previous:
+                try:
+                    await asyncio.to_thread(
+                        _save_policy, tenant, policy_text, candidate.policy_version
+                    )
+                except OSError as exc:
+                    _LOG.error("could not save the policy for tenant %r: %s", tenant, exc)
+                    await self._record_policy_change(
+                        tenant, "failed", candidate.policy_version, previous, digest,
+                        f"could not be saved, so it was not applied: {exc}",
+                    )
+                    raise
+            result = self._registry.apply_if_newer(tenant, candidate)
+            await self._record_policy_change(
+                tenant,
+                "applied" if result.reloaded else "ignored",
+                candidate.policy_version,
+                previous,
+                digest,
+                result.detail,
+            )
+        return result
+
+    async def _record_policy_change(
+        self,
+        tenant: str,
+        outcome: PolicyOutcome,
+        version: int | None,
+        previous: int | None,
+        digest: str,
+        detail: str,
+    ) -> None:
+        """One trace per attempt, in the history of the tenant whose policy it is.
+
+        Best effort once the decision is made: an applied policy is already in
+        force, so failing the request because the record could not be written
+        would misreport what happened. The security log line is written either way.
+        """
+        trace_id = self._emitter.new_trace_id()
+        record = RunRecord(
+            run_id=trace_id,
+            trace_id=trace_id,
+            agent_id="operator",
+            scenario="policy-change",
+            tenant=tenant,
+            status="completed",
+        )
+        self._runs[trace_id] = record
+        self._persist_owner(record)
+        try:
+            await self._emitter.emit(
+                PolicyChanged(
+                    tenant=tenant,
+                    outcome=outcome,
+                    version=version,
+                    previous_version=previous,
+                    policy_sha256=digest,
+                    # A load error can quote the submitted text; keep it short.
+                    detail=detail[:300],
+                ),
+                trace_id=trace_id,
+            )
+        except Exception:  # noqa: BLE001 - see the docstring; logged, not swallowed
+            _LOG.exception("could not record a policy change for tenant %r", tenant)
+        finally:
+            self._emitter.forget(trace_id)
+            self._prune_finished_runs()
+        log_policy_change(
+            tenant, outcome=outcome, version=version, previous=previous, trace_id=trace_id
+        )
 
     # --- SOC audit (the classifier's ONE consumer) -----------------------------
 
@@ -716,6 +918,9 @@ async def audit(self, trace_id: str) -> list[dict[str, Any]]:
                 severity = "high"
             elif isinstance(p, AgentQuarantined):
                 severity = "critical"
+            elif isinstance(p, PolicyChanged):
+                # A change to what is enforced, or an attempt at one.
+                severity = "medium"
             alert: dict[str, Any] = {
                 "timestamp": s.timestamp.isoformat(),
                 "source": "SENTINEL",
@@ -740,6 +945,14 @@ async def audit(self, trace_id: str) -> list[dict[str, Any]]:
                 alert["attack_class"] = label.attack_class
                 alert["attack_rationale"] = label.rationale
                 alert["classifier"] = label.classifier
+            elif isinstance(p, PolicyChanged):
+                alert.update(
+                    tenant=p.tenant,
+                    outcome=p.outcome,
+                    policy_version=p.version,
+                    previous_version=p.previous_version,
+                    policy_sha256=p.policy_sha256,
+                )
             alerts.append(alert)
         return alerts
 

src/sentinel/control/static/index.html

@@ -521,6 +521,7 @@ <h1>SENTINEL<span class="c">_</span></h1>
   else if(t==='TrustUpdated'){ lvl='TRUST'; cls='trust'; const d=p.delta>0?'+'+p.delta.toFixed(0):p.delta.toFixed(0); msg=esc(cleanReason(p.reason))+'   score '+p.previous_score.toFixed(0)+'→'+p.new_score.toFixed(0)+' ('+d+')'; }
   else if(t==='AgentQuarantined'&&p.enforced===false){ lvl='TRUST'; msg='trust below threshold — score '+p.score_at_quarantine.toFixed(1)+'; quarantine not enforced, calls continue'; }
   else if(t==='AgentQuarantined'){ lvl='QUAR'; cls='block'; msg='agent quarantined — score '+p.score_at_quarantine.toFixed(1)+' < threshold, lane disabled'; }
+  else if(t==='PolicyChanged'){ lvl='POLICY'; cls=p.outcome==='applied'?'allow':(p.outcome==='ignored'?'dim':'deny'); msg='tenant '+esc(p.tenant)+'  '+esc(p.outcome)+'  v'+(p.previous_version===null?'-':p.previous_version)+' → v'+(p.version===null?'-':p.version)+'  '+esc(trunc(p.detail,80)); }
   else { lvl=t.slice(0,6).toUpperCase(); msg=''; }
   return '<div class="ln '+cls+'"><span class="seq">'+seq+'</span><span class="lvl">'+lvl+'</span><span class="msg">'+msg+'</span></div>';
 }