diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index 539af3d..bc04c9f 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `658804e835d4710ad27ace5ab7250c6f7cc0a0e3`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `4f159a7e60d5becc5218ef4c541e2005d733a537`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index c20a022..fe19ee7 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "658804e835d4710ad27ace5ab7250c6f7cc0a0e3", + "sourceRevision": "4f159a7e60d5becc5218ef4c541e2005d733a537", "files": [ { "path": ".dockerignore", @@ -524,8 +524,8 @@ }, { "path": "backend/app/services/deployment_target_registry.py", - "sha256": "322208e3793306d72d47685ee1567a568a64078abe7547de3f2d0721ec888851", - "bytes": 20168 + "sha256": "2266540e1820997df473e9d65922b07171becbf0c300f527e939718f8317b4a6", + "bytes": 23483 }, { "path": "backend/app/services/discovery_scheduler.py", @@ -854,8 +854,8 @@ }, { "path": "backend/tests/test_controller_artifact_bootstrap.py", - "sha256": "fcb29fae7778e00a3c84e6bcdb51a20ffaea73c24b5a8bd4304058825e30d359", - "bytes": 42896 + "sha256": "e198bfef1b404005472f56e7c399055b665e0004317bbc890bc7ba73f7e74fa9", + "bytes": 44333 }, { "path": "backend/tests/test_controller_artifact_promoter.py", @@ -869,8 +869,8 @@ }, { "path": "backend/tests/test_controller_config_paths.py", - "sha256": "1081ecf7206e59be24aebf92284d3833743b1a84eeb1cefc3891a340dea883ae", - "bytes": 8254 + "sha256": "5c20d8b533b1e3aa78078ca8a520600fa530d1ad0e134aa5242a6ed344055f0e", + "bytes": 8738 }, { "path": "backend/tests/test_controller_host_dependencies.py", @@ -924,8 +924,8 @@ }, { "path": "backend/tests/test_deployment_target_registry.py", - "sha256": "2c28d05cbf6156d4ab53e2b52a770e2b0bda34138a3d715ba5d97bca0b9e8ce7", - "bytes": 9429 + "sha256": "a4582c70b0adb23437481c9f639e5bd94fb2314495777d1a936f2a1f5c28acef", + "bytes": 9502 }, { "path": "backend/tests/test_dockdeck_classifier.py", @@ -1097,6 +1097,11 @@ "sha256": "cf6043f3cc775d1cdc68dfdddc36d1a460c6dbc6f84ffea607246daa572509c3", "bytes": 26469 }, + { + "path": "backend/tests/test_mediarr_validation_policy.py", + "sha256": "f2dc9c8bec831fd30cfe1affe4994d4d2fcaf35840f7c20cc95d3576a5b03657", + "bytes": 22513 + }, { "path": "backend/tests/test_mediarr_validation_window.py", "sha256": "9a3f8e768eaf4b77e3687fa1a4af2f970cc70def50ddbdf4e693fbd63e4c5acd", diff --git a/backend/app/services/deployment_target_registry.py b/backend/app/services/deployment_target_registry.py index cc1ceb1..3428afd 100644 --- a/backend/app/services/deployment_target_registry.py +++ b/backend/app/services/deployment_target_registry.py @@ -21,12 +21,11 @@ _MAX_CONTAINERS = 32 # Fixed application-owned mutation. Callers never provide a command or host # path. Repository-derived topology is validated before it reaches this script, # then re-bound to the exact immutable Git revision inside the controller. -_REGISTER_SCRIPT = r''' +_REGISTER_SCRIPT = r""" import fcntl import json import os from pathlib import Path -import shutil import stat import subprocess import sys @@ -38,11 +37,35 @@ lock_path = Path("/mnt/cache/appdata/gitea-deploy-runners/deploy.lock") repository, revision, branch, raw_project = sys.argv[1:5] project = json.loads(raw_project) -if path.is_symlink() or not path.is_file(): - raise SystemExit(20) -info = path.stat() -if not stat.S_ISREG(info.st_mode): - raise SystemExit(21) +def _validation_compatible(repository, validation): + expected_validation = { + "validation_gate_version": 1, + "required": True, + "contexts": ["Managed validation / full (push)"], + "timeout_seconds": 3600 if repository == "jens/mediarr" else 1800, + "poll_interval_seconds": 5, + "max_poll_interval_seconds": 30, + } + if not isinstance(validation, dict) or validation.get("required") is not True or any( + type(validation.get(key)) is not int + for key in ("validation_gate_version", "timeout_seconds", "poll_interval_seconds", "max_poll_interval_seconds") + ): + return False + if validation == expected_validation: + return True + return repository == "jens/mediarr" and validation == { + **expected_validation, + "contexts": ["CI / gates (push)", "Managed validation / full (push)"], + } + +for directory in (path.parent, host_path.parent, host_path.parent.parent): + info = directory.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o022: + raise SystemExit(21) +for registry_path in (path, host_path): + info = registry_path.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o600: + raise SystemExit(21) try: if host_path.is_symlink() or not host_path.is_file() or not os.path.samefile(path, host_path): @@ -117,7 +140,18 @@ with lock_handle as lock: raise SystemExit(30) try: - projects = json.loads(mutation_path.read_text(encoding="utf-8")) + registry_info = mutation_path.lstat() + registry_identity = (registry_info.st_dev, registry_info.st_ino, registry_info.st_mode, + registry_info.st_uid, registry_info.st_gid, registry_info.st_mtime_ns, + registry_info.st_ctime_ns, registry_info.st_size) + if not stat.S_ISREG(registry_info.st_mode) or registry_info.st_uid != 0 or stat.S_IMODE(registry_info.st_mode) != 0o600: + raise SystemExit(31) + descriptor = os.open(mutation_path, os.O_RDONLY | os.O_NOFOLLOW) + with os.fdopen(descriptor, "rb") as stream: + original = stream.read(4 * 1024 * 1024 + 1) + if len(original) > 4 * 1024 * 1024: + raise SystemExit(31) + projects = json.loads(original) except Exception: raise SystemExit(31) if not isinstance(projects, dict): @@ -185,23 +219,32 @@ with lock_handle as lock: raise SystemExit(23) validation = current.get("validation") if validation is not None: - expected_validation = { - "validation_gate_version": 1, - "required": True, - "contexts": ["Managed validation / full (push)"], - "timeout_seconds": 3600 if repository == "jens/mediarr" else 1800, - "poll_interval_seconds": 5, - "max_poll_interval_seconds": 30, - } - if validation != expected_validation: + if not _validation_compatible(repository, validation): raise SystemExit(23) if changed: - backup = mutation_path.with_name( - f"projects.json.appops-backup-{int(time.time())}-{os.getpid()}" - ) try: - shutil.copy2(mutation_path, backup) + backup_root = Path("/mnt/cache/appdata/gitea-deploy-runners/state/registry-backups") + for directory in (backup_root.parent, backup_root.parent.parent): + directory_info = directory.lstat() + if not stat.S_ISDIR(directory_info.st_mode) or directory_info.st_uid != 0 or directory_info.st_mode & 0o022: + raise SystemExit(32) + backup_root.mkdir(mode=0o700, exist_ok=True) + directory_info = backup_root.lstat() + if not stat.S_ISDIR(directory_info.st_mode) or directory_info.st_uid != 0 or stat.S_IMODE(directory_info.st_mode) != 0o700: + raise SystemExit(32) + backup = backup_root / f"projects.json.appops-backup-{time.time_ns()}-{os.getpid()}" + fd = os.open(backup, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, "wb") as handle: + handle.write(original) + handle.flush() + os.fsync(handle.fileno()) + for directory in (backup_root.parent, backup_root): + fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + os.fsync(fd) + finally: + os.close(fd) except Exception: raise SystemExit(32) @@ -210,11 +253,11 @@ with lock_handle as lock: temporary = mutation_path.with_name(f".{mutation_path.name}.{os.getpid()}.tmp") payload = json.dumps(updated, indent=2, sort_keys=True) + "\n" try: - with temporary.open("w", encoding="utf-8") as handle: + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: handle.write(payload) handle.flush() os.fsync(handle.fileno()) - os.chmod(temporary, stat.S_IMODE(info.st_mode)) except Exception: try: temporary.unlink(missing_ok=True) @@ -223,6 +266,11 @@ with lock_handle as lock: raise SystemExit(33) try: + latest = mutation_path.lstat() + latest_identity = (latest.st_dev, latest.st_ino, latest.st_mode, latest.st_uid, + latest.st_gid, latest.st_mtime_ns, latest.st_ctime_ns, latest.st_size) + if latest_identity != registry_identity or mutation_path.read_bytes() != original or not os.path.samefile(path, mutation_path): + raise SystemExit(34) os.replace(temporary, mutation_path) except Exception: try: @@ -242,7 +290,7 @@ with lock_handle as lock: os.close(directory_fd) print("registered" if changed else "already-registered", flush=True) -''' +""" def _safe_compose_file(value: str) -> str: @@ -314,10 +362,14 @@ class DeploymentTargetRegistry: normalized_revision = str(revision or "").strip().lower() normalized_branch = str(branch or "").strip() if not _REPOSITORY.fullmatch(normalized_repository): - raise ValueError("Repository must be a canonical Jens owner/name identifier") + raise ValueError( + "Repository must be a canonical Jens owner/name identifier" + ) slug = DeploymentTargetRegistry._slug(normalized_repository) if not _SLUG.fullmatch(slug): - raise ValueError("Repository name cannot be used as a bounded deployment slug") + raise ValueError( + "Repository name cannot be used as a bounded deployment slug" + ) if not _REVISION.fullmatch(normalized_revision): raise ValueError("Revision must be a full lowercase 40-character Git SHA") if ( @@ -328,15 +380,25 @@ class DeploymentTargetRegistry: ): raise ValueError("Default branch is outside the bounded Git ref contract") normalized_compose_file = _safe_compose_file(compose_file) - normalized_project = str( - compose_project - or DeploymentTargetRegistry.default_compose_project(normalized_repository) - ).strip().lower() + normalized_project = ( + str( + compose_project + or DeploymentTargetRegistry.default_compose_project( + normalized_repository + ) + ) + .strip() + .lower() + ) if not _COMPOSE_PROJECT.fullmatch(normalized_project): raise ValueError("Compose project is outside the bounded naming contract") - normalized_containers = [str(item or "").strip() for item in (containers or [slug])] + normalized_containers = [ + str(item or "").strip() for item in (containers or [slug]) + ] if not normalized_containers or len(normalized_containers) > _MAX_CONTAINERS: - raise ValueError("Deployment target must contain between 1 and 32 containers") + raise ValueError( + "Deployment target must contain between 1 and 32 containers" + ) if any(not _CONTAINER.fullmatch(item) for item in normalized_containers): raise ValueError("Container name is outside the bounded naming contract") if len(set(normalized_containers)) != len(normalized_containers): @@ -396,7 +458,10 @@ class DeploymentTargetRegistry: ), } controller = dict(status.get("container") or {}) - if controller.get("name") != _CONTROLLER or controller.get("state") != "running": + if ( + controller.get("name") != _CONTROLLER + or controller.get("state") != "running" + ): return { "status": "controller_unavailable", "reason": "deployment controller is not running", diff --git a/backend/tests/test_controller_artifact_bootstrap.py b/backend/tests/test_controller_artifact_bootstrap.py index 313ad57..b1933a2 100644 --- a/backend/tests/test_controller_artifact_bootstrap.py +++ b/backend/tests/test_controller_artifact_bootstrap.py @@ -909,7 +909,22 @@ def test_child_logs_are_bounded_and_redacted_before_result_output(): assert len(result["logTail"].encode("utf-8")) <= 4096 -def test_legacy_upgrade_signature_fetches_bootstrap_only_from_protected_git(tmp_path: Path): +@pytest.mark.parametrize( + "temporary_root_info,accepted", + [ + ("1777:0:0", True), + ("700:0:0", True), + ("777:0:0", False), + ("755:0:0", False), + ("770:0:0", False), + ("1700:0:0", False), + ("1777:1000:0", False), + ("700:0:1000", False), + ], +) +def test_legacy_upgrade_signature_fetches_bootstrap_only_from_protected_git( + tmp_path: Path, temporary_root_info: str, accepted: bool +): bash = shutil.which("bash") if bash is None: pytest.skip("bash is required to exercise the upgrade script") @@ -941,7 +956,7 @@ def test_legacy_upgrade_signature_fetches_bootstrap_only_from_protected_git(tmp_ "import pathlib, sys\n" "args = sys.argv[1:]\n" "target = args[-1]\n" - "if target == '/tmp': print('1777:0:0')\n" + f"if target == '/tmp': print({temporary_root_info!r})\n" "elif args[1] == '%u:%a': print('0:700')\n" "elif args[1] == '%u:%a:%s': print('0:600:' + str(pathlib.Path(target).stat().st_size))\n" "else: raise SystemExit(1)\n", @@ -985,6 +1000,13 @@ def test_legacy_upgrade_signature_fetches_bootstrap_only_from_protected_git(tmp_ check=False, ) + if not accepted: + assert result.returncode == 66, result.stderr + assert "temporary root is unsafe" in result.stderr + assert not python_trace.exists() + calls = [json.loads(line) for line in git_trace.read_text(encoding="utf-8").splitlines()] + assert not any("cat-file" in args for args in calls) + return assert result.returncode == 0, result.stderr bootstrap_call = json.loads(python_trace.read_text(encoding="utf-8")) assert bootstrap_call["script"] == protected_source @@ -1062,12 +1084,23 @@ def test_upgrade_script_rejects_mutated_legacy_call_without_bootstrap( @pytest.mark.parametrize("modern_call", ["missing-image", "shared-archive"]) +@pytest.mark.parametrize("temporary_root_info", ["1777:0:0", "700:0:0", "777:0:0"]) def test_modern_upgrade_path_still_requires_private_archive_and_image_id( - tmp_path: Path, modern_call: str + tmp_path: Path, modern_call: str, temporary_root_info: str ): bash = shutil.which("bash") if bash is None: pytest.skip("bash is required to exercise the upgrade script") + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + stat_shim = bin_dir / "stat" + stat_shim.write_text( + f"#!{sys.executable}\nimport sys\n" + f"if sys.argv[-1] == '/tmp': print({temporary_root_info!r})\n" + "else: raise SystemExit(91)\n", + encoding="utf-8", + ) + stat_shim.chmod(0o755) archive = ( f"/tmp/itworx-controller-artifact-promoter-{REVISION[:12]}.AAAAAA/controller.tar" if modern_call == "missing-image" @@ -1094,13 +1127,17 @@ def test_modern_upgrade_path_still_requires_private_archive_and_image_id( result = subprocess.run( [bash, str(ROOT / "scripts/upgrade_deployment_controller.sh"), *args], cwd=ROOT, + env={**os.environ, "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}"}, capture_output=True, text=True, check=False, ) assert result.returncode != 0 + if modern_call == "shared-archive": + if temporary_root_info == "777:0:0": + assert "helper temporary root is unsafe" in result.stderr + else: + assert "archive is outside the helper private temporary directory" in result.stderr if modern_call == "missing-image": assert "image ID" in result.stderr - else: - assert "archive is outside" in result.stderr diff --git a/backend/tests/test_controller_config_paths.py b/backend/tests/test_controller_config_paths.py index a226c83..1edbf07 100644 --- a/backend/tests/test_controller_config_paths.py +++ b/backend/tests/test_controller_config_paths.py @@ -149,12 +149,16 @@ def test_embedded_registry_writer_uses_protected_aliases_and_preserves_code( repositories = tmp_path / "repositories" bare = repositories / "jens" / "example.git" bare.mkdir(parents=True) + state = tmp_path / "controller-state" + state.mkdir(mode=0o700) + backups = state / "registry-backups" paths = { "/controller-config/projects.json": readonly, "/controller-hub-releases/controller-config/projects.json": registry, "/opt/gitea-deploy/projects.json": old_registry, "/mnt/cache/appdata/gitea-deploy-runners/controller/projects.json": old_host_registry, "/mnt/cache/appdata/gitea-deploy-runners/deploy.lock": tmp_path / "deploy.lock", + "/mnt/cache/appdata/gitea-deploy-runners/state/registry-backups": backups, "/gitea-repositories": repositories, } @@ -175,7 +179,12 @@ def test_embedded_registry_writer_uses_protected_aliases_and_preserves_code( exec(compile(tree, "registry-writer", "exec"), {"Path": fixture_path}) assert json.loads(registry.read_text()) == {"jens/example": project} assert registry.stat().st_mode & 0o777 == 0o600 - assert list(registry.parent.glob("projects.json.appops-backup-*")) + backup_files = list(backups.glob("projects.json.appops-backup-*")) + assert len(backup_files) == 1 + assert backup_files[0].read_bytes() == b"{}\n" + assert backup_files[0].stat().st_mode & 0o777 == 0o600 + assert backups.stat().st_mode & 0o777 == 0o700 + assert not list(registry.parent.glob("projects.json.appops-backup-*")) else: with pytest.raises(SystemExit) as error: exec(compile(tree, "registry-writer", "exec"), {"Path": fixture_path}) diff --git a/backend/tests/test_deployment_target_registry.py b/backend/tests/test_deployment_target_registry.py index 3b7a2a4..7426da7 100644 --- a/backend/tests/test_deployment_target_registry.py +++ b/backend/tests/test_deployment_target_registry.py @@ -88,6 +88,7 @@ def test_derives_bounded_single_container_topology_and_uses_fixed_exec(): assert '"containers":["certscope"]' in command[6] script = command[2] assert '/controller-hub-releases/controller-config/projects.json' in script + assert '/controller-config/projects.json' in script assert '/srv/example/appdata/gitea-deploy-runners/deploy.lock' not in script assert '/mnt/cache/appdata/gitea-deploy-runners/deploy.lock' in script assert 'os.path.samefile(path, host_path)' in script @@ -224,7 +225,7 @@ def test_runnerless_validation_compatibility_is_fixed_push_only(): assert '"timeout_seconds": 3600 if repository == "jens/mediarr" else 1800' in script assert '"poll_interval_seconds": 5' in script assert '"max_poll_interval_seconds": 30' in script - assert "if validation != expected_validation" in script + assert "if not _validation_compatible(repository, validation)" in script def test_runnerless_compatibility_requires_the_complete_metadata_triplet(): fake = FakePortainer() diff --git a/backend/tests/test_mediarr_validation_policy.py b/backend/tests/test_mediarr_validation_policy.py new file mode 100644 index 0000000..2a9af9f --- /dev/null +++ b/backend/tests/test_mediarr_validation_policy.py @@ -0,0 +1,616 @@ +from __future__ import annotations + +import ast +import copy +import importlib.util +import json +import os +import shutil +import stat +import sys +import tempfile +import threading +from pathlib import Path +from typing import Any + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts/reconcile_mediarr_validation_policy.py" +REGISTRY_SOURCE = REPO_ROOT / "backend/app/services/deployment_target_registry.py" +SPEC = importlib.util.spec_from_file_location("mediarr_validation_policy", SCRIPT) +assert SPEC is not None and SPEC.loader is not None +policy = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(policy) + +PUSH_CONTEXTS = ["CI / gates (push)", "Managed validation / full (push)"] +ACTIVE_STATUSES = ( + "pending", + "running", + "awaiting_validation", + "validation_passed", + "retry_wait", + "deploying", + "future_unrecognized_state", +) +TERMINAL_STATUSES = ( + "deployed", + "non_forward", + "failed", + "validation_failed", + "validation_timeout", + "superseded", + "rejected", +) + + +def _projects(validation: dict[str, Any] | None = None) -> dict[str, Any]: + project = copy.deepcopy(policy.TARGET_IDENTITY) + project["validation"] = copy.deepcopy( + policy.LEGACY_POLICY if validation is None else validation + ) + return {policy.REPOSITORY: project} + + +def _extract_registry_compatibility(): + """Load only the inline predicate; importing the registry module has host side effects.""" + registry_tree = ast.parse( + REGISTRY_SOURCE.read_text(encoding="utf-8"), filename=str(REGISTRY_SOURCE) + ) + register_script = next( + ( + node + for node in registry_tree.body + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) and target.id == "_REGISTER_SCRIPT" + for target in node.targets + ) + ), + None, + ) + assert register_script is not None, "registry must keep its fixed registration script inline" + source = ast.literal_eval(register_script.value) + tree = ast.parse(source, filename=f"{REGISTRY_SOURCE}:_REGISTER_SCRIPT") + function = next( + ( + node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == "_validation_compatible" + ), + None, + ) + assert function is not None, "inline registration script must define _validation_compatible" + + # The predicate may use module-level literal policy constants. Extract only literals and the + # function, deliberately excluding imports, filesystem setup, controller state, and CLI code. + namespace: dict[str, Any] = {"__builtins__": __builtins__} + body: list[ast.stmt] = [] + for node in tree.body: + if isinstance(node, ast.ImportFrom) and node.module == "__future__": + body.append(node) + elif isinstance(node, ast.Assign): + try: + ast.literal_eval(node.value) + except (ValueError, TypeError, SyntaxError): + continue + if all(isinstance(target, ast.Name) for target in node.targets): + body.append(node) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + try: + ast.literal_eval(node.value) + except (ValueError, TypeError, SyntaxError, AttributeError): + continue + body.append(node) + body.append(function) + module = ast.fix_missing_locations(ast.Module(body=body, type_ignores=[])) + exec(compile(module, str(REGISTRY_SOURCE), "exec"), namespace) # noqa: S102 + return namespace["_validation_compatible"] + + +def _registry_bytes(projects: dict[str, Any]) -> bytes: + return (json.dumps(projects, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _write_json(path: Path, value: Any, *, mode: int = 0o600) -> bytes: + payload = _registry_bytes(value) if path.name == "projects.json" else ( + json.dumps(value, sort_keys=True) + "\n" + ).encode("utf-8") + path.write_bytes(payload) + if os.name == "posix": + path.chmod(mode) + return payload + + +def _prepare_filesystem( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + *, + validation: dict[str, Any] | None = None, + inbox_records: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Path | bytes]: + runtime_root = tmp_path / "runner-root" + config_root = tmp_path / "protected-config" + state_root = runtime_root / "state" + inbox = state_root / "webhook-inbox" + lock_root = state_root / "locks" + backup_root = state_root / "registry-backups" + registry = config_root / "projects.json" + project_lock = lock_root / "mediarr.lock" + global_lock = runtime_root / "deploy.lock" + receipt_root = state_root / "receipts" + + for directory in ( + runtime_root, + config_root, + state_root, + inbox, + lock_root, + receipt_root, + backup_root, + ): + directory.mkdir(parents=True, exist_ok=True) + if os.name == "posix": + directory.chmod(0o700) + + original = _write_json(registry, _projects(validation)) + for lock in (project_lock, global_lock): + lock.touch(mode=0o600) + if os.name == "posix": + lock.chmod(0o600) + + # A completed delivery is terminal history, not active work. It must remain byte-for-byte. + records = inbox_records + if records is None: + records = { + f"terminal-{status}.json": { + "repository": policy.REPOSITORY, + "status": status, + "receipt": f"fixture-{status}-receipt", + } + for status in TERMINAL_STATUSES + } + records["other-repository.json"] = { + "repository": "jens/another-project", + "status": "running", + "receipt": "fixture-other-receipt", + } + for filename, record in records.items(): + _write_json(inbox / filename, record) + receipt = receipt_root / "last.json" + _write_json(receipt, {"receipt": "fixture-state-receipt", "status": "unchanged"}) + + monkeypatch.setattr(policy, "ROOT", runtime_root, raising=False) + monkeypatch.setattr(policy, "CONFIG_ROOT", config_root, raising=False) + monkeypatch.setattr(policy, "REGISTRY", registry) + monkeypatch.setattr(policy, "BACKUP_ROOT", backup_root, raising=False) + monkeypatch.setattr(policy, "INBOX", inbox) + monkeypatch.setattr(policy, "PROJECT_LOCK", project_lock) + monkeypatch.setattr(policy, "GLOBAL_LOCK", global_lock) + # The production proof shells out to Docker. This suite exercises only the policy + # reconciliation against temporary fixtures and must never contact a live controller. + monkeypatch.setattr(policy, "_verify_volume_contract", lambda: "a" * 64) + return { + "root": runtime_root, + "state": state_root, + "inbox": inbox, + "registry": registry, + "project_lock": project_lock, + "global_lock": global_lock, + "receipt": receipt, + "backup_root": backup_root, + "original": original, + } + + +def _tree_bytes(root: Path) -> dict[str, bytes]: + return { + path.relative_to(root).as_posix(): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() and not path.is_symlink() + } + + +@pytest.fixture +def root_owned_tmp_path(): + if os.name != "posix" or not hasattr(os, "geteuid"): + pytest.skip("filesystem contract uses POSIX ownership, flock and fsync") + if os.geteuid() != 0: + pytest.skip("filesystem contract requires root-owned temporary fixtures") + base = Path(tempfile.mkdtemp(prefix="mediarr-validation-policy-", dir="/root")) + try: + yield base + finally: + shutil.rmtree(base) + + +def test_accepts_exact_legacy_and_required_policies_without_downgrade() -> None: + required = policy._validate(_projects(policy.REQUIRED_POLICY)) + legacy = policy._validate(_projects(policy.LEGACY_POLICY)) + assert required == "required" + assert legacy == "legacy" + + +@pytest.mark.parametrize( + "mutate", + [ + lambda project: project.update(branch="main"), + lambda project: project.update(source_dir="/tmp/mediarr"), + lambda project: project.update(compose_files=["other.yml"]), + lambda project: project.update(validation={"contexts": ["*"]}), + lambda project: project.update( + validation={**policy.REQUIRED_POLICY, "contexts": ["CI / gates (pull_request)"]} + ), + lambda project: project.update( + validation={**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + ["extra"]} + ), + lambda project: project.update( + validation={**policy.REQUIRED_POLICY, "timeout_seconds": 1800} + ), + ], + ids=["branch", "source", "compose", "wildcard", "pull-request", "extra-context", "wrong-timeout"], +) +def test_rejects_unreviewed_target_or_validation_policy(mutate) -> None: + projects = _projects(policy.REQUIRED_POLICY) + mutate(projects[policy.REPOSITORY]) + with pytest.raises(RuntimeError): + policy._validate(projects) + + +def test_registry_inline_validator_accepts_only_mediarr_legacy_or_dual_push_policy() -> None: + compatible = _extract_registry_compatibility() + legacy = copy.deepcopy(policy.LEGACY_POLICY) + required = copy.deepcopy(policy.REQUIRED_POLICY) + other_managed_only = {**legacy, "timeout_seconds": 1800} + + assert compatible(policy.REPOSITORY, legacy) is True + assert compatible(policy.REPOSITORY, required) is True + assert compatible("jens/another-project", other_managed_only) is True + assert compatible("jens/another-project", required) is False + + +@pytest.mark.parametrize( + "validation", + [ + {**policy.REQUIRED_POLICY, "contexts": ["*"]}, + {**policy.REQUIRED_POLICY, "contexts": ["CI / gates (pull_request)"]}, + {**policy.REQUIRED_POLICY, "contexts": []}, + {**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + ["extra"]}, + {**policy.REQUIRED_POLICY, "contexts": PUSH_CONTEXTS + [PUSH_CONTEXTS[0]]}, + {**policy.REQUIRED_POLICY, "required": False}, + None, + ], + ids=["wildcard", "pull-request", "missing", "extra", "duplicate", "not-required", "missing-policy"], +) +def test_registry_inline_validator_rejects_mediarr_policy_drift(validation) -> None: + compatible = _extract_registry_compatibility() + assert compatible(policy.REPOSITORY, validation) is False + + +def test_registry_inline_validator_rejects_non_mediarr_policy_drift() -> None: + compatible = _extract_registry_compatibility() + assert compatible("jens/another-project", {**policy.LEGACY_POLICY, "contexts": ["*"]}) is False + + +@pytest.mark.parametrize("operation", ["check", "apply"]) +@pytest.mark.parametrize("status", ACTIVE_STATUSES) +def test_reconcile_refuses_active_or_unknown_mediarr_delivery_status( + root_owned_tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + status: str, + operation: str, +) -> None: + fs = _prepare_filesystem( + root_owned_tmp_path, + monkeypatch, + validation=policy.REQUIRED_POLICY, + inbox_records={ + "mediarr.json": {"repository": policy.REPOSITORY, "status": status}, + "terminal.json": {"repository": policy.REPOSITORY, "status": "deployed"}, + }, + ) + registry = fs["registry"] + assert isinstance(registry, Path) + before_registry = registry.read_bytes() + before_state = _tree_bytes(fs["state"]) + + with pytest.raises(RuntimeError): + policy.reconcile(operation) + + assert registry.read_bytes() == before_registry + assert _tree_bytes(fs["state"]) == before_state + + +def test_check_accepts_legacy_without_mutation_and_returns_only_reviewed_contexts( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch) + registry = fs["registry"] + assert isinstance(registry, Path) + before = registry.read_bytes() + before_identity = registry.stat() + before_state = _tree_bytes(fs["state"]) + + result = policy.reconcile("check") + + assert result == {"status": "ready", "contexts": PUSH_CONTEXTS} + assert registry.read_bytes() == before + assert (registry.stat().st_dev, registry.stat().st_ino) == ( + before_identity.st_dev, + before_identity.st_ino, + ) + assert _tree_bytes(fs["state"]) == before_state + assert not list(registry.parent.glob("*.mediarr-policy-*")) + + +def test_apply_updates_only_mediarr_and_keeps_private_backup_and_state_history( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch) + registry = fs["registry"] + state = fs["state"] + backup_root = fs["backup_root"] + assert ( + isinstance(registry, Path) + and isinstance(state, Path) + and isinstance(backup_root, Path) + ) + original = registry.read_bytes() + original_projects = json.loads(original) + state_before = _tree_bytes(state) + real_replace = policy.os.replace + temporary_metadata: list[tuple[str, int, int]] = [] + + def inspect_temporary( + source: os.PathLike[str] | str, target: os.PathLike[str] | str + ) -> None: + source_path = Path(source) + info = source_path.stat() + temporary_metadata.append( + (source_path.name, stat.S_IMODE(info.st_mode), info.st_uid) + ) + real_replace(source, target) + + monkeypatch.setattr(policy.os, "replace", inspect_temporary) + + result = policy.reconcile("apply") + + assert set(result) == {"status", "contexts", "backup"} + assert result["status"] == "updated" + assert result["contexts"] == PUSH_CONTEXTS + assert isinstance(result["backup"], str) + assert Path(result["backup"]).name == result["backup"] + backup = backup_root / result["backup"] + assert backup.read_bytes() == original + assert stat.S_IMODE(backup.stat().st_mode) == 0o600 + assert backup.stat().st_uid == 0 + assert len(temporary_metadata) == 1 + assert temporary_metadata[0][1:] == (0o600, 0) + current_projects = json.loads(registry.read_bytes()) + assert current_projects[policy.REPOSITORY]["validation"] == policy.REQUIRED_POLICY + original_projects[policy.REPOSITORY]["validation"] = policy.REQUIRED_POLICY + assert current_projects == original_projects + assert stat.S_IMODE(registry.stat().st_mode) == 0o600 + assert registry.stat().st_uid == 0 + assert sorted(path.name for path in registry.parent.iterdir()) == [registry.name] + assert _tree_bytes(state) == { + **state_before, + f"registry-backups/{backup.name}": original, + } + assert not list(registry.parent.glob(f".{registry.name}.*.tmp")) + + +def test_apply_is_idempotent_for_required_policy_but_still_checks_active_delivery( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem( + root_owned_tmp_path, monkeypatch, validation=policy.REQUIRED_POLICY + ) + registry = fs["registry"] + assert isinstance(registry, Path) + before = registry.read_bytes() + + checked = policy.reconcile("check") + assert checked == {"status": "ready", "contexts": PUSH_CONTEXTS} + assert registry.read_bytes() == before + + result = policy.reconcile("apply") + + assert result == {"status": "ready", "contexts": PUSH_CONTEXTS} + assert registry.read_bytes() == before + assert not list(registry.parent.glob("*.mediarr-policy-*")) + + +def test_non_root_refuses_before_reading_or_mutating_registry( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + registry = tmp_path / "must-not-be-read.json" + registry.write_text("sentinel", encoding="utf-8") + monkeypatch.setattr(policy, "REGISTRY", registry) + monkeypatch.setattr(policy.os, "geteuid", lambda: 1000, raising=False) + + with pytest.raises(RuntimeError): + policy.reconcile("check") + + assert registry.read_text(encoding="utf-8") == "sentinel" + + +def test_registry_symlink_is_refused_without_touching_its_target( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch) + registry = fs["registry"] + assert isinstance(registry, Path) + outside = registry.with_name("outside.json") + outside.write_bytes(registry.read_bytes()) + outside.chmod(0o600) + registry.unlink() + registry.symlink_to(outside) + target_before = outside.read_bytes() + + with pytest.raises(RuntimeError): + policy.reconcile("apply") + + assert outside.read_bytes() == target_before + + +def test_inbox_symlink_is_refused_even_when_registry_is_already_required( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem( + root_owned_tmp_path, + monkeypatch, + validation=policy.REQUIRED_POLICY, + inbox_records={}, + ) + inbox = fs["inbox"] + assert isinstance(inbox, Path) + outside = inbox.with_name("outside-inbox") + outside.mkdir(mode=0o700) + inbox.rmdir() + inbox.symlink_to(outside, target_is_directory=True) + + with pytest.raises(RuntimeError): + policy.reconcile("check") + + +def test_inbox_record_symlink_is_refused_without_touching_its_target( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem( + root_owned_tmp_path, + monkeypatch, + validation=policy.REQUIRED_POLICY, + inbox_records={}, + ) + inbox = fs["inbox"] + assert isinstance(inbox, Path) + outside = inbox.with_name("outside-active-delivery.json") + target_before = _write_json( + outside, + {"repository": policy.REPOSITORY, "status": "running"}, + ) + (inbox / "active-delivery.json").symlink_to(outside) + + with pytest.raises(RuntimeError): + policy.reconcile("check") + + assert outside.read_bytes() == target_before + + +@pytest.mark.parametrize("race", ["content", "identity"]) +def test_apply_refuses_registry_cas_race_before_atomic_replace( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch, race: str +) -> None: + fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch) + registry = fs["registry"] + assert isinstance(registry, Path) + original_fsync = policy.os.fsync + original_replace = policy.os.replace + original_identity = registry.stat() + external = registry.with_name("concurrent-registry.json") + injected = False + + def race_after_durable_temp(fd: int) -> None: + nonlocal injected + original_fsync(fd) + if injected: + return + injected = True + if race == "content": + registry.write_bytes(b'{"external":"concurrent update"}\n') + registry.chmod(0o600) + else: + external.write_bytes(registry.read_bytes()) + external.chmod(0o600) + original_replace(external, registry) + + monkeypatch.setattr(policy.os, "fsync", race_after_durable_temp) + with pytest.raises((RuntimeError, OSError)): + policy.reconcile("apply") + + assert injected + assert ( + json.loads(registry.read_bytes()).get(policy.REPOSITORY, {}).get("validation") + != policy.REQUIRED_POLICY + ) + if race == "identity": + assert (registry.stat().st_dev, registry.stat().st_ino) != ( + original_identity.st_dev, + original_identity.st_ino, + ) + assert not list(registry.parent.glob(f".{registry.name}.*.tmp")) + + +def test_apply_replace_failure_preserves_registry_and_cleans_temporary_file( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fs = _prepare_filesystem(root_owned_tmp_path, monkeypatch) + registry = fs["registry"] + assert isinstance(registry, Path) + original = registry.read_bytes() + + def reject_replace(_source: os.PathLike[str] | str, _target: os.PathLike[str] | str) -> None: + raise OSError("synthetic atomic replace refusal") + + monkeypatch.setattr(policy.os, "replace", reject_replace) + with pytest.raises((RuntimeError, OSError)): + policy.reconcile("apply") + + assert registry.read_bytes() == original + assert not list(registry.parent.glob(f".{registry.name}.*.tmp")) + + +def test_project_lock_is_acquired_before_global_lock( + root_owned_tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + if not sys.platform.startswith("linux"): + pytest.skip("lock-order contention fixture observes Linux flock inode ownership") + import fcntl + + fs = _prepare_filesystem( + root_owned_tmp_path, monkeypatch, validation=policy.REQUIRED_POLICY + ) + project_lock = fs["project_lock"] + global_lock = fs["global_lock"] + assert isinstance(project_lock, Path) and isinstance(global_lock, Path) + project_identity = (project_lock.stat().st_dev, project_lock.stat().st_ino) + project_attempted = threading.Event() + thread_errors: list[RuntimeError] = [] + real_flock = fcntl.flock + + def observed_flock(file_or_fd, operation: int) -> None: + descriptor = file_or_fd if isinstance(file_or_fd, int) else file_or_fd.fileno() + info = os.fstat(descriptor) + if (info.st_dev, info.st_ino) == project_identity and operation & fcntl.LOCK_EX: + project_attempted.set() + real_flock(file_or_fd, operation) + + monkeypatch.setattr(fcntl, "flock", observed_flock) + + def run_check() -> None: + try: + policy.reconcile("check") + except RuntimeError as exc: # surfaced in the parent assertion below + thread_errors.append(exc) + + with project_lock.open("a+") as held_project: + real_flock(held_project, fcntl.LOCK_EX) + thread = threading.Thread(target=run_check, daemon=True) + thread.start() + try: + assert project_attempted.wait(timeout=3), "reconcile never attempted the project lock" + with global_lock.open("a+") as probe_global: + try: + real_flock(probe_global, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + pytest.fail("reconcile acquired the global lock before blocking on the project lock") + else: + real_flock(probe_global, fcntl.LOCK_UN) + finally: + real_flock(held_project, fcntl.LOCK_UN) + thread.join(timeout=5) + + assert not thread.is_alive(), "reconcile remained blocked after the project lock was released" + assert not thread_errors