Publish curated Unraid-AppOps-Gateway source from 451646fee806a47e3256731766980a05a262dcdd

backend/tests/test_forge_secret_bundle.py

@@ -0,0 +1,747 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+BROKER = ROOT / "ops" / "gitea-deploy-broker"
+if str(BROKER) not in sys.path:
+    sys.path.insert(0, str(BROKER))
+
+
+@pytest.fixture
+def bundle_module():
+    spec = importlib.util.spec_from_file_location(
+        "forge_secret_bundle_test",
+        BROKER / "forge_secret_bundle.py",
+    )
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def _config(**overrides: str) -> bytes:
+    value = {
+        "schema": "itworx.forge-first-install-config/v4",
+        "issuer": "https://auth.example.test/application/o/forge/",
+    }
+    value.update(overrides)
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def _initialize_test_roots(roots) -> None:
+    for path in dict.fromkeys(
+        (roots.config_root, roots.state_parent, roots.backup_root)
+    ):
+        path.mkdir(mode=0o700)
+        path.chmod(0o700)
+
+
+def test_validator_must_return_literal_true_before_any_filesystem_effect(
+    bundle_module, tmp_path
+):
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    roots = bundle_module._Roots.for_test(tmp_path)
+
+    for result in (False, None, 1):
+        with pytest.raises(bundle_module.ForgeSecretBundleError):
+            bundle_module._prepare_with_roots(
+                oidc,
+                authentik,
+                config,
+                validate_config=lambda _raw, result=result: result,
+                roots=roots,
+            )
+        assert list(tmp_path.iterdir()) == []
+
+    def raises_with_secret(_raw: bytes) -> bool:
+        raise RuntimeError("validation failed: " + oidc.decode("ascii"))
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError) as raised:
+        bundle_module._prepare_with_roots(
+            oidc,
+            authentik,
+            config,
+            validate_config=raises_with_secret,
+            roots=roots,
+        )
+    assert oidc.decode("ascii") not in str(raised.value)
+    assert list(tmp_path.iterdir()) == []
+
+
+@pytest.mark.parametrize(
+    "config",
+    [
+        b'{"schema":"v4","schema":"v4"}',
+        b'{ "schema":"v4"}',
+        b'{"schema":NaN}',
+        b"[]",
+        b"\xff",
+    ],
+)
+def test_rejects_duplicate_invalid_noncanonical_or_nonobject_config_without_effects(
+    bundle_module, tmp_path, config
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            config,
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_rejects_supplied_secret_embedded_in_config_before_validator_or_writes(
+    bundle_module, tmp_path
+):
+    oidc = b"oidc-secret-sentinel-123456789"
+    roots = bundle_module._Roots.for_test(tmp_path)
+    called = False
+
+    def validate(_raw: bytes) -> bool:
+        nonlocal called
+        called = True
+        return True
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            oidc,
+            b"authentik-token-sentinel-123456789",
+            _config(note=oidc.decode("ascii")),
+            validate_config=validate,
+            roots=roots,
+        )
+    assert called is False
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_rejects_escaped_secret_value_without_calling_validator(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    secret = b'oidc-"quoted"-secret-123456789'
+    encoded = _config(note=secret.decode("ascii"))
+    called = False
+
+    def validate(_raw: bytes) -> bool:
+        nonlocal called
+        called = True
+        return True
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            secret,
+            b"authentik-token-sentinel-123456789",
+            encoded,
+            validate_config=validate,
+            roots=roots,
+        )
+    assert called is False
+    assert list(tmp_path.iterdir()) == []
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_creates_fixed_files_and_backup_and_retry_only_verifies(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    def validate(raw):
+        return raw == config
+
+
+    result = bundle_module._prepare_with_roots(
+        oidc, authentik, config, validate_config=validate, roots=roots
+    )
+
+    assert result == {
+        "status": "bundle_ready",
+        "secretFilesReady": 7,
+        "masterKeyBackupVerified": True,
+        "applyPreflight": "not_evaluated",
+    }
+    assert oidc.decode("ascii") not in repr(result)
+    assert authentik.decode("ascii") not in repr(result)
+    assert sorted(path.name for path in roots.secrets_root.iterdir()) == sorted(
+        f"{name}.secret"
+        for name in (
+            "postgres_bootstrap_password",
+            "runtime_database_url",
+            "migrator_database_url",
+            "oidc_client_secret",
+            "session_secret",
+            "master_key",
+            "authentik_read_token",
+        )
+    )
+    assert (roots.secrets_root / "oidc_client_secret.secret").read_bytes() == oidc
+    assert (
+        roots.secrets_root / "authentik_read_token.secret"
+    ).read_bytes() == authentik
+    master_key = (roots.secrets_root / "master_key.secret").read_bytes()
+    assert len(master_key) == 32
+    assert (roots.backup_root / "master_key.backup").read_bytes() == master_key
+
+    import urllib.parse
+
+    generated = {
+        (roots.secrets_root / "postgres_bootstrap_password.secret").read_bytes(),
+        urllib.parse.urlsplit(
+            (roots.secrets_root / "runtime_database_url.secret").read_text(
+                encoding="ascii"
+            )
+        ).password.encode("ascii"),
+        urllib.parse.urlsplit(
+            (roots.secrets_root / "migrator_database_url.secret").read_text(
+                encoding="ascii"
+            )
+        ).password.encode("ascii"),
+        (roots.secrets_root / "session_secret.secret").read_bytes(),
+    }
+    assert len(generated) == 4
+    assert all(len(value) >= 32 for value in generated)
+
+    from forge_first_install import FORGE_SECRETS
+
+    for name, (_logical, uid, gid, minimum, maximum) in FORGE_SECRETS.items():
+        path = roots.secrets_root / f"{name}.secret"
+        info = path.stat()
+        assert (info.st_uid, info.st_gid) == (uid, gid)
+        assert info.st_mode & 0o777 == 0o400
+        assert minimum <= info.st_size <= maximum
+    assert (roots.secrets_root.stat().st_mode & 0o777) == 0o700
+    assert (roots.config_path.stat().st_mode & 0o777) == 0o600
+    assert (roots.backup_root / "master_key.backup").stat().st_mode & 0o777 == 0o400
+
+    journal = roots.journal_path.read_text(encoding="utf-8")
+    assert "oidc-secret-sentinel-123456789" not in journal
+    assert "authentik-token-sentinel-123456789" not in journal
+    assert "sha256" not in journal.lower()
+
+    before = {
+        path.name: (path.stat().st_ino, path.read_bytes())
+        for path in roots.secrets_root.iterdir()
+    }
+    backup_identity = (roots.backup_root / "master_key.backup").stat().st_ino
+    config_identity = roots.config_path.stat().st_ino
+    original_write = bundle_module._write_all
+
+    def reject_write(_fd: int, _payload: bytes) -> None:
+        raise AssertionError("a completed retry must not write")
+
+    bundle_module._write_all = reject_write
+    retry = bundle_module._prepare_with_roots(
+        oidc, authentik, config, validate_config=validate, roots=roots
+    )
+    bundle_module._write_all = original_write
+    assert retry == result
+    assert before == {
+        path.name: (path.stat().st_ino, path.read_bytes())
+        for path in roots.secrets_root.iterdir()
+    }
+    assert (roots.backup_root / "master_key.backup").stat().st_ino == backup_identity
+    assert roots.config_path.stat().st_ino == config_identity
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_partial_retry_reuses_already_committed_password_and_recovers_backup(
+    bundle_module, monkeypatch, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    original_write = bundle_module._write_all
+    failed = False
+
+    def fail_runtime_write(fd: int, payload: bytes) -> None:
+        nonlocal failed
+        if payload.startswith(b"postgresql://") and not failed:
+            failed = True
+            raise OSError("injected write failure")
+        original_write(fd, payload)
+
+    monkeypatch.setattr(bundle_module, "_write_all", fail_runtime_write)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            oidc,
+            authentik,
+            config,
+            validate_config=lambda raw: raw == config,
+            roots=roots,
+        )
+    bootstrap = roots.secrets_root / "postgres_bootstrap_password.secret"
+    original_bootstrap = bootstrap.read_bytes()
+    original_inode = bootstrap.stat().st_ino
+
+    monkeypatch.setattr(bundle_module, "_write_all", original_write)
+    result = bundle_module._prepare_with_roots(
+        oidc,
+        authentik,
+        config,
+        validate_config=lambda raw: raw == config,
+        roots=roots,
+    )
+    assert result["status"] == "bundle_ready"
+    assert bootstrap.read_bytes() == original_bootstrap
+    assert bootstrap.stat().st_ino == original_inode
+    master = (roots.secrets_root / "master_key.secret").read_bytes()
+    assert (roots.backup_root / "master_key.backup").read_bytes() == master
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_link_failure_after_durable_stage_keeps_and_reuses_original_secret(
+    bundle_module, monkeypatch, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    original_link = os.link
+    failed = False
+
+    def fail_runtime_link(source, destination, **kwargs):
+        nonlocal failed
+        if destination == "runtime_database_url.secret" and not failed:
+            failed = True
+            raise OSError("injected link failure")
+        return original_link(source, destination, **kwargs)
+
+    monkeypatch.setattr(os, "link", fail_runtime_link)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            oidc,
+            authentik,
+            config,
+            validate_config=lambda raw: raw == config,
+            roots=roots,
+        )
+
+    journal = json.loads(roots.journal_path.read_bytes())
+    stage = journal["staging"]
+    assert stage["kind"] == "secret"
+    assert stage["name"] == "runtime_database_url"
+    staged_file = roots.secrets_root / stage["temporary"]
+    assert staged_file.is_file()
+    original_payload = staged_file.read_bytes()
+    assert original_payload.startswith(b"postgresql://forge_runtime:")
+
+    monkeypatch.setattr(os, "link", original_link)
+    result = bundle_module._prepare_with_roots(
+        oidc,
+        authentik,
+        config,
+        validate_config=lambda raw: raw == config,
+        roots=roots,
+    )
+
+    assert result["status"] == "bundle_ready"
+    runtime_password = (roots.secrets_root / "runtime_database_url.secret").read_bytes()
+    assert runtime_password == original_payload
+    assert not staged_file.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_journal_directory_fsync_uncertainty_preserves_referenced_stage(
+    bundle_module, monkeypatch, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    original_fsync = os.fsync
+    failed = False
+
+    def fail_stage_directory_fsync(fd: int) -> None:
+        nonlocal failed
+        if not failed and stat.S_ISDIR(os.fstat(fd).st_mode):
+            try:
+                journal = json.loads(roots.journal_path.read_bytes())
+            except (OSError, ValueError):
+                journal = {}
+            stage = journal.get("staging")
+            if stage is not None and stage.get("name") == "runtime_database_url":
+                failed = True
+                raise OSError("injected directory fsync uncertainty")
+        original_fsync(fd)
+
+    monkeypatch.setattr(os, "fsync", fail_stage_directory_fsync)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            oidc,
+            authentik,
+            config,
+            validate_config=lambda raw: raw == config,
+            roots=roots,
+        )
+
+    journal = json.loads(roots.journal_path.read_bytes())
+    stage = journal["staging"]
+    assert stage["name"] == "runtime_database_url"
+    staged_file = roots.secrets_root / stage["temporary"]
+    assert staged_file.is_file()
+    original_payload = staged_file.read_bytes()
+
+    monkeypatch.setattr(os, "fsync", original_fsync)
+    result = bundle_module._prepare_with_roots(
+        oidc,
+        authentik,
+        config,
+        validate_config=lambda raw: raw == config,
+        roots=roots,
+    )
+    assert result["status"] == "bundle_ready"
+    assert (
+        roots.secrets_root / "runtime_database_url.secret"
+    ).read_bytes() == original_payload
+    assert not staged_file.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_backup_link_failure_reuses_same_master_key_and_completes_backup(
+    bundle_module, monkeypatch, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    oidc = b"oidc-secret-sentinel-123456789"
+    authentik = b"authentik-token-sentinel-123456789"
+    config = _config()
+    original_link = os.link
+    failed = False
+
+    def fail_backup_link(source, destination, **kwargs):
+        nonlocal failed
+        if destination == "master_key.backup" and not failed:
+            failed = True
+            raise OSError("injected backup link failure")
+        return original_link(source, destination, **kwargs)
+
+    monkeypatch.setattr(os, "link", fail_backup_link)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            oidc,
+            authentik,
+            config,
+            validate_config=lambda raw: raw == config,
+            roots=roots,
+        )
+
+    journal = json.loads(roots.journal_path.read_bytes())
+    stage = journal["staging"]
+    assert stage["kind"] == "backup"
+    assert stage["name"] == "master_key.backup"
+    staged_file = roots.backup_root / stage["temporary"]
+    original_payload = staged_file.read_bytes()
+    original_master = (roots.secrets_root / "master_key.secret").read_bytes()
+    original_master_inode = (roots.secrets_root / "master_key.secret").stat().st_ino
+    assert original_payload == original_master
+    assert len(original_payload) == 32
+
+    monkeypatch.setattr(os, "link", original_link)
+    result = bundle_module._prepare_with_roots(
+        oidc,
+        authentik,
+        config,
+        validate_config=lambda raw: raw == config,
+        roots=roots,
+    )
+
+    assert result["status"] == "bundle_ready"
+    assert (roots.secrets_root / "master_key.secret").read_bytes() == original_master
+    assert (
+        roots.secrets_root / "master_key.secret"
+    ).stat().st_ino == original_master_inode
+    assert (roots.backup_root / "master_key.backup").read_bytes() == original_payload
+    assert not staged_file.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_existing_secret_residue_is_never_overwritten_or_adopted(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    roots.secrets_root.mkdir(mode=0o700)
+    residue = roots.secrets_root / "session_secret.secret"
+    residue.write_bytes(b"foreign-residue-must-survive")
+    residue.chmod(0o400)
+    before = residue.read_bytes()
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            _config(),
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert residue.read_bytes() == before
+    assert not roots.journal_path.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_unsafe_ancestor_and_symlinked_secret_root_fail_without_chmod(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    roots.config_root.chmod(0o777)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            _config(),
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert roots.config_root.stat().st_mode & 0o777 == 0o777
+
+    roots.config_root.chmod(0o700)
+    target = tmp_path / "foreign-secret-dir"
+    target.mkdir(mode=0o700)
+    roots.secrets_root.symlink_to(target, target_is_directory=True)
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            _config(),
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert list(target.iterdir()) == []
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_stale_journal_generation_is_rejected_by_compare_and_swap(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    roots.state_root.mkdir(mode=0o700)
+    descriptor = os.open(roots.state_root, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        record = bundle_module._empty_journal()
+        bundle_module._write_journal(descriptor, record, expected_generation=None)
+        stale = dict(record)
+        record["status"] = "progress"
+        bundle_module._persist(descriptor, record)
+        before = roots.journal_path.read_bytes()
+        with pytest.raises(bundle_module.ForgeSecretBundleError):
+            bundle_module._write_journal(descriptor, stale, expected_generation=0)
+        assert roots.journal_path.read_bytes() == before
+    finally:
+        os.close(descriptor)
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_backup_root_rejects_untracked_files_before_creating_journal(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    foreign = roots.backup_root / "unrelated-backup-residue"
+    foreign.write_bytes(b"leave this untouched")
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            _config(),
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert foreign.read_bytes() == b"leave this untouched"
+    assert not roots.state_root.exists()
+    assert not roots.secrets_root.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+@pytest.mark.parametrize("target", ["config", "backup"])
+def test_existing_config_or_backup_files_are_never_overwritten(
+    bundle_module, tmp_path, target
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    path = (
+        roots.config_path
+        if target == "config"
+        else roots.backup_root / "master_key.backup"
+    )
+    path.write_bytes(b"foreign-config-or-backup")
+    path.chmod(0o600)
+    original = path.read_bytes()
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            b"oidc-secret-sentinel-123456789",
+            b"authentik-token-sentinel-123456789",
+            _config(),
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert path.read_bytes() == original
+    assert not roots.state_root.exists()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_hardlinked_existing_secret_fails_closed_without_rekeying(
+    bundle_module, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    args = (
+        b"oidc-secret-sentinel-123456789",
+        b"authentik-token-sentinel-123456789",
+        _config(),
+    )
+    bundle_module._prepare_with_roots(
+        *args,
+        validate_config=lambda _raw: True,
+        roots=roots,
+    )
+    master = roots.secrets_root / "master_key.secret"
+    original = master.read_bytes()
+    foreign_link = tmp_path / "foreign-master-link"
+    os.link(master, foreign_link)
+    journal_before = roots.journal_path.read_bytes()
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            *args,
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert master.read_bytes() == original
+    assert roots.journal_path.read_bytes() == journal_before
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_pinned_file_reader_rejects_symlink_swap_between_stat_and_open(
+    bundle_module, monkeypatch, tmp_path
+):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    roots.secrets_root.mkdir(mode=0o700)
+    target = roots.secrets_root / "trusted-content"
+    target.write_bytes(b"trusted")
+    target.chmod(0o400)
+    race = roots.secrets_root / "race-target"
+    race.write_bytes(b"before-race")
+    race.chmod(0o400)
+    directory_fd = os.open(roots.secrets_root, os.O_RDONLY | os.O_DIRECTORY)
+    original_stat_at = bundle_module._stat_at
+    swapped = False
+
+    def swap_after_stat(fd: int, name: str):
+        nonlocal swapped
+        info = original_stat_at(fd, name)
+        if name == "race-target" and info is not None and not swapped:
+            swapped = True
+            os.unlink(name, dir_fd=fd)
+            os.symlink("trusted-content", name, dir_fd=fd)
+        return info
+
+    monkeypatch.setattr(bundle_module, "_stat_at", swap_after_stat)
+    try:
+        with pytest.raises(bundle_module.ForgeSecretBundleError):
+            bundle_module._read_file_at(
+                directory_fd,
+                "race-target",
+                uid=0,
+                gid=0,
+                min_size=1,
+                max_size=32,
+                allowed_modes=frozenset({0o400}),
+            )
+    finally:
+        os.close(directory_fd)
+    assert swapped is True
+    assert race.is_symlink()
+
+
+@pytest.mark.skipif(
+    os.name != "posix" or getattr(os, "geteuid", lambda: 1)() != 0,
+    reason="the production helper requires POSIX root ownership",
+)
+def test_changed_master_key_backup_is_not_silently_rewritten(bundle_module, tmp_path):
+    roots = bundle_module._Roots.for_test(tmp_path)
+    _initialize_test_roots(roots)
+    args = (
+        b"oidc-secret-sentinel-123456789",
+        b"authentik-token-sentinel-123456789",
+        _config(),
+    )
+    bundle_module._prepare_with_roots(
+        *args,
+        validate_config=lambda _raw: True,
+        roots=roots,
+    )
+    backup = roots.backup_root / "master_key.backup"
+    original = backup.read_bytes()
+    backup.write_bytes(bytes(byte ^ 0xFF for byte in original))
+    journal_before = roots.journal_path.read_bytes()
+
+    with pytest.raises(bundle_module.ForgeSecretBundleError):
+        bundle_module._prepare_with_roots(
+            *args,
+            validate_config=lambda _raw: True,
+            roots=roots,
+        )
+    assert backup.read_bytes() == bytes(byte ^ 0xFF for byte in original)
+    assert (roots.secrets_root / "master_key.secret").read_bytes() == original
+    assert roots.journal_path.read_bytes() == journal_before