335 lines
13 KiB
Python
335 lines
13 KiB
Python
"""Canonical fail-closed capability semantics for task eligibility."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any
|
|
|
|
|
|
_TASK_ALIASES: dict[str, frozenset[str]] = {
|
|
"repository_read": frozenset({"repo.read"}),
|
|
"git.read": frozenset({"repo.read"}),
|
|
"repo.read": frozenset({"repo.read"}),
|
|
"test_execution": frozenset({"test.execute"}),
|
|
"git.test": frozenset({"test.execute"}),
|
|
"test.execute": frozenset({"test.execute"}),
|
|
"governed_git_write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
|
|
"governed.git.write": frozenset({"governed.git.write.branch", "governed.git.write.file"}),
|
|
"governed.write.branch": frozenset({"governed.git.write.branch"}),
|
|
"governed.write.file": frozenset({"governed.git.write.file"}),
|
|
"shell": frozenset({"runtime.shell"}),
|
|
"git": frozenset({"runtime.git"}),
|
|
"python": frozenset({"runtime.python"}),
|
|
"code.read": frozenset({"legacy.code.read"}),
|
|
"code.review": frozenset({"legacy.code.review"}),
|
|
"code.write": frozenset({"legacy.code.write"}),
|
|
"code.write.patch": frozenset({"legacy.code.write.patch"}),
|
|
"test.python": frozenset({"legacy.test.python"}),
|
|
"test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
|
|
"git.commit": frozenset({"legacy.git.commit"}),
|
|
"git.diff.read": frozenset({"legacy.git.diff.read"}),
|
|
"git.force_push": frozenset({"legacy.git.force_push"}),
|
|
"deploy.production": frozenset({"legacy.deploy.production"}),
|
|
"secret.transform": frozenset({"legacy.secret.transform"}),
|
|
"secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
|
|
"runtime.shadow": frozenset({"legacy.runtime.shadow"}),
|
|
"review.security": frozenset({"legacy.review.security"}),
|
|
"review.api-contract": frozenset({"legacy.review.api-contract"}),
|
|
"gpu.train": frozenset({"legacy.gpu.train"}),
|
|
"agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
|
|
"agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
|
|
"release:candidate": frozenset({"release:candidate"}),
|
|
}
|
|
|
|
_WORKER_ALIASES: dict[str, frozenset[str]] = {
|
|
"git.read": frozenset({"repo.read"}),
|
|
"repo.read": frozenset({"repo.read"}),
|
|
"repository_read": frozenset({"repo.read"}),
|
|
"git.test": frozenset({"test.execute"}),
|
|
"test.execute": frozenset({"test.execute"}),
|
|
"test_execution": frozenset({"test.execute"}),
|
|
"governed.write.branch": frozenset({"governed.git.write.branch"}),
|
|
"governed.git.write.branch": frozenset({"governed.git.write.branch"}),
|
|
"governed.write.file": frozenset({"governed.git.write.file"}),
|
|
"governed.git.write.file": frozenset({"governed.git.write.file"}),
|
|
"shell": frozenset({"runtime.shell"}),
|
|
"git": frozenset({"runtime.git"}),
|
|
"python": frozenset({"runtime.python"}),
|
|
"code.read": frozenset({"legacy.code.read"}),
|
|
"code.review": frozenset({"legacy.code.review"}),
|
|
"code.write": frozenset({"legacy.code.write"}),
|
|
"code.write.patch": frozenset({"legacy.code.write.patch"}),
|
|
"test.python": frozenset({"legacy.test.python"}),
|
|
"test.shell.allowlisted": frozenset({"legacy.test.shell.allowlisted"}),
|
|
"git.commit": frozenset({"legacy.git.commit"}),
|
|
"git.diff.read": frozenset({"legacy.git.diff.read"}),
|
|
"git.force_push": frozenset({"legacy.git.force_push"}),
|
|
"deploy.production": frozenset({"legacy.deploy.production"}),
|
|
"secret.transform": frozenset({"legacy.secret.transform"}),
|
|
"secure_file_transformer": frozenset({"legacy.secure_file_transformer"}),
|
|
"runtime.shadow": frozenset({"legacy.runtime.shadow"}),
|
|
"review.security": frozenset({"legacy.review.security"}),
|
|
"review.api-contract": frozenset({"legacy.review.api-contract"}),
|
|
"gpu.train": frozenset({"legacy.gpu.train"}),
|
|
"agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
|
|
"agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
|
|
"release:candidate": frozenset({"release:candidate"}),
|
|
}
|
|
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CapabilityContract:
|
|
"""Versioned runtime contract attached to one canonical semantic capability."""
|
|
|
|
name: str
|
|
min_version: int = 1
|
|
max_version: int = 1
|
|
permissions: tuple[str, ...] = ()
|
|
forbidden: tuple[str, ...] = ()
|
|
dependencies: tuple[str, ...] = ()
|
|
health_requirements: tuple[str, ...] = ()
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
# This registry is the semantic source of truth. Alias tables below are retained
|
|
# as compatibility adapters; startup/tests validate every alias target against
|
|
# this registry so a name cannot be accepted in one layer and be unknown in
|
|
# another.
|
|
_CAPABILITY_CONTRACTS: dict[str, CapabilityContract] = {
|
|
"repo.read": CapabilityContract("repo.read", permissions=("repository_read",)),
|
|
"test.execute": CapabilityContract("test.execute", permissions=("test_execution",)),
|
|
"governed.git.write.branch": CapabilityContract(
|
|
"governed.git.write.branch",
|
|
permissions=("governed_branch_write",),
|
|
dependencies=("writer_lease", "governed_gitops"),
|
|
),
|
|
"governed.git.write.file": CapabilityContract(
|
|
"governed.git.write.file",
|
|
permissions=("governed_file_write",),
|
|
dependencies=("writer_lease", "governed_gitops"),
|
|
),
|
|
"governed.git.write": CapabilityContract(
|
|
"governed.git.write",
|
|
permissions=("governed_git_write",),
|
|
dependencies=("writer_lease", "governed_gitops"),
|
|
),
|
|
"runtime.shell": CapabilityContract("runtime.shell", permissions=("runtime_shell",)),
|
|
"runtime.git": CapabilityContract("runtime.git", permissions=("runtime_git",)),
|
|
"runtime.python": CapabilityContract("runtime.python", permissions=("runtime_python",)),
|
|
"legacy.code.read": CapabilityContract("legacy.code.read"),
|
|
"legacy.code.review": CapabilityContract("legacy.code.review"),
|
|
"legacy.code.write": CapabilityContract("legacy.code.write"),
|
|
"legacy.code.write.patch": CapabilityContract("legacy.code.write.patch"),
|
|
"legacy.test.python": CapabilityContract("legacy.test.python"),
|
|
"legacy.test.shell.allowlisted": CapabilityContract("legacy.test.shell.allowlisted"),
|
|
"legacy.git.commit": CapabilityContract("legacy.git.commit"),
|
|
"legacy.git.diff.read": CapabilityContract("legacy.git.diff.read"),
|
|
"legacy.git.force_push": CapabilityContract(
|
|
"legacy.git.force_push",
|
|
forbidden=("protected_ref_bypass",),
|
|
),
|
|
"legacy.deploy.production": CapabilityContract(
|
|
"legacy.deploy.production",
|
|
permissions=("production_deploy",),
|
|
dependencies=("owner_approval", "deployment_controller"),
|
|
health_requirements=("approval_envelope_verified", "runtime_acceptance_available"),
|
|
),
|
|
"legacy.secret.transform": CapabilityContract("legacy.secret.transform"),
|
|
"legacy.secure_file_transformer": CapabilityContract("legacy.secure_file_transformer"),
|
|
"legacy.runtime.shadow": CapabilityContract("legacy.runtime.shadow"),
|
|
"legacy.review.security": CapabilityContract("legacy.review.security"),
|
|
"legacy.review.api-contract": CapabilityContract("legacy.review.api-contract"),
|
|
"legacy.gpu.train": CapabilityContract("legacy.gpu.train"),
|
|
"agent.runtime.hermes": CapabilityContract(
|
|
"agent.runtime.hermes",
|
|
dependencies=("hermes_runtime",),
|
|
health_requirements=("worker_liveness",),
|
|
),
|
|
"agent.runtime.openclaw": CapabilityContract(
|
|
"agent.runtime.openclaw",
|
|
dependencies=("openclaw_runtime",),
|
|
health_requirements=("worker_liveness",),
|
|
),
|
|
"release:candidate": CapabilityContract(
|
|
"release:candidate",
|
|
min_version=1,
|
|
max_version=1,
|
|
permissions=("prepare_release",),
|
|
forbidden=(
|
|
"production_deploy",
|
|
"submit_owner_approval",
|
|
"tag_publish",
|
|
"baseline_registry_finalize",
|
|
"runtime_switch",
|
|
),
|
|
dependencies=(
|
|
"credential_projection",
|
|
"trusted_builder",
|
|
),
|
|
health_requirements=(
|
|
"authenticated_probe",
|
|
"release_api_health",
|
|
"worker_liveness",
|
|
),
|
|
),
|
|
}
|
|
|
|
|
|
def capability_contract(name: object) -> CapabilityContract | None:
|
|
"""Return the contract for one canonical or registered capability name."""
|
|
|
|
text = _clean(name)
|
|
direct = _CAPABILITY_CONTRACTS.get(text)
|
|
if direct is not None:
|
|
return direct
|
|
mapped = _TASK_ALIASES.get(text) or _WORKER_ALIASES.get(text)
|
|
if mapped is None or len(mapped) != 1:
|
|
return None
|
|
return _CAPABILITY_CONTRACTS.get(next(iter(mapped)))
|
|
|
|
|
|
def registered_capability(name: object, *, worker: bool = False) -> bool:
|
|
"""Fail-closed validation used by registration/task admission."""
|
|
|
|
text = _clean(name)
|
|
if "*" in text:
|
|
return False
|
|
aliases = _WORKER_ALIASES if worker else _TASK_ALIASES
|
|
mapped = aliases.get(text)
|
|
if mapped is None:
|
|
return text in _CAPABILITY_CONTRACTS
|
|
return all(item in _CAPABILITY_CONTRACTS for item in mapped)
|
|
|
|
|
|
def validate_capability_contract_registry() -> dict[str, object]:
|
|
"""Prove alias/contract closure and contract safety invariants."""
|
|
|
|
alias_targets: set[str] = set()
|
|
for mapping in (_TASK_ALIASES, _WORKER_ALIASES):
|
|
for values in mapping.values():
|
|
alias_targets.update(values)
|
|
missing = sorted(alias_targets - set(_CAPABILITY_CONTRACTS))
|
|
invalid_versions = sorted(
|
|
name
|
|
for name, contract in _CAPABILITY_CONTRACTS.items()
|
|
if contract.min_version < 1 or contract.max_version < contract.min_version
|
|
)
|
|
permission_forbidden_overlap = sorted(
|
|
name
|
|
for name, contract in _CAPABILITY_CONTRACTS.items()
|
|
if set(contract.permissions) & set(contract.forbidden)
|
|
)
|
|
return {
|
|
"schema": "aios.capability-contract-registry.v1",
|
|
"contract_count": len(_CAPABILITY_CONTRACTS),
|
|
"missing_alias_targets": missing,
|
|
"invalid_versions": invalid_versions,
|
|
"permission_forbidden_overlap": permission_forbidden_overlap,
|
|
"valid": not missing and not invalid_versions and not permission_forbidden_overlap,
|
|
}
|
|
|
|
|
|
def capability_contract_snapshot() -> dict[str, dict[str, Any]]:
|
|
return {
|
|
name: contract.as_dict()
|
|
for name, contract in sorted(_CAPABILITY_CONTRACTS.items())
|
|
}
|
|
|
|
|
|
_UNKNOWN_PREFIX = "exact:"
|
|
|
|
|
|
def _clean(value: object) -> str:
|
|
if not isinstance(value, str):
|
|
raise ValueError("capability must be a string")
|
|
text = value.strip().lower()
|
|
if not text:
|
|
raise ValueError("capability must not be empty")
|
|
return text
|
|
|
|
|
|
def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
|
|
"""Map task-facing names to one semantic vocabulary.
|
|
|
|
Unknown names remain explicit exact requirements. They are never treated
|
|
as wildcards or granted by a failed normalization.
|
|
"""
|
|
|
|
result: set[str] = set()
|
|
for value in values:
|
|
text = _clean(value)
|
|
if "*" in text:
|
|
result.add(_UNKNOWN_PREFIX + text)
|
|
else:
|
|
mapped = _TASK_ALIASES.get(text)
|
|
if mapped is None or any(item not in _CAPABILITY_CONTRACTS for item in mapped):
|
|
result.add(_UNKNOWN_PREFIX + text)
|
|
else:
|
|
result.update(mapped)
|
|
return frozenset(result)
|
|
|
|
|
|
def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
|
|
"""Map registered worker names to effective semantic capabilities."""
|
|
|
|
result: set[str] = set()
|
|
for value in values:
|
|
text = _clean(value)
|
|
if "*" in text:
|
|
continue
|
|
mapped = _WORKER_ALIASES.get(text)
|
|
if mapped is not None and all(item in _CAPABILITY_CONTRACTS for item in mapped):
|
|
result.update(mapped)
|
|
if {
|
|
"governed.git.write.branch",
|
|
"governed.git.write.file",
|
|
}.issubset(result):
|
|
result.add("governed.git.write")
|
|
return frozenset(result)
|
|
|
|
|
|
def has_capabilities(
|
|
worker_capabilities: Iterable[object],
|
|
required_capabilities: Iterable[object],
|
|
) -> bool:
|
|
"""Return the single eligibility decision used by poll and claim."""
|
|
|
|
required = normalize_required_capabilities(required_capabilities)
|
|
effective = normalize_worker_capabilities(worker_capabilities)
|
|
if any(value.startswith(_UNKNOWN_PREFIX) for value in required):
|
|
return False
|
|
return required.issubset(effective)
|
|
|
|
|
|
def required_capabilities_subset(
|
|
granted_requirements: Iterable[object],
|
|
required_requirements: Iterable[object],
|
|
) -> bool:
|
|
"""Check semantic capability containment for delegation constraints."""
|
|
granted = normalize_required_capabilities(granted_requirements)
|
|
required = normalize_required_capabilities(required_requirements)
|
|
if any(value.startswith(_UNKNOWN_PREFIX) for value in granted | required):
|
|
return False
|
|
return required.issubset(granted)
|
|
|
|
|
|
def normalized_capability_matrix(
|
|
worker_capabilities: Iterable[object],
|
|
required_capabilities: Iterable[object],
|
|
) -> dict[str, object]:
|
|
"""Expose non-secret decision evidence for tests and diagnostics."""
|
|
|
|
worker_values = tuple(worker_capabilities)
|
|
required_values = tuple(required_capabilities)
|
|
worker = normalize_worker_capabilities(worker_values)
|
|
required = normalize_required_capabilities(required_values)
|
|
return {
|
|
"worker_effective": sorted(worker),
|
|
"task_required": sorted(required),
|
|
"eligible": has_capabilities(worker_values, required_values),
|
|
}
|