Add repository grant v2 publish scope and adopt daily security flow
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 4s
Build and Publish Container Image / build-and-push (push) Successful in 1m30s

ACTIVITY-WP-0046: grant v2 carries a publish scope (remote, branch, path
prefixes inside allowed_paths, fast-forward only) in its canonical grant id;
v1 stays local-only with unchanged ids.

ACTIVITY-WP-0047: disabled daily-security-flow definition dispatching the
the-custodian runner as an ops_run (CUST-WP-0079-T03).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 296847@bnt-lap001
Assistant-Session: 4f997d46-1b9d-4131-9699-0c353a215155
This commit is contained in:
tegwick 2026-10-13 21:08:15 +02:00
parent f9f2640f38
commit 9737577f0b
9 changed files with 372 additions and 29 deletions

View file

@ -198,6 +198,9 @@ def _normalise_repository_grant_evidence(raw: Any) -> dict[str, Any]:
"acceptance_policy_id",
"version",
"allowed_paths_digest",
"remote",
"branch",
"pushed_head",
):
value = raw.get(key)
if isinstance(value, str):
@ -206,9 +209,10 @@ def _normalise_repository_grant_evidence(raw: Any) -> dict[str, Any]:
value = raw.get(key)
if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
result[key] = min(value, 1_000_000)
publish = raw.get("publish")
if isinstance(publish, bool):
result["publish"] = publish
for key in ("publish", "fast_forward_only", "published"):
value = raw.get(key)
if isinstance(value, bool):
result[key] = value
return result

View file

@ -1,5 +1,8 @@
"""Typed repository mutation authority carried by Activity Core.
Version 1 is local-only. Version 2 adds an explicit publish scope (remote,
branch, path prefixes, fast-forward only) for reins that push their result.
The v1 schema mirrors the accepted rein-aharness repository-grant contract,
but Activity Core owns admission and unchanged queue carriage at its boundary.
It deliberately does not infer grants from prose, labels, profiles, or
@ -10,6 +13,7 @@ from __future__ import annotations
import hashlib
import json
import re
from typing import Any, Literal
from pydantic import (
@ -42,15 +46,69 @@ class CommitCount(BaseModel):
return self
class RepositoryGrant(BaseModel):
"""Versioned, local-only repository authority envelope."""
_REMOTE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,199}$")
class PublishScope(BaseModel):
"""Where a version-2 grant may push: one remote branch, fast-forward only."""
model_config = ConfigDict(extra="forbid", frozen=True)
version: Literal["1"]
remote: str
branch: str
path_prefixes: list[str] = Field(min_length=1, max_length=20)
fast_forward_only: Literal[True]
@field_validator("remote")
@classmethod
def _validate_remote(cls, value: str) -> str:
if not _REMOTE_RE.fullmatch(value):
raise ValueError(f"remote must be a plain remote name: {value!r}")
return value
@field_validator("branch")
@classmethod
def _validate_branch(cls, value: str) -> str:
if (
not _BRANCH_RE.fullmatch(value)
or value == "HEAD"
or value.startswith("refs/")
or ".." in value
or "//" in value
or value.endswith(("/", ".", ".lock"))
):
raise ValueError(f"branch must be a plain branch name: {value!r}")
return value
@field_validator("path_prefixes")
@classmethod
def _validate_path_prefixes(cls, value: list[str]) -> list[str]:
if len(set(value)) != len(value):
raise ValueError("path_prefixes must not contain duplicates")
for pattern in value:
_validate_path_pattern(pattern)
return value
def canonical(self) -> dict[str, Any]:
return {
"branch": self.branch,
"fast_forward_only": self.fast_forward_only,
"path_prefixes": sorted(self.path_prefixes),
"remote": self.remote,
}
class RepositoryGrant(BaseModel):
"""Versioned repository authority envelope (v1 local-only, v2 publish)."""
model_config = ConfigDict(extra="forbid", frozen=True)
version: Literal["1", "2"]
allowed_paths: list[str] = Field(min_length=1, max_length=100)
commit_count: CommitCount
publish: StrictBool
publish_scope: PublishScope | None = None
@field_validator("allowed_paths")
@classmethod
@ -61,33 +119,44 @@ class RepositoryGrant(BaseModel):
_validate_path_pattern(pattern)
return value
@field_validator("publish")
@classmethod
def _local_only(cls, value: bool) -> bool:
if value:
@model_validator(mode="after")
def _publish_matches_version(self) -> RepositoryGrant:
if self.version == "1":
if self.publish or self.publish_scope is not None:
raise ValueError(
"version 1 does not grant publication; publish must be false"
)
return self
if not self.publish or self.publish_scope is None:
raise ValueError(
"version 1 does not grant publication; publish must be false"
"version 2 requires publish: true and a publish_scope"
)
return value
for prefix in self.publish_scope.path_prefixes:
if not any(_covers(allowed, prefix) for allowed in self.allowed_paths):
raise ValueError(
f"publish path prefix {prefix!r} is outside allowed_paths"
)
return self
def payload(self) -> dict[str, Any]:
"""Return the admitted field without semantic rewriting or inference."""
return self.model_dump(mode="python")
return self.model_dump(mode="python", exclude_none=True)
@property
def grant_id(self) -> str:
"""Match rein-aharness's canonical, path-order-independent grant id."""
return _digest(
{
"allowed_paths": sorted(self.allowed_paths),
"commit_count": {
"max": self.commit_count.max,
"min": self.commit_count.min,
},
"publish": self.publish,
"version": self.version,
}
)[:32]
identity: dict[str, Any] = {
"allowed_paths": sorted(self.allowed_paths),
"commit_count": {
"max": self.commit_count.max,
"min": self.commit_count.min,
},
"publish": self.publish,
"version": self.version,
}
if self.publish_scope is not None:
identity["publish_scope"] = self.publish_scope.canonical()
return _digest(identity)[:32]
@property
def acceptance_policy_id(self) -> str:
@ -128,6 +197,12 @@ def _validate_path_pattern(pattern: str) -> None:
raise ValueError("path patterns cannot grant protected .git metadata")
def _covers(allowed: str, prefix: str) -> bool:
allowed_parts = allowed.rstrip("/").split("/")
prefix_parts = prefix.rstrip("/").split("/")
return prefix_parts[: len(allowed_parts)] == allowed_parts
def _digest(value: dict[str, Any]) -> str:
payload = json.dumps(value, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(payload.encode("utf-8")).hexdigest()