rein-aharness/rein_aharness/repository_grant.py

"""Versioned authority contract for repository mutation.

The grant is parsed by TaskSpec but is not yet wired into run execution.  A
supplied grant therefore causes run_task to refuse before adapter dispatch.
This keeps the contract reviewable without implying enforcement that the live
runner does not yet provide.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from typing import Any

from rein_aharness.repository_transaction import RepositoryAcceptancePolicy

REPOSITORY_GRANT_VERSION = "1"
_GRANT_KEYS = frozenset({"version", "allowed_paths", "commit_count", "publish"})
_COMMIT_COUNT_KEYS = frozenset({"min", "max"})


class RepositoryGrantError(ValueError):
    """A repository grant is missing, ambiguous, or unsupported."""


@dataclass(frozen=True)
class RepositoryGrant:
    """One explicit, local-only repository mutation authority envelope."""

    version: str
    allowed_paths: tuple[str, ...]
    min_commits: int
    max_commits: int
    publish: bool

    def __post_init__(self) -> None:
        if self.version != REPOSITORY_GRANT_VERSION:
            raise RepositoryGrantError(
                f"unsupported version {self.version!r}; "
                f"expected {REPOSITORY_GRANT_VERSION!r}"
            )
        if (
            not isinstance(self.allowed_paths, tuple)
            or not self.allowed_paths
            or any(not isinstance(path, str) for path in self.allowed_paths)
        ):
            raise RepositoryGrantError(
                "allowed_paths must be a non-empty canonical string tuple"
            )
        if tuple(sorted(set(self.allowed_paths))) != self.allowed_paths:
            raise RepositoryGrantError(
                "allowed_paths must be unique and in canonical sorted order"
            )
        _positive_int(self.min_commits, "min_commits")
        _positive_int(self.max_commits, "max_commits")
        if not isinstance(self.publish, bool):
            raise RepositoryGrantError("publish must be a boolean")
        if self.publish:
            raise RepositoryGrantError(
                "version 1 does not grant publication; publish must be false"
            )
        try:
            RepositoryAcceptancePolicy(
                allowed_paths=self.allowed_paths,
                min_commits=self.min_commits,
                max_commits=self.max_commits,
            )
        except ValueError as exc:
            raise RepositoryGrantError(str(exc)) from exc

    @classmethod
    def from_mapping(cls, value: Any) -> "RepositoryGrant":
        if not isinstance(value, dict):
            raise RepositoryGrantError("must be an object")
        _require_exact_keys(value, _GRANT_KEYS, "repository_grant")

        version = value["version"]
        if not isinstance(version, str) or version != REPOSITORY_GRANT_VERSION:
            raise RepositoryGrantError(
                f"unsupported version {version!r}; expected {REPOSITORY_GRANT_VERSION!r}"
            )

        raw_paths = value["allowed_paths"]
        if not isinstance(raw_paths, list) or not raw_paths:
            raise RepositoryGrantError("allowed_paths must be a non-empty array")
        if any(not isinstance(path, str) for path in raw_paths):
            raise RepositoryGrantError("allowed_paths entries must be strings")
        if len(set(raw_paths)) != len(raw_paths):
            raise RepositoryGrantError("allowed_paths must not contain duplicates")
        allowed_paths = tuple(sorted(raw_paths))

        raw_commits = value["commit_count"]
        if not isinstance(raw_commits, dict):
            raise RepositoryGrantError("commit_count must be an object")
        _require_exact_keys(raw_commits, _COMMIT_COUNT_KEYS, "commit_count")
        min_commits = _positive_int(raw_commits["min"], "commit_count.min")
        max_commits = _positive_int(raw_commits["max"], "commit_count.max")

        publish = value["publish"]
        if not isinstance(publish, bool):
            raise RepositoryGrantError("publish must be a boolean")
        if publish:
            raise RepositoryGrantError(
                "version 1 does not grant publication; publish must be false"
            )

        return cls(
            version=version,
            allowed_paths=allowed_paths,
            min_commits=min_commits,
            max_commits=max_commits,
            publish=publish,
        )

    @property
    def grant_id(self) -> str:
        """Return a stable digest identifier without exposing granted paths."""
        return hashlib.sha256(self._canonical_json().encode("utf-8")).hexdigest()[:32]

    def acceptance_policy(self) -> RepositoryAcceptancePolicy:
        """Translate the authority envelope into the local validator policy."""
        return RepositoryAcceptancePolicy(
            allowed_paths=self.allowed_paths,
            min_commits=self.min_commits,
            max_commits=self.max_commits,
        )

    def evidence(self) -> dict[str, Any]:
        """Return bounded, value-safe grant identity and authority facts."""
        path_payload = "\0".join(self.allowed_paths)
        return {
            "grant_id": self.grant_id,
            "acceptance_policy_id": self.acceptance_policy().policy_id,
            "version": self.version,
            "allowed_path_count": len(self.allowed_paths),
            "allowed_paths_digest": hashlib.sha256(
                path_payload.encode("utf-8", errors="replace")
            ).hexdigest(),
            "min_commits": self.min_commits,
            "max_commits": self.max_commits,
            "publish": self.publish,
        }

    def _canonical_json(self) -> str:
        return json.dumps(
            {
                "allowed_paths": self.allowed_paths,
                "commit_count": {
                    "max": self.max_commits,
                    "min": self.min_commits,
                },
                "publish": self.publish,
                "version": self.version,
            },
            sort_keys=True,
            separators=(",", ":"),
        )


def _require_exact_keys(
    value: dict[str, Any],
    expected: frozenset[str],
    context: str,
) -> None:
    if any(not isinstance(key, str) for key in value):
        raise RepositoryGrantError(f"{context} field names must be strings")
    actual = set(value)
    missing = sorted(expected - actual)
    unknown = sorted(actual - expected)
    details: list[str] = []
    if missing:
        details.append(f"missing: {', '.join(missing)}")
    if unknown:
        details.append(f"unknown: {', '.join(unknown)}")
    if details:
        raise RepositoryGrantError(f"{context} fields invalid ({'; '.join(details)})")


def _positive_int(value: Any, field_name: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        raise RepositoryGrantError(f"{field_name} must be a positive integer")
    return value