Compare commits

..
15 changed files with 1056 additions and 971 deletions
@@ -0,0 +1,352 @@
"""Cryptographic Owner-approval boundary for governed existing-branch publication.
Only a complete signed envelope from an authoritative trust-root adapter is
accepted. References, chat assertions, and locally supplied keys are not an
approval. Missing public-key or revocation evidence fails closed.
"""
from __future__ import annotations
import base64
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
from typing import Any, Mapping, Protocol
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
APPROVAL_SCHEMA = "aios.existing-branch-owner-approval.v1"
PUBLISH_OPERATION = "publishExistingBranchGoverned"
OID_RE = re.compile(r"^[0-9a-f]{40}$")
DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@:+/-]{0,254}$")
BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
class OwnerApprovalRejected(RuntimeError):
"""Stable, audit-safe error for a rejected Owner approval."""
def __init__(self, code: str) -> None:
super().__init__(code)
self.code = code
def canonical_json(value: Any) -> bytes:
"""Encode a signed payload without implementation-dependent spacing."""
try:
encoded = json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
allow_nan=False,
)
except (TypeError, ValueError) as exc:
raise OwnerApprovalRejected("OWNER_APPROVAL_CANONICALIZATION_REJECTED") from exc
return (encoded + "\n").encode("utf-8")
def digest_json(value: Any) -> str:
return hashlib.sha256(canonical_json(value)).hexdigest()
def _require_identifier(field: str, value: object) -> str:
if not isinstance(value, str) or not IDENTIFIER_RE.fullmatch(value):
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
return value
def _require_digest(field: str, value: object) -> str:
if not isinstance(value, str) or not DIGEST_RE.fullmatch(value):
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
return value
def _require_oid(field: str, value: object) -> str:
if not isinstance(value, str) or not OID_RE.fullmatch(value):
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
return value
def _parse_utc(field: str, value: object) -> datetime:
if not isinstance(value, str):
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
try:
instant = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED") from exc
if instant.tzinfo is None:
raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
return instant.astimezone(timezone.utc)
@dataclass(frozen=True, slots=True)
class PublicationApprovalBinding:
repository: str
branch: str
expected_branch_oid: str
operation: str
request_digest: str
idempotency_digest: str
def validate(self) -> "PublicationApprovalBinding":
_require_identifier("repository", self.repository)
if not BRANCH_RE.fullmatch(self.branch):
raise OwnerApprovalRejected("OWNER_APPROVAL_BRANCH_REJECTED")
_require_oid("expected_branch_oid", self.expected_branch_oid)
if self.operation != PUBLISH_OPERATION:
raise OwnerApprovalRejected("OWNER_APPROVAL_OPERATION_REJECTED")
_require_digest("request_digest", self.request_digest)
_require_digest("idempotency_digest", self.idempotency_digest)
return self
@property
def as_dict(self) -> dict[str, str]:
return {
"repository": self.repository,
"branch": self.branch,
"expected_branch_oid": self.expected_branch_oid,
"operation": self.operation,
"request_digest": self.request_digest,
"idempotency_digest": self.idempotency_digest,
}
@dataclass(frozen=True, slots=True)
class OwnerApprovalEnvelope:
approval_id: str
repository: str
branch: str
expected_branch_oid: str
operation: str
issued_at: str
not_before: str
expires_at: str
nonce: str
owner_identity: str
trust_domain: str
key_id: str
request_digest: str
idempotency_digest: str
signature: str
@property
def canonical_payload(self) -> dict[str, str]:
return {
"schema": APPROVAL_SCHEMA,
"approval_id": self.approval_id,
"repository": self.repository,
"branch": self.branch,
"expected_branch_oid": self.expected_branch_oid,
"operation": self.operation,
"issued_at": self.issued_at,
"not_before": self.not_before,
"expires_at": self.expires_at,
"nonce": self.nonce,
"owner_identity": self.owner_identity,
"trust_domain": self.trust_domain,
"key_id": self.key_id,
"request_digest": self.request_digest,
"idempotency_digest": self.idempotency_digest,
}
def canonical_bytes(self) -> bytes:
return canonical_json(self.canonical_payload)
@property
def fingerprint(self) -> str:
return hashlib.sha256(self.canonical_bytes()).hexdigest()
@property
def binding(self) -> PublicationApprovalBinding:
return PublicationApprovalBinding(
repository=self.repository,
branch=self.branch,
expected_branch_oid=self.expected_branch_oid,
operation=self.operation,
request_digest=self.request_digest,
idempotency_digest=self.idempotency_digest,
)
@classmethod
def from_mapping(cls, value: object) -> "OwnerApprovalEnvelope":
if not isinstance(value, Mapping):
raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REQUIRED")
expected_fields = {
"schema", "approval_id", "repository", "branch", "expected_branch_oid",
"operation", "issued_at", "not_before", "expires_at", "nonce",
"owner_identity", "trust_domain", "key_id", "request_digest",
"idempotency_digest", "signature",
}
if set(value) != expected_fields or value.get("schema") != APPROVAL_SCHEMA:
raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REJECTED")
envelope = cls(
approval_id=_require_identifier("approval_id", value["approval_id"]),
repository=_require_identifier("repository", value["repository"]),
branch=str(value["branch"]),
expected_branch_oid=_require_oid("expected_branch_oid", value["expected_branch_oid"]),
operation=str(value["operation"]),
issued_at=str(value["issued_at"]),
not_before=str(value["not_before"]),
expires_at=str(value["expires_at"]),
nonce=_require_identifier("nonce", value["nonce"]),
owner_identity=_require_identifier("owner_identity", value["owner_identity"]),
trust_domain=_require_identifier("trust_domain", value["trust_domain"]),
key_id=_require_identifier("key_id", value["key_id"]),
request_digest=_require_digest("request_digest", value["request_digest"]),
idempotency_digest=_require_digest("idempotency_digest", value["idempotency_digest"]),
signature=str(value["signature"]),
)
envelope.binding.validate()
_parse_utc("issued_at", envelope.issued_at)
_parse_utc("not_before", envelope.not_before)
_parse_utc("expires_at", envelope.expires_at)
if not envelope.signature:
raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED")
return envelope
@dataclass(frozen=True, slots=True)
class TrustRootEvidence:
"""Public-key and revocation evidence from the authoritative trust root."""
key_id: str
owner_identity: str
trust_domain: str
public_key: bytes | None
authority_confirmed: bool
revocation_state: str
class AuthoritativeTrustRootAdapter(Protocol):
def resolve(
self, *, key_id: str, owner_identity: str, trust_domain: str
) -> TrustRootEvidence | None: ...
class ReplayLedger(Protocol):
def consume(
self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
) -> None: ...
class DurableNonceReplayLedger:
"""Single-use nonce storage that remains conservative across restarts."""
def __init__(self, root: Path) -> None:
self.root = Path(root)
def consume(
self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
) -> None:
binding.validate()
if self.root.is_symlink():
raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
try:
self.root.mkdir(parents=True, exist_ok=True)
if not self.root.is_dir() or self.root.is_symlink():
raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
name = hashlib.sha256(approval.nonce.encode("utf-8")).hexdigest()
record_path = self.root / name
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(record_path, flags, 0o600)
except FileExistsError as exc:
raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_REJECTED") from exc
except OwnerApprovalRejected:
raise
except OSError as exc:
raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
record = {
"schema": "aios.existing-branch-owner-approval-replay.v1",
"nonce_sha256": name,
"approval_fingerprint": approval.fingerprint,
"binding_digest": digest_json(binding.as_dict),
}
try:
with os.fdopen(descriptor, "wb", closefd=True) as stream:
stream.write(canonical_json(record))
stream.flush()
os.fsync(stream.fileno())
directory = os.open(self.root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(directory)
finally:
os.close(directory)
except OSError as exc:
raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
class OwnerApprovalVerifier:
def __init__(
self,
trust_root: AuthoritativeTrustRootAdapter,
*,
allowed_trust_domains: set[str] | frozenset[str],
maximum_lifetime_seconds: int = 900,
) -> None:
if not allowed_trust_domains or maximum_lifetime_seconds < 1:
raise ValueError("OWNER_APPROVAL_VERIFIER_CONFIGURATION_REJECTED")
self.trust_root = trust_root
self.allowed_trust_domains = frozenset(allowed_trust_domains)
self.maximum_lifetime_seconds = maximum_lifetime_seconds
def verify(
self,
value: object,
binding: PublicationApprovalBinding,
*,
now: datetime | None = None,
) -> OwnerApprovalEnvelope:
binding.validate()
approval = OwnerApprovalEnvelope.from_mapping(value)
if approval.binding != binding:
raise OwnerApprovalRejected("OWNER_APPROVAL_BINDING_REJECTED")
instant = datetime.now(timezone.utc) if now is None else now
if instant.tzinfo is None:
raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
instant = instant.astimezone(timezone.utc)
issued_at = _parse_utc("issued_at", approval.issued_at)
not_before = _parse_utc("not_before", approval.not_before)
expires_at = _parse_utc("expires_at", approval.expires_at)
if not issued_at <= not_before <= expires_at:
raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
if not not_before <= instant <= expires_at:
raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
if (expires_at - issued_at).total_seconds() > self.maximum_lifetime_seconds:
raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
if approval.trust_domain not in self.allowed_trust_domains:
raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_DOMAIN_REJECTED")
evidence = self.trust_root.resolve(
key_id=approval.key_id,
owner_identity=approval.owner_identity,
trust_domain=approval.trust_domain,
)
if evidence is None or not evidence.authority_confirmed:
raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
if (
evidence.key_id != approval.key_id
or evidence.owner_identity != approval.owner_identity
or evidence.trust_domain != approval.trust_domain
):
raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_BINDING_REJECTED")
if evidence.revocation_state == "REVOKED":
raise OwnerApprovalRejected("OWNER_APPROVAL_REVOKED")
if evidence.revocation_state != "ACTIVE":
raise OwnerApprovalRejected("OWNER_APPROVAL_REVOCATION_EVIDENCE_MISSING")
if evidence.public_key is None or len(evidence.public_key) != 32:
raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
try:
signature = base64.b64decode(approval.signature, validate=True)
Ed25519PublicKey.from_public_bytes(evidence.public_key).verify(
signature, approval.canonical_bytes()
)
except (ValueError, TypeError, InvalidSignature) as exc:
raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED") from exc
return approval
@@ -0,0 +1,331 @@
"""Narrow governed publisher for an already-existing task branch.
Its only writing port is a compare-and-swap publication after a complete
signed Owner approval, a stable remote observation, and a single-use nonce
claim. It neither owns trust material nor accepts direct reference strings.
"""
from __future__ import annotations
from dataclasses import dataclass
import re
from typing import Any, Mapping, Protocol
from aios_existing_branch_approval import (
OwnerApprovalEnvelope,
OwnerApprovalRejected,
OwnerApprovalVerifier,
PublicationApprovalBinding,
ReplayLedger,
canonical_json,
digest_json,
)
REQUEST_SCHEMA = "aios.existing-branch-publication.v1"
PUBLISH_OPERATION = "publishExistingBranchGoverned"
OID_RE = re.compile(r"^[0-9a-f]{40}$")
BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
class ExistingBranchPublicationRejected(RuntimeError):
def __init__(self, code: str) -> None:
super().__init__(code)
self.code = code
class RemoteOidConflict(RuntimeError):
pass
class RemoteObservationAmbiguous(RuntimeError):
pass
@dataclass(frozen=True, slots=True)
class ExistingBranchPublicationRequest:
repository: str
branch: str
expected_branch_oid: str
idempotency_key: str
payload: Mapping[str, Any]
@classmethod
def from_mapping(cls, value: object) -> "ExistingBranchPublicationRequest":
if not isinstance(value, Mapping):
raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
expected_fields = {
"schema",
"repository",
"branch",
"expected_branch_oid",
"operation",
"idempotency_key",
"payload",
}
if set(value) != expected_fields or value.get("schema") != REQUEST_SCHEMA:
raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
repository = value["repository"]
branch = value["branch"]
expected_oid = value["expected_branch_oid"]
key = value["idempotency_key"]
payload = value["payload"]
if (
not isinstance(repository, str)
or repository != "aios/aios"
or not BRANCH_RE.fullmatch(str(branch))
or not isinstance(expected_oid, str)
or not OID_RE.fullmatch(expected_oid)
or value.get("operation") != PUBLISH_OPERATION
or not isinstance(key, str)
or not key.strip()
or not isinstance(payload, Mapping)
):
raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
try:
canonical_json(payload)
except OwnerApprovalRejected as exc:
raise ExistingBranchPublicationRejected("PUBLICATION_PAYLOAD_REJECTED") from exc
return cls(repository, str(branch), expected_oid, key, dict(payload))
@property
def request_digest(self) -> str:
return digest_json(
{
"schema": REQUEST_SCHEMA,
"repository": self.repository,
"branch": self.branch,
"expected_branch_oid": self.expected_branch_oid,
"operation": PUBLISH_OPERATION,
"payload": self.payload,
}
)
@property
def idempotency_digest(self) -> str:
return digest_json(
{
"idempotency_key": self.idempotency_key,
"request_digest": self.request_digest,
}
)
@property
def approval_binding(self) -> PublicationApprovalBinding:
return PublicationApprovalBinding(
repository=self.repository,
branch=self.branch,
expected_branch_oid=self.expected_branch_oid,
operation=PUBLISH_OPERATION,
request_digest=self.request_digest,
idempotency_digest=self.idempotency_digest,
)
@dataclass(frozen=True, slots=True)
class RemoteBranchObservation:
state: str
git_oid: str | None
api_oid: str | None
expected_commit_visible: bool | None
@dataclass(frozen=True, slots=True)
class RemotePublicationEvidence:
authority_oid: str | None
gitea_git_oid: str | None
gitea_api_oid: str | None
commit_visible: bool | None
class ExistingBranchRemotePort(Protocol):
def observe(self, branch: str) -> RemoteBranchObservation: ...
def publish_compare_and_swap(
self, branch: str, expected_branch_oid: str, commit_oid: str
) -> RemotePublicationEvidence: ...
class AuthorityBranchPort(Protocol):
def read_branch_oid(self, branch: str) -> str | None: ...
class CredentialProjection(Protocol):
def project(self) -> None: ...
def revoke(self) -> None: ...
class PublicationWorkspace(Protocol):
def head_oid(self) -> str: ...
def commit_payload(self, payload: Mapping[str, Any]) -> str: ...
class DetachedWorkspaceFactory(Protocol):
def open_detached(self, expected_branch_oid: str) -> PublicationWorkspace: ...
class AuditSink(Protocol):
def record(self, event: Mapping[str, Any]) -> None: ...
class ExistingBranchPublisher:
"""Validate once, observe once, then make one bounded publication attempt."""
def __init__(
self,
*,
verifier: OwnerApprovalVerifier,
replay_ledger: ReplayLedger,
authority: AuthorityBranchPort,
remote: ExistingBranchRemotePort,
credentials: CredentialProjection,
workspaces: DetachedWorkspaceFactory,
audit: AuditSink,
) -> None:
self.verifier = verifier
self.replay_ledger = replay_ledger
self.authority = authority
self.remote = remote
self.credentials = credentials
self.workspaces = workspaces
self.audit = audit
def _audit(
self,
*,
status: str,
request: ExistingBranchPublicationRequest,
approval: OwnerApprovalEnvelope | None = None,
commit_oid: str | None = None,
) -> None:
event: dict[str, Any] = {
"event": "governed_existing_branch_publication",
"status": status,
"repository": request.repository,
"branch": request.branch,
"expected_branch_oid": request.expected_branch_oid,
"operation": PUBLISH_OPERATION,
"request_digest": request.request_digest,
"idempotency_digest": request.idempotency_digest,
}
if approval is not None:
event.update(
{
"approval_id": approval.approval_id,
"owner_identity": approval.owner_identity,
"trust_domain": approval.trust_domain,
"approval_fingerprint": approval.fingerprint,
}
)
if commit_oid is not None:
event["commit_oid"] = commit_oid
self.audit.record(event)
def _preflight_remote(
self, request: ExistingBranchPublicationRequest
) -> RemoteBranchObservation | dict[str, str]:
try:
authority_oid = self.authority.read_branch_oid(request.branch)
except Exception:
return {"status": "RECONCILIATION_REQUIRED"}
if authority_oid != request.expected_branch_oid:
return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
try:
observation = self.remote.observe(request.branch)
except RemoteObservationAmbiguous:
return {"status": "RECONCILIATION_REQUIRED"}
if observation.state == "AMBIGUOUS":
return {"status": "RECONCILIATION_REQUIRED"}
if observation.state != "PRESENT":
return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
if (
observation.git_oid is not None
and observation.api_oid is not None
and observation.git_oid != observation.api_oid
):
return {"status": "RECONCILIATION_REQUIRED"}
if (
observation.git_oid != request.expected_branch_oid
or observation.api_oid != request.expected_branch_oid
):
return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
if observation.expected_commit_visible is not True:
return {"status": "RECONCILIATION_REQUIRED"}
return observation
def publish(
self, request_value: object, approval_value: object
) -> dict[str, str | bool]:
request = ExistingBranchPublicationRequest.from_mapping(request_value)
try:
approval = self.verifier.verify(approval_value, request.approval_binding)
except OwnerApprovalRejected:
self._audit(status="REJECTED", request=request)
raise
preflight = self._preflight_remote(request)
if isinstance(preflight, dict):
self._audit(status=preflight["status"], request=request, approval=approval)
return preflight
self.replay_ledger.consume(approval, request.approval_binding)
workspace = self.workspaces.open_detached(request.expected_branch_oid)
if workspace.head_oid() != request.expected_branch_oid:
self._audit(status="REJECTED", request=request, approval=approval)
raise ExistingBranchPublicationRejected("DETACHED_WORKSPACE_HEAD_REJECTED")
projected = False
try:
self.credentials.project()
projected = True
commit_oid = workspace.commit_payload(request.payload)
if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
raise ExistingBranchPublicationRejected("PUBLICATION_COMMIT_REJECTED")
try:
evidence = self.remote.publish_compare_and_swap(
request.branch, request.expected_branch_oid, commit_oid
)
except RemoteOidConflict:
self._audit(
status="MANUAL_OWNER_REVIEW_REQUIRED",
request=request,
approval=approval,
commit_oid=commit_oid,
)
return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
except RemoteObservationAmbiguous:
self._audit(
status="RECONCILIATION_REQUIRED",
request=request,
approval=approval,
commit_oid=commit_oid,
)
return {"status": "RECONCILIATION_REQUIRED"}
if (
evidence.authority_oid != commit_oid
or evidence.gitea_git_oid != commit_oid
or evidence.gitea_api_oid != commit_oid
or evidence.commit_visible is not True
):
self._audit(
status="RECONCILIATION_REQUIRED",
request=request,
approval=approval,
commit_oid=commit_oid,
)
return {"status": "RECONCILIATION_REQUIRED"}
self._audit(
status="PUBLISHED",
request=request,
approval=approval,
commit_oid=commit_oid,
)
return {
"status": "PUBLISHED",
"commit_oid": commit_oid,
"authority_oid": evidence.authority_oid,
"gitea_git_oid": evidence.gitea_git_oid,
"gitea_api_oid": evidence.gitea_api_oid,
"commit_visible": evidence.commit_visible,
}
finally:
if projected:
self.credentials.revoke()
-86
View File
@@ -1,86 +0,0 @@
"""Narrow, non-root client contract for the privileged signer broker."""
from __future__ import annotations
import base64
import hashlib
import json
from pathlib import Path
import re
import socket
from typing import Any, Protocol
DEFAULT_SIGNER_BROKER_SOCKET = Path("/run/aios/signer-broker.sock")
MAX_SIGNING_PAYLOAD = 1024 * 1024
CODE_RE = re.compile(r"^[A-Z0-9_.:-]{3,96}$")
class SignerBrokerError(RuntimeError):
def __init__(self, code: str):
self.code = code
super().__init__(code)
class SignerBroker(Protocol):
def self_test(self) -> None: ...
def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]: ...
class UnixSignerBroker:
"""AF_UNIX request client; it never accepts commands, paths, or shell text."""
def __init__(self, socket_path: Path = DEFAULT_SIGNER_BROKER_SOCKET, timeout: float = 15.0):
path = Path(socket_path)
if path != DEFAULT_SIGNER_BROKER_SOCKET or not path.is_absolute():
raise SignerBrokerError("SIGNER_BROKER_SOCKET_INVALID")
self.socket_path = path
self.timeout = timeout
def _request(self, operation: str, **payload: Any) -> dict[str, Any]:
if operation not in {"self_test", "sign"}:
raise SignerBrokerError("SIGNER_OPERATION_REJECTED")
request = {
"schema": "aios.signer-broker.request.v1",
"operation": operation,
**payload,
}
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel:
channel.settimeout(self.timeout)
channel.connect(str(self.socket_path))
channel.sendall((json.dumps(request, sort_keys=True, separators=(",", ":")) + "\n").encode())
raw = channel.makefile("rb").readline()
except (OSError, TimeoutError) as exc:
raise SignerBrokerError("SIGNER_BROKER_UNAVAILABLE") from exc
try:
response = json.loads(raw.decode())
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID") from exc
if not isinstance(response, dict) or response.get("schema") != "aios.signer-broker.response.v1":
raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID")
if response.get("status") != "PASS":
code = str(response.get("error_code") or "SIGNER_BROKER_REJECTED")
raise SignerBrokerError(code if CODE_RE.fullmatch(code) else "SIGNER_BROKER_REJECTED")
return response
def self_test(self) -> None:
self._request("self_test")
def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
if payload_kind not in {"approval_plan", "grant_payload"}:
raise SignerBrokerError("SIGNER_PAYLOAD_KIND_REJECTED")
if not isinstance(payload, bytes) or not payload or len(payload) > MAX_SIGNING_PAYLOAD:
raise SignerBrokerError("SIGNER_PAYLOAD_INVALID")
response = self._request(
"sign",
payload_kind=payload_kind,
payload_sha256=hashlib.sha256(payload).hexdigest(),
payload_base64=base64.b64encode(payload).decode("ascii"),
)
try:
signature = base64.b64decode(str(response["signature_base64"]), validate=True)
key_id = str(response["key_id"]).strip()
except (KeyError, ValueError) as exc:
raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID") from exc
if not signature or not key_id or len(key_id) > 256:
raise SignerBrokerError("SIGNER_BROKER_RESPONSE_INVALID")
return signature, key_id
@@ -1,358 +0,0 @@
"""Privileged, fixed-function signer broker service."""
from __future__ import annotations
import argparse
import base64
import binascii
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import socket
import socketserver
import stat
import struct
import subprocess
import tempfile
import threading
from typing import Any, Callable, Mapping, Protocol, Sequence
REQUEST_SCHEMA = "aios.signer-broker.request.v1"
RESPONSE_SCHEMA = "aios.signer-broker.response.v1"
AUDIT_SCHEMA = "aios.signer-broker.audit.v1"
DEFAULT_SOCKET_PATH = Path("/run/aios/signer-broker.sock")
DEFAULT_SIGNER_PATH = Path("/usr/local/libexec/aios-grant-signer-adapter")
DEFAULT_AUDIT_PATH = Path("/var/lib/aios-gitops/audit/signer-broker.jsonl")
DEFAULT_STATE_ROOT = Path("/var/lib/aios-gitops/signer-broker")
ALLOWED_OPERATIONS = frozenset({"self_test", "sign"})
ALLOWED_PAYLOAD_KINDS = frozenset({"approval_plan", "grant_payload"})
MAX_SIGNING_PAYLOAD = 1024 * 1024
MAX_REQUEST_BYTES = 2 * MAX_SIGNING_PAYLOAD + 8192
CODE_RE = re.compile(r"^[A-Z0-9_.:-]{3,96}$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
class SignerBrokerServiceError(RuntimeError):
def __init__(self, code: str):
self.code = code if CODE_RE.fullmatch(code) else "SIGNER_INTERNAL_ERROR"
super().__init__(self.code)
class SignerRunner(Protocol):
def self_test(self) -> None: ...
def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]: ...
@dataclass(frozen=True, slots=True)
class SignerBrokerPolicy:
socket_path: Path = DEFAULT_SOCKET_PATH
signer_path: Path = DEFAULT_SIGNER_PATH
audit_path: Path = DEFAULT_AUDIT_PATH
state_root: Path = DEFAULT_STATE_ROOT
allowed_uids: frozenset[int] = frozenset()
allowed_gids: frozenset[int] = frozenset()
signer_timeout: float = 60.0
max_payload_bytes: int = MAX_SIGNING_PAYLOAD
def __post_init__(self) -> None:
for name in ("socket_path", "signer_path", "audit_path", "state_root"):
if not Path(getattr(self, name)).is_absolute():
raise SignerBrokerServiceError("SIGNER_POLICY_PATH_INVALID")
if (
not self.allowed_uids
or not self.allowed_gids
or any(not isinstance(uid, int) or uid < 0 for uid in self.allowed_uids)
or any(not isinstance(gid, int) or gid < 0 for gid in self.allowed_gids)
):
raise SignerBrokerServiceError("SIGNER_POLICY_CALLER_INVALID")
if not 0 < self.signer_timeout <= 300:
raise SignerBrokerServiceError("SIGNER_POLICY_TIMEOUT_INVALID")
if not 0 < self.max_payload_bytes <= MAX_SIGNING_PAYLOAD:
raise SignerBrokerServiceError("SIGNER_POLICY_PAYLOAD_LIMIT_INVALID")
class FixedSignerRunner:
"""Use one configured signer executable and broker-owned temporary paths."""
def __init__(self, policy: SignerBrokerPolicy):
self.policy = policy
def _run(self, argv: Sequence[str]) -> None:
try:
result = subprocess.run(list(argv), capture_output=True, text=True, check=False, timeout=self.policy.signer_timeout)
except FileNotFoundError as exc:
raise SignerBrokerServiceError("SIGNER_UNAVAILABLE") from exc
except subprocess.TimeoutExpired as exc:
raise SignerBrokerServiceError("SIGNER_TIMEOUT") from exc
except OSError as exc:
raise SignerBrokerServiceError("SIGNER_UNAVAILABLE") from exc
if result.returncode != 0:
raise SignerBrokerServiceError("SIGNER_FAILED")
def self_test(self) -> None:
self._run((str(self.policy.signer_path), "self-test"))
def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
if payload_kind not in ALLOWED_PAYLOAD_KINDS:
raise SignerBrokerServiceError("SIGNER_PAYLOAD_KIND_REJECTED")
try:
self.policy.state_root.mkdir(parents=True, exist_ok=True)
os.chmod(self.policy.state_root, 0o700)
with tempfile.TemporaryDirectory(prefix="request-", dir=self.policy.state_root) as directory:
root = Path(directory)
input_path = root / "payload.canonical"
signature_path = root / "payload.sig"
key_id_path = root / "payload.key-id"
input_path.write_bytes(payload)
os.chmod(input_path, 0o600)
self._run((str(self.policy.signer_path), "sign", "--namespace", payload_kind, "--input", str(input_path), "--signature-out", str(signature_path), "--key-id-out", str(key_id_path)))
signature = signature_path.read_bytes()
key_id = key_id_path.read_text(encoding="utf-8").strip()
except SignerBrokerServiceError:
raise
except (OSError, UnicodeError) as exc:
raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID") from exc
if not signature or not key_id or len(key_id) > 256 or "\n" in key_id or "\r" in key_id:
raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID")
return signature, key_id
class SignerBrokerService:
def __init__(self, policy: SignerBrokerPolicy, runner: SignerRunner | None = None, *, now: Callable[[], datetime] | None = None) -> None:
self.policy = policy
self.runner = runner or FixedSignerRunner(policy)
self.now = now or (lambda: datetime.now(timezone.utc))
self._audit_lock = threading.Lock()
self._signer_lock = threading.Lock()
@staticmethod
def _operation(value: Any) -> str:
return value if isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_.:-]{1,64}", value) else "INVALID"
def _audit(self, *, peer_uid: int, peer_gid: int, operation: str, status: str, error_code: str | None = None, payload_sha256: str | None = None) -> None:
event: dict[str, Any] = {"schema": AUDIT_SCHEMA, "timestamp": self.now().astimezone(timezone.utc).isoformat(), "peer_uid": peer_uid, "peer_gid": peer_gid, "operation": self._operation(operation), "status": status}
if isinstance(payload_sha256, str) and SHA256_RE.fullmatch(payload_sha256):
event["payload_sha256"] = payload_sha256
if error_code:
event["error_code"] = error_code if CODE_RE.fullmatch(error_code) else "SIGNER_INTERNAL_ERROR"
try:
with self._audit_lock:
self.policy.audit_path.parent.mkdir(parents=True, exist_ok=True)
with self.policy.audit_path.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n")
handle.flush()
os.fsync(handle.fileno())
os.chmod(self.policy.audit_path, 0o600)
except OSError as exc:
raise SignerBrokerServiceError("SIGNER_AUDIT_UNAVAILABLE") from exc
@staticmethod
def _failure(code: str) -> dict[str, Any]:
return {"schema": RESPONSE_SCHEMA, "status": "FAIL", "error_code": code if CODE_RE.fullmatch(code) else "SIGNER_INTERNAL_ERROR"}
def _dispatch(self, request: Mapping[str, Any], peer_uid: int, peer_gid: int) -> tuple[dict[str, Any], str, str | None]:
operation = self._operation(request.get("operation"))
if peer_uid not in self.policy.allowed_uids or peer_gid not in self.policy.allowed_gids:
raise SignerBrokerServiceError("CALLER_UNAUTHORIZED")
if request.get("schema") != REQUEST_SCHEMA:
raise SignerBrokerServiceError("REQUEST_SCHEMA_INVALID")
if operation not in ALLOWED_OPERATIONS:
raise SignerBrokerServiceError("SIGNER_OPERATION_REJECTED")
if operation == "self_test":
if set(request) != {"schema", "operation"}:
raise SignerBrokerServiceError("REQUEST_FIELDS_REJECTED")
with self._signer_lock:
self.runner.self_test()
return {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": operation}, operation, None
if set(request) != {"schema", "operation", "payload_kind", "payload_sha256", "payload_base64"}:
raise SignerBrokerServiceError("REQUEST_FIELDS_REJECTED")
payload_kind = request.get("payload_kind")
encoded = request.get("payload_base64")
claimed_hash = request.get("payload_sha256")
if payload_kind not in ALLOWED_PAYLOAD_KINDS or not isinstance(encoded, str) or not isinstance(claimed_hash, str) or not SHA256_RE.fullmatch(claimed_hash):
raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
if len(encoded) > self.policy.max_payload_bytes * 2 + 8:
raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
try:
payload = base64.b64decode(encoded.encode("ascii"), validate=True)
except (ValueError, UnicodeError, binascii.Error) as exc:
raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID") from exc
if not payload or len(payload) > self.policy.max_payload_bytes:
raise SignerBrokerServiceError("SIGNER_PAYLOAD_INVALID")
actual_hash = hashlib.sha256(payload).hexdigest()
if actual_hash != claimed_hash:
raise SignerBrokerServiceError("SIGNER_PAYLOAD_HASH_MISMATCH")
with self._signer_lock:
signature, key_id = self.runner.sign(payload_kind, payload)
if not isinstance(signature, bytes) or not signature or not isinstance(key_id, str) or not key_id.strip() or len(key_id) > 256:
raise SignerBrokerServiceError("SIGNER_OUTPUT_INVALID")
return {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": operation, "signature_base64": base64.b64encode(signature).decode("ascii"), "key_id": key_id.strip()}, operation, actual_hash
def handle_request(self, request: Mapping[str, Any], peer_uid: int, peer_gid: int) -> dict[str, Any]:
operation = self._operation(request.get("operation")) if isinstance(request, Mapping) else "INVALID"
payload_hash = request.get("payload_sha256") if isinstance(request, Mapping) else None
status = "FAIL"
error_code: str | None = None
try:
if not isinstance(request, Mapping):
raise SignerBrokerServiceError("REQUEST_OBJECT_REQUIRED")
response, operation, payload_hash = self._dispatch(request, peer_uid, peer_gid)
status = "PASS"
except SignerBrokerServiceError as exc:
error_code = exc.code
response = self._failure(exc.code)
except Exception:
error_code = "SIGNER_INTERNAL_ERROR"
response = self._failure(error_code)
try:
self._audit(peer_uid=peer_uid, peer_gid=peer_gid, operation=operation, status=status, error_code=error_code, payload_sha256=payload_hash if isinstance(payload_hash, str) else None)
except SignerBrokerServiceError as exc:
response = self._failure(exc.code)
return response
def handle_json_line(self, raw: bytes, peer_uid: int, peer_gid: int) -> dict[str, Any]:
if len(raw) > MAX_REQUEST_BYTES:
return self.handle_request({"operation": "invalid"}, peer_uid, peer_gid) | {"error_code": "REQUEST_TOO_LARGE", "status": "FAIL"}
try:
request = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
return self.handle_request({"operation": "invalid"}, peer_uid, peer_gid) | {"error_code": "REQUEST_JSON_INVALID", "status": "FAIL"}
return self.handle_request(request, peer_uid, peer_gid)
class _UnixRequestHandler(socketserver.StreamRequestHandler):
def handle(self) -> None:
service: SignerBrokerService = self.server.service # type: ignore[attr-defined]
peer_uid, peer_gid = _peer_credentials(self.request)
response = service.handle_json_line(self.rfile.readline(MAX_REQUEST_BYTES + 1), peer_uid, peer_gid)
self.wfile.write((json.dumps(response, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8"))
self.wfile.flush()
class SignerBrokerUnixServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
daemon_threads = True
allow_reuse_address = True
def __init__(
self,
socket_path: Path,
service: SignerBrokerService,
*,
activated_socket: socket.socket | None = None,
):
self.socket_path = Path(socket_path)
self.service = service
self._socket_is_systemd_owned = activated_socket is not None
if activated_socket is None:
super().__init__(str(self.socket_path), _UnixRequestHandler)
else:
socketserver.BaseServer.__init__(
self, str(self.socket_path), _UnixRequestHandler
)
self.socket = activated_socket
def server_close(self) -> None:
super().server_close()
if self._socket_is_systemd_owned:
return
try:
if self.socket_path.exists() and stat.S_ISSOCK(self.socket_path.stat().st_mode):
self.socket_path.unlink()
except OSError:
pass
def _peer_credentials(channel: socket.socket) -> tuple[int, int]:
if not hasattr(socket, "SO_PEERCRED"):
return -1, -1
try:
_pid, uid, gid = struct.unpack(
"3i",
channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i")),
)
return int(uid), int(gid)
except (OSError, struct.error):
return -1, -1
def _systemd_socket() -> socket.socket | None:
"""Return exactly one validated systemd-provided AF_UNIX listener, or none."""
raw_fds = os.environ.get("LISTEN_FDS")
if raw_fds is None:
return None
try:
count = int(raw_fds)
pid = int(os.environ.get("LISTEN_PID", "0"))
except ValueError as exc:
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID") from exc
if count != 1 or pid != os.getpid():
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID")
try:
listener = socket.socket(fileno=3)
except OSError as exc:
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID") from exc
if listener.family != socket.AF_UNIX or listener.type != socket.SOCK_STREAM:
listener.close()
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID")
return listener
def build_server(
policy: SignerBrokerPolicy,
runner: SignerRunner | None = None,
*,
activated_socket: socket.socket | None = None,
) -> SignerBrokerUnixServer:
socket_path = policy.socket_path
if activated_socket is None:
socket_path.parent.mkdir(parents=True, exist_ok=True)
if socket_path.exists():
if not stat.S_ISSOCK(socket_path.stat().st_mode):
raise SignerBrokerServiceError("SIGNER_BROKER_SOCKET_CONFLICT")
socket_path.unlink()
else:
try:
bound_path = activated_socket.getsockname()
except OSError as exc:
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID") from exc
if bound_path != str(socket_path):
raise SignerBrokerServiceError("SIGNER_BROKER_ACTIVATION_INVALID")
server = SignerBrokerUnixServer(
socket_path,
SignerBrokerService(policy, runner),
activated_socket=activated_socket,
)
if activated_socket is None:
os.chmod(socket_path, 0o660)
return server
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="AIOS fixed-function signer broker")
parser.add_argument("command", choices=("serve",))
parser.add_argument("--socket", default=str(DEFAULT_SOCKET_PATH))
parser.add_argument("--signer", default=str(DEFAULT_SIGNER_PATH))
parser.add_argument("--audit", default=str(DEFAULT_AUDIT_PATH))
parser.add_argument("--state-root", default=str(DEFAULT_STATE_ROOT))
parser.add_argument("--allowed-uid", action="append", type=int, required=True)
parser.add_argument("--allowed-gid", action="append", type=int, required=True)
parser.add_argument("--timeout", type=float, default=60.0)
return parser
def main(argv: Sequence[str] | None = None) -> int:
args = _parser().parse_args(argv)
policy = SignerBrokerPolicy(socket_path=Path(args.socket), signer_path=Path(args.signer), audit_path=Path(args.audit), state_root=Path(args.state_root), allowed_uids=frozenset(args.allowed_uid), allowed_gids=frozenset(args.allowed_gid), signer_timeout=args.timeout)
server = build_server(policy, activated_socket=_systemd_socket())
try:
server.serve_forever()
finally:
server.server_close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -15,11 +15,6 @@ import subprocess
from pathlib import Path
from typing import Any, Mapping, Sequence
from aios_signer_broker import SignerBroker, SignerBrokerError, UnixSignerBroker
DEPLOYED_ROOT_SIGNER_ADAPTER = Path("/usr/local/libexec/aios-grant-signer-adapter")
_SIGNER_FLAGS = (
"--namespace",
@@ -160,36 +155,6 @@ def _sign(
raise SignerContractError("signer_output_empty")
def run_signer_self_test(broker: SignerBroker | None = None) -> dict[str, str]:
active = broker or UnixSignerBroker()
try:
active.self_test()
except SignerBrokerError as exc:
raise SignerContractError(exc.code) from exc
return {"status": "PASS", "contract": "aios.signer-broker.v1"}
def _sign_with_broker(
broker: SignerBroker,
payload_kind: str,
input_path: Path,
signature_path: Path,
key_id_path: Path,
) -> None:
if signature_path.exists() or key_id_path.exists():
raise SignerContractError("contract_output_exists")
try:
signature, key_id = broker.sign(payload_kind, input_path.read_bytes())
except SignerBrokerError as exc:
raise SignerContractError(exc.code) from exc
if not signature or not key_id.strip():
raise SignerContractError("signer_output_empty")
signature_path.write_bytes(signature)
key_id_path.write_text(key_id.strip() + "\n", encoding="utf-8")
os.chmod(signature_path, 0o600)
os.chmod(key_id_path, 0o600)
def _manifest(grant_dir: Path, names: Sequence[str]) -> None:
rows = []
for name in names:
@@ -315,30 +280,13 @@ def _prepare(grant_dir: Path) -> dict[str, Any]:
}
def signer_broker_required(
signer_command: Sequence[str],
broker: SignerBroker | None,
require_broker: bool,
) -> bool:
"""Root adapter and explicit callers are never eligible for direct fallback."""
if require_broker or broker is not None:
return True
return bool(signer_command) and Path(signer_command[0]) == DEPLOYED_ROOT_SIGNER_ADAPTER
def run_compatible_signer(
signer_command: Sequence[str],
grant_dir: Path,
result_path: Path,
*,
broker: SignerBroker | None = None,
require_broker: bool = False,
) -> dict[str, str]:
try:
broker_required = signer_broker_required(signer_command, broker, require_broker)
if not _supports_deployed_signer(signer_command):
if broker_required:
raise SignerContractError("SIGNER_BROKER_REQUIRED")
_run(
[
*signer_command,
@@ -370,21 +318,22 @@ def run_compatible_signer(
"signing_key_identity": signer,
}
prepared = _prepare(grant_dir)
active_broker = broker if broker is not None else UnixSignerBroker()
try:
active_broker.self_test()
except SignerBrokerError as exc:
raise SignerContractError(exc.code) from exc
_sign_with_broker(
active_broker,
"approval_plan",
config_path = Path(
os.environ.get("AIOS_GRANT_ROOT_CONFIG", "/etc/aios/grant-root/config.json")
)
config = _read(config_path)
grant_namespace = str(config["grant_namespace"])
approval_namespace = str(config["approval_namespace"])
_sign(
signer_command,
approval_namespace,
grant_dir / "approval-plan.canonical.json",
grant_dir / "approval-plan.sig",
grant_dir / "approval-plan.key-id",
)
_sign_with_broker(
active_broker,
"grant_payload",
_sign(
signer_command,
grant_namespace,
grant_dir / "grant-payload.canonical.json",
grant_dir / "signed-grant.sig",
grant_dir / "signed-grant.key-id",
-33
View File
@@ -1,33 +0,0 @@
[Unit]
Description=AIOS fixed-function privileged signer broker
Requires=aios-signer-broker.socket
After=aios-signer-broker.socket
[Service]
Type=simple
User=root
Group=root
UMask=0077
Environment=PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=/opt/aios-gitops/active-runtime/app AIOS_SIGNER_BROKER_CALLER_UID=1000 AIOS_SIGNER_BROKER_CALLER_GID=1000 AIOS_SIGNER_BROKER_CALLER_GROUP=gitops AIOS_SIGNER_BROKER_REQUIRED=1 AIOS_SIGNER_BROKER_USER=root
ExecStart=/usr/bin/python3 -B /opt/aios-gitops/active-runtime/app/aios_signer_broker_service.py serve --socket /run/aios/signer-broker.sock --allowed-uid 1000 --allowed-gid 1000 --signer /usr/local/libexec/aios-grant-signer-adapter --audit /var/lib/aios-gitops/audit/signer-broker.jsonl --state-root /var/lib/aios-gitops/signer-broker --timeout 60
Restart=on-failure
RestartSec=2
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictSUIDSGID=true
RestrictRealtime=true
MemoryDenyWriteExecute=true
RestrictAddressFamilies=AF_UNIX
SystemCallArchitectures=native
ReadOnlyPaths=/opt/aios-gitops/active-runtime/app /usr/local/libexec/aios-grant-signer-adapter
ReadWritePaths=/var/lib/aios-gitops/audit /var/lib/aios-gitops/signer-broker
[Install]
WantedBy=multi-user.target
-14
View File
@@ -1,14 +0,0 @@
[Unit]
Description=AIOS fixed-function signer broker socket
[Socket]
ListenStream=/run/aios/signer-broker.sock
Service=aios-signer-broker.service
SocketUser=root
SocketGroup=gitops
SocketMode=0660
DirectoryMode=0750
RemoveOnStop=true
[Install]
WantedBy=sockets.target
-24
View File
@@ -1,24 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
if (( EUID != 0 )); then
printf '%s\n' 'SIGNER_BROKER_INSTALL_REQUIRES_ROOT' >&2
exit 1
fi
repo_root=$(cd -- "$(dirname -- "$0")/.." && pwd -P)
validator="$repo_root/scripts/validate_aios_signer_broker_config.py"
unit_root="$repo_root/deploy/systemd"
destination=/etc/systemd/system
python3 "$validator"
if command -v systemd-analyze >/dev/null 2>&1; then
systemd-analyze verify "$unit_root/aios-signer-broker.socket" "$unit_root/aios-signer-broker.service"
fi
install -D -o root -g root -m 0644 "$unit_root/aios-signer-broker.socket" "$destination/aios-signer-broker.socket"
install -D -o root -g root -m 0644 "$unit_root/aios-signer-broker.service" "$destination/aios-signer-broker.service"
printf '%s\n' 'SIGNER_BROKER_UNITS_INSTALLED_ONLY'
printf '%s\n' 'No daemon-reload, enable, restart, or start action was performed.'
printf '%s\n' 'After reviewed deployment: systemctl daemon-reload && systemctl enable --now aios-signer-broker.socket'
@@ -1,90 +0,0 @@
#!/usr/bin/env python3
"""Fail-closed static contract validator for the signer broker units."""
from __future__ import annotations
import configparser
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[1]
SOCKET = ROOT / "deploy/systemd/aios-signer-broker.socket"
SERVICE = ROOT / "deploy/systemd/aios-signer-broker.service"
class ConfigurationError(RuntimeError):
pass
def load(path: Path) -> configparser.ConfigParser:
parser = configparser.ConfigParser(interpolation=None, strict=True)
parser.optionxform = str
try:
with path.open("r", encoding="utf-8") as handle:
parser.read_file(handle)
except (OSError, configparser.Error) as exc:
raise ConfigurationError(f"READ_REJECTED:{path.name}") from exc
return parser
def exact(section: configparser.SectionProxy, expected: dict[str, str], label: str) -> None:
if set(section) != set(expected):
raise ConfigurationError(f"{label}_FIELDS_REJECTED")
for key, value in expected.items():
if section.get(key) != value:
raise ConfigurationError(f"{label}_{key}_REJECTED")
def validate() -> None:
socket_unit = load(SOCKET)
if set(socket_unit.sections()) != {"Unit", "Socket", "Install"}:
raise ConfigurationError("SOCKET_SECTIONS_REJECTED")
exact(socket_unit["Unit"], {"Description": "AIOS fixed-function signer broker socket"}, "SOCKET_UNIT")
exact(socket_unit["Socket"], {
"ListenStream": "/run/aios/signer-broker.sock",
"Service": "aios-signer-broker.service",
"SocketUser": "root", "SocketGroup": "gitops", "SocketMode": "0660",
"DirectoryMode": "0750", "RemoveOnStop": "true",
}, "SOCKET")
exact(socket_unit["Install"], {"WantedBy": "sockets.target"}, "SOCKET_INSTALL")
service_unit = load(SERVICE)
if set(service_unit.sections()) != {"Unit", "Service", "Install"}:
raise ConfigurationError("SERVICE_SECTIONS_REJECTED")
exact(service_unit["Unit"], {
"Description": "AIOS fixed-function privileged signer broker",
"Requires": "aios-signer-broker.socket", "After": "aios-signer-broker.socket",
}, "SERVICE_UNIT")
service = service_unit["Service"]
expected = {
"Type": "simple", "User": "root", "Group": "root", "UMask": "0077",
"Environment": "PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=/opt/aios-gitops/active-runtime/app AIOS_SIGNER_BROKER_CALLER_UID=1000 AIOS_SIGNER_BROKER_CALLER_GID=1000 AIOS_SIGNER_BROKER_CALLER_GROUP=gitops AIOS_SIGNER_BROKER_REQUIRED=1 AIOS_SIGNER_BROKER_USER=root",
"Restart": "on-failure", "RestartSec": "2", "NoNewPrivileges": "true",
"PrivateTmp": "true", "PrivateDevices": "true", "ProtectSystem": "strict",
"ProtectHome": "true", "ProtectKernelTunables": "true", "ProtectKernelModules": "true",
"ProtectControlGroups": "true", "LockPersonality": "true", "RestrictSUIDSGID": "true",
"RestrictRealtime": "true", "MemoryDenyWriteExecute": "true",
"RestrictAddressFamilies": "AF_UNIX", "SystemCallArchitectures": "native",
"ReadOnlyPaths": "/opt/aios-gitops/active-runtime/app /usr/local/libexec/aios-grant-signer-adapter",
"ReadWritePaths": "/var/lib/aios-gitops/audit /var/lib/aios-gitops/signer-broker",
}
for key, value in expected.items():
if service.get(key) != value:
raise ConfigurationError(f"SERVICE_{key}_REJECTED")
argv = service.get("ExecStart", "").split()
if argv != [
"/usr/bin/python3", "-B", "/opt/aios-gitops/active-runtime/app/aios_signer_broker_service.py", "serve",
"--socket", "/run/aios/signer-broker.sock", "--allowed-uid", "1000", "--allowed-gid", "1000", "--signer",
"/usr/local/libexec/aios-grant-signer-adapter", "--audit", "/var/lib/aios-gitops/audit/signer-broker.jsonl",
"--state-root", "/var/lib/aios-gitops/signer-broker", "--timeout", "60",
]:
raise ConfigurationError("SERVICE_EXECSTART_REJECTED")
if any(token in service["ExecStart"] for token in ("sudo", "setuid", "/bin/sh", "bash", "--command", "--output")):
raise ConfigurationError("SERVICE_GENERIC_EXECUTION_REJECTED")
exact(service_unit["Install"], {"WantedBy": "multi-user.target"}, "SERVICE_INSTALL")
def main() -> int:
try:
validate()
except ConfigurationError as exc:
print(f"SIGNER_BROKER_CONFIG_INVALID:{exc}", file=sys.stderr)
return 1
print("SIGNER_BROKER_CONFIG_VALID")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,285 @@
from __future__ import annotations
import base64
from datetime import datetime, timedelta, timezone
import json
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
import pytest
from aios_existing_branch_approval import (
DurableNonceReplayLedger,
OwnerApprovalEnvelope,
OwnerApprovalRejected,
OwnerApprovalVerifier,
TrustRootEvidence,
)
from aios_existing_branch_publisher import (
ExistingBranchPublicationRejected,
ExistingBranchPublisher,
RemoteBranchObservation,
RemotePublicationEvidence,
)
BASE_OID = "a" * 40
COMMIT_OID = "b" * 40
BRANCH = "agent/owner-approval-contract-20260810"
class TrustRoot:
def __init__(self, public_key: bytes, *, revocation: str = "ACTIVE") -> None:
self.public_key = public_key
self.revocation = revocation
self.calls: list[str] = []
def resolve(self, *, key_id: str, owner_identity: str, trust_domain: str):
self.calls.append("trust-root")
return TrustRootEvidence(
key_id=key_id,
owner_identity=owner_identity,
trust_domain=trust_domain,
public_key=self.public_key,
authority_confirmed=True,
revocation_state=self.revocation,
)
class Credentials:
def __init__(self, events: list[str]) -> None:
self.events = events
def project(self) -> None:
self.events.append("credential-project")
def revoke(self) -> None:
self.events.append("credential-revoke")
class Workspace:
def __init__(self, events: list[str]) -> None:
self.events = events
def head_oid(self) -> str:
self.events.append("workspace-head")
return BASE_OID
def commit_payload(self, payload):
self.events.append("workspace-commit")
return COMMIT_OID
class Workspaces:
def __init__(self, events: list[str]) -> None:
self.events = events
def open_detached(self, expected_branch_oid: str) -> Workspace:
assert expected_branch_oid == BASE_OID
self.events.append("workspace-open")
return Workspace(self.events)
class Remote:
def __init__(self, events: list[str]) -> None:
self.events = events
self.observation = RemoteBranchObservation("PRESENT", BASE_OID, BASE_OID, True)
self.authority_oid = BASE_OID
self.pushes: list[tuple[str, str, str]] = []
def read_branch_oid(self, branch: str) -> str:
self.events.append("authority-observe")
assert branch == BRANCH
return self.authority_oid
def observe(self, branch: str) -> RemoteBranchObservation:
self.events.append("remote-observe")
assert branch == BRANCH
return self.observation
def publish_compare_and_swap(self, branch: str, expected: str, commit: str):
self.events.append("remote-publish")
self.pushes.append((branch, expected, commit))
return RemotePublicationEvidence(COMMIT_OID, COMMIT_OID, COMMIT_OID, True)
class Audit:
def __init__(self) -> None:
self.rows: list[dict] = []
def record(self, event) -> None:
self.rows.append(dict(event))
def request(*, payload=None, branch=BRANCH, expected_oid=BASE_OID):
return {
"schema": "aios.existing-branch-publication.v1",
"repository": "aios/aios",
"branch": branch,
"expected_branch_oid": expected_oid,
"operation": "publishExistingBranchGoverned",
"idempotency_key": "owner-approval-binding-20260810",
"payload": {"files": ["active-runtime/app/example.py"]} if payload is None else payload,
}
def signed_approval(private: Ed25519PrivateKey, request_value, **replacements):
from aios_existing_branch_publisher import ExistingBranchPublicationRequest
parsed = ExistingBranchPublicationRequest.from_mapping(request_value)
now = datetime.now(timezone.utc)
raw = {
"schema": "aios.existing-branch-owner-approval.v1",
"approval_id": "approval-owner-binding-20260810",
"repository": parsed.repository,
"branch": parsed.branch,
"expected_branch_oid": parsed.expected_branch_oid,
"operation": "publishExistingBranchGoverned",
"issued_at": (now - timedelta(seconds=30)).isoformat(),
"not_before": (now - timedelta(seconds=1)).isoformat(),
"expires_at": (now + timedelta(minutes=5)).isoformat(),
"nonce": "nonce-owner-binding-20260810",
"owner_identity": "owner.aios",
"trust_domain": "aios-owner-root",
"key_id": "owner-key-20260810",
"request_digest": parsed.request_digest,
"idempotency_digest": parsed.idempotency_digest,
"signature": "AA==",
}
raw.update(replacements)
envelope = OwnerApprovalEnvelope.from_mapping(raw)
raw["signature"] = base64.b64encode(private.sign(envelope.canonical_bytes())).decode("ascii")
return raw
def service(tmp_path, *, revocation="ACTIVE"):
private = Ed25519PrivateKey.generate()
public_key = private.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw
)
root = TrustRoot(public_key, revocation=revocation)
events: list[str] = []
remote = Remote(events)
audit = Audit()
publisher = ExistingBranchPublisher(
verifier=OwnerApprovalVerifier(root, allowed_trust_domains={"aios-owner-root"}),
replay_ledger=DurableNonceReplayLedger(tmp_path / "replay"),
authority=remote,
remote=remote,
credentials=Credentials(events),
workspaces=Workspaces(events),
audit=audit,
)
return private, root, events, remote, audit, publisher
def test_valid_signed_owner_approval_is_verified_before_capability_projection(tmp_path):
private, root, events, remote, audit, publisher = service(tmp_path)
value = request()
approval = signed_approval(private, value)
result = publisher.publish(value, approval)
assert result["status"] == "PUBLISHED"
assert result["commit_oid"] == COMMIT_OID
assert remote.pushes == [(BRANCH, BASE_OID, COMMIT_OID)]
assert events == [
"authority-observe",
"remote-observe",
"workspace-open",
"workspace-head",
"credential-project",
"workspace-commit",
"remote-publish",
"credential-revoke",
]
assert root.calls == ["trust-root"]
assert approval["signature"] not in json.dumps(audit.rows)
assert "signature" not in audit.rows[-1]
@pytest.mark.parametrize(
"case",
[
"invalid_signature",
"expired",
"revoked",
"missing_key_evidence",
"missing_revocation_evidence",
"wrong_repository",
"wrong_branch",
"wrong_oid",
"wrong_operation",
"arbitrary_reference",
],
)
def test_invalid_owner_approval_inputs_are_rejected_without_publication(tmp_path, case):
private, root, events, remote, _, publisher = service(
tmp_path, revocation="REVOKED" if case == "revoked" else "ACTIVE"
)
value = request()
approval = signed_approval(private, value)
if case == "invalid_signature":
approval["signature"] = "AAAA"
elif case == "expired":
approval = signed_approval(
private,
value,
issued_at="2000-01-01T00:00:00+00:00",
not_before="2000-01-01T00:00:01+00:00",
expires_at="2000-01-01T00:01:00+00:00",
)
elif case == "wrong_repository":
approval = signed_approval(private, value, repository="other/repository")
elif case == "wrong_branch":
approval = signed_approval(private, value, branch="agent/another-approved-branch")
elif case == "wrong_oid":
approval = signed_approval(private, value, expected_branch_oid="c" * 40)
elif case == "wrong_operation":
approval["operation"] = "differentOperation"
elif case == "arbitrary_reference":
approval = "owner-approval-reference-123"
elif case == "missing_key_evidence":
root.public_key = None
elif case == "missing_revocation_evidence":
root.revocation = "UNKNOWN"
with pytest.raises((OwnerApprovalRejected, ExistingBranchPublicationRejected)):
publisher.publish(value, approval)
assert remote.pushes == []
assert "credential-project" not in events
assert "workspace-open" not in events
@pytest.mark.parametrize(
"changed",
[
lambda value: {**value, "branch": "agent/approval-replay-other-branch"},
lambda value: {**value, "expected_branch_oid": "c" * 40},
lambda value: {**value, "payload": {"files": ["other.py"]}},
],
)
def test_signed_approval_cannot_be_replayed_for_another_branch_oid_or_payload(tmp_path, changed):
private, _, events, remote, _, publisher = service(tmp_path)
original = request()
approval = signed_approval(private, original)
with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_BINDING_REJECTED"):
publisher.publish(changed(original), approval)
assert remote.pushes == []
assert events == []
def test_signed_approval_nonce_is_single_use(tmp_path):
private, _, events, remote, _, publisher = service(tmp_path)
value = request()
approval = signed_approval(private, value)
assert publisher.publish(value, approval)["status"] == "PUBLISHED"
with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_REPLAY_REJECTED"):
publisher.publish(value, approval)
assert len(remote.pushes) == 1
assert events.count("credential-project") == 1
@@ -0,0 +1,72 @@
from __future__ import annotations
import inspect
from aios_existing_branch_publisher import (
ExistingBranchPublisher,
RemoteBranchObservation,
)
from test_existing_branch_owner_approval import (
BASE_OID,
COMMIT_OID,
request,
service,
signed_approval,
)
def test_remote_oid_conflict_requires_manual_owner_review_before_capability_projection(tmp_path):
private, _, events, remote, _, publisher = service(tmp_path)
remote.observation = RemoteBranchObservation("PRESENT", "c" * 40, "c" * 40, True)
value = request()
result = publisher.publish(value, signed_approval(private, value))
assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
assert events == ["authority-observe", "remote-observe"]
assert remote.pushes == []
def test_ambiguous_remote_requires_reconciliation_after_one_observation(tmp_path):
private, _, events, remote, _, publisher = service(tmp_path)
remote.observation = RemoteBranchObservation("AMBIGUOUS", None, None, None)
value = request()
result = publisher.publish(value, signed_approval(private, value))
assert result == {"status": "RECONCILIATION_REQUIRED"}
assert events == ["authority-observe", "remote-observe"]
assert remote.pushes == []
def test_authority_oid_conflict_requires_manual_owner_review(tmp_path):
private, _, events, remote, _, publisher = service(tmp_path)
remote.authority_oid = "c" * 40
value = request()
result = publisher.publish(value, signed_approval(private, value))
assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
assert events == ["authority-observe"]
assert remote.pushes == []
def test_success_reports_all_remote_agreement_fields(tmp_path):
private, _, _, _, _, publisher = service(tmp_path)
value = request()
result = publisher.publish(value, signed_approval(private, value))
assert result == {
"status": "PUBLISHED",
"commit_oid": COMMIT_OID,
"authority_oid": COMMIT_OID,
"gitea_git_oid": COMMIT_OID,
"gitea_api_oid": COMMIT_OID,
"commit_visible": True,
}
def test_publisher_exposes_only_compare_and_swap_publication_path():
source = inspect.getsource(ExistingBranchPublisher).lower()
assert "force" not in source
-105
View File
@@ -1,105 +0,0 @@
from __future__ import annotations
import base64
from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
import os
from pathlib import Path
import socket
import stat
import threading
import time
import pytest
from aios_signer_broker import SignerBrokerError
from aios_signer_broker_service import REQUEST_SCHEMA, SignerBrokerPolicy, SignerBrokerService, SignerBrokerServiceError, build_server
from signer_contract_compatibility import DEPLOYED_ROOT_SIGNER_ADAPTER, SignerContractError, run_signer_self_test, signer_broker_required
class Runner:
def __init__(self):
self.active = 0
self.maximum = 0
self.lock = threading.Lock()
def self_test(self) -> None:
return None
def sign(self, _kind: str, payload: bytes) -> tuple[bytes, str]:
with self.lock:
self.active += 1
self.maximum = max(self.maximum, self.active)
try:
time.sleep(0.01)
return b"sig:" + payload, "nonroot-test-key"
finally:
with self.lock:
self.active -= 1
class Unavailable:
def __init__(self, code: str = "SIGNER_BROKER_UNAVAILABLE"):
self.code = code
def self_test(self) -> None:
raise SignerBrokerError(self.code)
def sign(self, _kind: str, _payload: bytes) -> tuple[bytes, str]:
raise SignerBrokerError(self.code)
def policy(root: Path, uids: frozenset[int], gids: frozenset[int] = frozenset({1000})) -> SignerBrokerPolicy:
return SignerBrokerPolicy(
socket_path=root / "signer.sock",
signer_path=Path("/usr/local/libexec/aios-grant-signer-adapter"),
audit_path=root / "audit.jsonl",
state_root=root / "state",
allowed_uids=uids,
allowed_gids=gids,
)
def request(payload: bytes = b"secret-shaped payload") -> dict[str, str]:
return {
"schema": REQUEST_SCHEMA, "operation": "sign", "payload_kind": "approval_plan",
"payload_sha256": hashlib.sha256(payload).hexdigest(),
"payload_base64": base64.b64encode(payload).decode("ascii"),
}
def test_non_root_uid_passes_and_wrong_uid_fails_closed(tmp_path: Path):
service = SignerBrokerService(policy(tmp_path, frozenset({1000})), Runner())
assert service.handle_request(request(), 1000, 1000)["status"] == "PASS"
assert service.handle_request(request(), 1001, 1000)["error_code"] == "CALLER_UNAUTHORIZED"
def test_deployed_root_adapter_is_always_broker_required():
assert signer_broker_required([str(DEPLOYED_ROOT_SIGNER_ADAPTER)], None, False)
@pytest.mark.parametrize("code", ["SIGNER_BROKER_UNAVAILABLE", "SIGNER_TIMEOUT"])
def test_broker_missing_or_timeout_is_non_root_compatibility_failure(code: str):
with pytest.raises(SignerContractError, match=code):
run_signer_self_test(Unavailable(code))
def test_signer_is_serialized_and_audit_is_redacted(tmp_path: Path):
runner = Runner()
service = SignerBrokerService(policy(tmp_path, frozenset({1000})), runner)
with ThreadPoolExecutor(max_workers=8) as pool:
results = list(pool.map(lambda n: service.handle_request(request(f"secret-{n}".encode()), 1000, 1000), range(16)))
assert all(result["status"] == "PASS" for result in results)
assert runner.maximum == 1
audit = (tmp_path / "audit.jsonl").read_text(encoding="utf-8")
assert "secret-" not in audit and "payload_base64" not in audit
def test_systemd_owned_socket_keeps_restart_permissions(tmp_path: Path):
path = tmp_path / "signer.sock"
listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
listener.bind(str(path))
listener.listen()
os.chmod(path, 0o660)
server = build_server(policy(tmp_path, frozenset({os.getuid()}), frozenset({os.getgid()})), Runner(), activated_socket=listener)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
client.connect(str(path))
client.sendall((json.dumps({"schema": REQUEST_SCHEMA, "operation": "self_test"}) + "\n").encode())
assert json.loads(client.makefile("rb").readline())["status"] == "PASS"
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
assert path.exists() and stat.S_IMODE(path.stat().st_mode) == 0o660
path.unlink()
-130
View File
@@ -1,130 +0,0 @@
from __future__ import annotations
import base64
from concurrent.futures import ThreadPoolExecutor
import hashlib
import json
from pathlib import Path
import threading
from aios_signer_broker_service import (
REQUEST_SCHEMA,
RESPONSE_SCHEMA,
SignerBrokerPolicy,
SignerBrokerService,
SignerBrokerServiceError,
)
class FakeRunner:
def __init__(self, failure: str | None = None):
self.failure = failure
self.calls: list[tuple[str, str]] = []
self.lock = threading.Lock()
def self_test(self) -> None:
with self.lock:
self.calls.append(("self_test", ""))
if self.failure:
raise SignerBrokerServiceError(self.failure)
def sign(self, payload_kind: str, payload: bytes) -> tuple[bytes, str]:
with self.lock:
self.calls.append(("sign", payload_kind))
if self.failure:
raise SignerBrokerServiceError(self.failure)
return b"signature:" + payload, "test-broker-key"
def make_policy(tmp_path: Path, *, allowed_uids: frozenset[int] = frozenset({1000}), allowed_gids: frozenset[int] = frozenset({1000})) -> SignerBrokerPolicy:
return SignerBrokerPolicy(
socket_path=tmp_path / "signer.sock",
signer_path=Path("/usr/local/libexec/aios-grant-signer-adapter"),
audit_path=tmp_path / "audit.jsonl",
state_root=tmp_path / "state",
allowed_uids=allowed_uids,
allowed_gids=allowed_gids,
)
def sign_request(payload: bytes = b"canonical payload") -> dict[str, str]:
return {
"schema": REQUEST_SCHEMA,
"operation": "sign",
"payload_kind": "approval_plan",
"payload_sha256": hashlib.sha256(payload).hexdigest(),
"payload_base64": base64.b64encode(payload).decode("ascii"),
}
def read_audit(path: Path) -> list[dict[str, object]]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
def test_authorized_request_and_self_test_emit_metadata_only_audit(tmp_path: Path):
service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
self_test = service.handle_request({"schema": REQUEST_SCHEMA, "operation": "self_test"}, 1000, 1000)
signed = service.handle_request(sign_request(b"secret-shaped canonical input"), 1000, 1000)
assert self_test == {"schema": RESPONSE_SCHEMA, "status": "PASS", "operation": "self_test"}
assert signed["status"] == "PASS"
assert base64.b64decode(signed["signature_base64"]) == b"signature:secret-shaped canonical input"
assert signed["key_id"] == "test-broker-key"
audit_text = (tmp_path / "audit.jsonl").read_text(encoding="utf-8")
assert "secret-shaped canonical input" not in audit_text
assert "payload_base64" not in audit_text
allowed = {"schema", "timestamp", "peer_uid", "peer_gid", "operation", "status", "error_code", "payload_sha256"}
assert all(set(event) <= allowed for event in read_audit(tmp_path / "audit.jsonl"))
def test_unauthorized_caller_is_rejected_before_signer_use(tmp_path: Path):
runner = FakeRunner()
service = SignerBrokerService(make_policy(tmp_path), runner)
result = service.handle_request(sign_request(), 1001, 1000)
assert result["status"] == "FAIL"
assert result["error_code"] == "CALLER_UNAUTHORIZED"
assert runner.calls == []
def test_wrong_gid_is_rejected_before_signer_use(tmp_path: Path):
runner = FakeRunner()
service = SignerBrokerService(make_policy(tmp_path), runner)
result = service.handle_request(sign_request(), 1000, 1001)
assert result["status"] == "FAIL"
assert result["error_code"] == "CALLER_UNAUTHORIZED"
assert runner.calls == []
def test_arbitrary_command_and_invalid_operation_are_rejected(tmp_path: Path):
runner = FakeRunner()
service = SignerBrokerService(make_policy(tmp_path), runner)
injected = {"schema": REQUEST_SCHEMA, "operation": "sign", "command": ["/bin/sh", "-c", "id"]}
invalid = {"schema": REQUEST_SCHEMA, "operation": "run", "command": ["id"]}
assert service.handle_request(injected, 1000, 1000)["error_code"] == "REQUEST_FIELDS_REJECTED"
assert service.handle_request(invalid, 1000, 1000)["error_code"] == "SIGNER_OPERATION_REJECTED"
assert runner.calls == []
def test_invalid_payload_and_hash_fail_closed(tmp_path: Path):
service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
invalid = sign_request()
invalid["payload_base64"] = "not-base64"
assert service.handle_request(invalid, 1000, 1000)["error_code"] == "SIGNER_PAYLOAD_INVALID"
mismatch = sign_request()
mismatch["payload_sha256"] = "0" * 64
assert service.handle_request(mismatch, 1000, 1000)["error_code"] == "SIGNER_PAYLOAD_HASH_MISMATCH"
def test_signer_unavailable_and_timeout_are_stable_failures(tmp_path: Path):
for code in ("SIGNER_UNAVAILABLE", "SIGNER_TIMEOUT"):
result = SignerBrokerService(make_policy(tmp_path / code), FakeRunner(code)).handle_request(
{"schema": REQUEST_SCHEMA, "operation": "self_test"}, 1000, 1000
)
assert result == {"schema": RESPONSE_SCHEMA, "status": "FAIL", "error_code": code}
def test_concurrent_calls_are_safe_and_audited(tmp_path: Path):
service = SignerBrokerService(make_policy(tmp_path), FakeRunner())
with ThreadPoolExecutor(max_workers=8) as pool:
results = list(pool.map(lambda index: service.handle_request(sign_request(f"payload-{index}".encode()), 1000, 1000), range(24)))
assert all(result["status"] == "PASS" for result in results)
assert len(read_audit(tmp_path / "audit.jsonl")) == 24
@@ -1,50 +0,0 @@
from __future__ import annotations
import configparser
from pathlib import Path
import shutil
import subprocess
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
SOCKET = ROOT / "deploy/systemd/aios-signer-broker.socket"
SERVICE = ROOT / "deploy/systemd/aios-signer-broker.service"
VALIDATOR = ROOT / "scripts/validate_aios_signer_broker_config.py"
def unit(path: Path) -> configparser.ConfigParser:
parser = configparser.ConfigParser(interpolation=None)
parser.optionxform = str
parser.read(path, encoding="utf-8")
return parser
def test_validator_and_socket_ownership_contract():
result = subprocess.run([sys.executable, str(VALIDATOR)], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stderr
assert result.stdout.strip() == "SIGNER_BROKER_CONFIG_VALID"
socket_unit = unit(SOCKET)["Socket"]
assert socket_unit["SocketUser"] == "root"
assert socket_unit["SocketGroup"] == "gitops"
assert socket_unit["SocketMode"] == "0660"
assert socket_unit["RemoveOnStop"] == "true"
def test_fixed_function_root_service_contract():
service = unit(SERVICE)["Service"]
assert service["User"] == "root" and service["Group"] == "root"
assert "AIOS_SIGNER_BROKER_CALLER_UID=1000" in service["Environment"]
assert "AIOS_SIGNER_BROKER_CALLER_GID=1000" in service["Environment"]
assert "AIOS_SIGNER_BROKER_REQUIRED=1" in service["Environment"]
argv = service["ExecStart"].split()
assert argv[0:4] == ["/usr/bin/python3", "-B", "/opt/aios-gitops/active-runtime/app/aios_signer_broker_service.py", "serve"]
assert argv[argv.index("--allowed-uid") + 1] == "1000"
assert argv[argv.index("--allowed-gid") + 1] == "1000"
assert argv[argv.index("--signer") + 1] == "/usr/local/libexec/aios-grant-signer-adapter"
assert not {"sudo", "setuid", "/bin/sh", "bash", "--command", "--output"}.intersection(argv)
def test_systemd_analyze_verify_if_available():
analyzer = shutil.which("systemd-analyze")
if analyzer is None:
pytest.skip("systemd-analyze unavailable")
result = subprocess.run([analyzer, "verify", str(SOCKET), str(SERVICE)], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stdout + result.stderr
+4 -18
View File
@@ -20,7 +20,6 @@ from signer_contract_compatibility import ( # noqa: E402
canonical_json_bytes,
deployed_signer_help,
run_compatible_signer,
run_signer_self_test,
resolve_ref_oid,
run_compatible_validator,
sha256_json,
@@ -114,13 +113,9 @@ class SignerContractCompatibilityTests(unittest.TestCase):
help_result = subprocess.run([signer, "sign", "--help"], capture_output=True, text=True, check=False)
self.assertEqual(help_result.returncode, 0)
self.assertTrue(deployed_signer_help(help_result.stdout, help_result.stderr))
try:
broker_result = run_signer_self_test()
except SignerContractError as exc:
if str(exc) == "SIGNER_BROKER_UNAVAILABLE":
self.skipTest("signer broker is not deployed in this engineering phase")
raise
self.assertEqual(broker_result["status"], "PASS")
self_test = subprocess.run([signer, "self-test"], capture_output=True, text=True, check=False)
self.assertEqual(self_test.returncode, 0)
self.assertIn("SIGNER_SELF_TEST=PASS", self_test.stdout)
def test_resolve_ref_oid_uses_full_synthetic_commit_and_fails_closed(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
@@ -193,18 +188,9 @@ class SignerContractCompatibilityTests(unittest.TestCase):
(grant_dir / "approval-plan.json").write_text("{}", encoding="utf-8")
(grant_dir / "grant-request.json").write_text("{}", encoding="utf-8")
with self.assertRaises(SignerContractError) as captured:
run_compatible_signer([str(signer)], grant_dir, root / "result.json", require_broker=False)
run_compatible_signer([str(signer)], grant_dir, root / "result.json")
self.assertEqual(str(captured.exception), "legacy_signature_missing")
def test_broker_required_never_falls_back_to_legacy_signer(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
signer = root / "legacy-signer"
signer.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
signer.chmod(0o700)
with self.assertRaisesRegex(SignerContractError, "SIGNER_BROKER_REQUIRED"):
run_compatible_signer([str(signer)], root, root / "result.json", require_broker=True)
@unittest.skipUnless(
os.environ.get("AIOS_RUN_REAL_SIGNER_TESTS") == "1",
"live signer integration is opt-in",