from __future__ import annotations import importlib.util import os from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] MODULE_PATH = ROOT / "ops/gitea-deploy-broker/forge_backup_paths.py" if MODULE_PATH.is_file(): spec = importlib.util.spec_from_file_location("forge_backup_paths_test", MODULE_PATH) assert spec is not None and spec.loader is not None paths = importlib.util.module_from_spec(spec) spec.loader.exec_module(paths) else: paths = None def _paths(): if paths is None: pytest.fail("fixed Forge backup path verifier has not been implemented", pytrace=False) return paths def volume_metadata(): return { "Name": "itworx-forge-master-key-backup", "Driver": "local", "Scope": "local", "Options": None, "Labels": { "itworx.managed-by": "deployment-controller", "itworx.purpose": "forge-master-key-backup", }, "Mountpoint": "/var/lib/docker/volumes/itworx-forge-master-key-backup/_data", } def controller_mounts(): return [ { "Type": "volume", "Name": "itworx-forge-master-key-backup", "Driver": "local", "Source": "/var/lib/docker/volumes/itworx-forge-master-key-backup/_data", "Destination": "/forge-master-key-backup", "RW": True, }, { "Type": "bind", "Name": "", "Source": "/var/lib/docker", "Destination": "/var/lib/docker", "RW": False, }, { "Type": "volume", "Name": "itworx-managed-appdata", "Driver": "local", "Source": "/var/lib/docker/volumes/itworx-managed-appdata/_data", "Destination": "/var/lib/docker/volumes/itworx-managed-appdata/_data", "RW": True, }, { "Type": "volume", "Name": "itworx-controller-hub-releases", "Driver": "local", "Source": "/var/lib/docker/volumes/itworx-controller-hub-releases/_data", "Destination": "/controller-hub-releases", "RW": True, }, { "Type": "bind", "Name": "", "Source": "/var/lib/docker/volumes/itworx-controller-hub-releases/_data/controller-config", "Destination": "/controller-config", "RW": False, }, ] def mountinfo(*, backup_options="rw,nosuid,nodev", docker_options="ro,relatime"): base = [ "35 25 8:1 / / rw,relatime - ext4 /dev/sda1 rw,relatime", f"41 35 8:2 / /var/lib/docker {docker_options} - ext4 /dev/sda2 ro,relatime", f"42 35 8:3 / /forge-master-key-backup {backup_options} - ext4 /dev/sda3 rw,relatime", ] base.extend( f"{50 + index} 41 8:{index + 4} / /var/lib/docker/containers/c{index}/mount{index} " f"{'ro' if index % 2 else 'rw'},relatime - ext4 /dev/sdb{index} rw,relatime" for index in range(35) ) return "\n".join(base) def test_exact_local_backup_volume_metadata_is_accepted(): module = _paths() assert module.validate_volume_metadata(volume_metadata(), "/var/lib/docker") is None @pytest.mark.parametrize( ("field", "value"), [ ("Name", "foreign"), ("Driver", "nfs"), ("Scope", "global"), ("Options", {"device": "/mnt/share"}), ("Options", []), ("Labels", {"itworx.managed-by": "deployment-controller"}), ("Labels", { "itworx.managed-by": "deployment-controller", "itworx.purpose": "managed-appdata", }), ("Mountpoint", "/mnt/foreign"), ], ) def test_backup_volume_metadata_drift_is_rejected(field, value): module = _paths() item = volume_metadata() item[field] = value with pytest.raises(module.UnsafeForgeBackupVolume): module.validate_volume_metadata(item, "/var/lib/docker") def test_backup_volume_metadata_requires_the_fixed_docker_root(): module = _paths() with pytest.raises(module.UnsafeForgeBackupVolume): module.validate_volume_metadata(volume_metadata(), "/mnt/docker") def test_mount_contract_allows_only_the_fixed_rw_backup_and_ro_docker_root(): module = _paths() assert module.validate_controller_mounts( controller_mounts(), controller_id_before="a" * 64, controller_id_after="a" * 64, ) is None def test_mount_contract_allows_managed_appdata_to_be_absent(): module = _paths() mounts = controller_mounts() mounts.pop(2) assert module.validate_controller_mounts( mounts, controller_id_before="a" * 64, controller_id_after="a" * 64, ) is None def test_mount_contract_allows_only_the_exact_managed_appdata_nested_volume(): module = _paths() mounts = controller_mounts() module.validate_controller_mounts( mounts, controller_id_before="a" * 64, controller_id_after="a" * 64, ) mounts[-3]["Destination"] = "/var/lib/docker/volumes/foreign/_data" with pytest.raises(module.UnsafeForgeBackupVolume): module.validate_controller_mounts( mounts, controller_id_before="a" * 64, controller_id_after="a" * 64, ) @pytest.mark.parametrize( "damage", [ "controller_changed", "backup_rw_disabled", "backup_name_changed", "backup_source_changed", "backup_destination_changed", "backup_driver_changed", "duplicate_backup_mount", "backup_source_alias", "backup_target_child_alias", "backup_target_parent_alias", "docker_root_rw", "docker_root_child_mount", "managed_appdata_duplicate", "managed_appdata_source_changed", "managed_appdata_driver_changed", "managed_appdata_target_alias", "managed_appdata_name_alias", "root_exposure", ], ) def test_mount_contract_rejects_aliases_and_mount_drift(damage): module = _paths() mounts = controller_mounts() before = after = "a" * 64 if damage == "controller_changed": after = "b" * 64 elif damage == "backup_rw_disabled": mounts[0]["RW"] = False elif damage == "backup_name_changed": mounts[0]["Name"] = "foreign" elif damage == "backup_source_changed": mounts[0]["Source"] = "/mnt/foreign" elif damage == "backup_destination_changed": mounts[0]["Destination"] = "/backup-copy" elif damage == "backup_driver_changed": mounts[0]["Driver"] = "nfs" elif damage == "duplicate_backup_mount": mounts.append(dict(mounts[0])) elif damage == "backup_source_alias": mounts.append({ "Type": "bind", "Source": volume_metadata()["Mountpoint"], "Destination": "/backup-copy", "RW": False, }) elif damage == "backup_target_child_alias": mounts.append({ "Type": "bind", "Source": "/foreign", "Destination": "/forge-master-key-backup/state", "RW": True, }) elif damage == "backup_target_parent_alias": mounts.append({ "Type": "bind", "Source": "/foreign", "Destination": "/", "RW": True, }) elif damage == "docker_root_rw": mounts[1]["RW"] = True elif damage == "docker_root_child_mount": mounts.append({ "Type": "bind", "Source": "/var/lib/docker/volumes/other", "Destination": "/var/lib/docker/volumes/other", "RW": True, }) elif damage == "managed_appdata_duplicate": mounts.append(dict(mounts[2])) elif damage == "managed_appdata_source_changed": mounts[2]["Source"] = "/var/lib/docker/volumes/foreign/_data" elif damage == "managed_appdata_driver_changed": mounts[2]["Driver"] = "nfs" elif damage == "managed_appdata_target_alias": mounts[2]["Destination"] = "/appdata-copy" elif damage == "managed_appdata_name_alias": mounts[2].update({ "Source": "/var/lib/docker/volumes/foreign/_data", "Destination": "/appdata-copy", }) else: mounts.append({ "Type": "bind", "Source": "/", "Destination": "/", "RW": True, }) with pytest.raises(module.UnsafeForgeBackupVolume): module.validate_controller_mounts( mounts, controller_id_before=before, controller_id_after=after ) def test_proc_mountinfo_requires_the_exact_rw_backup_and_ro_docker_mounts(): module = _paths() assert module._validate_mountinfo_contract(mountinfo()) is None for unsafe in ( mountinfo(backup_options="ro,nosuid,nodev"), mountinfo(docker_options="rw,relatime"), mountinfo().replace("/forge-master-key-backup", "/backup-copy"), mountinfo() + "\n43 35 8:4 / /forge-master-key-backup rw - ext4 /dev/sda4 rw", ): with pytest.raises(module.UnsafeForgeBackupVolume): module._validate_mountinfo_contract(unsafe) @pytest.mark.skipif(os.name == "nt", reason="directory inode checks require POSIX descriptors") def test_directory_identity_binds_backup_mount_to_the_fixed_docker_volume_source(tmp_path): module = _paths() source = tmp_path / "source" mounted = tmp_path / "mounted" source.mkdir() mounted.mkdir() source_fd = os.open(source, os.O_RDONLY) mounted_fd = os.open(mounted, os.O_RDONLY) same_fd = os.open(source, os.O_RDONLY) try: module._require_same_directory(source_fd, same_fd) with pytest.raises(module.UnsafeForgeBackupVolume, match="mount_source_mismatch"): module._require_same_directory(source_fd, mounted_fd) finally: os.close(source_fd) os.close(mounted_fd) os.close(same_fd) @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="directory ownership checks require root on POSIX", ) def test_fixed_directory_walk_rejects_symlinks_and_unsafe_modes(tmp_path): module = _paths() volume_root = tmp_path / "volume" volume_root.mkdir(mode=0o755) volume_root.chmod(0o755) fd = module._open_directory_no_symlinks(volume_root, anchor=tmp_path) os.close(fd) linked = tmp_path / "linked" linked.symlink_to(volume_root, target_is_directory=True) with pytest.raises(module.UnsafeForgeBackupVolume): module._open_directory_no_symlinks(linked, anchor=tmp_path) volume_root.chmod(0o777) with pytest.raises(module.UnsafeForgeBackupVolume): module._open_directory_no_symlinks(volume_root, anchor=tmp_path) @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="directory ACL checks require root on POSIX", ) def test_fixed_directory_walk_rejects_access_acl(tmp_path, monkeypatch): module = _paths() volume_root = tmp_path / "volume" volume_root.mkdir(mode=0o755) volume_root.chmod(0o755) monkeypatch.setattr(module.os, "listxattr", lambda _fd: ["system.posix_acl_access"]) with pytest.raises(module.UnsafeForgeBackupVolume): module._open_directory_no_symlinks(volume_root, anchor=tmp_path) def test_read_only_identity_check_fails_closed_without_creating_or_repairing_paths(tmp_path, monkeypatch): module = _paths() sentinel = tmp_path / "unchanged" sentinel.write_bytes(b"keep") bad = volume_metadata() bad["Labels"]["itworx.purpose"] = "foreign-purpose" monkeypatch.setattr(module, "_read_volume_snapshot", lambda: (bad, "/var/lib/docker")) monkeypatch.setattr( module, "_open_fixed_backup_root", lambda: pytest.fail("unsafe volume metadata must fail before opening the data root"), ) with pytest.raises(module.UnsafeForgeBackupVolume): module.validate_forge_backup_volume() assert sentinel.read_bytes() == b"keep" assert sorted(path.name for path in tmp_path.iterdir()) == ["unchanged"]