informed-decision/informed_decision/lifecycle_service.py

"""Headless application service consuming fresh, exact external policy decisions.

Identity adapters supply Caller after authentication; no request body can mint
one. Runtime composition and lifecycle policy admission remain separate work.
"""

from dataclasses import dataclass
import hashlib
import json
import math
import time
import uuid

from .disposition import Actor, ActorKind
from .lifecycle import Attribution, LifecycleRepository
from .oidc import HumanSession
from .policy import PolicyError
from .provenance import Route
from .records import dumps, memo_from
from .store import _text


@dataclass(frozen=True)
class Caller:
    attribution: Attribution
    expires_at: float
    attributes_json: str

    def __post_init__(self):
        if (type(self.expires_at) not in (int, float) or not math.isfinite(self.expires_at)
                or not isinstance(self.attributes_json, str) or len(self.attributes_json) > 16384):
            raise ValueError("invalid authenticated caller")
        attributes = json.loads(self.attributes_json)
        if not isinstance(attributes, dict):
            raise ValueError("invalid caller attributes")

    @classmethod
    def from_human_session(cls, session: HumanSession):
        if (not isinstance(session, HumanSession)
                or session.principal_type.value != "human"
                or session.principal_type.route is not Route.AUTHENTICATION
                or session.tenant.route not in (Route.DIRECTORY, Route.REGISTRATION)):
            raise ValueError("verified human identity required")
        return cls(Attribution(session.tenant.value, Actor(session.subject, ActorKind.PERSON)),
                   session.expires_at, dumps({
                       "tenant_source": session.tenant.route.value,
                       "principal_type_source": session.principal_type.route.value,
                       "assurance": session.assurance,
                       "roles": list(session.roles), "groups": list(session.groups)}))


class LifecycleError(RuntimeError):
    def __init__(self, status, code):
        self.status, self.code = status, code
        super().__init__(code)


class LifecycleService:
    def __init__(self, repository: LifecycleRepository, policy, *, clock=time.time):
        self.repository, self.policy, self.clock = repository, policy, clock

    def _caller(self, caller):
        if not isinstance(caller, Caller) or caller.expires_at <= self.clock():
            raise LifecycleError(401, "identity_expired")

    def _authorize(self, caller, memo_id, operation, payload):
        self._caller(caller)
        _text(memo_id, "memo id")
        actor = caller.attribution.actor
        tenant = caller.attribution.tenant
        command = dumps(payload).encode()
        if len(command) > 256 * 1024:
            raise ValueError("lifecycle command too large")
        request = {"id": str(uuid.uuid4()), "tenant": tenant,
                   "subject": {"id": actor.sub, "type": "human" if actor.kind is ActorKind.PERSON else actor.kind.value,
                               "tenant": tenant, "attributes": json.loads(caller.attributes_json)},
                   "action": "lifecycle." + operation,
                   "resource": {"id": "memo:" + memo_id, "type": "decision-memo",
                                "system": "informed-decision", "tenant": tenant},
                   "context": {"contract": "informed-decision.lifecycle.v1",
                               "memo_version": payload.get("memo").version if payload.get("memo") is not None else payload.get("version"),
                               "command_digest": "sha256:" + hashlib.sha256(command).hexdigest()},
                   "policy_version": self.policy.version}
        observation = self.policy.check(request)
        # Defend the application boundary even with an incorrectly composed client.
        if json.loads(observation.request_json) != request:
            raise LifecycleError(503, "policy_request_mismatch")
        policy_id = self.repository.store.observe_policy(observation)

        def current():
            self._caller(caller)
            try:
                observation.require_current(self.clock())
            except PolicyError:
                raise LifecycleError(403 if observation.outcome == "policy_denied" else 503,
                                     observation.outcome if observation.outcome != "allow" else "decision_expired") from None

        def guard(db):
            current()
            recorded = db.execute("SELECT request,outcome,usable_until FROM policy_observations WHERE id=?",
                                  (policy_id,)).fetchone()
            if (recorded is None or recorded["request"] != observation.request_json
                    or recorded["outcome"] != "allow" or recorded["usable_until"] <= self.clock()):
                raise LifecycleError(503, "policy_observation_unavailable")
            return policy_id

        current()
        return current, guard

    def register(self, caller, memo, *, key):
        # Snapshot before the PDP call: mutable caller-owned objects cannot change
        # the command between authorization and persistence.
        memo = memo_from(json.loads(dumps(memo)))
        _, guard = self._authorize(caller, memo.id, "register", {"memo": memo, "key": key})
        return self.repository.register(caller.attribution, memo, key=key, guard=guard)

    def revise(self, caller, memo, *, expected_version, summary, key):
        memo = memo_from(json.loads(dumps(memo)))
        _, guard = self._authorize(caller, memo.id, "revise", {
            "memo": memo, "expected_version": expected_version, "summary": summary, "key": key})
        return self.repository.revise(caller.attribution, memo, expected_version=expected_version,
                                      summary=summary, key=key, guard=guard)

    def annotate(self, caller, memo_id, version, *, kind, body, key):
        body = json.loads(dumps(body))
        self.repository._validate_annotation(kind, body)
        _, guard = self._authorize(caller, memo_id, kind, {
            "memo_id": memo_id, "version": version, "body": body, "key": key})
        return self.repository.annotate(caller.attribution, memo_id, version, kind=kind,
                                        body=body, key=key, guard=guard)

    def retrieve(self, caller, memo_id, version=None):
        current, _ = self._authorize(caller, memo_id, "read", {"version": version})
        result = self.repository.retrieve(caller.attribution.tenant, memo_id, version)
        current()
        return result

    def events(self, caller, memo_id, *, after=0, limit=100):
        current, _ = self._authorize(caller, memo_id, "events", {"after": after, "limit": limit})
        result = self.repository.events(caller.attribution.tenant, memo_id, after=after, limit=limit)
        current()
        return result