748 lines
25 KiB
Python
748 lines
25 KiB
Python
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
|