Compare commits

...
11 changed files with 971 additions and 16 deletions
+86
View File
@@ -0,0 +1,86 @@
"""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
@@ -0,0 +1,358 @@
"""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,6 +15,11 @@ 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",
@@ -155,6 +160,36 @@ 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:
@@ -280,13 +315,30 @@ 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,
@@ -318,22 +370,21 @@ def run_compatible_signer(
"signing_key_identity": signer,
}
prepared = _prepare(grant_dir)
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,
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",
grant_dir / "approval-plan.canonical.json",
grant_dir / "approval-plan.sig",
grant_dir / "approval-plan.key-id",
)
_sign(
signer_command,
grant_namespace,
_sign_with_broker(
active_broker,
"grant_payload",
grant_dir / "grant-payload.canonical.json",
grant_dir / "signed-grant.sig",
grant_dir / "signed-grant.key-id",
+33
View File
@@ -0,0 +1,33 @@
[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
@@ -0,0 +1,14 @@
[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
@@ -0,0 +1,24 @@
#!/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'
@@ -0,0 +1,90 @@
#!/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())
+105
View File
@@ -0,0 +1,105 @@
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
@@ -0,0 +1,130 @@
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
@@ -0,0 +1,50 @@
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
+18 -4
View File
@@ -20,6 +20,7 @@ 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,
@@ -113,9 +114,13 @@ 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))
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)
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")
def test_resolve_ref_oid_uses_full_synthetic_commit_and_fails_closed(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
@@ -188,9 +193,18 @@ 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")
run_compatible_signer([str(signer)], grant_dir, root / "result.json", require_broker=False)
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",