1087 lines
39 KiB
Python
1087 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
from contextlib import contextmanager
|
|
from dataclasses import replace
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
BROKER_DIR = REPO_ROOT / "ops" / "gitea-deploy-broker"
|
|
EFFECTS_PATH = BROKER_DIR / "repopulse_adoption_effects.py"
|
|
REPOSITORY = "jens/repopulse"
|
|
APPLICATION_ID = "repopulse"
|
|
REVISION = "a" * 40
|
|
TREE = "b" * 40
|
|
CONTEXT = "Managed validation / full (push)"
|
|
DATA_SOURCE = "/srv/example/appdata/repopulse/data"
|
|
REPOPULSE_SCAN_SOURCE = "/mnt/cache/appdata/gitea-deploy-runners/sources/repopulse"
|
|
PRIVATEMUSE_SCAN_SOURCE = "/mnt/cache/appdata/gitea-deploy-runners/sources/privatemuse"
|
|
|
|
|
|
def _load_module(name: str, path: Path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _sha(value: str) -> str:
|
|
return hashlib.sha256(value.encode("ascii")).hexdigest()
|
|
|
|
|
|
def _canonical_sha(value: object) -> str:
|
|
payload = json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=True,
|
|
allow_nan=False,
|
|
).encode("ascii")
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def _managed_project(
|
|
enrollment_id: str = "enr-0123456789abcdef0123",
|
|
) -> dict[str, object]:
|
|
return {
|
|
"branch": "main",
|
|
"slug": "repopulse",
|
|
"source_dir": REPOPULSE_SCAN_SOURCE,
|
|
"exact_source": True,
|
|
"env_source": "/srv/example/appdata/repopulse/deploy.managed.env",
|
|
"compose_files": [".itworx/deployment.yml"],
|
|
"compose_project": "repopulse",
|
|
"containers": ["repopulse"],
|
|
"managed_plan_version": "itworx.deployment/v1",
|
|
"managed_enrollment_id": enrollment_id,
|
|
"managed_profile": "repopulse-authenticated",
|
|
"managed_application_id": "repopulse",
|
|
"managed_appdata_path": "/srv/example/appdata/repopulse",
|
|
"managed_image_repository": "repopulse-managed",
|
|
"managed_image_tag": "production",
|
|
"managed_proxy_network": "itworx-mcp-hub-npm",
|
|
"managed_internal_port": 8000,
|
|
"managed_proxy_hostname": "repopulse.example.com",
|
|
"managed_oidc_application_slug": None,
|
|
"managed_runtime_environment_keys": [
|
|
"REPOPULSE_GITEA_URL",
|
|
"REPOPULSE_GITEA_TOKEN",
|
|
"REPOPULSE_MAINTENANCE_TOKEN",
|
|
],
|
|
"managed_runtime_uid": 10001,
|
|
"managed_runtime_gid": 10001,
|
|
"managed_health_path": "/health",
|
|
"appops_deployment_secret_file": "jens__repopulse",
|
|
"environment": {},
|
|
"appops_managed_target": True,
|
|
"managed_repository": "jens/repopulse",
|
|
"managed_build_context": ".",
|
|
"managed_build_dockerfile": "deploy/managed/Dockerfile",
|
|
"gitea_repository_id": 64,
|
|
"gitea_webhook_secret_file": "jens__repopulse",
|
|
"health_timeout": 180,
|
|
"validation": {
|
|
"validation_gate_version": 1,
|
|
"required": True,
|
|
"contexts": [CONTEXT],
|
|
"timeout_seconds": 1800,
|
|
"poll_interval_seconds": 5,
|
|
"max_poll_interval_seconds": 30,
|
|
},
|
|
"verify_oci_revision": False,
|
|
}
|
|
|
|
|
|
def _deployment_receipt(revision: str = REVISION) -> dict[str, object]:
|
|
return {
|
|
"schema": "gitea-deploy-receipt.v1",
|
|
"repository": REPOSITORY,
|
|
"revision": revision,
|
|
"tree": TREE,
|
|
"treeDigest": _sha(f"git-tree:{TREE}"),
|
|
"build": {
|
|
"id": revision[:12],
|
|
"repository": REPOSITORY,
|
|
"revision": revision,
|
|
"tree": TREE,
|
|
"treeDigest": _sha(f"git-tree:{TREE}"),
|
|
"builtAt": "2026-09-23T10:00:00.000Z",
|
|
},
|
|
"containers": {
|
|
"repopulse": {
|
|
"container_id": "c" * 64,
|
|
"image_id": "sha256:" + "d" * 64,
|
|
"image_ref": "repopulse-managed:production",
|
|
"container_revision": revision,
|
|
"image_revision": revision,
|
|
"image_source": REPOSITORY,
|
|
"image_tree": TREE,
|
|
"image_version": revision[:12],
|
|
"image_created": "2026-09-23T10:00:00.000Z",
|
|
}
|
|
},
|
|
"live": None,
|
|
}
|
|
|
|
|
|
def _registration_receipt() -> dict[str, object]:
|
|
return {
|
|
"apiVersion": "itworx.repopulse-controller-registration-receipt/v1",
|
|
"intentId": "rpa-0123456789abcdef0123",
|
|
"repository": REPOSITORY,
|
|
"applicationId": APPLICATION_ID,
|
|
"phase": "registry_committed",
|
|
"intentBindingSha256": _sha("binding"),
|
|
"legacyVolumeRetained": True,
|
|
"intentDurable": True,
|
|
"managedEnrollmentId": "enr-0123456789abcdef0123",
|
|
"projectsSha256": _sha("projects"),
|
|
"snapshotFileSha256": _sha("snapshot"),
|
|
"receiptSha256": _sha("receipt"),
|
|
}
|
|
|
|
|
|
class FakeControllerStoreModule:
|
|
PROJECT_LOCK = Path("/controller/state/locks/repopulse.lock")
|
|
GLOBAL_DEPLOY_LOCK = Path("/controller/state/deploy.lock")
|
|
|
|
def __init__(self, rw_root: Path):
|
|
self.CONTROLLER_RW_ROOT = rw_root
|
|
self.PROJECTS_FILENAME = "projects.json"
|
|
self.SECRET_ROOT_NAME = "appops-deployment-secrets"
|
|
self.locks_held = True
|
|
self.lock_checks: list[Path] = []
|
|
self.managed_target = _managed_project()
|
|
self.capabilities = {
|
|
"repository": REPOSITORY,
|
|
"executionEnabled": False,
|
|
"controllerAliasesReady": True,
|
|
"registryReady": True,
|
|
"secretRootReady": True,
|
|
}
|
|
self.projects = {REPOSITORY: dict(self.managed_target)}
|
|
self.registration_receipt: dict[str, object] | None = None
|
|
|
|
def transaction_lock_held(self, path: Path) -> bool:
|
|
self.lock_checks.append(path)
|
|
return self.locks_held and path in {self.PROJECT_LOCK, self.GLOBAL_DEPLOY_LOCK}
|
|
|
|
def _managed_target(self, enrollment_id: str) -> dict[str, object]:
|
|
return _managed_project(enrollment_id)
|
|
|
|
def capability_report(self) -> dict[str, object]:
|
|
return dict(self.capabilities)
|
|
|
|
def _check_controller_aliases(self):
|
|
return (
|
|
b"",
|
|
{key: dict(value) for key, value in self.projects.items()},
|
|
b"secret",
|
|
)
|
|
|
|
|
|
class FakeStore:
|
|
def __init__(self, controller_store: FakeControllerStoreModule):
|
|
self.controller_store = controller_store
|
|
self.registration_receipt: dict[str, object] | None = None
|
|
|
|
@contextmanager
|
|
def transaction(self):
|
|
self.controller_store.locks_held = True
|
|
yield self
|
|
|
|
def load_receipt(self):
|
|
return self.registration_receipt
|
|
|
|
def load_intent(self):
|
|
return {"intentId": "rpa-0123456789abcdef0123"}
|
|
|
|
|
|
class FakeStorageEffects:
|
|
def __init__(self):
|
|
self.enrollment_id = "enr-0123456789abcdef0123"
|
|
self.credential = b"C" * 64
|
|
self.registration_material_calls: list[str] = []
|
|
|
|
def registration_material(self, *, intent_id: str):
|
|
self.registration_material_calls.append(intent_id)
|
|
return self.enrollment_id, self.credential
|
|
|
|
|
|
class FakeStatusServer:
|
|
def __init__(self):
|
|
self.bootstrap_secret: bytes | None = (
|
|
b"bootstrap-secret-material-that-is-never-returned"
|
|
)
|
|
self.target_secret: bytes | None = (
|
|
b"target-secret-material-that-is-never-returned"
|
|
)
|
|
self.tls_ready = True
|
|
self.tls_configuration = {"ready": True}
|
|
self.validation_status = ("pass", [{"context": CONTEXT, "status": "success"}])
|
|
self.validation_calls: list[tuple[str, str, list[dict[str, object]]]] = []
|
|
|
|
def _enrollment_bootstrap_secret(self):
|
|
return self.bootstrap_secret
|
|
|
|
def _private_tls_configuration(self, *, required: bool):
|
|
assert required is True
|
|
return self.tls_configuration if self.tls_ready else None
|
|
|
|
def _private_tls_context(self, _configuration):
|
|
if not self.tls_ready:
|
|
raise RuntimeError("unavailable")
|
|
return object()
|
|
|
|
def _appops_policy(self, repository: str, project: dict[str, object]):
|
|
if (
|
|
repository == REPOSITORY
|
|
and project.get("managed_repository") == REPOSITORY
|
|
and project.get("managed_application_id") == APPLICATION_ID
|
|
and project.get("appops_deployment_secret_file") == "jens__repopulse"
|
|
):
|
|
return APPLICATION_ID, "jens__repopulse"
|
|
return None
|
|
|
|
def _appops_deployment_secret(self, repository: str, project: dict[str, object]):
|
|
return self.target_secret if self._appops_policy(repository, project) else None
|
|
|
|
def _validation_policy(self, repository: str, project: dict[str, object]):
|
|
if repository != REPOSITORY or not isinstance(project.get("validation"), dict):
|
|
return None
|
|
validation = dict(project["validation"])
|
|
return {
|
|
"repository": repository,
|
|
"contexts": list(validation.get("contexts", [])),
|
|
}
|
|
|
|
def _fetch_gitea_validation(self, repository: str, revision: str, policy):
|
|
self.validation_calls.append((repository, revision, list(policy["contexts"])))
|
|
return self.validation_status
|
|
|
|
|
|
class FakeDeployCore:
|
|
def __init__(self, project: dict[str, object] | None = None):
|
|
self.REPOSITORY_ROOT = Path("/gitea-repositories")
|
|
self.EXACT_SOURCE_ROOT = Path("/mnt/cache/appdata/gitea-deploy-runners/sources")
|
|
self.MANAGED_APPCONFIG_ROOT = Path("/srv/example/appdata")
|
|
self.STATE_ROOT = Path("/controller/state")
|
|
self.project = dict(project or _managed_project())
|
|
self.main_revision = REVISION
|
|
self.receipt = _deployment_receipt()
|
|
self.container = self._container()
|
|
self.container_ids = ["c" * 64]
|
|
self.commands: list[list[str]] = []
|
|
self.events: list[tuple[str, object]] = []
|
|
self.runtime_meta = {
|
|
"health": "ok",
|
|
"name": "RepoPulse",
|
|
"queue": {"queued": 0, "active": 0},
|
|
"databaseRecoveryStatus": "healthy",
|
|
}
|
|
|
|
def _container(self) -> dict[str, object]:
|
|
return {
|
|
"Id": "c" * 64,
|
|
"Image": "sha256:" + "d" * 64,
|
|
"State": {"Running": True, "Health": {"Status": "healthy"}},
|
|
"Config": {
|
|
"Image": "repopulse-managed:production",
|
|
"Labels": {
|
|
"com.docker.compose.project": "repopulse",
|
|
"com.docker.compose.service": "repopulse",
|
|
"org.opencontainers.image.revision": REVISION,
|
|
"org.opencontainers.image.source": REPOSITORY,
|
|
},
|
|
},
|
|
"HostConfig": {"PortBindings": {}},
|
|
"Mounts": [
|
|
{
|
|
"Type": "bind",
|
|
"Source": DATA_SOURCE,
|
|
"Destination": "/data",
|
|
"RW": True,
|
|
},
|
|
{
|
|
"Type": "bind",
|
|
"Source": REPOPULSE_SCAN_SOURCE,
|
|
"Destination": "/scan/repopulse",
|
|
"RW": False,
|
|
},
|
|
{
|
|
"Type": "bind",
|
|
"Source": PRIVATEMUSE_SCAN_SOURCE,
|
|
"Destination": "/scan/privatemuse",
|
|
"RW": False,
|
|
},
|
|
],
|
|
}
|
|
|
|
def output(self, command: list[str]) -> str:
|
|
self.commands.append(list(command))
|
|
if command[:2] == [
|
|
"git",
|
|
f"--git-dir={self.REPOSITORY_ROOT / 'jens' / 'repopulse.git'}",
|
|
]:
|
|
if command[-1] == "refs/heads/main":
|
|
return self.main_revision
|
|
if command[-1] == f"{self.main_revision}^{{commit}}":
|
|
return self.main_revision
|
|
if command[:3] == ["docker", "ps", "--all"]:
|
|
return "\n".join(self.container_ids)
|
|
if command[:3] == ["docker", "volume", "inspect"]:
|
|
return json.dumps([{"Name": "repopulse_data"}])
|
|
if command[:3] == ["docker", "exec", "repopulse"]:
|
|
return json.dumps(self.runtime_meta)
|
|
raise AssertionError(f"unexpected fixed controller command: {command!r}")
|
|
|
|
def run(self, command: list[str]) -> None:
|
|
self.events.append(("run", list(command)))
|
|
if command[:2] == ["docker", "stop"]:
|
|
self.container["State"]["Running"] = False
|
|
|
|
def load_projects(self):
|
|
return {REPOSITORY: dict(self.project)}
|
|
|
|
def deployment_metadata(self, repo: Path, revision: str, repository: str):
|
|
assert repository == REPOSITORY
|
|
assert revision == self.main_revision
|
|
return {"repository": repository, "revision": revision, "tree": TREE}
|
|
|
|
def repository_path(self, repository: str, _project):
|
|
assert repository == REPOSITORY
|
|
return self.REPOSITORY_ROOT / "jens" / "repopulse.git"
|
|
|
|
def validate_project_config(self, repo: Path, project, revision: str):
|
|
assert repo == self.REPOSITORY_ROOT / "jens" / "repopulse.git"
|
|
assert project["managed_repository"] == REPOSITORY
|
|
assert revision == self.main_revision
|
|
|
|
def exact_source_matches_revision(self, _repo: Path, revision: str, _target: Path):
|
|
return revision == self.main_revision
|
|
|
|
def validate_managed_v1_runtime(self, project, revision: str, repository: str):
|
|
assert project["managed_profile"] == "repopulse-authenticated"
|
|
assert revision == self.main_revision
|
|
assert repository == REPOSITORY
|
|
|
|
def managed_runtime_environment(self, _project):
|
|
return {
|
|
"REPOPULSE_GITEA_URL": "https://gitea.example.com",
|
|
"REPOPULSE_GITEA_TOKEN": "private-token",
|
|
"REPOPULSE_MAINTENANCE_TOKEN": "private-maintenance-token",
|
|
}
|
|
|
|
def deploy_repopulse_under_adoption_transaction(self, revision: str):
|
|
self.events.append(("deploy", revision))
|
|
assert revision == self.main_revision
|
|
return dict(self.receipt)
|
|
|
|
def verified_current_receipt(self, repository, project, revision, tree):
|
|
self.events.append(("verified_current_receipt", revision))
|
|
if (
|
|
repository == REPOSITORY
|
|
and project.get("managed_profile") == "repopulse-authenticated"
|
|
and revision == self.receipt["revision"]
|
|
and tree == self.receipt["tree"]
|
|
):
|
|
return dict(self.receipt)
|
|
return None
|
|
|
|
def read_state(self, _project):
|
|
return {"live_receipt": dict(self.receipt)}
|
|
|
|
def inspect_container(self, name: str):
|
|
if name not in {"repopulse", "c" * 64}:
|
|
raise AssertionError("adapter inspected a non-fixed container")
|
|
return (
|
|
self.container
|
|
if name == "repopulse" or name == self.container["Id"]
|
|
else None
|
|
)
|
|
|
|
|
|
class EffectsFixture:
|
|
def __init__(self, tmp_path: Path):
|
|
controller_root = tmp_path / "controller"
|
|
controller_root.mkdir(mode=0o700)
|
|
projects_file = controller_root / "projects.json"
|
|
projects_file.write_text("{}\n", encoding="utf-8")
|
|
projects_file.chmod(0o600)
|
|
(controller_root / "appops-deployment-secrets").mkdir(mode=0o700)
|
|
self.controller_store = FakeControllerStoreModule(controller_root)
|
|
self.store = FakeStore(self.controller_store)
|
|
self.storage_effects = FakeStorageEffects()
|
|
self.status = FakeStatusServer()
|
|
self.deploy = FakeDeployCore(project=self.controller_store.managed_target)
|
|
|
|
self.effects_module = _load_module(
|
|
"repopulse_adoption_effects_test", EFFECTS_PATH
|
|
)
|
|
self.effects = self.effects_module.FixedControllerAdoptionEffects(
|
|
store=self.store,
|
|
storage_effects=self.storage_effects,
|
|
controller_store=self.controller_store,
|
|
deployment_core=self.deploy,
|
|
status_server=self.status,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def effects_env(tmp_path):
|
|
return EffectsFixture(tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime_readable_scan_root():
|
|
# The pytest tmp_path parent is mode 0700 on Linux, so runtime UID 10001
|
|
# cannot traverse it even when the leaf scan directories are mode 0755.
|
|
# A private TMPDIR must not override the public Linux parent needed by UID 10001.
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="repopulse-scan-", dir="/tmp" if os.name == "posix" else None,
|
|
) as directory:
|
|
root = Path(directory)
|
|
root.chmod(0o755)
|
|
yield root
|
|
|
|
|
|
def test_target_gates_report_only_proved_controller_facts_and_keep_unavailable_external_facts_false(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
assert gates["profileId"] == "repopulse-authenticated"
|
|
assert gates["profileVersion"] == "enrollment-profile/v1"
|
|
assert re.fullmatch(r"[0-9a-f]{64}", gates["planSha256"])
|
|
assert gates["profileReady"] is True
|
|
assert gates["exactPolicyReady"] is True
|
|
assert gates["controllerBootstrapSecretImported"] is True
|
|
assert gates["targetSecretReady"] is True
|
|
assert gates["controllerAdoptionCapabilityReady"] is True
|
|
assert gates["controllerRegistryWriterReady"] is True
|
|
assert gates["controllerSecretWriterReady"] is True
|
|
assert gates["controllerTrustReady"] is False
|
|
assert gates["typedEnrollmentTrustReady"] is False
|
|
assert gates["authenticatedRouteReady"] is False
|
|
assert gates["accessListReady"] is False
|
|
assert gates["maintenanceAuthReady"] is False
|
|
assert "authenticatedRouteReady" in env.effects.target_gate_gaps
|
|
assert "accessListReady" in env.effects.target_gate_gaps
|
|
assert any(
|
|
"certificate" in gap.lower() for gap in env.effects.target_gate_gap_details
|
|
)
|
|
assert any(
|
|
"access-list" in gap.lower() and "rule" in gap.lower()
|
|
for gap in env.effects.target_gate_gap_details
|
|
)
|
|
|
|
|
|
def test_target_secret_gate_uses_validated_cli_credential_before_registration(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
env.status.target_secret = None
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
assert gates["targetSecretReady"] is True
|
|
assert env.storage_effects.registration_material_calls == [
|
|
"rpa-0123456789abcdef0123"
|
|
]
|
|
|
|
|
|
def test_registration_reuses_the_same_preflighted_target_credential(effects_env):
|
|
env = effects_env
|
|
intent_id = "rpa-0123456789abcdef0123"
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
assert gates["targetSecretReady"] is True
|
|
enrollment_id, credential = env.effects.registration_material(
|
|
intent_id=intent_id
|
|
)
|
|
|
|
assert enrollment_id == env.storage_effects.enrollment_id
|
|
assert credential == env.storage_effects.credential
|
|
|
|
env.storage_effects.credential = b"D" * 64
|
|
with (
|
|
env.store.transaction(),
|
|
pytest.raises(env.effects_module.ControllerEffectsError),
|
|
):
|
|
env.effects.registration_material(intent_id=intent_id)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"credential",
|
|
[b"short", b"C" * 47, b"C" * 257, b"C" * 63 + b"\x20", b"C" * 63 + b"\x7f"],
|
|
)
|
|
def test_target_secret_gate_rejects_credentials_outside_store_registration_contract(
|
|
effects_env, credential
|
|
):
|
|
env = effects_env
|
|
env.storage_effects.credential = credential
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
assert gates["targetSecretReady"] is False
|
|
assert "targetSecretReady" in env.effects.target_gate_gaps
|
|
|
|
|
|
class FakeTargetAttestationVerifier:
|
|
def __init__(self, attestation):
|
|
self.attestation = attestation
|
|
self.expected: dict[str, str] | None = None
|
|
|
|
def verify_target_attestation(self, *, expected):
|
|
self.expected = dict(expected)
|
|
return self.attestation
|
|
|
|
|
|
def _valid_attestation(env, certificate_sha256: str):
|
|
effects = env.effects
|
|
target = effects._target_template()
|
|
target.pop("managed_enrollment_id", None)
|
|
plan_sha256 = _canonical_sha(
|
|
{
|
|
"profileId": "repopulse-authenticated",
|
|
"profileVersion": "enrollment-profile/v1",
|
|
"target": target,
|
|
}
|
|
)
|
|
now = datetime.now(timezone.utc).replace(microsecond=0)
|
|
rule_type = env.effects_module.NpmAccessRule
|
|
rules = (rule_type("authenticate", "basic_user_sha256", _sha("npm-auth-username")),)
|
|
rules_payload = [
|
|
{
|
|
"action": rule.action,
|
|
"subjectType": rule.subject_type,
|
|
"subject": rule.subject,
|
|
}
|
|
for rule in rules
|
|
]
|
|
def timestamp(value):
|
|
return value.isoformat().replace("+00:00", "Z")
|
|
return env.effects_module.VerifiedTargetAttestation(
|
|
schema="itworx.repopulse-target-attestation/v1",
|
|
attestation_id="attest-0123456789",
|
|
intent_id="rpa-0123456789abcdef0123",
|
|
nonce="a" * 32,
|
|
issued_at=timestamp(now),
|
|
expires_at=timestamp(now + timedelta(minutes=2)),
|
|
controller_identity="appops-controller-prod-01",
|
|
controller_trust_profile_id="appops-controller-trust",
|
|
controller_trust_profile_version="v1",
|
|
controller_bootstrap_key_ref="repopulse-enrollment-hmac",
|
|
controller_tls_certificate_sha256=certificate_sha256,
|
|
repository=REPOSITORY,
|
|
application_id=APPLICATION_ID,
|
|
profile_id="repopulse-authenticated",
|
|
profile_version="enrollment-profile/v1",
|
|
plan_sha256=plan_sha256,
|
|
target_secret_ref="jens__repopulse",
|
|
npm_hostname="repopulse.example.com",
|
|
npm_access_list_route_hostname="repopulse.example.com",
|
|
npm_upstream_host="repopulse",
|
|
npm_upstream_network="itworx-mcp-hub-npm",
|
|
npm_upstream_port=8000,
|
|
npm_certificate_sha256=_sha("npm-leaf-certificate"),
|
|
npm_certificate_sans=("repopulse.example.com",),
|
|
npm_certificate_not_before=timestamp(now - timedelta(days=5)),
|
|
npm_certificate_not_after=timestamp(now + timedelta(days=60)),
|
|
npm_access_list_id=1,
|
|
npm_access_list_satisfy_any=0,
|
|
npm_access_list_is_deleted=0,
|
|
npm_access_list_pass_auth=1,
|
|
npm_access_list_client_count=0,
|
|
npm_access_list_rule_count=len(rules),
|
|
npm_access_list_rules_sha256=_canonical_sha(rules_payload),
|
|
npm_access_list_rules=rules,
|
|
maintenance_method="GET",
|
|
maintenance_path="/api/v1/operator/maintenance",
|
|
maintenance_status_code=200,
|
|
maintenance_auth_mechanism="operator_token",
|
|
maintenance_authenticated_principal="operator:jens",
|
|
maintenance_response_sha256=_sha("maintenance-response"),
|
|
maintenance_observed_at=timestamp(now),
|
|
)
|
|
|
|
|
|
def test_target_gates_accept_only_verified_immutable_concrete_external_facts(
|
|
effects_env, tmp_path
|
|
):
|
|
env = effects_env
|
|
certificate_file = tmp_path / "controller.pem"
|
|
certificate_file.write_text(
|
|
"-----BEGIN CERTIFICATE-----\nYWJjZA==\n-----END CERTIFICATE-----\n",
|
|
encoding="ascii",
|
|
)
|
|
certificate_file.chmod(0o600)
|
|
env.status.tls_configuration = {"certificate_file": certificate_file}
|
|
local_certificate_sha256 = hashlib.sha256(b"abcd").hexdigest()
|
|
attestation = _valid_attestation(env, local_certificate_sha256)
|
|
verifier = FakeTargetAttestationVerifier(attestation)
|
|
env.effects._external_attestation_verifier = verifier
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
external_gates = (
|
|
"controllerTrustReady",
|
|
"typedEnrollmentTrustReady",
|
|
"authenticatedRouteReady",
|
|
"accessListReady",
|
|
"maintenanceAuthReady",
|
|
)
|
|
assert all(gates[name] is True for name in external_gates)
|
|
assert verifier.expected == {
|
|
"repository": REPOSITORY,
|
|
"applicationId": APPLICATION_ID,
|
|
"profileId": "repopulse-authenticated",
|
|
"profileVersion": "enrollment-profile/v1",
|
|
"planSha256": attestation.plan_sha256,
|
|
"intentId": attestation.intent_id,
|
|
"targetSecretRef": "jens__repopulse",
|
|
"controllerTlsCertificateSha256": local_certificate_sha256,
|
|
}
|
|
assert not any(name in env.effects.target_gate_gaps for name in external_gates)
|
|
with pytest.raises(AttributeError):
|
|
attestation.npm_hostname = "attacker.example"
|
|
|
|
|
|
def test_target_gates_ignore_untyped_boolean_external_attestation(
|
|
effects_env, tmp_path
|
|
):
|
|
env = effects_env
|
|
certificate_file = tmp_path / "controller.pem"
|
|
certificate_file.write_text(
|
|
"-----BEGIN CERTIFICATE-----\nYWJjZA==\n-----END CERTIFICATE-----\n",
|
|
encoding="ascii",
|
|
)
|
|
certificate_file.chmod(0o600)
|
|
env.status.tls_configuration = {"certificate_file": certificate_file}
|
|
env.effects._external_attestation_verifier = FakeTargetAttestationVerifier(
|
|
{
|
|
name: True
|
|
for name in (
|
|
"controllerTrustReady",
|
|
"typedEnrollmentTrustReady",
|
|
"authenticatedRouteReady",
|
|
"accessListReady",
|
|
"maintenanceAuthReady",
|
|
)
|
|
}
|
|
)
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
assert gates["controllerTrustReady"] is False
|
|
assert gates["typedEnrollmentTrustReady"] is False
|
|
assert gates["authenticatedRouteReady"] is False
|
|
assert gates["accessListReady"] is False
|
|
assert gates["maintenanceAuthReady"] is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "gate"),
|
|
[
|
|
("intent_id", "rpa-ffffffffffffffffffff", "typedEnrollmentTrustReady"),
|
|
("plan_sha256", _sha("different-plan"), "typedEnrollmentTrustReady"),
|
|
(
|
|
"controller_tls_certificate_sha256",
|
|
_sha("different-controller-cert"),
|
|
"controllerTrustReady",
|
|
),
|
|
("npm_certificate_sans", ("wrong.example",), "authenticatedRouteReady"),
|
|
(
|
|
"npm_certificate_not_after",
|
|
"2020-01-01T00:00:00Z",
|
|
"authenticatedRouteReady",
|
|
),
|
|
("npm_access_list_id", 2, "accessListReady"),
|
|
("npm_access_list_satisfy_any", 1, "accessListReady"),
|
|
("npm_access_list_is_deleted", 1, "accessListReady"),
|
|
("npm_access_list_pass_auth", 0, "accessListReady"),
|
|
("npm_access_list_client_count", 1, "accessListReady"),
|
|
("npm_access_list_rules_sha256", _sha("different-rules"), "accessListReady"),
|
|
("maintenance_observed_at", "2020-01-01T00:00:00Z", "maintenanceAuthReady"),
|
|
("maintenance_authenticated_principal", "", "maintenanceAuthReady"),
|
|
],
|
|
)
|
|
def test_target_gates_reject_attestations_with_unbound_concrete_facts(
|
|
effects_env, tmp_path, field, value, gate
|
|
):
|
|
env = effects_env
|
|
certificate_file = tmp_path / "controller.pem"
|
|
certificate_file.write_text(
|
|
"-----BEGIN CERTIFICATE-----\nYWJjZA==\n-----END CERTIFICATE-----\n",
|
|
encoding="ascii",
|
|
)
|
|
certificate_file.chmod(0o600)
|
|
env.status.tls_configuration = {"certificate_file": certificate_file}
|
|
local_certificate_sha256 = hashlib.sha256(b"abcd").hexdigest()
|
|
attestation = replace(
|
|
_valid_attestation(env, local_certificate_sha256), **{field: value}
|
|
)
|
|
env.effects._external_attestation_verifier = FakeTargetAttestationVerifier(
|
|
attestation
|
|
)
|
|
|
|
with env.store.transaction():
|
|
gates = env.effects.target_gates()
|
|
|
|
assert gates[gate] is False
|
|
assert gate in env.effects.target_gate_gaps
|
|
|
|
|
|
def test_exact_sha_ci_accepts_only_current_main_with_the_single_push_context(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
|
|
with env.store.transaction():
|
|
evidence = env.effects.validate_exact_sha_ci_locked(
|
|
repository=REPOSITORY, revision=REVISION
|
|
)
|
|
|
|
assert evidence == {
|
|
"repository": REPOSITORY,
|
|
"revision": REVISION,
|
|
"contexts": {CONTEXT: "success"},
|
|
}
|
|
assert env.status.validation_calls == [(REPOSITORY, REVISION, [CONTEXT])]
|
|
git_dir = f"--git-dir={env.deploy.REPOSITORY_ROOT / 'jens' / 'repopulse.git'}"
|
|
assert env.deploy.commands == [
|
|
["git", git_dir, "rev-parse", "refs/heads/main"],
|
|
["git", git_dir, "rev-parse", f"{REVISION}^{{commit}}"],
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"repository",
|
|
"revision",
|
|
"main_revision",
|
|
"validation_status",
|
|
"validation_contexts",
|
|
),
|
|
[
|
|
(
|
|
"jens/other",
|
|
REVISION,
|
|
REVISION,
|
|
("pass", [{"context": CONTEXT, "status": "success"}]),
|
|
[CONTEXT],
|
|
),
|
|
(
|
|
REPOSITORY,
|
|
"a" * 39,
|
|
REVISION,
|
|
("pass", [{"context": CONTEXT, "status": "success"}]),
|
|
[CONTEXT],
|
|
),
|
|
(
|
|
REPOSITORY,
|
|
REVISION,
|
|
"f" * 40,
|
|
("pass", [{"context": CONTEXT, "status": "success"}]),
|
|
[CONTEXT],
|
|
),
|
|
(
|
|
REPOSITORY,
|
|
REVISION,
|
|
REVISION,
|
|
("pass", [{"context": CONTEXT, "status": "success"}]),
|
|
[CONTEXT, "Other (push)"],
|
|
),
|
|
(
|
|
REPOSITORY,
|
|
REVISION,
|
|
REVISION,
|
|
("wait", [{"context": CONTEXT, "status": "pending"}]),
|
|
[CONTEXT],
|
|
),
|
|
(
|
|
REPOSITORY,
|
|
REVISION,
|
|
REVISION,
|
|
("pass", [{"context": "Other (push)", "status": "success"}]),
|
|
[CONTEXT],
|
|
),
|
|
],
|
|
)
|
|
def test_exact_sha_ci_rejects_wrong_identity_head_or_validation_context(
|
|
effects_env,
|
|
repository,
|
|
revision,
|
|
main_revision,
|
|
validation_status,
|
|
validation_contexts,
|
|
):
|
|
env = effects_env
|
|
env.deploy.main_revision = main_revision
|
|
env.status.validation_status = validation_status
|
|
env.controller_store.managed_target["validation"]["contexts"] = list(
|
|
validation_contexts
|
|
)
|
|
env.controller_store.projects[REPOSITORY] = dict(
|
|
env.controller_store.managed_target
|
|
)
|
|
|
|
with (
|
|
env.store.transaction(),
|
|
pytest.raises(env.effects_module.ControllerEffectsError),
|
|
):
|
|
env.effects.validate_exact_sha_ci_locked(
|
|
repository=repository, revision=revision
|
|
)
|
|
|
|
|
|
def test_deploy_calls_the_existing_locked_core_and_returns_a_receipt_bound_to_live_runtime(
|
|
effects_env, tmp_path, runtime_readable_scan_root, monkeypatch
|
|
):
|
|
env = effects_env
|
|
env.store.registration_receipt = _registration_receipt()
|
|
env.controller_store.registration_receipt = env.store.registration_receipt
|
|
env.deploy.project = _managed_project()
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
|
|
legacy_env = tmp_path / "deploy.env"
|
|
legacy_env.write_text("PRIVATE=preserved\n", encoding="utf-8")
|
|
legacy_env.chmod(0o600)
|
|
managed_env = tmp_path / "deploy.managed.env"
|
|
managed_env.write_text(
|
|
"REPOPULSE_GITEA_URL=https://gitea.example.com\n"
|
|
"REPOPULSE_GITEA_TOKEN=private-token\n"
|
|
"REPOPULSE_MAINTENANCE_TOKEN=private-maintenance-token\n",
|
|
encoding="utf-8",
|
|
)
|
|
managed_env.chmod(0o600)
|
|
monkeypatch.setattr(env.effects_module, "LEGACY_ENVIRONMENT_FILE", legacy_env)
|
|
monkeypatch.setattr(env.effects_module, "MANAGED_ENVIRONMENT_FILE", managed_env)
|
|
monkeypatch.setattr(env.effects_module, "MANAGED_DATA_PATH", Path(DATA_SOURCE))
|
|
repopulse_scan = runtime_readable_scan_root / "repopulse"
|
|
privatemuse_scan = runtime_readable_scan_root / "privatemuse"
|
|
repopulse_scan.mkdir(parents=True)
|
|
privatemuse_scan.mkdir(parents=True)
|
|
repopulse_scan.chmod(0o755)
|
|
privatemuse_scan.chmod(0o755)
|
|
monkeypatch.setattr(env.effects_module, "REPOPULSE_SCAN_SOURCE", repopulse_scan)
|
|
monkeypatch.setattr(env.effects_module, "PRIVATEMUSE_SCAN_SOURCE", privatemuse_scan)
|
|
env.deploy.container["Mounts"][1]["Source"] = str(repopulse_scan)
|
|
env.deploy.container["Mounts"][2]["Source"] = str(privatemuse_scan)
|
|
|
|
snapshot = {
|
|
"intentId": "rpa-0123456789abcdef0123",
|
|
"snapshotFileSha256": _sha("snapshot"),
|
|
"snapshotLogicalSha256": _sha("logical"),
|
|
"snapshotQueueSha256": _sha("queue"),
|
|
"sourceAuxiliaryTreeSha256": _sha("aux"),
|
|
"sourceAuxiliaryFileCount": 2,
|
|
"snapshotScanCount": 14,
|
|
"snapshotCompactedScanCount": 4,
|
|
}
|
|
copied = {
|
|
"intentId": "rpa-0123456789abcdef0123",
|
|
"snapshotFileSha256": snapshot["snapshotFileSha256"],
|
|
"targetLogicalSha256": snapshot["snapshotLogicalSha256"],
|
|
"targetScanCount": 14,
|
|
"targetCompactedScanCount": 4,
|
|
"targetQueueSha256": snapshot["snapshotQueueSha256"],
|
|
"sourceAuxiliaryTreeSha256": snapshot["sourceAuxiliaryTreeSha256"],
|
|
"targetAuxiliaryTreeSha256": snapshot["sourceAuxiliaryTreeSha256"],
|
|
"sourceAuxiliaryFileCount": 2,
|
|
"targetAuxiliaryFileCount": 2,
|
|
"targetQuickCheck": "ok",
|
|
"targetIntegrityCheck": "ok",
|
|
"targetDataUid": 10001,
|
|
"targetDataGid": 10001,
|
|
"targetRecoveryStatus": "healthy",
|
|
"managedDataMountMatches": True,
|
|
"copySourceIsVerifiedSnapshot": True,
|
|
"legacyVolumeRetained": True,
|
|
"legacyEnvironmentRetained": True,
|
|
"managedEnvironmentFilePrivate": True,
|
|
}
|
|
|
|
with env.store.transaction():
|
|
proof = env.effects.deploy_exact_sha_locked(
|
|
intent_id="rpa-0123456789abcdef0123",
|
|
revision=REVISION,
|
|
controller_registry_receipt=env.store.registration_receipt,
|
|
snapshot_evidence=snapshot,
|
|
copied_data_evidence=copied,
|
|
)
|
|
|
|
assert env.deploy.events[0] == ("deploy", REVISION)
|
|
assert any(name == "verified_current_receipt" for name, _value in env.deploy.events)
|
|
assert proof["repository"] == REPOSITORY
|
|
assert proof["revision"] == REVISION
|
|
assert proof["containerRevision"] == REVISION
|
|
assert proof["deploymentReceiptVerified"] is True
|
|
assert proof["deploymentReceiptSha256"] == _canonical_sha(env.deploy.receipt)
|
|
assert proof["runtimeHealthy"] is True
|
|
assert proof["publishedHostPorts"] == []
|
|
assert proof["managedDataMountMatches"] is True
|
|
assert proof["targetQueuedJobs"] == proof["targetActiveJobs"] == 0
|
|
assert proof["legacyVolumeRetained"] is True
|
|
assert "private-token" not in json.dumps(proof)
|
|
|
|
|
|
def test_deploy_rejects_a_receipt_whose_runtime_container_has_a_different_revision(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
env.store.registration_receipt = {"intentId": "rpa-0123456789abcdef0123"}
|
|
env.controller_store.registration_receipt = env.store.registration_receipt
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
env.deploy.container["Config"]["Labels"]["org.opencontainers.image.revision"] = (
|
|
"f" * 40
|
|
)
|
|
|
|
with (
|
|
env.store.transaction(),
|
|
pytest.raises(env.effects_module.ControllerEffectsError),
|
|
):
|
|
env.effects.deploy_exact_sha_locked(
|
|
intent_id="rpa-0123456789abcdef0123",
|
|
revision=REVISION,
|
|
controller_registry_receipt=env.store.registration_receipt,
|
|
snapshot_evidence={},
|
|
copied_data_evidence={},
|
|
)
|
|
|
|
|
|
def test_stop_managed_writer_stops_only_the_fixed_matching_managed_container_and_proves_volume_retention(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
env.deploy.project = _managed_project()
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
|
|
with env.store.transaction():
|
|
proof = env.effects.stop_managed_writer(
|
|
intent_id="rpa-0123456789abcdef0123", revision=REVISION
|
|
)
|
|
|
|
assert proof["managedWriterStopped"] is True
|
|
assert proof["activeWriterCount"] == 0
|
|
assert proof["legacyVolumeRetained"] is True
|
|
assert ("run", ["docker", "stop", "c" * 64]) in env.deploy.events
|
|
|
|
|
|
def test_stop_managed_writer_refuses_to_stop_a_different_revision(effects_env):
|
|
env = effects_env
|
|
env.deploy.project = _managed_project()
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
env.deploy.container["Config"]["Labels"]["org.opencontainers.image.revision"] = (
|
|
"f" * 40
|
|
)
|
|
|
|
with (
|
|
env.store.transaction(),
|
|
pytest.raises(env.effects_module.ControllerEffectsError),
|
|
):
|
|
env.effects.stop_managed_writer(
|
|
intent_id="rpa-0123456789abcdef0123", revision=REVISION
|
|
)
|
|
|
|
assert not any(event[0] == "run" for event in env.deploy.events)
|
|
|
|
|
|
def test_live_completion_revalidates_receipt_container_mount_health_and_exact_identity(
|
|
effects_env, tmp_path, monkeypatch
|
|
):
|
|
env = effects_env
|
|
env.deploy.project = _managed_project()
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
env.store.registration_receipt = _registration_receipt()
|
|
env.controller_store.registration_receipt = env.store.registration_receipt
|
|
legacy_env = tmp_path / "deploy.env"
|
|
legacy_env.write_text("PRIVATE=preserved\n", encoding="utf-8")
|
|
legacy_env.chmod(0o600)
|
|
managed_env = tmp_path / "deploy.managed.env"
|
|
managed_env.write_text("REPOPULSE_GITEA_TOKEN=private-token\n", encoding="utf-8")
|
|
managed_env.chmod(0o600)
|
|
monkeypatch.setattr(env.effects_module, "LEGACY_ENVIRONMENT_FILE", legacy_env)
|
|
monkeypatch.setattr(env.effects_module, "MANAGED_ENVIRONMENT_FILE", managed_env)
|
|
receipt_sha = _canonical_sha(env.deploy.receipt)
|
|
completion = {
|
|
"mainRevision": REVISION,
|
|
"deploymentReceiptSha256": receipt_sha,
|
|
"controllerRegistrationReceiptSha256": env.store.registration_receipt[
|
|
"receiptSha256"
|
|
],
|
|
"containerIdSha256": _sha("c" * 64),
|
|
"containerImageSha256": "d" * 64,
|
|
"legacyVolume": "repopulse_data",
|
|
"legacyVolumeRetained": True,
|
|
}
|
|
|
|
with env.store.transaction():
|
|
proof = env.effects.verify_live_completion_locked(completion=completion)
|
|
|
|
assert proof == {
|
|
"repository": REPOSITORY,
|
|
"revision": REVISION,
|
|
"containerRevision": REVISION,
|
|
"deploymentReceiptSha256": receipt_sha,
|
|
"containerIdSha256": _sha("c" * 64),
|
|
"containerImageSha256": "d" * 64,
|
|
"runtimeHealthy": True,
|
|
"legacyVolume": "repopulse_data",
|
|
"legacyVolumeRetained": True,
|
|
}
|
|
|
|
|
|
def test_live_completion_rejects_changed_container_id_and_refuses_unlocked_calls(
|
|
effects_env,
|
|
):
|
|
env = effects_env
|
|
env.deploy.project = _managed_project()
|
|
env.controller_store.projects = {REPOSITORY: dict(env.deploy.project)}
|
|
env.store.registration_receipt = _registration_receipt()
|
|
env.controller_store.registration_receipt = env.store.registration_receipt
|
|
completion = {
|
|
"mainRevision": REVISION,
|
|
"deploymentReceiptSha256": _canonical_sha(env.deploy.receipt),
|
|
"controllerRegistrationReceiptSha256": env.store.registration_receipt[
|
|
"receiptSha256"
|
|
],
|
|
"containerIdSha256": _sha("other-container"),
|
|
"containerImageSha256": "d" * 64,
|
|
"legacyVolume": "repopulse_data",
|
|
"legacyVolumeRetained": True,
|
|
}
|
|
|
|
with (
|
|
env.store.transaction(),
|
|
pytest.raises(env.effects_module.ControllerEffectsError),
|
|
):
|
|
env.effects.verify_live_completion_locked(completion=completion)
|
|
|
|
env.controller_store.locks_held = False
|
|
with pytest.raises(env.effects_module.ControllerEffectsError):
|
|
env.effects.verify_live_completion_locked(completion=completion)
|
|
assert env.controller_store.lock_checks[-2:] == [
|
|
env.controller_store.PROJECT_LOCK,
|
|
env.controller_store.GLOBAL_DEPLOY_LOCK,
|
|
]
|