diff --git a/scripts/build_c5_staging_artifact.py b/scripts/build_c5_staging_artifact.py index 4453dbf..4ab8fe9 100644 --- a/scripts/build_c5_staging_artifact.py +++ b/scripts/build_c5_staging_artifact.py @@ -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)) diff --git a/tests/test_c5_staging_artifact.py b/tests/test_c5_staging_artifact.py index a017bbb..6719173 100644 --- a/tests/test_c5_staging_artifact.py +++ b/tests/test_c5_staging_artifact.py @@ -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, + )