aios/aios
1
0
Fork 0
Code Issues Pull Requests 5 Actions Packages Projects Releases Wiki Activity

Bind staging artifacts to exact dual-target installer contracts

Browse Source
This commit is contained in:
Codex Security
2026-09-26 14:02:52 +08:00
parent 8fd457f8f5
commit 6582e8a5e1
2 changed files with 184 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/build_c5_staging_artifact.py
+159 -48
View File
@@ -26,11 +26,17 @@ from typing import Any, Iterable, Mapping
REPOSITORY = "aios/aios"
RELEASE_BASE = "/var/lib/aios-gitops/bridge-v2-staging/releases"
UNIT = "aios-chatgpt-bridge-v2-staging.service"
DROPIN_NAME = "32-tildes-c5-source.conf"
BRIDGE_TARGET = "bridge-18892"
ISSUER_TARGET = "principal-issuer-18946"
ISSUER_UNIT = "aios-task-principal-issuer.service"
DROPIN_NAME = "~" * 32 + "-c5-source.conf"
DROPIN = f"/etc/systemd/system/{UNIT}.d/{DROPIN_NAME}"
ISSUER_DROPIN = f"/etc/systemd/system/{ISSUER_UNIT}.d/{DROPIN_NAME}"
RUNTIME_URL = "http://127.0.0.1:18892/v2/runtime"
ISSUER_RUNTIME_URL = "http://127.0.0.1:18946/health"
PORT = 18892
REQUEST_SCHEMA = "aios.c5-staging-installation-request.v1"
ISSUER_PORT = 18946
REQUEST_SCHEMA = "aios.c5-staging-installation-request.v2"
OWNER_NAMESPACE = "aios-owner-approval-v1"
MAX_TTL = 900
DEFAULT_TIMEOUT = 3
@@ -53,6 +59,7 @@ _HEX40 = re.compile(r"^[0-9a-f]{40}$")
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
_NONCE = re.compile(r"^[A-Za-z0-9._:-]{16,128}$")
_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{2,255}$")
_SOURCE_MODES = {0o444, 0o640, 0o644, 0o755}
class BuildError(ValueError):
@@ -170,15 +177,30 @@ def _root(support_commit: str) -> str:
return f"{RELEASE_BASE}/c5-source-{support_commit}"
def _read_existing(snapshot_path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
def _read_existing(snapshot_path: Path, target: str = BRIDGE_TARGET) -> tuple[dict[str, Any], dict[str, Any]]:
snapshot, _ = _json(snapshot_path)
value = snapshot.get("existing", snapshot)
profiles = snapshot.get("profiles")
if isinstance(profiles, Mapping):
profile = profiles.get(target)
if not isinstance(profile, Mapping):
_error("EXISTING_SNAPSHOT_TARGET_MISSING")
value = profile.get("existing", profile)
selected_snapshot: dict[str, Any] = dict(profile)
else:
value = snapshot.get("existing", snapshot)
selected_snapshot = dict(snapshot)
if not isinstance(value, Mapping):
_error("EXISTING_SNAPSHOT_INVALID")
required = {"unit_active", "effective_config_sha256", "dropin", "pythonpath", "source_files"}
if set(value) != required:
required = {
"unit_active", "effective_config_sha256", "dropin", "pythonpath", "source_files",
"activation_file", "principal_scope_config",
}
if not required.issubset(value):
_error("EXISTING_SNAPSHOT_FIELDS_INVALID")
existing = dict(value)
# Keep only the installer-owned snapshot projection. Live capture files
# also carry target/unit/readiness metadata; copying those keys into a
# request would fail the typed request's exact-field check.
existing = {key: value[key] for key in required}
if existing["unit_active"] is not True:
_error("EXISTING_UNIT_NOT_ACTIVE")
if not isinstance(existing["effective_config_sha256"], str) or not _HEX64.fullmatch(existing["effective_config_sha256"]):
@@ -189,6 +211,8 @@ def _read_existing(snapshot_path: Path) -> tuple[dict[str, Any], dict[str, Any]]
for item in pythonpath:
if not isinstance(item, str) or not item.startswith("/") or ".." in PurePosixPath(item).parts:
_error("EXISTING_PYTHONPATH_INVALID")
if value["activation_file"] is not None or value["principal_scope_config"] is not None:
_error("EXISTING_ACTIVATION_ALREADY_PRESENT")
dropin = existing["dropin"]
if not isinstance(dropin, Mapping) or set(dropin) != {"exists", "sha256"}:
_error("EXISTING_DROPIN_INVALID")
@@ -210,19 +234,27 @@ def _read_existing(snapshot_path: Path) -> tuple[dict[str, Any], dict[str, Any]]
or "\\" in record["path"]
or not _HEX64.fullmatch(str(record["sha256"]))
or type(record["size"]) is not int or record["size"] < 0
or record["mode"] not in {0o644, 0o755} or record["kind"] != "existing_source"
or record["mode"] not in _SOURCE_MODES or record["kind"] != "existing_source"
):
_error("EXISTING_SOURCE_FILES_INVALID")
if not any(record["path"] != base and record["path"].startswith(base.rstrip("/") + "/") for base in pythonpath):
_error("EXISTING_SOURCE_OUTSIDE_PYTHONPATH")
return existing, snapshot
return existing, selected_snapshot
def _old_module(existing: Mapping[str, Any], target: str) -> dict[str, Any]:
suffix = (
"/aios_chatgpt_bridge_v2_server.py"
if target == BRIDGE_TARGET else "/aios_task_principal_issuer_server.py"
)
for record in existing["source_files"]:
if str(record["path"]).endswith(suffix):
return dict(record)
_error("EXISTING_MODULE_IDENTITY_MISSING")
def _old_bridge(existing: Mapping[str, Any]) -> dict[str, Any]:
for record in existing["source_files"]:
if str(record["path"]).endswith("/aios_chatgpt_bridge_v2_server.py"):
return dict(record)
_error("EXISTING_BRIDGE_IDENTITY_MISSING")
return _old_module(existing, BRIDGE_TARGET)
def _authority_inputs(authority: Path, support_commit: str) -> tuple[list[ArtifactFile], str, dict[str, Any]]:
@@ -354,57 +386,116 @@ def _verify_overlays(overlay_app: Path, repo: Path, support_commit: str) -> list
def _rewrite_c5_imports(data: bytes, name: str) -> bytes:
text = data.decode("utf-8")
old = "from aios_single_host_authority import ("
new = "from c5_source_authority import ("
if text.count(old) != 1:
old = "from aios_single_host_authority import"
new = "from c5_source_authority import"
if text.count(old) < 1:
_error(f"C5_IMPORT_ANCHOR_INVALID:{name}")
return text.replace(old, new, 1).encode("utf-8")
return text.replace(old, new).encode("utf-8")
def _dropin(root: str, pythonpath: Iterable[str], principal: str | None) -> bytes:
def _dropin(
root: str, pythonpath: Iterable[str], principal: str | None,
*, target: str = BRIDGE_TARGET,
) -> bytes:
path = ":".join((f"{root}/app", *pythonpath))
value = (
"[Service]\n"
f"Environment=PYTHONPATH={path}\n"
f"Environment=AIOS_SOURCE_PUBLICATION_ACTIVATION_FILE={root}/source-activation.json\n"
)
value = "[Service]\n" f"Environment=PYTHONPATH={path}\n"
if target == BRIDGE_TARGET:
value += f"Environment=AIOS_SOURCE_PUBLICATION_ACTIVATION_FILE={root}/source-activation.json\n"
elif target != ISSUER_TARGET:
_error("INSTALL_TARGET_INVALID")
if principal is not None:
value += f"Environment=AIOS_C5_PRINCIPAL_SCOPE_CONFIG={principal}\n"
return value.encode("utf-8")
def _captured_rollback(
snapshot: Mapping[str, Any], existing: Mapping[str, Any], target: str,
) -> dict[str, Any] | None:
"""Return the captured pre-activation readiness projection when present.
The live capture contains richer profile metadata around ``existing``;
only the installer readiness object is carried into the request. Keep it
byte-for-byte as captured after checking the fields that bind it to the
selected existing snapshot.
"""
value = snapshot.get("rollback_readiness")
if not isinstance(value, Mapping):
if "existing" in snapshot:
_error("ROLLBACK_READINESS_MISSING")
return None
required = {
"url", "timeout_seconds", "unit_active", "pythonpath", "activation_file",
"principal_scope_config", "module_path", "module_sha256", "runtime_fields",
}
if set(value) != required or not isinstance(value["runtime_fields"], Mapping):
_error("ROLLBACK_READINESS_INVALID")
profile_url = RUNTIME_URL if target == BRIDGE_TARGET else ISSUER_RUNTIME_URL
if (
value["url"] != profile_url or value["timeout_seconds"] != DEFAULT_TIMEOUT
or value["unit_active"] is not True
or value["pythonpath"] != existing["pythonpath"]
or value["activation_file"] != existing["activation_file"]
or value["principal_scope_config"] != existing["principal_scope_config"]
):
_error("ROLLBACK_READINESS_BINDING_INVALID")
old = _old_module(existing, target)
if value["module_path"] != old["path"] or value["module_sha256"] != old["sha256"]:
_error("ROLLBACK_READINESS_BINDING_INVALID")
return dict(value)
def _request(
*, source_commit: str, support_commit: str, files: list[ArtifactFile], existing: Mapping[str, Any],
root: str, activation: Mapping[str, Any], owner_signer_id: str, contract_id: str,
nonce: str, ttl_seconds: int, issued_at: int, principal_path: str | None,
target: str = BRIDGE_TARGET, rollback_readiness: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
bridge = next(file for file in files if file.path == f"{root}/app/aios_chatgpt_bridge_v2_server.py")
old_bridge = _old_bridge(existing)
module_name = (
"aios_chatgpt_bridge_v2_server.py"
if target == BRIDGE_TARGET else "aios_task_principal_issuer_server.py"
)
module = next((file for file in files if file.path == f"{root}/app/{module_name}"), None)
if module is None:
_error("TARGET_MODULE_MISSING")
old_module = _old_module(existing, target)
new_pythonpath = [f"{root}/app", *existing["pythonpath"]]
new_module = f"{root}/app/aios_chatgpt_bridge_v2_server.py"
mandatory = {
"status": "active", "health": 200,
"deployment_state": "STAGING", "repository": REPOSITORY,
}
new_module = f"{root}/app/{module_name}"
new_activation = f"{root}/source-activation.json" if target == BRIDGE_TARGET else None
new_principal = principal_path
if target == BRIDGE_TARGET:
mandatory = {"status": "active", "health": 200, "deployment_state": "STAGING", "repository": REPOSITORY}
runtime_url = RUNTIME_URL
elif target == ISSUER_TARGET:
mandatory = {"ready": True}
if principal_path is not None:
mandatory.update(c5_scope_enabled=True)
runtime_url = ISSUER_RUNTIME_URL
else:
_error("INSTALL_TARGET_INVALID")
readiness_fields = {
**mandatory, "source_module_path": new_module,
"source_module_sha256": _sha(bridge.data),
"source_module_sha256": _sha(module.data),
}
old_fields = dict(mandatory)
old_fields = {"status": "active", "health": 200, "deployment_state": "STAGING", "repository": REPOSITORY} if target == BRIDGE_TARGET else {"ready": True}
if target == ISSUER_TARGET and principal_path is not None:
readiness_fields.update(c5_scope_role="issuer", c5_scope_enrolled=False)
readiness = {
"url": RUNTIME_URL, "timeout_seconds": DEFAULT_TIMEOUT, "unit_active": True,
"pythonpath": new_pythonpath, "activation_file": f"{root}/source-activation.json",
"module_path": new_module, "module_sha256": _sha(bridge.data),
"url": runtime_url, "timeout_seconds": DEFAULT_TIMEOUT, "unit_active": True,
"pythonpath": new_pythonpath, "activation_file": new_activation,
"principal_scope_config": new_principal,
"module_path": new_module, "module_sha256": _sha(module.data),
"runtime_fields": readiness_fields,
}
rollback = {
"url": RUNTIME_URL, "timeout_seconds": DEFAULT_TIMEOUT, "unit_active": True,
"pythonpath": list(existing["pythonpath"]), "activation_file": None,
"module_path": old_bridge["path"], "module_sha256": old_bridge["sha256"],
rollback = dict(rollback_readiness) if rollback_readiness is not None else {
"url": runtime_url, "timeout_seconds": DEFAULT_TIMEOUT, "unit_active": True,
"pythonpath": list(existing["pythonpath"]), "activation_file": existing["activation_file"],
"principal_scope_config": existing["principal_scope_config"],
"module_path": old_module["path"], "module_sha256": old_module["sha256"],
"runtime_fields": old_fields,
}
return {
"schema": REQUEST_SCHEMA,
"target": target,
"contract_id": contract_id, "owner_signer_id": owner_signer_id,
"source_commit": source_commit, "support_commit": support_commit,
"files": [file.record for file in sorted(files, key=lambda item: item.path)],
@@ -418,11 +509,14 @@ def build_artifact(
snapshot: Path, output: Path, *, owner_signer_id: str = "local-owner-authority",
contract_id: str | None = None, nonce: str | None = None, ttl_seconds: int = MAX_TTL,
issued_at: int | None = None, principal_scope_config: Path | None = None,
target: str = BRIDGE_TARGET,
) -> dict[str, Any]:
"""Build the artifact and unsigned request, returning their paths and hashes."""
repo, overlay_app, authority_inputs, snapshot, output = map(Path, (repo, overlay_app, authority_inputs, snapshot, output))
if not repo.is_dir() or repo.is_symlink():
_error("REPOSITORY_INVALID")
if target not in {BRIDGE_TARGET, ISSUER_TARGET}:
_error("INSTALL_TARGET_INVALID")
support_commit = _support_oid(repo, support_commit)
if not _ID.fullmatch(owner_signer_id):
_error("OWNER_SIGNER_ID_INVALID")
@@ -436,9 +530,18 @@ def build_artifact(
_error("ISSUED_AT_INVALID")
if principal_scope_config is not None and (principal_scope_config.is_symlink() or not principal_scope_config.is_file()):
_error("PRINCIPAL_CONFIG_INVALID")
existing, snapshot_value = _read_existing(snapshot)
existing, snapshot_value = _read_existing(snapshot, target)
authority_files, source_commit, activation = _authority_inputs(authority_inputs, support_commit)
overlay_files = _verify_overlays(overlay_app, repo, support_commit)
rewritten_overlays: list[ArtifactFile] = []
for item in overlay_files:
if item.path == "app/c5_governed_principal_scope.py":
rewritten_overlays.append(
ArtifactFile(item.path, _rewrite_c5_imports(item.data, item.path), item.mode, item.kind)
)
else:
rewritten_overlays.append(item)
overlay_files = rewritten_overlays
root = _root(support_commit)
c5_files: list[ArtifactFile] = []
for source_path, destination in sorted(SOURCE_MODULES.items()):
@@ -460,8 +563,11 @@ def build_artifact(
if any(item.path == principal_path for item in files):
_error("PRINCIPAL_CONFIG_DUPLICATE")
files.append(ArtifactFile(principal_path, data, 0o644))
dropin_data = _dropin(root, existing["pythonpath"], principal_path)
files.append(ArtifactFile(DROPIN, dropin_data, 0o644, "systemd_dropin"))
if target == ISSUER_TARGET and principal_path is None:
_error("PRINCIPAL_CONFIG_REQUIRED")
dropin_path = DROPIN if target == BRIDGE_TARGET else ISSUER_DROPIN
dropin_data = _dropin(root, existing["pythonpath"], principal_path, target=target)
files.append(ArtifactFile(dropin_path, dropin_data, 0o644, "systemd_dropin"))
names = [item.path for item in files]
if len(names) != len(set(names)):
_error("ARTIFACT_DUPLICATE_PATH")
@@ -474,25 +580,29 @@ def build_artifact(
existing=existing, root=root, activation=activation,
owner_signer_id=owner_signer_id, contract_id=contract_id,
nonce=nonce, ttl_seconds=ttl_seconds, issued_at=issued_at,
principal_path=principal_path,
principal_path=principal_path, target=target,
rollback_readiness=_captured_rollback(snapshot_value, existing, target),
)
manifest = {
"schema": "aios.c5-staging-artifact-manifest.v1",
"repository": REPOSITORY, "source_commit": source_commit,
"support_commit": support_commit, "release_root": root,
"unit": UNIT, "listen": {"host": "127.0.0.1", "port": PORT},
"target": target,
"unit": UNIT if target == BRIDGE_TARGET else ISSUER_UNIT,
"listen": {"host": "127.0.0.1", "port": PORT if target == BRIDGE_TARGET else ISSUER_PORT},
"files": [item.record for item in sorted(files, key=lambda item: item.path)],
}
output.mkdir(parents=True, exist_ok=True)
artifact_path = output / f"c5-source-{support_commit}.tar"
request_path = output / "installation-request.json"
suffix = "" if target == BRIDGE_TARGET else "-principal-issuer"
artifact_path = output / f"c5-source-{support_commit}{suffix}.tar"
request_path = output / ("installation-request.json" if target == BRIDGE_TARGET else "principal-issuer-installation-request.json")
manifest_path = output / "artifact-manifest.json"
sums_path = output / "SHA256SUMS"
if any(path.exists() for path in (artifact_path, request_path, manifest_path, sums_path)):
_error("OUTPUT_ALREADY_EXISTS")
with tarfile.open(artifact_path, "w", format=tarfile.PAX_FORMAT) as archive:
for item in sorted(files, key=lambda value: value.path):
member_name = "dropin.conf" if item.path == DROPIN else item.path.removeprefix(root + "/")
member_name = "dropin.conf" if item.path == dropin_path else item.path.removeprefix(root + "/")
info = tarfile.TarInfo(member_name)
info.type = tarfile.REGTYPE
info.mode = item.mode
@@ -533,6 +643,7 @@ def _parser() -> argparse.ArgumentParser:
parser.add_argument("--ttl-seconds", type=int, default=MAX_TTL)
parser.add_argument("--issued-at", type=int)
parser.add_argument("--principal-scope-config", type=Path)
parser.add_argument("--target", choices=(BRIDGE_TARGET, ISSUER_TARGET), default=BRIDGE_TARGET)
return parser
@@ -544,7 +655,7 @@ def main(argv: list[str] | None = None) -> int:
args.snapshot, args.output, owner_signer_id=args.owner_signer_id,
contract_id=args.contract_id, nonce=args.nonce,
ttl_seconds=args.ttl_seconds, issued_at=args.issued_at,
principal_scope_config=args.principal_scope_config,
principal_scope_config=args.principal_scope_config, target=args.target,
)
except (BuildError, OSError, ValueError) as exc:
print(json.dumps({"status": "REJECTED", "error": str(exc)}, sort_keys=True))
tests/test_c5_staging_artifact.py
+25 -1
View File
@@ -6,6 +6,7 @@ import json
from pathlib import Path
import shutil
import subprocess
import sys
import tarfile
import pytest
@@ -61,6 +62,8 @@ def _fixture(tmp_path: Path) -> tuple[Path, str, Path, Path, Path]:
old_module = "/opt/aios/current/app/aios_chatgpt_bridge_v2_server.py"
snapshot_payload = {
"unit_active": True,
"activation_file": None,
"principal_scope_config": None,
"effective_config_sha256": "1" * 64,
"dropin": {"exists": True, "sha256": "2" * 64},
"pythonpath": ["/opt/aios/current", "/opt/aios/current/app"],
@@ -84,7 +87,8 @@ def test_builds_regular_only_archive_and_typed_request(tmp_path: Path) -> None:
request = json.loads(Path(result["request"]).read_text())
assert result["status"] == "BUILT_UNSIGNED"
assert result["signed"] is False and result["installed"] is False
assert request["schema"] == "aios.c5-staging-installation-request.v1"
assert request["schema"] == "aios.c5-staging-installation-request.v2"
assert request["target"] == "bridge-18892"
paths = {record["path"] for record in request["files"]}
assert any(path.endswith("/app/c5_source_authority.py") for path in paths)
assert any(path.endswith("/app/aios_task_admission_live_adapters.py") for path in paths)
@@ -101,3 +105,23 @@ def test_rejects_bytecode_in_overlay(tmp_path: Path) -> None:
(overlay / "bad.pyc").write_bytes(b"bytecode")
with pytest.raises(BuildError, match="OVERLAY_BYTECODE_FORBIDDEN"):
build_artifact(repo, support, overlay, authority, snapshot, tmp_path / "out")
def test_generated_request_and_archive_use_installer_producer_validators(tmp_path: Path) -> None:
"""Exercise the real typed request and artifact producer checks when crypto is available."""
repo, support, overlay, authority, snapshot = _fixture(tmp_path)
result = build_artifact(
repo, support, overlay, authority, snapshot, tmp_path / "out",
nonce="c5-validator-nonce-20260926", issued_at=1000, ttl_seconds=900,
)
sys.path.insert(0, str((repo / "active-runtime" / "app").resolve()))
c5 = pytest.importorskip("c5_staging_installation")
request = c5.load_installation_request(Path(result["request"]), require_root_owner=False)
layout = c5._Layout(test_only=True)
payload = c5._request_payload(
request, owner_key_id="SHA256:ephemeral-validator", issued_at=1000, layout=layout,
)
c5._validate_payload(payload, 1000, layout=layout)
c5._verified_artifact(
Path(result["artifact"]), payload, layout=layout, require_root_owner=False,
)