Compare commits

...
6 changed files with 770 additions and 9 deletions
@@ -18,12 +18,15 @@ import hmac
from http.server import BaseHTTPRequestHandler
import json
import os
import pwd
from pathlib import Path
import re
import sqlite3
import socket
import socketserver
import stat
import tempfile
import struct
import threading
import urllib.error
import urllib.request
@@ -34,6 +37,8 @@ RECEIPT_SCHEMA = "aios.owner-durable-authorization-receipt.v1"
DECISION_SCHEMA = "aios.owner-durable-authorization-decision.v1"
PROJECTION_SCHEMA = "aios.external-capability-projection.request.v1"
PROJECTION_LIST_SCHEMA = "aios.external-capability-projection.list.v1"
GATE_AMENDMENT_SCHEMA = "aios.release-preparation.gate-amendment.v1"
GATE_AMENDMENT_CLASSIFICATION = "SUPERVISOR_GLOBAL_EXTERNAL_ACK_GATE_REMEDIATED"
ALLOWED_PERSISTENT_CAPABILITIES = frozenset({
"governed-task:project",
@@ -64,6 +69,28 @@ TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
JOB_RE = re.compile(r"^job-[0-9a-f]{32}$")
MAX_BODY = 512 * 1024
SO_PEERCRED_SIZE = struct.calcsize("3i")
def peer_user_from_connection(connection: object) -> str:
if not hasattr(socket, "SO_PEERCRED"):
raise AuthorizationError("SO_PEERCRED_UNAVAILABLE", 500)
try:
raw = connection.getsockopt(
socket.SOL_SOCKET,
socket.SO_PEERCRED,
SO_PEERCRED_SIZE,
)
except OSError as exc:
raise AuthorizationError("SO_PEERCRED_UNAVAILABLE", 403) from exc
try:
_pid, uid, _gid = struct.unpack("3i", raw)
except struct.error as exc:
raise AuthorizationError("SO_PEERCRED_INVALID", 403) from exc
try:
return pwd.getpwuid(uid).pw_name
except (KeyError, OSError) as exc:
raise AuthorizationError("SO_PEERCRED_UID_UNKNOWN", 403) from exc
class AuthorizationError(RuntimeError):
@@ -447,6 +474,16 @@ class ReceiptStore:
ON projection_outbox(state, capability, created_at)
"""
)
con.execute(
"""
CREATE TABLE IF NOT EXISTS gate_amendments(
idempotency_key TEXT PRIMARY KEY,
amendment_sha256 TEXT NOT NULL,
amendment_json TEXT NOT NULL,
created_at TEXT NOT NULL
)
"""
)
con.commit()
self._lock = threading.RLock()
@@ -519,6 +556,37 @@ class ReceiptStore:
)
con.commit()
def put_gate_amendment(
self,
idempotency_key: str,
amendment: Mapping[str, Any],
) -> dict[str, Any]:
encoded = json.dumps(amendment, sort_keys=True, separators=(",", ":"))
lowered = encoded.lower()
if any(marker in lowered for marker in (
"bearer ", "access_token", "refresh_token", "private_key", "credential_file"
)):
raise AuthorizationError("GATE_AMENDMENT_SECRET_MATERIAL_DENY", 403)
digest = sha256(encoded.encode())
with self._lock, sqlite3.connect(self.path) as con:
con.execute("BEGIN IMMEDIATE")
row = con.execute(
"SELECT amendment_sha256,amendment_json FROM gate_amendments WHERE idempotency_key=?",
(idempotency_key,),
).fetchone()
if row is not None:
if row[0] != digest or row[1] != encoded:
con.rollback()
raise AuthorizationError("GATE_AMENDMENT_IDEMPOTENCY_CONFLICT", 409)
con.commit()
return {"amendment_sha256": digest, "idempotent": True}
con.execute(
"INSERT INTO gate_amendments(idempotency_key,amendment_sha256,amendment_json,created_at) VALUES(?,?,?,?)",
(idempotency_key, digest, encoded, datetime.now(timezone.utc).isoformat()),
)
con.commit()
return {"amendment_sha256": digest, "idempotent": False}
@staticmethod
def projection_id(
grant_id: str,
@@ -716,12 +784,14 @@ class Broker:
store: ReceiptStore,
upstream: Upstream,
expected_account: str,
expected_peer_user: str,
) -> None:
self.policy_path = policy_path
self.revocation_path = revocation_path
self.store = store
self.upstream = upstream
self.expected_account = expected_account
self.expected_peer_user = expected_peer_user
def policy(self) -> DurablePolicy:
policy = DurablePolicy.load(self.policy_path, self.revocation_path)
@@ -758,24 +828,90 @@ class Broker:
"separate_owner_release_approval_still_required": True,
}
@staticmethod
def validate_gate_amendment(
amendment: Mapping[str, Any],
*,
policy: DurablePolicy,
request: Mapping[str, Any],
) -> dict[str, Any]:
expected = {
"schema": GATE_AMENDMENT_SCHEMA,
"classification": GATE_AMENDMENT_CLASSIFICATION,
"supervisor_task_id": "r42-release-preparation-autonomous-supervisor-20260820",
"supervisor_execution_digest": "491457595667a95809f6f3cd19e9e7c9e74181070fd8038b86f30eb41bdfc231",
"actor_type": "credentialed_host_coordinator",
"operation": "createGovernedTask",
"repository": "aios/aios",
"original_external_request_sha256": "08fc2f3d1b64a2ca11da9f22da48de834cff3442fa7fa2d3bc390f8dbf9b254a",
"external_connector_rule": "ACK_STILL_REQUIRED",
"host_coordinator_rule": "DURABLE_OWNER_GOVERNED_TASK_PROJECT",
"owner_grant_id": policy.grant_id,
"owner_policy_sha256": policy.policy_sha256,
"builder_task_id": "r42-trusted-candidate-builder-runtime-wiring-20260820",
"builder_base_oid": "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"builder_expected_branch_oid": "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"builder_target_branch": "agent/r42-trusted-candidate-builder-runtime-wiring-20260820",
"builder_generation": 1,
"builder_idempotency_key": "r42-trusted-candidate-builder-runtime-wiring-binding-v1",
}
value = dict(amendment)
if value != expected:
raise AuthorizationError("GATE_AMENDMENT_BINDING_INVALID", 403)
request_coordinates = {
"task_id": expected["builder_task_id"],
"base_oid": expected["builder_base_oid"],
"expected_branch_oid": expected["builder_expected_branch_oid"],
"target_branch": expected["builder_target_branch"],
"generation": expected["builder_generation"],
"idempotency_key": expected["builder_idempotency_key"],
"repository": expected["repository"],
}
if any(request.get(key) != item for key, item in request_coordinates.items()):
raise AuthorizationError("GATE_AMENDMENT_BUILDER_BINDING_INVALID", 403)
return value
def execute(
self,
capability: str,
request: Mapping[str, Any],
*,
idempotency_key: str,
amendment: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
if not GRANT_RE.fullmatch(idempotency_key):
raise AuthorizationError("IDEMPOTENCY_KEY_INVALID")
policy = self.policy()
decision = policy.authorize(capability, request)
request_sha = decision["request_sha256"]
amendment_value: dict[str, Any] | None = None
if amendment is not None:
if capability != "governed-task:project":
raise AuthorizationError("GATE_AMENDMENT_CAPABILITY_INVALID", 403)
amendment_value = self.validate_gate_amendment(
amendment,
policy=policy,
request=request,
)
prior = self.store.get(idempotency_key)
if prior is not None:
if prior["request_sha256"] != request_sha:
raise AuthorizationError("IDEMPOTENCY_CONFLICT", 409)
response = prior["response"]
if amendment_value is not None:
amendment_result = self.store.put_gate_amendment(
idempotency_key,
amendment_value,
)
response = {
**response,
"gate_amendment": {
"classification": GATE_AMENDMENT_CLASSIFICATION,
"amendment_sha256": amendment_result["amendment_sha256"],
},
}
self.store.update_response(idempotency_key, request_sha, response)
if (
capability == "release:candidate"
and not response.get("projection_id")
@@ -862,6 +998,15 @@ class Broker:
),
"separate_owner_release_approval_still_required": True,
}
if amendment_value is not None:
amendment_result = self.store.put_gate_amendment(
idempotency_key,
amendment_value,
)
response["gate_amendment"] = {
"classification": GATE_AMENDMENT_CLASSIFICATION,
"amendment_sha256": amendment_result["amendment_sha256"],
}
self.store.put(idempotency_key, capability, request_sha, response)
return response
@@ -893,6 +1038,19 @@ class Handler(BaseHTTPRequestHandler):
def caller_account(self) -> str:
return self.headers.get("X-AIOS-Account", "").strip()
def peer_user(self) -> str:
if self.broker is None:
return ""
return peer_user_from_connection(self.connection)
def peer_authorized(self) -> bool:
if self.broker is None:
return False
try:
return self.peer_user() == self.broker.expected_peer_user
except AuthorizationError:
return False
def account_authorized(self) -> bool:
if self.broker is None:
return False
@@ -924,6 +1082,9 @@ class Handler(BaseHTTPRequestHandler):
200,
)
return
if not self.peer_authorized():
self.send_json({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)
return
if not self.account_authorized():
self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
return
@@ -938,6 +1099,12 @@ class Handler(BaseHTTPRequestHandler):
"grant_id": policy.grant_id,
"account": policy.account,
"capabilities": sorted(policy.capabilities),
"policy_sha256": policy.policy_sha256,
"constraints": {
"repository": policy.constraints.repository,
"allowed_target_prefix": policy.constraints.allowed_target_prefix,
"max_task_risk": policy.constraints.max_task_risk,
},
"persistent_until_revoked": True,
"separate_owner_release_approval_still_required": True,
}
@@ -957,6 +1124,9 @@ class Handler(BaseHTTPRequestHandler):
self.send_json({"error": exc.code}, exc.status)
def do_POST(self) -> None:
if not self.peer_authorized():
self.send_json({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)
return
if not self.account_authorized():
self.send_json({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)
return
@@ -970,14 +1140,24 @@ class Handler(BaseHTTPRequestHandler):
return
try:
body = self.body()
if set(body) != {"idempotency_key", "request"}:
if set(body) not in (
{"idempotency_key", "request"},
{"idempotency_key", "request", "amendment"},
):
raise AuthorizationError("BROKER_FIELDS_INVALID")
if not isinstance(body["request"], Mapping):
raise AuthorizationError("BROKER_REQUEST_INVALID")
if "amendment" in body and not isinstance(body["amendment"], Mapping):
raise AuthorizationError("GATE_AMENDMENT_INVALID")
value = self.broker.execute(
capability,
dict(body["request"]),
idempotency_key=str(body["idempotency_key"]),
amendment=(
dict(body["amendment"])
if isinstance(body.get("amendment"), Mapping)
else None
),
)
self.send_json(value, 200)
except AuthorizationError as exc:
@@ -989,6 +1169,14 @@ class UnixHTTPServer(socketserver.ThreadingUnixStreamServer):
def serve() -> None:
credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
allowed_peer_user = os.environ.get(
"AIOS_OWNER_AUTH_BROKER_ALLOWED_USER",
"aios-autonomous-agent-worker",
).strip()
if not allowed_peer_user:
raise RuntimeError("BROKER_ALLOWED_USER_MISSING")
allowed_peer_uid = pwd.getpwnam(allowed_peer_user).pw_uid
allowed_peer_gid = pwd.getpwnam(allowed_peer_user).pw_gid
policy_path = Path(
os.environ.get(
"AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY",
@@ -1029,6 +1217,7 @@ def serve() -> None:
bridge_token=bridge_token,
),
expected_account=account,
expected_peer_user=allowed_peer_user,
)
socket_path = Path(
os.environ.get(
@@ -1044,6 +1233,7 @@ def serve() -> None:
raise RuntimeError("BROKER_SOCKET_CONFLICT")
socket_path.unlink()
server = UnixHTTPServer(str(socket_path), Handler)
os.chown(socket_path, allowed_peer_uid, allowed_peer_gid)
os.chmod(socket_path, 0o660)
try:
server.serve_forever()
@@ -9,9 +9,12 @@ heartbeat lease while later phases observe and repair through governed APIs.
from __future__ import annotations
from datetime import datetime, timezone
import http.client
import json
import os
from pathlib import Path
import re
import socket
import tempfile
import time
from typing import Any, Mapping
@@ -47,6 +50,18 @@ STATE_ROOT = Path(
"/var/lib/aios-release-preparation-supervisor-coordinator",
)
)
BROKER_SOCKET = Path(os.environ.get(
"AIOS_OWNER_AUTH_BROKER_SOCKET",
"/run/aios-owner-authorization-broker/broker.sock",
))
BROKER_ACCOUNT = "chatgpt-governed-implementer-singlehost"
EXTERNAL_REQUEST_SHA256 = "08fc2f3d1b64a2ca11da9f22da48de834cff3442fa7fa2d3bc390f8dbf9b254a"
SUPERVISOR_EXECUTION_DIGEST = "491457595667a95809f6f3cd19e9e7c9e74181070fd8038b86f30eb41bdfc231"
BUILDER_TASK_ID = "r42-trusted-candidate-builder-runtime-wiring-20260820"
BUILDER_BASE_OID = "d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3"
BUILDER_TARGET_BRANCH = "agent/r42-trusted-candidate-builder-runtime-wiring-20260820"
BUILDER_IDEMPOTENCY_KEY = "r42-trusted-candidate-builder-runtime-wiring-binding-v1"
GATE_AMENDMENT_CLASSIFICATION = "SUPERVISOR_GLOBAL_EXTERNAL_ACK_GATE_REMEDIATED"
class CoordinatorError(RuntimeError):
@@ -55,6 +70,18 @@ class CoordinatorError(RuntimeError):
self.code = code
class UnixHTTPConnection(http.client.HTTPConnection):
def __init__(self, socket_path: Path, *, timeout: int = 30) -> None:
super().__init__("localhost", timeout=timeout)
self.socket_path = socket_path
def connect(self) -> None:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
sock.connect(str(self.socket_path))
self.sock = sock
def now() -> str:
return datetime.now(timezone.utc).isoformat()
@@ -127,6 +154,225 @@ def api_call(
return 599, {"error": "AGENT_TASK_API_UNAVAILABLE"}
def broker_call(
method: str,
route: str,
body: Mapping[str, Any] | None = None,
) -> tuple[int, dict[str, Any]]:
if not BROKER_SOCKET.is_absolute():
raise CoordinatorError("OWNER_BROKER_SOCKET_INVALID")
try:
if BROKER_SOCKET.is_symlink():
raise CoordinatorError("OWNER_BROKER_SOCKET_INVALID")
except OSError as exc:
raise CoordinatorError("OWNER_BROKER_UNAVAILABLE") from exc
connection = UnixHTTPConnection(BROKER_SOCKET)
payload = None if body is None else canonical(dict(body))
headers = {"Accept": "application/json", "X-AIOS-Account": BROKER_ACCOUNT}
if payload is not None:
headers["Content-Type"] = "application/json"
try:
connection.request(method, route, body=payload, headers=headers)
response = connection.getresponse()
raw = response.read()
value = json.loads(raw.decode("utf-8"))
if not isinstance(value, Mapping):
raise CoordinatorError("OWNER_BROKER_RESPONSE_INVALID")
return response.status, dict(value)
except (OSError, TimeoutError, http.client.HTTPException, ValueError) as exc:
raise CoordinatorError("OWNER_BROKER_UNAVAILABLE") from exc
finally:
connection.close()
def authorize_actor(
actor_type: str,
*,
external_ack: Mapping[str, Any] | None = None,
host_authorization: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
if actor_type == "external_connector":
expected = {
"account": BROKER_ACCOUNT,
"capability": "governed-task:project",
"operation": "createGovernedTask",
"single_use": True,
"request_sha256": EXTERNAL_REQUEST_SHA256,
}
if external_ack is None:
raise CoordinatorError("WAIT_EXTERNAL_ACK")
if dict(external_ack) != expected:
raise CoordinatorError("EXTERNAL_ACK_INVALID")
return {
"actor_type": actor_type,
"decision": "ALLOW_EXTERNAL_CONNECTOR_EXACT_ACK",
"external_ack_required": True,
"request_sha256": EXTERNAL_REQUEST_SHA256,
"authorized_capabilities": ["governed-task:project"],
}
if actor_type == "credentialed_host_coordinator":
if not isinstance(host_authorization, Mapping):
raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_REQUIRED")
constraints = host_authorization.get("constraints")
capabilities = host_authorization.get("capabilities")
if (
host_authorization.get("account") != BROKER_ACCOUNT
or host_authorization.get("persistent_until_revoked") is not True
or host_authorization.get("separate_owner_release_approval_still_required") is not True
or not isinstance(capabilities, list)
or "governed-task:project" not in capabilities
or not isinstance(constraints, Mapping)
or constraints.get("repository") != "aios/aios"
or constraints.get("allowed_target_prefix") != "agent/"
or constraints.get("max_task_risk") not in {"low", "engineering"}
or re.fullmatch(r"[0-9a-f]{64}", str(host_authorization.get("policy_sha256") or "")) is None
or not str(host_authorization.get("grant_id") or "")
):
raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_DENY")
return {
"actor_type": actor_type,
"decision": "ALLOW_HOST_COORDINATOR_GOVERNED_PROJECTION",
"external_ack_required": False,
"durable_owner_governed_task_project": "verified",
"grant_id": str(host_authorization["grant_id"]),
"policy_sha256": str(host_authorization["policy_sha256"]),
# Do not inherit release:candidate even if the durable policy also has it.
"authorized_capabilities": ["governed-task:project"],
}
raise CoordinatorError("UNKNOWN_ACTOR_DENY")
def builder_request() -> dict[str, Any]:
capabilities = [
"git.read",
"git.test",
"governed.write.branch",
"governed.write.file",
]
return {
"task_id": BUILDER_TASK_ID,
"repository": "aios/aios",
"base_oid": BUILDER_BASE_OID,
"expected_branch_oid": BUILDER_BASE_OID,
"target_branch": BUILDER_TARGET_BRANCH,
"generation": 1,
"branch_mode": "CREATE_NEW",
"risk": "engineering",
"description": "Wire TrustedCandidateBuilder through the canonical authenticated Release API and fixed candidate retry chain.",
"idempotency_key": BUILDER_IDEMPOTENCY_KEY,
"priority": 100,
"max_attempts": 3,
"require_online_task_projection": True,
"online_task": {
"goal": "Implement and verify TrustedCandidateBuilder runtime wiring from the fixed d957a216 lineage.",
"instructions": (
"Capture the real Builder and Release API contracts, implement only the fixed governed branch, "
"prove authenticated POST {} returns HTTP 400, and stop before Owner release approval or Production. "
"No main write, force push, tag, baseline, production deploy, runtime switch, credential forwarding, "
"or legacy candidate input."
),
"success_criteria": [
{"id": "builder_wiring", "description": "TrustedCandidateBuilder wiring is verified."},
{"id": "release_api", "description": "Authenticated Release API POST {} returns HTTP 400."},
{"id": "governed_candidate", "description": "All modified files are committed only to the fixed governed branch."},
],
"constraints": {
"main_write": False,
"production_write": False,
"owner_release_approval": False,
"tag_publish": False,
"baseline_finalize": False,
"runtime_switch": False,
"credential_forwarding": False,
"legacy_candidate_input": False,
},
"required_capabilities": capabilities,
"priority": 100,
"max_attempts": 3,
},
"execution": {
"action": "source_capture",
"adapter": "source_capture",
"capability_scope": capabilities,
"source_signals": [
"TrustedCandidateBuilder runtime wiring",
"Release API authenticated POST empty object HTTP 400",
"fixed V2R2C2 retry-extension attempt 2",
"V2R2C2 and V2R2C3 candidate verification",
],
},
}
def gate_amendment(host_decision: Mapping[str, Any]) -> dict[str, Any]:
return {
"schema": "aios.release-preparation.gate-amendment.v1",
"classification": GATE_AMENDMENT_CLASSIFICATION,
"supervisor_task_id": TASK_ID,
"supervisor_execution_digest": SUPERVISOR_EXECUTION_DIGEST,
"actor_type": "credentialed_host_coordinator",
"operation": "createGovernedTask",
"repository": "aios/aios",
"original_external_request_sha256": EXTERNAL_REQUEST_SHA256,
"external_connector_rule": "ACK_STILL_REQUIRED",
"host_coordinator_rule": "DURABLE_OWNER_GOVERNED_TASK_PROJECT",
"owner_grant_id": host_decision["grant_id"],
"owner_policy_sha256": host_decision["policy_sha256"],
"builder_task_id": BUILDER_TASK_ID,
"builder_base_oid": BUILDER_BASE_OID,
"builder_expected_branch_oid": BUILDER_BASE_OID,
"builder_target_branch": BUILDER_TARGET_BRANCH,
"builder_generation": 1,
"builder_idempotency_key": BUILDER_IDEMPOTENCY_KEY,
}
def ensure_builder_projection() -> dict[str, Any]:
status, authorization = broker_call("GET", "/v1/capabilities")
if status != 200:
raise CoordinatorError("HOST_COORDINATOR_DURABLE_OWNER_DENY")
decision = authorize_actor(
"credentialed_host_coordinator",
host_authorization=authorization,
)
request = builder_request()
status, receipt = broker_call(
"POST",
"/v1/project-governed-task",
{
"idempotency_key": BUILDER_IDEMPOTENCY_KEY,
"request": request,
"amendment": gate_amendment(decision),
},
)
upstream = receipt.get("upstream_response")
amendment = receipt.get("gate_amendment")
if (
status != 200
or receipt.get("operation_state") != "EXECUTED"
or receipt.get("capability") != "governed-task:project"
or receipt.get("grant_id") != decision["grant_id"]
or receipt.get("policy_sha256") != decision["policy_sha256"]
or receipt.get("external_capability_projection_required") is not False
or not isinstance(upstream, Mapping)
or upstream.get("task_id") != BUILDER_TASK_ID
or upstream.get("base_oid") != BUILDER_BASE_OID
or upstream.get("expected_branch_oid") != BUILDER_BASE_OID
or upstream.get("target_branch") != BUILDER_TARGET_BRANCH
or upstream.get("generation") != 1
or not isinstance(amendment, Mapping)
or amendment.get("classification") != GATE_AMENDMENT_CLASSIFICATION
or re.fullmatch(r"[0-9a-f]{64}", str(amendment.get("amendment_sha256") or "")) is None
):
raise CoordinatorError("BUILDER_PROJECTION_RECEIPT_INVALID")
return {
**decision,
"builder_task_id": BUILDER_TASK_ID,
"builder_binding_verified": True,
"gate_amendment_sha256": amendment["amendment_sha256"],
}
def load_agent_token() -> str:
path = Path(os.environ.get("CREDENTIALS_DIRECTORY", "")) / "agent_token"
if path.is_symlink() or not path.is_file():
@@ -174,11 +420,14 @@ def validate_task(task: Mapping[str, Any]) -> dict[str, Any]:
return dict(execution)
def heartbeat_progress(sequence: int) -> dict[str, Any]:
return {
def heartbeat_progress(
sequence: int,
gate: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
progress = {
"stage": "supervisor_coordinator_active",
"phase": "external_caller_projection",
"state": "PENDING_EXTERNAL_CONNECTOR_ACK",
"state": "PENDING_ACTOR_AWARE_ADMISSION",
"sequence": sequence,
"control_plane_access_granted": True,
"external_planner_role": "UNPRIVILEGED_PLANNER",
@@ -187,6 +436,17 @@ def heartbeat_progress(sequence: int) -> dict[str, Any]:
"production_write": False,
"owner_approval_submitted": False,
}
if gate is not None:
progress.update({
"state": str(gate.get("decision") or "DENY"),
"actor_type": gate.get("actor_type"),
"external_ack_required": gate.get("external_ack_required"),
"durable_owner_governed_task_project": gate.get("durable_owner_governed_task_project"),
"builder_task_id": gate.get("builder_task_id"),
"builder_binding_verified": gate.get("builder_binding_verified", False),
"gate_amendment_sha256": gate.get("gate_amendment_sha256"),
})
return progress
def coordinate_claim(
@@ -199,8 +459,20 @@ def coordinate_claim(
if not isinstance(lease_token, str) or len(lease_token) < 24:
raise CoordinatorError("COORDINATOR_LEASE_TOKEN_MISSING")
sequence = 0
gate: dict[str, Any] | None = None
while max_heartbeats is None or sequence < max_heartbeats:
sequence += 1
if gate is None:
try:
gate = ensure_builder_projection()
audit("SUPERVISOR_ACTOR_AWARE_GATE_ALLOWED", **gate)
except CoordinatorError as exc:
audit(
"SUPERVISOR_ACTOR_AWARE_GATE_DENIED",
actor_type="credentialed_host_coordinator",
decision="DENY",
error=exc.code,
)
status, response = api_call(
"POST",
f"/v1/tasks/{TASK_ID}/heartbeat",
@@ -208,7 +480,7 @@ def coordinate_claim(
{
"lease_token": lease_token,
"lease_seconds": LEASE_SECONDS,
"progress": heartbeat_progress(sequence),
"progress": heartbeat_progress(sequence, gate),
},
)
if status != 200:
@@ -6,7 +6,7 @@ Wants=aios-chatgpt-bridge-v2-staging.service
[Service]
Type=simple
User=root
Group=aios-executor-exchange
Group=aios-autonomous-agent-worker
RuntimeDirectory=aios-owner-authorization-broker
RuntimeDirectoryMode=0750
StateDirectory=aios-owner-authorization-broker
@@ -16,6 +16,7 @@ Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY=/etc/aios-gitops/owner-durable-authorization.json
Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_REVOCATIONS=/etc/aios-gitops/owner-durable-authorization-revocations.json
Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
Environment=AIOS_OWNER_AUTH_BROKER_ALLOWED_USER=aios-autonomous-agent-worker
Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_durable_owner_authorization.py
Restart=on-failure
@@ -1,7 +1,7 @@
[Unit]
Description=AIOS credential-isolated release-preparation supervisor coordinator
After=network-online.target aios-agent-task-api.service aios-autonomous-agent-register.service
Wants=network-online.target aios-agent-task-api.service
After=network-online.target aios-agent-task-api.service aios-autonomous-agent-register.service aios-owner-durable-authorization-broker.service
Wants=network-online.target aios-agent-task-api.service aios-owner-durable-authorization-broker.service
Requires=aios-autonomous-agent-register.service
[Service]
@@ -18,6 +18,7 @@ Environment=AIOS_SUPERVISOR_COORDINATOR_STATE_ROOT=/var/lib/aios-release-prepara
Environment=AIOS_SUPERVISOR_COORDINATOR_POLL_SECONDS=5
Environment=AIOS_SUPERVISOR_COORDINATOR_HEARTBEAT_SECONDS=10
Environment=AIOS_SUPERVISOR_COORDINATOR_LEASE_SECONDS=300
Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_preparation_supervisor_coordinator.py
Restart=always
RestartSec=5
+166 -1
View File
@@ -1,10 +1,12 @@
from __future__ import annotations
from datetime import datetime, timezone
import struct
import json
from pathlib import Path
import sqlite3
import sys
from types import SimpleNamespace
import pytest
@@ -134,6 +136,7 @@ def broker(tmp_path,monkeypatch):
store=store,
upstream=upstream,
expected_account=ACCOUNT,
expected_peer_user="aios-autonomous-agent-worker",
)
monkeypatch.setattr(
auth.DurablePolicy,
@@ -573,16 +576,110 @@ def test_receipt_store_contains_no_owner_policy_or_credentials(
assert "policy_json" not in columns
def gate_amendment(policy):
return {
"schema":auth.GATE_AMENDMENT_SCHEMA,
"classification":auth.GATE_AMENDMENT_CLASSIFICATION,
"supervisor_task_id":"r42-release-preparation-autonomous-supervisor-20260820",
"supervisor_execution_digest":"491457595667a95809f6f3cd19e9e7c9e74181070fd8038b86f30eb41bdfc231",
"actor_type":"credentialed_host_coordinator",
"operation":"createGovernedTask",
"repository":"aios/aios",
"original_external_request_sha256":"08fc2f3d1b64a2ca11da9f22da48de834cff3442fa7fa2d3bc390f8dbf9b254a",
"external_connector_rule":"ACK_STILL_REQUIRED",
"host_coordinator_rule":"DURABLE_OWNER_GOVERNED_TASK_PROJECT",
"owner_grant_id":policy.grant_id,
"owner_policy_sha256":policy.policy_sha256,
"builder_task_id":"r42-trusted-candidate-builder-runtime-wiring-20260820",
"builder_base_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"builder_expected_branch_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"builder_target_branch":"agent/r42-trusted-candidate-builder-runtime-wiring-20260820",
"builder_generation":1,
"builder_idempotency_key":"r42-trusted-candidate-builder-runtime-wiring-binding-v1",
}
def builder_governed_request():
value=governed_request()
value.update({
"task_id":"r42-trusted-candidate-builder-runtime-wiring-20260820",
"base_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"expected_branch_oid":"d957a216a6e9bd4c2292ffc6e44876a1c08fa1e3",
"target_branch":"agent/r42-trusted-candidate-builder-runtime-wiring-20260820",
"generation":1,
"idempotency_key":"r42-trusted-candidate-builder-runtime-wiring-binding-v1",
})
return value
def test_task_bound_gate_amendment_is_validated_and_persisted(tmp_path,monkeypatch):
instance,upstream,store,policy=broker(tmp_path,monkeypatch)
request=builder_governed_request()
response=instance.execute(
"governed-task:project",
request,
idempotency_key=request["idempotency_key"],
amendment=gate_amendment(policy),
)
assert response["operation_state"]=="EXECUTED"
assert response["gate_amendment"]["classification"]==auth.GATE_AMENDMENT_CLASSIFICATION
assert len(upstream.project_calls)==1
con=sqlite3.connect(store.path)
row=con.execute(
"select amendment_sha256,amendment_json from gate_amendments where idempotency_key=?",
(request["idempotency_key"],),
).fetchone()
assert row[0]==response["gate_amendment"]["amendment_sha256"]
persisted=json.loads(row[1])
assert persisted["supervisor_task_id"]=="r42-release-preparation-autonomous-supervisor-20260820"
assert persisted["builder_task_id"]==request["task_id"]
for forbidden in ("bearer ","access_token","refresh_token","private_key","credential_file"):
assert forbidden not in row[1].lower()
def test_gate_amendment_rejects_wrong_builder_or_policy_binding(tmp_path,monkeypatch):
instance,_,_,policy=broker(tmp_path,monkeypatch)
request=builder_governed_request()
for bad in (
{**gate_amendment(policy),"builder_base_oid":"0"*40},
{**gate_amendment(policy),"owner_policy_sha256":"0"*64},
{**gate_amendment(policy),"actor_type":"unknown"},
):
with pytest.raises(auth.AuthorizationError,match="GATE_AMENDMENT_BINDING_INVALID"):
instance.execute(
"governed-task:project",
request,
idempotency_key=request["idempotency_key"],
amendment=bad,
)
def test_gate_amendment_does_not_inherit_release_candidate_authority(tmp_path,monkeypatch):
instance,_,_,policy=broker(tmp_path,monkeypatch)
request=builder_governed_request()
response=instance.execute(
"governed-task:project",
request,
idempotency_key=request["idempotency_key"],
amendment=gate_amendment(policy),
)
assert response["capability"]=="governed-task:project"
assert response["external_capability_projection_required"] is False
assert "release:candidate" not in json.dumps(response["gate_amendment"])
def test_systemd_broker_uses_unix_socket_and_only_host_control_credential():
unit=(
ROOT/"deploy/systemd/aios-owner-durable-authorization-broker.service"
).read_text()
assert "LoadCredential=agent_task_control_token:" in unit
assert "AIOS_OWNER_AUTH_BROKER_ALLOWED_USER=aios-autonomous-agent-worker" in unit
assert "release_candidate_token" not in unit
assert "18947" not in unit
assert "AIOS_RELEASE_CANDIDATE_API_URL" not in unit
assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" in unit
assert "Group=aios-executor-exchange" in unit
assert "User=root" in unit
assert "Group=aios-autonomous-agent-worker" in unit
assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
assert "IPAddressDeny=any" in unit
assert "IPAddressAllow=127.0.0.0/8" in unit
@@ -609,6 +706,74 @@ def test_source_contains_no_release_builder_endpoint_or_high_risk_execution():
assert "INITIAL_EXECUTION_ACTION_NOT_AUTHORIZED" in source
def test_peer_credential_mapping_extracts_expected_user(monkeypatch):
class FakeSocket:
def getsockopt(self, *_args, **_kwargs):
return struct.pack("3i", 0, 555, 0)
monkeypatch.setattr(
auth.pwd,
"getpwuid",
lambda uid: SimpleNamespace(pw_name="aios-autonomous-agent-worker"),
)
assert (
auth.peer_user_from_connection(FakeSocket())
== "aios-autonomous-agent-worker"
)
def test_unknown_or_shared_aos_worker_user_is_rejected_before_business_logic(monkeypatch):
handler = object.__new__(auth.Handler)
observed = []
handler.path = "/v1/authorize-release-candidate"
handler.headers = {"X-AIOS-Account": ACCOUNT}
handler.send_json = lambda body, status: observed.append((body, status))
handler.connection = object()
handler.broker = SimpleNamespace(
expected_account=ACCOUNT,
expected_peer_user="aios-autonomous-agent-worker",
execute=lambda *args, **kwargs: observed.append("executed"),
)
handler.body = lambda: observed.append(("should_never_read_body",))
monkeypatch.setattr(auth, "peer_user_from_connection", lambda *_args: "aios-engineering-worker")
handler.do_POST()
assert observed == [({"error": "SOCKUSER_NOT_AUTHORIZED"}, 403)]
def test_account_check_stays_secondary_after_peer_auth(monkeypatch):
handler = object.__new__(auth.Handler)
observed = []
handler.path = "/v1/authorize-release-candidate"
handler.headers = {"X-AIOS-Account": "wrong-account"}
handler.send_json = lambda body, status: observed.append((body, status))
handler.connection = object()
handler.broker = SimpleNamespace(
expected_account=ACCOUNT,
expected_peer_user="aios-autonomous-agent-worker",
execute=lambda *args, **kwargs: observed.append("executed"),
)
handler.body = lambda: observed.append(("should_never_read_body",))
monkeypatch.setattr(
auth,
"peer_user_from_connection",
lambda *_args: "aios-autonomous-agent-worker",
)
handler.do_POST()
assert observed == [({"error": "ACCOUNT_NOT_AUTHORIZED"}, 403)]
def test_health_endpoint_remains_unauthenticated():
handler = object.__new__(auth.Handler)
observed = []
handler.path = "/health"
handler.send_json = lambda body, status: observed.append((body, status))
handler.do_GET()
assert observed[0][1] == 200
assert observed[0][0]["status"] == "ok"
def test_docs_avoid_new_broker_bearer_secret():
doc=(
@@ -80,6 +80,136 @@ def test_heartbeat_is_control_plane_host_only_and_never_forwards_credentials():
assert "lease_token" not in encoded
def exact_external_ack():
return {
"account":coordinator.BROKER_ACCOUNT,
"capability":"governed-task:project",
"operation":"createGovernedTask",
"single_use":True,
"request_sha256":coordinator.EXTERNAL_REQUEST_SHA256,
}
def durable_host_authorization(**overrides):
value={
"grant_id":"owner-permanent-preparation-20260819",
"account":coordinator.BROKER_ACCOUNT,
"capabilities":["governed-task:project","release:candidate"],
"policy_sha256":"a"*64,
"constraints":{
"repository":"aios/aios",
"allowed_target_prefix":"agent/",
"max_task_risk":"engineering",
},
"persistent_until_revoked":True,
"separate_owner_release_approval_still_required":True,
}
value.update(overrides)
return value
def test_external_connector_requires_exact_single_use_ack():
with pytest.raises(coordinator.CoordinatorError,match="WAIT_EXTERNAL_ACK"):
coordinator.authorize_actor("external_connector")
wrong=exact_external_ack();wrong["request_sha256"]="0"*64
with pytest.raises(coordinator.CoordinatorError,match="EXTERNAL_ACK_INVALID"):
coordinator.authorize_actor("external_connector",external_ack=wrong)
decision=coordinator.authorize_actor(
"external_connector",external_ack=exact_external_ack()
)
assert decision["decision"]=="ALLOW_EXTERNAL_CONNECTOR_EXACT_ACK"
assert decision["external_ack_required"] is True
assert decision["request_sha256"]==coordinator.EXTERNAL_REQUEST_SHA256
def test_host_coordinator_requires_live_durable_owner_scope_without_external_ack():
with pytest.raises(coordinator.CoordinatorError,match="DURABLE_OWNER_REQUIRED"):
coordinator.authorize_actor("credentialed_host_coordinator")
for denied in (
durable_host_authorization(capabilities=["release:candidate"]),
durable_host_authorization(constraints={"repository":"other/repo","allowed_target_prefix":"agent/","max_task_risk":"engineering"}),
durable_host_authorization(constraints={"repository":"aios/aios","allowed_target_prefix":"release/","max_task_risk":"engineering"}),
durable_host_authorization(constraints={"repository":"aios/aios","allowed_target_prefix":"agent/","max_task_risk":"high"}),
durable_host_authorization(persistent_until_revoked=False),
):
with pytest.raises(coordinator.CoordinatorError,match="DURABLE_OWNER_DENY"):
coordinator.authorize_actor(
"credentialed_host_coordinator",host_authorization=denied
)
decision=coordinator.authorize_actor(
"credentialed_host_coordinator",
host_authorization=durable_host_authorization(),
)
assert decision["decision"]=="ALLOW_HOST_COORDINATOR_GOVERNED_PROJECTION"
assert decision["external_ack_required"] is False
assert decision["authorized_capabilities"]==["governed-task:project"]
assert "release:candidate" not in decision["authorized_capabilities"]
def test_unknown_actor_is_fail_closed():
with pytest.raises(coordinator.CoordinatorError,match="UNKNOWN_ACTOR_DENY"):
coordinator.authorize_actor("unknown")
def test_builder_request_and_amendment_preserve_fixed_binding():
request=coordinator.builder_request()
assert request["task_id"]==coordinator.BUILDER_TASK_ID
assert request["base_oid"]==coordinator.BUILDER_BASE_OID
assert request["expected_branch_oid"]==coordinator.BUILDER_BASE_OID
assert request["target_branch"]==coordinator.BUILDER_TARGET_BRANCH
assert request["generation"]==1
assert request["idempotency_key"]==coordinator.BUILDER_IDEMPOTENCY_KEY
decision=coordinator.authorize_actor(
"credentialed_host_coordinator",
host_authorization=durable_host_authorization(),
)
amendment=coordinator.gate_amendment(decision)
assert amendment["supervisor_task_id"]==coordinator.TASK_ID
assert amendment["supervisor_execution_digest"]==coordinator.SUPERVISOR_EXECUTION_DIGEST
assert amendment["external_connector_rule"]=="ACK_STILL_REQUIRED"
assert amendment["host_coordinator_rule"]=="DURABLE_OWNER_GOVERNED_TASK_PROJECT"
assert amendment["builder_task_id"]==request["task_id"]
encoded=json.dumps(amendment,sort_keys=True).lower()
for forbidden in ("bearer ","access_token","refresh_token","private_key"):
assert forbidden not in encoded
def test_host_coordinator_projects_builder_through_broker_and_verifies_receipt(monkeypatch):
calls=[]
authorization=durable_host_authorization()
def fake(method,route,body=None):
calls.append((method,route,body))
if method=="GET":
return 200,authorization
request=body["request"]
return 200,{
"operation_state":"EXECUTED",
"capability":"governed-task:project",
"grant_id":authorization["grant_id"],
"policy_sha256":authorization["policy_sha256"],
"external_capability_projection_required":False,
"upstream_response":{
"task_id":request["task_id"],
"base_oid":request["base_oid"],
"expected_branch_oid":request["expected_branch_oid"],
"target_branch":request["target_branch"],
"generation":request["generation"],
},
"gate_amendment":{
"classification":coordinator.GATE_AMENDMENT_CLASSIFICATION,
"amendment_sha256":"b"*64,
},
}
monkeypatch.setattr(coordinator,"broker_call",fake)
decision=coordinator.ensure_builder_projection()
assert decision["builder_binding_verified"] is True
assert decision["external_ack_required"] is False
assert [item[1] for item in calls]==[
"/v1/capabilities","/v1/project-governed-task"
]
assert calls[1][2]["amendment"]["owner_grant_id"]==authorization["grant_id"]
def test_coordinate_claim_heartbeats_without_submitting_result(monkeypatch,tmp_path):
calls=[]
def fake(method,route,token,body=None):
@@ -111,6 +241,8 @@ def test_units_use_existing_encrypted_agent_credential_and_no_planner_credential
assert "LoadCredentialEncrypted=agent_token:" in coordinator_unit
assert "control_token" not in coordinator_unit
assert "SupplementaryGroups=aios-agent-runtime" in coordinator_unit
assert "aios-owner-durable-authorization-broker.service" in coordinator_unit
assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" in coordinator_unit
assert "LoadCredential=agent_task_control_token:" in projection_unit
assert "User=aios-agent-task" in projection_unit
assert "ReadWritePaths=/var/lib/aios-agent-task" in projection_unit