fix: make privacy scrubbing fail closed

openadapt_privacy/__init__.py

@@ -11,7 +11,7 @@
     ScrubbingProviderUnavailable,
     TextScrubbingMixin,
 )
-from openadapt_privacy.config import PrivacyConfig, config
+from openadapt_privacy.config import PrivacyConfig, ScrubbingPolicyChanged, config
 from openadapt_privacy.loaders import (
     Action,
     DictRecordingLoader,
@@ -79,11 +79,13 @@ def __dir__() -> list[str]:
     """Return the package attribute names, including lazy re-exports."""
     return sorted(set(globals()) | set(_LAZY_EXPORTS))
 
+
 __all__ = [
     # Base classes
     "Modality",
     "ScrubbingProvider",
     "ScrubbingProviderFactory",
+    "ScrubbingPolicyChanged",
     "ScrubbingProviderUnavailable",
     "TextScrubbingMixin",
     # Config

openadapt_privacy/base.py

@@ -7,12 +7,13 @@
 from __future__ import annotations
 
 import importlib
-from typing import Any, List
+from importlib import metadata
+from typing import Any, List, Optional
 
 from PIL import Image
 from pydantic import BaseModel
 
-from openadapt_privacy.config import config
+from openadapt_privacy.config import effective_config
 
 # Provider modules that must be imported before the subclass registry is read.
 # ``ScrubbingProvider.__subclasses__()`` only sees classes whose module has been
@@ -51,6 +52,36 @@ class ScrubbingProvider(BaseModel):
 
     model_config = {"arbitrary_types_allowed": True}
 
+    def validate_ready(self, modalities: list[str]) -> None:
+        """Fail before processing when this provider cannot scrub a modality."""
+        unsupported = sorted(set(modalities) - set(self.capabilities))
+        if unsupported:
+            raise ScrubbingProviderUnavailable(
+                f"Provider {self.name!r} cannot scrub required modalities {unsupported}; "
+                "no scrub was attempted."
+            )
+
+    def evidence(
+        self,
+        modalities: list[str],
+        *,
+        policy_sha256: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """Return privacy-safe provenance for a completed scrub operation."""
+        try:
+            package_version = metadata.version("openadapt-privacy")
+        except metadata.PackageNotFoundError:  # pragma: no cover - source checkout only
+            package_version = "unknown"
+        return {
+            "schema_version": 1,
+            "provider": self.name,
+            "provider_class": f"{type(self).__module__}.{type(self).__qualname__}",
+            "package_version": package_version,
+            "policy_sha256": policy_sha256 or effective_config().policy_digest(),
+            "modalities": sorted(set(modalities)),
+            "status": "completed",
+        }
+
     def scrub_text(self, text: str, is_separated: bool = False) -> str:
         """Scrub PII/PHI from text.
 
@@ -152,7 +183,7 @@ def scrub_text_all(self, text: str) -> str:
         """
         if text is None:
             return None
-        return config.SCRUB_CHAR * len(text)
+        return effective_config().SCRUB_CHAR * len(text)
 
     def scrub_dict(
         self,
@@ -178,7 +209,7 @@ def scrub_dict(
             Scrubbed dictionary with PII/PHI removed.
         """
         if list_keys is None:
-            list_keys = config.SCRUB_KEYS_HTML
+            list_keys = effective_config().SCRUB_KEYS_HTML
 
         scrubbed_dict: dict[str, Any] = {}
         for key, value in input_dict.items():

openadapt_privacy/config.py

@@ -7,8 +7,55 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass, field
-from typing import Sequence
+import hashlib
+import json
+from contextlib import contextmanager
+from contextvars import ContextVar
+from copy import deepcopy
+from dataclasses import asdict, dataclass, field
+from threading import RLock
+from typing import Iterator, Sequence
+
+
+class ScrubbingPolicyChanged(RuntimeError):
+    """A scrubber attempted to mutate policy during a scrub operation."""
+
+
+class _FrozenList(list):
+    """List-compatible policy value that refuses in-operation mutation."""
+
+    def _refuse(self, *_args, **_kwargs) -> None:
+        raise ScrubbingPolicyChanged("privacy policy is immutable during a scrub operation")
+
+    __setitem__ = __delitem__ = append = clear = extend = insert = pop = remove = reverse = sort = (
+        _refuse
+    )
+    __iadd__ = __imul__ = _refuse
+
+    def __deepcopy__(self, _memo):
+        return self
+
+
+class _FrozenDict(dict):
+    """Dict-compatible policy value that refuses in-operation mutation."""
+
+    def _refuse(self, *_args, **_kwargs) -> None:
+        raise ScrubbingPolicyChanged("privacy policy is immutable during a scrub operation")
+
+    __setitem__ = __delitem__ = clear = pop = popitem = setdefault = update = _refuse
+
+    def __deepcopy__(self, _memo):
+        return self
+
+
+def _freeze_policy_value(value):
+    if isinstance(value, dict):
+        return _FrozenDict({key: _freeze_policy_value(item) for key, item in value.items()})
+    if isinstance(value, list):
+        return _FrozenList(_freeze_policy_value(item) for item in value)
+    if isinstance(value, tuple):
+        return tuple(_freeze_policy_value(item) for item in value)
+    return value
 
 
 @dataclass
@@ -76,6 +123,66 @@ class PrivacyConfig:
     # SpaCy model name
     SPACY_MODEL_NAME: str = "en_core_web_sm"
 
+    def __setattr__(self, name: str, value) -> None:
+        if self.__dict__.get("_policy_locked", False) and not name.startswith("_"):
+            raise ScrubbingPolicyChanged("privacy policy is immutable during a scrub operation")
+        object.__setattr__(self, name, value)
+
+    def policy_digest(self) -> str:
+        """Return a stable digest of the complete effective scrub policy."""
+        payload = json.dumps(
+            asdict(self),
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+        ).encode("utf-8")
+        return hashlib.sha256(payload).hexdigest()
+
 
 # Global default configuration instance
 config = PrivacyConfig()
+
+_operation_lock = RLock()
+
+_operation_config: ContextVar[PrivacyConfig | None] = ContextVar(
+    "openadapt_privacy_operation_config",
+    default=None,
+)
+
+
+def effective_config() -> PrivacyConfig:
+    """Return the immutable-per-operation policy, or the process default."""
+    return _operation_config.get() or config
+
+
+@contextmanager
+def privacy_operation() -> Iterator[PrivacyConfig]:
+    """Bind one deep-copied policy snapshot for a complete scrub operation."""
+    existing = _operation_config.get()
+    if existing is not None:
+        yield existing
+        return
+
+    # Serialize policy admission. During the operation both the explicit
+    # snapshot and the legacy public ``config`` object are read-compatible but
+    # mutation-proof, so older third-party providers cannot silently mix two
+    # policies and still receive completed evidence.
+    with _operation_lock:
+        snapshot = deepcopy(config)
+        global_values = {name: getattr(config, name) for name in asdict(config)}
+        for name, value in asdict(snapshot).items():
+            object.__setattr__(snapshot, name, _freeze_policy_value(value))
+        object.__setattr__(snapshot, "_policy_locked", True)
+
+        for name, value in global_values.items():
+            object.__setattr__(config, name, _freeze_policy_value(deepcopy(value)))
+        object.__setattr__(config, "_policy_locked", True)
+
+        token = _operation_config.set(snapshot)
+        try:
+            yield snapshot
+        finally:
+            _operation_config.reset(token)
+            object.__setattr__(config, "_policy_locked", False)
+            for name, value in global_values.items():
+                object.__setattr__(config, name, value)

openadapt_privacy/loaders.py

@@ -36,6 +36,7 @@ def save(self, recording: Recording, path: str) -> None:
 from PIL import Image
 
 from openadapt_privacy.base import ScrubbingProvider
+from openadapt_privacy.config import privacy_operation
 
 
 class UnscrubbedScreenshot(RuntimeError):
@@ -195,22 +196,43 @@ def scrub(
         Returns:
             New Recording instance with all content scrubbed.
         """
-        scrubbed_actions = [action.scrub(scrubber) for action in self.actions]
-        scrubbed_screenshots = (
-            [screenshot.scrub(scrubber) for screenshot in self.screenshots]
-            if scrub_images
-            else self.screenshots
-        )
+        with privacy_operation() as policy:
+            policy_sha256 = policy.policy_digest()
+            required_modalities = ["TEXT"]
+            if scrub_images and self.screenshots:
+                required_modalities.append("PIL_IMAGE")
+            scrubber.validate_ready(required_modalities)
+
+            scrubbed_task_description = (
+                scrubber.scrub_text(self.task_description) if self.task_description else None
+            )
+            scrubbed_actions = [action.scrub(scrubber) for action in self.actions]
+            scrubbed_screenshots = (
+                [screenshot.scrub(scrubber) for screenshot in self.screenshots]
+                if scrub_images
+                else [
+                    Screenshot(
+                        id=screenshot.id,
+                        action_id=screenshot.action_id,
+                        timestamp=screenshot.timestamp,
+                    )
+                    for screenshot in self.screenshots
+                ]
+            )
+
+            scrubbed_metadata = scrubber.scrub_dict(self.metadata) if self.metadata else {}
+            scrubbed_metadata["_openadapt_privacy"] = {
+                **scrubber.evidence(required_modalities, policy_sha256=policy_sha256),
+                "omitted_modalities": [] if scrub_images else ["PIL_IMAGE"],
+            }
 
         return Recording(
             id=self.id,
-            task_description=(
-                scrubber.scrub_text(self.task_description) if self.task_description else None
-            ),
+            task_description=scrubbed_task_description,
             timestamp=self.timestamp,
             actions=scrubbed_actions,
             screenshots=scrubbed_screenshots,
-            metadata=scrubber.scrub_dict(self.metadata) if self.metadata else {},
+            metadata=scrubbed_metadata,
         )
 
     def iter_actions_with_screenshots(self) -> Iterator[tuple[Action, Optional[Screenshot]]]:

openadapt_privacy/providers/__init__.py

@@ -35,12 +35,12 @@ def as_options(cls) -> dict[str, str]:
 
     @classmethod
     def get_available_providers(cls) -> list[str]:
-        """Return the list of available provider IDs.
+        """Return provider IDs implemented by this installed package.
 
         Returns:
             List of provider ID strings.
         """
-        return [cls.PRESIDIO, cls.PRIVATE_AI]
+        return [cls.PRESIDIO]
 
     @classmethod
     def get_scrubber(cls, provider: str) -> "ScrubbingProvider":
@@ -55,17 +55,14 @@ def get_scrubber(cls, provider: str) -> "ScrubbingProvider":
         Raises:
             ValueError: If the provider is not supported.
         """
-        if provider not in cls.get_available_providers():
-            raise ValueError(f"Provider {provider} is not supported.")
-
         if provider == cls.PRESIDIO:
             from openadapt_privacy.providers.presidio import PresidioScrubbingProvider
 
             return PresidioScrubbingProvider()
-        elif provider == cls.PRIVATE_AI:
+        if provider == cls.PRIVATE_AI:
             raise NotImplementedError(
                 "Private AI provider is not yet implemented in openadapt-privacy. "
                 "Please use PRESIDIO or contribute an implementation."
             )
 
-        raise ValueError(f"Unknown provider: {provider}")
+        raise ValueError(f"Provider {provider} is not supported.")