secrets-engine/src/secrets_engine/execution_contract.py

"""Closed, immutable metadata descriptor; shape is never execution authority.

This source-only compiler input cannot grant an action, authenticate a principal,
prove owner source authenticity or open a backend. Those integration gates remain
outside this type. Existing actions cannot borrow the new descriptor.
"""
from __future__ import annotations

import copy
import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import PurePosixPath
from urllib.parse import urlsplit
from uuid import UUID

from secrets_engine.errors import DecisionError

VERSION = "SECRETS-EXEC-20261011-v1"
HASH = re.compile(r"^sha256:[0-9a-f]{64}$")
KEYS = {
    "contract_version", "operation_id", "operation", "execution_actor", "ccr_id",
    "ccr_immutable_sha256", "execution_grant_ccr_id", "grant_immutable_sha256",
    "catalog_sha256", "source_bundle", "backend_origin", "cluster_uid",
    "operations", "operation_deadline", "backend_budget", "expected_before",
    "expected_after",
}
SOURCE_ROLES = {"engine_dispatcher", "platform_mediator", "typed_ccr", "policy_auth"}


def _refuse() -> None:
    # Do not interpolate untrusted descriptors into messages/evidence.
    raise DecisionError("invalid execution_contract descriptor")


def _closed(value, keys):
    if type(value) is not dict or set(value) != set(keys):
        _refuse()
    return value


def _text(value):
    if type(value) is not str or not value or len(value) > 4096 or any(ord(c) < 32 for c in value):
        _refuse()
    return value


def _hash(value):
    if type(value) is not str or not HASH.fullmatch(value):
        _refuse()


def _path(value, *, absolute=False):
    _text(value)
    if (value.startswith("/") != absolute or "\\" in value or "%" in value
            or "?" in value or "#" in value or ":" in value
            or any(p in {"", ".", ".."} for p in value.lstrip("/").split("/"))
            or str(PurePosixPath(value)) != value):
        _refuse()


def _validated(value):
    value = copy.deepcopy(_closed(value, KEYS))
    if value['contract_version'] != VERSION or value['operation'] != 'metadata-apply-ccr' or value['execution_actor'] != 'service:secrets-engine':
        _refuse()
    try:
        if str(UUID(_text(value['operation_id']))) != value['operation_id']:
            _refuse()
    except (ValueError, AttributeError):
        _refuse()
    for key in ['ccr_id', 'execution_grant_ccr_id', 'cluster_uid']:
        _text(value[key])
    if value['ccr_id'] == value['execution_grant_ccr_id']:
        _refuse()
    for key in ['ccr_immutable_sha256', 'grant_immutable_sha256', 'catalog_sha256', 'expected_before', 'expected_after']:
        _hash(value[key])
    sources = _closed(value['source_bundle'], SOURCE_ROLES)
    for source in sources.values():
        _closed(source, {'path', 'sha256'})
        _path(source['path'], absolute=True)
        _hash(source['sha256'])
    try:
        origin = urlsplit(_text(value['backend_origin']))
        if (origin.scheme not in {'http', 'https'} or not origin.hostname
                or origin.username is not None or origin.password is not None
                or origin.path or origin.query or origin.fragment or origin.port is None):
            _refuse()
    except ValueError:
        _refuse()
    try:
        deadline = datetime.fromisoformat(_text(value['operation_deadline']).replace('Z', '+00:00'))
        if deadline.tzinfo is None:
            _refuse()
    except ValueError:
        _refuse()
    budget = _closed(value['backend_budget'], {'ttl_seconds', 'uses'})
    if (type(budget['ttl_seconds']) is not int or not 30 <= budget['ttl_seconds'] <= 120
            or type(budget['uses']) is not int or not 1 <= budget['uses'] <= 12):
        _refuse()
    operations = value['operations']
    if type(operations) is not list or not 1 <= len(operations) < budget['uses']:
        _refuse()  # Reserve at least one cleanup use; no silent retries.
    for operation in operations:
        _closed(operation, {'method', 'path', 'payload_sha256'})
        if type(operation['method']) is not str or operation['method'] not in {'GET', 'PUT'}:
            _refuse()
        _path(operation['path'])
        # Only metadata pilot endpoint shapes, never KV/mount/token/secretid.
        if not re.fullmatch(r'(sys/policies/acl/[A-Za-z0-9_-]+|auth/[A-Za-z0-9_-]+/role/[A-Za-z0-9_-]+)', operation['path']):
            _refuse()
        _hash(operation['payload_sha256'])
    return value


def catalog_target_digest(target: dict) -> str:
    """Descriptor-local sorted JSON fingerprint, distinct from PDP digest."""
    encoded = json.dumps(target, sort_keys=True, ensure_ascii=False, allow_nan=False,
                         separators=(',', ':')).encode('utf-8')
    return 'sha256:' + hashlib.sha256(encoded).hexdigest()


@dataclass(frozen=True)
class ExecutionContract:
    _serialized: str

    @classmethod
    def parse(cls, descriptor: dict) -> 'ExecutionContract':
        validated = _validated(descriptor)
        return cls(json.dumps(validated, sort_keys=True, separators=(',', ':'), allow_nan=False))

    def as_dict(self) -> dict:
        # Revalidate even a directly constructed instance: no trusted label bypass.
        try:
            return _validated(json.loads(self._serialized))
        except (ValueError, TypeError, RecursionError):
            _refuse()


def attach_execution_contract(request: dict, contract: ExecutionContract | None) -> None:
    action = request['action']
    if action != 'execute-ccr':
        if contract is not None:
            raise DecisionError('execution_contract requires execute-ccr action')
        return
    if type(contract) is not ExecutionContract:
        raise DecisionError('execute-ccr requires typed execution_contract')
    value = contract.as_dict()
    if request['resource']['attributes']['stage'] != 'prod':
        raise DecisionError('execute-ccr requires prod stage')
    if request['tenant'] != 'tenant:platform':
        raise DecisionError('execute-ccr requires platform tenant')
    if value['catalog_sha256'] != catalog_target_digest(request['context']['catalog_target']):
        raise DecisionError('execution_contract catalog target does not match')
    request['context']['execution_contract'] = value
    request['context']['human_control'] = True


def require_execution_deadline(request: dict, *, time_window=None, bounds=()) -> None:
    """Check trusted-time freshness and served upper bounds before CAS binding."""
    contract = request.get('context', {}).get('execution_contract')
    if contract is None:
        return
    from secrets_engine.application_time import validity_bounds
    deadline = datetime.fromisoformat(contract['operation_deadline'].replace('Z', '+00:00'))
    _, upper = validity_bounds(None, time_window)
    if upper >= deadline:
        raise DecisionError('execution_contract deadline has expired')
    for bound in bounds:
        try:
            limit = datetime.fromisoformat(bound.replace('Z', '+00:00'))
            if limit.tzinfo is None or deadline > limit:
                raise DecisionError('execution_contract deadline exceeds served authority')
        except (ValueError, TypeError, AttributeError):
            raise DecisionError('execution_contract requires served deadline') from None