grants.py

Go to the documentation of this file.

"""Durable operator-consent and risk-scoped connector grants.
 
Dependencies
------------
- oc_diagnostics (_dbg structured logging of grant creation, revocation and send checks)
- option_c_connector_contracts (ConnectorGrant, OperatorConsent - the grant and consent records the store persists)
- oc_connect.workspace (state_dir - the workspace state directory that holds the grant store)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
from collections.abc import Mapping
from pathlib import Path
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import ConnectorGrant, OperatorConsent
 
from .workspace import state_dir
 
GRANT_FILENAME = "grants.json"
 
 
def default_grant_path(root: str | Path | None = None) -> Path:
    """Return the workspace-anchored grant store path (CONNECT-02 §7.1)."""
    return state_dir(root) / GRANT_FILENAME
 
 
class GrantFileError(ValueError):
    """The grant store is unreadable or malformed."""
 
 
class GrantRequiredError(PermissionError):
    """A named grant is missing, revoked, or outside its exact scope."""
 
 
class GrantStore:
    """Atomic JSON store for grants and the consent records that created them."""
 
    def __init__(
        self,
        path: str | Path | None = None,
        *,
        root: str | Path | None = None,
    ) -> None:
        self.path = Path(path) if path is not None else default_grant_path(root)
 
    def _load(self) -> tuple[dict[str, ConnectorGrant], dict[str, OperatorConsent]]:
        if not self.path.is_file():
            return {}, {}
        try:
            payload = json.loads(self.path.read_text(encoding="utf-8"))
            if not isinstance(payload, Mapping):
                raise TypeError("not a mapping")
            grants = {
                item.grant_id: item
                for item in (
                    ConnectorGrant.model_validate_json(json.dumps(raw))
                    for raw in payload.get("grants", [])
                )
            }
            consents = {
                item.grant_id: item
                for item in (
                    OperatorConsent.model_validate_json(json.dumps(raw))
                    for raw in payload.get("consents", [])
                )
            }
        except (OSError, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc:
            raise GrantFileError(
                f"could not parse connector grant store at {self.path} "
                f"({type(exc).__name__})"
            ) from None
        return grants, consents
 
    def _write(
        self,
        grants: Mapping[str, ConnectorGrant],
        consents: Mapping[str, OperatorConsent],
    ) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.path.with_name(
            f".tmp-{self.path.name}-{_secrets.token_hex(8)}"
        )
        payload = {
            "version": 1,
            "grants": [
                grants[key].model_dump(mode="json") for key in sorted(grants)
            ],
            "consents": [
                consents[key].model_dump(mode="json") for key in sorted(consents)
            ],
        }
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                json.dump(payload, handle, indent=2, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            os.replace(temporary, self.path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def create(
        self,
        grant: ConnectorGrant,
        consent: OperatorConsent,
    ) -> ConnectorGrant:
        """Persist a grant only with explicit, matching operator consent."""
        if consent.grant_id != grant.grant_id:
            raise ValueError(
                f"operator consent for {consent.grant_id!r} does not match "
                f"grant {grant.grant_id!r}"
            )
        grants, consents = self._load()
        existing_grant = grants.get(grant.grant_id)
        existing_consent = consents.get(grant.grant_id)
        if existing_grant is not None:
            if existing_grant == grant and existing_consent == consent:
                return existing_grant
            raise ValueError(f"grant {grant.grant_id!r} already exists")
        grants[grant.grant_id] = grant
        consents[consent.grant_id] = consent
        self._write(grants, consents)
        _dbg(
            "OC_CONNECT.GRANTS",
            "create",
            f"created grant={grant.grant_id} destination={grant.destination} "
            f"risk_ceiling={grant.risk_ceiling}",
            2,
        )
        return grant
 
    def get(self, grant_id: str) -> ConnectorGrant | None:
        """Return one grant without interpreting its authorization scope."""
        grants, _consents = self._load()
        return grants.get(grant_id)
 
    def list_grants(self) -> tuple[ConnectorGrant, ...]:
        """Return all grants in stable id order."""
        grants, _consents = self._load()
        return tuple(grants[key] for key in sorted(grants))
 
    def revoke(self, grant_id: str, revoked_at_utc: str) -> ConnectorGrant:
        """Persist revocation; missing ids fail loudly."""
        if not revoked_at_utc.strip():
            raise ValueError("revocation timestamp must not be blank")
        grants, consents = self._load()
        current = grants.get(grant_id)
        if current is None:
            raise GrantRequiredError(f"cannot revoke missing grant {grant_id!r}")
        revoked = current.model_copy(update={"revoked_at_utc": revoked_at_utc})
        grants[grant_id] = revoked
        self._write(grants, consents)
        _dbg("OC_CONNECT.GRANTS", "revoke", f"revoked grant={grant_id}", 2)
        return revoked
 
    def resolve_send_risk(
        self,
        grant_id: str,
        *,
        default: str = "low",
    ) -> str:
        """Return the send risk implied by a live grant (CONNECT-02 §7.2).
 
        The send default now *follows* the operator's recorded authority
        instead of a hardcoded ``medium`` that sat above the real grant, which
        is why every plan-219 send needed an explicit ``--risk low``. This
        resolves downward only: a missing or revoked grant yields *default*
        rather than an invented ceiling, and the caller still has to pass
        :meth:`require_send_grant`, which refuses in both cases.
        """
        grant = self.get(grant_id)
        if grant is None or grant.revoked_at_utc is not None:
            _dbg(
                "OC_CONNECT.GRANTS",
                "resolve_send_risk",
                f"no live grant={grant_id}; risk falls back to {default}",
                3,
            )
            return default
        return grant.risk_ceiling
 
    def require_send_grant(
        self,
        *,
        grant_id: str,
        grantee: str,
        connector: str,
        destination: str,
        risk_class: str,
    ) -> ConnectorGrant:
        """Return an exact live send grant or fail closed with its name."""
        grant = self.get(grant_id)
        if grant is None or not grant.allows(
            grantee=grantee,
            connector=connector,
            destination=destination,
            capability="send",
            risk_class=risk_class,
        ):
            raise GrantRequiredError(
                f"send refused: required grant {grant_id!r} is missing, revoked, "
                "or does not authorize this grantee, connector, destination, and risk"
            )
        _dbg(
            "OC_CONNECT.GRANTS",
            "require_send_grant",
            f"authorized grant={grant_id} connector={connector} "
            f"destination={destination} risk={risk_class}",
            3,
        )
        return grant
 
 
__all__ = [
    "GRANT_FILENAME",
    "GrantFileError",
    "GrantRequiredError",
    "GrantStore",
    "default_grant_path",
]