779 lines
28 KiB
Python
779 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from ops import gcp_leoclean_nosend_iap_remote as remote
|
|
from ops import run_gcp_leoclean_nosend_iap as local
|
|
|
|
|
|
def completed(
|
|
argv: list[str],
|
|
*,
|
|
returncode: int = 0,
|
|
stdout: bytes = b"",
|
|
stderr: bytes = b"",
|
|
) -> subprocess.CompletedProcess[bytes]:
|
|
return subprocess.CompletedProcess(argv, returncode, stdout, stderr)
|
|
|
|
|
|
def private_directory(path: Path) -> Path:
|
|
path.mkdir(mode=0o700)
|
|
path.chmod(0o700)
|
|
return path
|
|
|
|
|
|
def fixed_commands(tmp_path: Path, operation: str = "install") -> tuple[list[str], list[str], list[str]]:
|
|
request_id = "iap-abcdefghijkl"
|
|
archive_name, key_name = local.remote_names(request_id)
|
|
archive = tmp_path / archive_name
|
|
ssh_key = tmp_path / key_name
|
|
archive.write_bytes(b"archive")
|
|
archive.chmod(0o600)
|
|
return local.build_gcloud_argv(
|
|
operation=operation,
|
|
request_id=request_id,
|
|
runner_revision="a" * 40,
|
|
release_revision="b" * 40,
|
|
archive=archive,
|
|
archive_sha256="c" * 64,
|
|
ssh_key=ssh_key,
|
|
helper_sha256="d" * 64,
|
|
)
|
|
|
|
|
|
def executable_plan(tmp_path: Path, operation: str) -> dict[str, Any]:
|
|
request_id = "iap-abcdefghijkl"
|
|
archive_name, key_name = local.remote_names(request_id)
|
|
archive = tmp_path / archive_name
|
|
ssh_key = tmp_path / key_name
|
|
archive.write_bytes(b"archive")
|
|
archive.chmod(0o600)
|
|
archive_sha256 = local._sha256_bytes(archive.read_bytes())
|
|
helper_sha256 = "c" * 64
|
|
commands = local.build_gcloud_argv(
|
|
operation=operation,
|
|
request_id=request_id,
|
|
runner_revision="a" * 40,
|
|
release_revision="b" * 40,
|
|
archive=archive,
|
|
archive_sha256=archive_sha256,
|
|
ssh_key=ssh_key,
|
|
helper_sha256=helper_sha256,
|
|
)
|
|
expected = {
|
|
"release_artifact_sha256": "d" * 64,
|
|
"release_sha256": "e" * 64,
|
|
"image_digest": f"sha256:{'f' * 64}",
|
|
"image_config_digest": f"sha256:{'1' * 64}",
|
|
}
|
|
return {
|
|
"schema": local.PLAN_SCHEMA,
|
|
"status": "pass",
|
|
"mode": "dry_run",
|
|
"operation": operation,
|
|
"request_id": "iap-abcdefghijkl",
|
|
"runner_revision": "a" * 40,
|
|
"release_revision": "b" * 40,
|
|
"target": {
|
|
"project": local.PROJECT,
|
|
"zone": local.ZONE,
|
|
"instance": local.INSTANCE,
|
|
"service": local.SERVICE,
|
|
"service_account": local.SERVICE_ACCOUNT,
|
|
},
|
|
"release": {
|
|
"release_sha256": expected["release_sha256"],
|
|
"image_reference": f"{local.package.IMAGE_REPOSITORY}@{expected['image_digest']}",
|
|
"image_digest": expected["image_digest"],
|
|
"image_config_digest": expected["image_config_digest"],
|
|
},
|
|
"release_artifact": {
|
|
"archive_sha256": expected["release_artifact_sha256"],
|
|
"archive_size": 1,
|
|
"members": [],
|
|
},
|
|
"expected": expected,
|
|
"request": {
|
|
"archive": str(archive),
|
|
"archive_sha256": archive_sha256,
|
|
"helper_sha256": helper_sha256,
|
|
"manifest_sha256": "d" * 64,
|
|
"ssh_key": str(ssh_key),
|
|
},
|
|
"actions": local._action_receipts(commands),
|
|
"authority": {
|
|
"gcp_mutation": operation in {"install", "rollback"},
|
|
"service_restart": operation in {"install", "rollback"},
|
|
"database_or_secret_provisioning": False,
|
|
"production_or_transport": False,
|
|
"vps": False,
|
|
},
|
|
}
|
|
|
|
|
|
def test_gcloud_commands_are_fixed_iap_only_and_expire_keys(tmp_path: Path) -> None:
|
|
scp, ssh, cleanup = fixed_commands(tmp_path)
|
|
|
|
assert scp[:3] == [local.GCLOUD, "compute", "scp"]
|
|
assert ssh[:4] == [local.GCLOUD, "compute", "ssh", local.INSTANCE]
|
|
for command in (scp, ssh, cleanup):
|
|
assert f"--project={local.PROJECT}" in command
|
|
assert f"--zone={local.ZONE}" in command
|
|
assert "--tunnel-through-iap" in command
|
|
assert "--ssh-key-expire-after=5m" in command
|
|
encoded = json.dumps({"scp": scp, "ssh": ssh, "cleanup": cleanup})
|
|
assert local.INSTANCE in encoded
|
|
assert local.SERVICE not in encoded
|
|
assert "teleo-prod-1" not in encoded
|
|
assert "77.42.65.182" not in encoded
|
|
assert "telegram" not in encoded.casefold()
|
|
assert "sudo -- /usr/bin/python3 -I -c" in ssh[-1]
|
|
assert "/tmp/livingip-leoclean-nosend-remote" not in ssh[-1]
|
|
|
|
|
|
def test_embedded_root_bootstrap_compiles() -> None:
|
|
compile(local.BOOTSTRAP_CODE, "<leoclean-iap-bootstrap>", "exec")
|
|
|
|
|
|
def test_live_origin_main_readback_is_exact() -> None:
|
|
revision = "a" * 40
|
|
responses = iter(
|
|
[
|
|
f"{local.ORIGIN_URL}\n".encode(),
|
|
f"{revision}\trefs/heads/main\n".encode(),
|
|
]
|
|
)
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
return completed(argv, stdout=next(responses))
|
|
|
|
local.validate_remote_main(revision, runner=runner)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"responses",
|
|
[
|
|
[b"git@github.com:someone-else/teleo-infrastructure.git\n"],
|
|
[
|
|
f"{local.ORIGIN_URL}\n".encode(),
|
|
f"{'b' * 40}\trefs/heads/main\n".encode(),
|
|
],
|
|
],
|
|
)
|
|
def test_live_origin_main_readback_rejects_different_authority(responses: list[bytes]) -> None:
|
|
output = iter(responses)
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
return completed(argv, stdout=next(output))
|
|
|
|
with pytest.raises(local.IapRequestError):
|
|
local.validate_remote_main("a" * 40, runner=runner)
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["main", "a" * 39, "A" * 40, "a" * 41, "a" * 39 + ";"])
|
|
def test_revision_input_is_exact(value: str, tmp_path: Path) -> None:
|
|
archive_name, key_name = local.remote_names("iap-abcdefghijkl")
|
|
with pytest.raises(local.IapRequestError, match="revision"):
|
|
local.build_gcloud_argv(
|
|
operation="install",
|
|
request_id="iap-abcdefghijkl",
|
|
runner_revision=value,
|
|
release_revision="b" * 40,
|
|
archive=tmp_path / archive_name,
|
|
archive_sha256="b" * 64,
|
|
ssh_key=tmp_path / key_name,
|
|
helper_sha256="c" * 64,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["iap-short", "iap-abcdefghijkl;id", "../iap-abcdefghijkl", "iap-ABCDEFGHIJKL"])
|
|
def test_request_id_rejects_shell_and_path_fragments(value: str) -> None:
|
|
with pytest.raises(local.IapRequestError, match="request ID"):
|
|
local.remote_names(value)
|
|
|
|
|
|
def test_mutable_release_reference_is_rejected(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
release = {
|
|
"image": {"reference": f"{local.package.IMAGE_REPOSITORY}:latest"},
|
|
"image_input": {"runtime": {"teleo_git_head": "a" * 40}},
|
|
}
|
|
monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit"))
|
|
|
|
with pytest.raises(local.IapRequestError, match="immutable"):
|
|
local.validate_release(
|
|
tmp_path,
|
|
"a" * 40,
|
|
expected_release_sha256="b" * 64,
|
|
expected_image_digest=f"sha256:{'c' * 64}",
|
|
expected_image_config_digest=f"sha256:{'d' * 64}",
|
|
)
|
|
|
|
|
|
def test_stable_reader_rejects_symlink(tmp_path: Path) -> None:
|
|
root = private_directory(tmp_path / "root")
|
|
target = root / "target"
|
|
target.write_text("safe", encoding="utf-8")
|
|
alias = root / "alias"
|
|
alias.symlink_to(target)
|
|
|
|
with pytest.raises(local.IapRequestError, match="unsafe"):
|
|
local._read_stable_file(alias.absolute(), root=root.absolute())
|
|
|
|
|
|
def archive_manifest(payloads: dict[str, tuple[bytes, int]]) -> dict[str, Any]:
|
|
files = [
|
|
{"path": path, "sha256": local._sha256_bytes(content), "size": len(content), "mode": mode}
|
|
for path, (content, mode) in sorted(payloads.items())
|
|
]
|
|
stable = {
|
|
"schema": remote.SCHEMA,
|
|
"operation": "verify",
|
|
"request_id": "iap-abcdefghijkl",
|
|
"runner_revision": "a" * 40,
|
|
"release_revision": "b" * 40,
|
|
"target": {
|
|
"project": remote.PROJECT,
|
|
"zone": remote.ZONE,
|
|
"instance": remote.INSTANCE,
|
|
"service": remote.SERVICE,
|
|
"service_account": remote.SERVICE_ACCOUNT,
|
|
},
|
|
"release": {
|
|
"release_sha256": "b" * 64,
|
|
"image_reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'c' * 64}",
|
|
"image_digest": f"sha256:{'c' * 64}",
|
|
"image_config_digest": f"sha256:{'d' * 64}",
|
|
},
|
|
"installer_source": {"revision": "a" * 40, "files": [], "sha256": "e" * 64},
|
|
"release_artifact": {"archive_sha256": "f" * 64, "archive_size": 1, "members": []},
|
|
"files": files,
|
|
}
|
|
return {**stable, "manifest_sha256": remote._canonical_sha256(stable)}
|
|
|
|
|
|
def exact_payloads() -> dict[str, tuple[bytes, int]]:
|
|
return {name: (f"payload:{name}\n".encode(), 0o644) for name in remote.EXPECTED_MEMBERS - {"request-manifest.json"}}
|
|
|
|
|
|
def test_archive_is_deterministic_and_extracts_exact_members(tmp_path: Path) -> None:
|
|
first_root = private_directory(tmp_path / "first")
|
|
second_root = private_directory(tmp_path / "second")
|
|
extracted = private_directory(tmp_path / "extracted")
|
|
payloads = exact_payloads()
|
|
manifest = archive_manifest(payloads)
|
|
|
|
first_hash = local.build_archive(first_root / "request.tar.gz", manifest=manifest, payloads=payloads)
|
|
second_hash = local.build_archive(second_root / "request.tar.gz", manifest=manifest, payloads=payloads)
|
|
observed = remote.extract_archive(first_root / "request.tar.gz", extracted)
|
|
|
|
assert first_hash == second_hash
|
|
assert (first_root / "request.tar.gz").read_bytes() == (second_root / "request.tar.gz").read_bytes()
|
|
assert observed == manifest
|
|
assert {str(path.relative_to(extracted)) for path in extracted.rglob("*") if path.is_file()} == (
|
|
remote.EXPECTED_MEMBERS - {"request-manifest.json"}
|
|
)
|
|
|
|
|
|
def write_unsafe_archive(path: Path, *, kind: str) -> None:
|
|
with tarfile.open(path, "w:gz") as archive:
|
|
for index, name in enumerate(sorted(remote.EXPECTED_MEMBERS)):
|
|
info = tarfile.TarInfo(name)
|
|
info.uid = 0
|
|
info.gid = 0
|
|
info.uname = "root"
|
|
info.gname = "root"
|
|
info.mtime = 0
|
|
info.mode = 0o600
|
|
content = b"{}"
|
|
if kind == "link" and index == 0:
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = "/etc/passwd"
|
|
info.size = 0
|
|
archive.addfile(info)
|
|
continue
|
|
if kind == "oversize" and index == 0:
|
|
content = b"x" * (remote.MAX_MEMBER_BYTES + 1)
|
|
info.size = len(content)
|
|
archive.addfile(info, io.BytesIO(content))
|
|
if kind == "duplicate":
|
|
info = tarfile.TarInfo("request-manifest.json")
|
|
info.size = 2
|
|
info.mode = 0o600
|
|
info.uid = info.gid = info.mtime = 0
|
|
info.uname = info.gname = "root"
|
|
archive.addfile(info, io.BytesIO(b"{}"))
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["link", "oversize", "duplicate"])
|
|
def test_archive_rejects_links_oversize_and_duplicate_members(tmp_path: Path, kind: str) -> None:
|
|
path = tmp_path / f"{kind}.tar.gz"
|
|
destination = private_directory(tmp_path / kind)
|
|
write_unsafe_archive(path, kind=kind)
|
|
|
|
with pytest.raises(remote.RemoteError):
|
|
remote.extract_archive(path, destination)
|
|
|
|
|
|
def test_archive_output_requires_private_directory(tmp_path: Path) -> None:
|
|
output = tmp_path / "public"
|
|
output.mkdir(mode=0o755)
|
|
output.chmod(0o755)
|
|
with pytest.raises(local.IapRequestError, match="private"):
|
|
local.build_archive(output / "request.tar.gz", manifest={}, payloads={})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("expected_artifact_sha256", "not-a-hash"),
|
|
("expected_release_sha256", "a" * 63),
|
|
("expected_image_digest", "latest"),
|
|
("expected_image_config_digest", "sha256:" + "A" * 64),
|
|
],
|
|
)
|
|
def test_expected_release_inputs_are_exact(field: str, value: str) -> None:
|
|
arguments = {
|
|
"expected_artifact_sha256": "a" * 64,
|
|
"expected_release_sha256": "b" * 64,
|
|
"expected_image_digest": f"sha256:{'c' * 64}",
|
|
"expected_image_config_digest": f"sha256:{'d' * 64}",
|
|
}
|
|
arguments[field] = value
|
|
with pytest.raises(local.IapRequestError, match="expected"):
|
|
local._validate_expected_bindings(**arguments)
|
|
|
|
|
|
def test_release_and_original_artifact_require_independent_expected_values(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
release = {
|
|
"release_sha256": "a" * 64,
|
|
"image": {
|
|
"reference": f"{local.package.IMAGE_REPOSITORY}@sha256:{'b' * 64}",
|
|
"digest": f"sha256:{'b' * 64}",
|
|
"config_digest": f"sha256:{'c' * 64}",
|
|
},
|
|
"image_input": {"runtime": {"teleo_git_head": "d" * 40}},
|
|
}
|
|
monkeypatch.setattr(local.installer, "validate_release_bundle", lambda _bundle: (release, b"unit"))
|
|
with pytest.raises(local.IapRequestError, match="independently supplied"):
|
|
local.validate_release(
|
|
tmp_path,
|
|
"d" * 40,
|
|
expected_release_sha256="e" * 64,
|
|
expected_image_digest=f"sha256:{'b' * 64}",
|
|
expected_image_config_digest=f"sha256:{'c' * 64}",
|
|
)
|
|
|
|
artifact = tmp_path / "artifact.zip"
|
|
artifact.write_bytes(b"not-the-approved-artifact")
|
|
artifact.chmod(0o600)
|
|
with pytest.raises(local.IapRequestError, match="independently supplied"):
|
|
local.validate_release_artifact(
|
|
artifact,
|
|
bundle=tmp_path,
|
|
release=release,
|
|
expected_artifact_sha256="f" * 64,
|
|
)
|
|
|
|
|
|
def test_dry_run_plan_records_only_bounded_action_hashes(tmp_path: Path) -> None:
|
|
plan = executable_plan(tmp_path, "install")
|
|
encoded = json.dumps(plan, sort_keys=True)
|
|
|
|
assert [item["label"] for item in plan["actions"]] == [
|
|
"upload_request",
|
|
"execute_remote_request",
|
|
"cleanup_failed_transfer",
|
|
]
|
|
assert all(set(item) == {"label", "argv_sha256"} for item in plan["actions"])
|
|
assert local.BOOTSTRAP_CODE not in encoded
|
|
assert "--command=" not in encoded
|
|
assert "sudo --" not in encoded
|
|
|
|
|
|
def test_gcloud_environment_selects_absolute_compatible_python(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
home = private_directory(tmp_path / "home")
|
|
config = private_directory(home / "gcloud")
|
|
python = tmp_path / "python311"
|
|
python.write_text("placeholder", encoding="utf-8")
|
|
python.chmod(0o700)
|
|
monkeypatch.setenv("HOME", str(home))
|
|
monkeypatch.setenv("CLOUDSDK_CONFIG", str(config))
|
|
monkeypatch.setenv("CLOUDSDK_PYTHON", str(python))
|
|
|
|
environment = local.gcloud_environment(
|
|
probe_runner=lambda argv, **_kwargs: completed(argv, stdout=b"Python 3.11.9\n")
|
|
)
|
|
|
|
assert environment["CLOUDSDK_PYTHON"] == str(python.resolve())
|
|
assert environment["CLOUDSDK_CONFIG"] == str(config.resolve())
|
|
|
|
|
|
def test_local_result_receipt_is_private_and_self_hashed(tmp_path: Path) -> None:
|
|
plan = executable_plan(tmp_path, "verify")
|
|
result = {"bounded": "remote-result"}
|
|
|
|
path, receipt = local._write_result_receipt(plan, result)
|
|
stable = {key: value for key, value in receipt.items() if key != "receipt_sha256"}
|
|
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
assert receipt["receipt_sha256"] == local.package.canonical_sha256(stable)
|
|
assert json.loads(path.read_text(encoding="utf-8")) == receipt
|
|
|
|
|
|
def test_execute_failure_does_not_forward_command_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
plan = executable_plan(tmp_path, "install")
|
|
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
return completed(argv, returncode=1, stderr=b"password=extremely-secret")
|
|
|
|
with pytest.raises(local.IapRequestError) as caught:
|
|
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
|
|
assert "extremely-secret" not in str(caught.value)
|
|
assert "password" not in str(caught.value).casefold()
|
|
|
|
|
|
def test_execute_rejects_unsanitized_remote_result(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
plan = executable_plan(tmp_path, "verify")
|
|
monkeypatch.setattr(local, "validate_remote_main", lambda *_args, **_kwargs: None)
|
|
calls = 0
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
return completed(argv)
|
|
payload = {
|
|
"schema": local.REMOTE_RESULT_SCHEMA,
|
|
"status": "pass",
|
|
"operation": "verify",
|
|
"request_id": "iap-abcdefghijkl",
|
|
"access_token": "not-allowed",
|
|
}
|
|
return completed(argv, stdout=json.dumps(payload).encode())
|
|
|
|
with pytest.raises(local.IapRequestError) as caught:
|
|
local.execute_plan(plan, runner=runner, gcloud_env={"PATH": "/usr/bin"})
|
|
assert "not-allowed" not in str(caught.value)
|
|
|
|
|
|
def test_execute_rederives_commands_and_rejects_plan_tampering(tmp_path: Path) -> None:
|
|
plan = executable_plan(tmp_path, "verify")
|
|
plan["actions"][1]["argv_sha256"] = "0" * 64
|
|
|
|
with pytest.raises(local.IapRequestError, match="fixed IAP plan"):
|
|
local.execute_plan(plan, runner=lambda argv, **_kwargs: completed(argv))
|
|
|
|
|
|
@pytest.mark.parametrize("fail_pull", [False, True])
|
|
def test_ephemeral_registry_credentials_are_always_removed(tmp_path: Path, fail_pull: bool) -> None:
|
|
calls: list[tuple[list[str], bytes | None]] = []
|
|
|
|
def runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
input_bytes = kwargs.get("input")
|
|
calls.append((argv, input_bytes))
|
|
if argv[0] == "/usr/bin/curl":
|
|
return completed(argv, stdout=b'{"access_token":"abcdefghijklmnop","token_type":"Bearer"}')
|
|
if "pull" in argv and fail_pull:
|
|
return completed(argv, returncode=1, stderr=b"registry said secret")
|
|
return completed(argv)
|
|
|
|
reference = f"{remote.IMAGE_REPOSITORY}@sha256:{'a' * 64}"
|
|
if fail_pull:
|
|
with pytest.raises(remote.RemoteError, match="immutable image pull"):
|
|
remote.pull_immutable_image(reference, tmp_path, runner=runner)
|
|
else:
|
|
remote.pull_immutable_image(reference, tmp_path, runner=runner)
|
|
assert not any(path.name.startswith("docker-auth-") for path in tmp_path.iterdir())
|
|
assert all("abcdefghijklmnop" not in " ".join(argv) for argv, _input in calls)
|
|
assert any(input_bytes == b"abcdefghijklmnop\n" for _argv, input_bytes in calls)
|
|
assert any("logout" in argv for argv, _input in calls)
|
|
|
|
|
|
def test_metadata_target_binds_project_zone_instance_and_service_account() -> None:
|
|
values = {
|
|
"instance/service-accounts/default/email": remote.SERVICE_ACCOUNT,
|
|
"project/project-id": remote.PROJECT,
|
|
"instance/name": remote.INSTANCE,
|
|
"instance/zone": f"projects/123/zones/{remote.ZONE}",
|
|
}
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
path = argv[-1].removeprefix(f"{remote.METADATA}/")
|
|
return completed(argv, stdout=values[path].encode())
|
|
|
|
remote.verify_service_account(runner=runner)
|
|
|
|
values["instance/name"] = "wrong-instance"
|
|
with pytest.raises(remote.RemoteError, match="target or service account"):
|
|
remote.verify_service_account(runner=runner)
|
|
|
|
|
|
def test_outer_lock_and_reused_request_id_fail_closed(tmp_path: Path) -> None:
|
|
lock = tmp_path / "runner.lock"
|
|
first = remote._acquire_lock(lock, expected_uid=os.geteuid())
|
|
try:
|
|
with pytest.raises(remote.RemoteError, match="another request"):
|
|
remote._acquire_lock(lock, expected_uid=os.geteuid())
|
|
finally:
|
|
os.close(first)
|
|
|
|
root = tmp_path / "requests"
|
|
claimed = remote._claim_request_directory(
|
|
root,
|
|
"iap-abcdefghijkl",
|
|
expected_uid=os.geteuid(),
|
|
trusted_parent=tmp_path,
|
|
)
|
|
assert stat.S_IMODE(claimed.stat().st_mode) == 0o700
|
|
with pytest.raises(remote.RemoteError, match="already used"):
|
|
remote._claim_request_directory(
|
|
root,
|
|
"iap-abcdefghijkl",
|
|
expected_uid=os.geteuid(),
|
|
trusted_parent=tmp_path,
|
|
)
|
|
|
|
|
|
def test_rollback_uses_only_explicit_receipt_bound_interface(tmp_path: Path) -> None:
|
|
observed: dict[str, Any] = {}
|
|
|
|
def rollback(release_sha256: str, **kwargs: Any) -> dict[str, Any]:
|
|
observed.update({"release_sha256": release_sha256, **kwargs})
|
|
return {"schema": "rollback", "status": "pass"}
|
|
|
|
installer = SimpleNamespace(rollback_release=rollback)
|
|
result = remote._execute_explicit_rollback(
|
|
installer_module=installer,
|
|
release_sha256="a" * 64,
|
|
paths=object(),
|
|
source_binding={"revision": "b" * 40},
|
|
runner=lambda argv, **_kwargs: completed(argv),
|
|
)
|
|
|
|
assert result["status"] == "pass"
|
|
assert observed["release_sha256"] == "a" * 64
|
|
assert observed["execute_rollback"] is True
|
|
assert "install_release" not in observed
|
|
|
|
|
|
def test_rollback_fails_when_explicit_interface_is_absent(tmp_path: Path) -> None:
|
|
with pytest.raises(remote.RemoteError, match="lacked the explicit rollback"):
|
|
remote._execute_explicit_rollback(
|
|
installer_module=SimpleNamespace(),
|
|
release_sha256="a" * 64,
|
|
paths=object(),
|
|
source_binding={},
|
|
runner=lambda argv, **_kwargs: completed(argv),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("failure_point", ["first_database", "restart", "second_database"])
|
|
def test_post_install_proof_failure_executes_explicit_rollback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
failure_point: str,
|
|
) -> None:
|
|
release = {
|
|
"release_sha256": "a" * 64,
|
|
"image": {
|
|
"reference": f"{remote.IMAGE_REPOSITORY}@sha256:{'b' * 64}",
|
|
"digest": f"sha256:{'b' * 64}",
|
|
"config_digest": f"sha256:{'c' * 64}",
|
|
},
|
|
"image_input": {"runtime": {"teleo_git_head": "d" * 40}},
|
|
}
|
|
manifest = {
|
|
"operation": "install",
|
|
"request_id": "iap-abcdefghijkl",
|
|
"runner_revision": "e" * 40,
|
|
"release_revision": "d" * 40,
|
|
"manifest_sha256": "f" * 64,
|
|
"release": {
|
|
"release_sha256": release["release_sha256"],
|
|
"image_reference": release["image"]["reference"],
|
|
"image_digest": release["image"]["digest"],
|
|
"image_config_digest": release["image"]["config_digest"],
|
|
},
|
|
"release_artifact": {"archive_sha256": "1" * 64},
|
|
"installer_source": {"revision": "e" * 40},
|
|
}
|
|
verifications = iter(
|
|
[
|
|
{
|
|
"main_pid": 10,
|
|
"invocation_id": "2" * 32,
|
|
"container_id": "3" * 64,
|
|
"n_restarts": 0,
|
|
},
|
|
{
|
|
"main_pid": 11,
|
|
"invocation_id": "4" * 32,
|
|
"container_id": "5" * 64,
|
|
"n_restarts": 0,
|
|
},
|
|
]
|
|
)
|
|
rollbacks: list[str] = []
|
|
|
|
class Paths:
|
|
@staticmethod
|
|
def under(_root: Path) -> object:
|
|
return object()
|
|
|
|
installer_module = SimpleNamespace(
|
|
SYSTEMCTL="/usr/bin/systemctl",
|
|
InstallPaths=Paths,
|
|
validate_release_bundle=lambda _bundle: (release, b"unit"),
|
|
_validate_source_binding=lambda _binding: None,
|
|
install_release=lambda *_args, **_kwargs: {"status": "pass"},
|
|
verify_running_release=lambda *_args, **_kwargs: next(verifications),
|
|
rollback_release=lambda release_sha256, **_kwargs: (
|
|
rollbacks.append(release_sha256)
|
|
or {
|
|
"schema": "livingip.leocleanNoSendRollbackResult.v1",
|
|
"status": "pass",
|
|
"mode": "executed",
|
|
"rolled_back_release_sha256": release_sha256,
|
|
"restored_release_sha256": None,
|
|
"restored_service_posture": "absent",
|
|
"verification": None,
|
|
}
|
|
),
|
|
)
|
|
package_module = SimpleNamespace(
|
|
render_systemd_unit=lambda _release: "unit",
|
|
canonical_sha256=local.package.canonical_sha256,
|
|
)
|
|
monkeypatch.setattr(remote, "validate_original_release_artifact", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(remote, "_load_contract_modules", lambda _root: (package_module, installer_module))
|
|
monkeypatch.setattr(remote, "pull_immutable_image", lambda *_args, **_kwargs: None)
|
|
database_calls = 0
|
|
|
|
def verify_database(**_kwargs: Any) -> dict[str, Any]:
|
|
nonlocal database_calls
|
|
database_calls += 1
|
|
if failure_point == "first_database" and database_calls == 1:
|
|
raise remote.RemoteError("database failed")
|
|
if failure_point == "second_database" and database_calls == 2:
|
|
raise remote.RemoteError("database failed")
|
|
return {"status": "pass"}
|
|
|
|
monkeypatch.setattr(remote, "_verify_database", verify_database)
|
|
|
|
def runner(argv: list[str], **_kwargs: Any) -> subprocess.CompletedProcess[bytes]:
|
|
if failure_point == "restart" and argv[:2] == ["/usr/bin/systemctl", "restart"]:
|
|
return completed(argv, returncode=1)
|
|
return completed(argv)
|
|
|
|
with pytest.raises(remote.RemoteError, match="install_failed_rolled_back"):
|
|
remote.execute_request(
|
|
manifest,
|
|
tmp_path,
|
|
tmp_path,
|
|
"6" * 64,
|
|
"7" * 64,
|
|
runner=runner,
|
|
)
|
|
assert rollbacks == [release["release_sha256"]]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("posture", "restored_release", "verification"),
|
|
[
|
|
(
|
|
"active",
|
|
"e" * 64,
|
|
{
|
|
"active_state": "active",
|
|
"sub_state": "running",
|
|
"container_id": "f" * 64,
|
|
},
|
|
),
|
|
("inactive", "e" * 64, None),
|
|
("absent", None, None),
|
|
],
|
|
)
|
|
def test_rollback_result_accepts_exact_recorded_postures(
|
|
posture: str,
|
|
restored_release: str | None,
|
|
verification: dict[str, Any] | None,
|
|
) -> None:
|
|
manifest = {
|
|
"request_id": "iap-abcdefghijkl",
|
|
"runner_revision": "a" * 40,
|
|
"release_revision": "d" * 40,
|
|
"manifest_sha256": "1" * 64,
|
|
"release_artifact": {"archive_sha256": "2" * 64},
|
|
}
|
|
release = {
|
|
"release_sha256": "b" * 64,
|
|
"image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"},
|
|
}
|
|
rollback = {
|
|
"schema": "livingip.leocleanNoSendRollbackResult.v1",
|
|
"status": "pass",
|
|
"mode": "executed",
|
|
"rolled_back_release_sha256": "b" * 64,
|
|
"restored_release_sha256": restored_release,
|
|
"restored_service_posture": posture,
|
|
"verification": verification,
|
|
}
|
|
|
|
result = remote._rollback_result(
|
|
manifest=manifest,
|
|
release=release,
|
|
rollback=rollback,
|
|
archive_sha256="3" * 64,
|
|
helper_sha256="4" * 64,
|
|
)
|
|
|
|
assert result["service"]["restored_posture"] == posture
|
|
assert result["database"] == {"status": "not_run_after_explicit_rollback"}
|
|
|
|
|
|
def test_absent_rollback_rejects_invented_database_or_runtime_verification() -> None:
|
|
manifest = {
|
|
"request_id": "iap-abcdefghijkl",
|
|
"runner_revision": "a" * 40,
|
|
"release_revision": "d" * 40,
|
|
"manifest_sha256": "1" * 64,
|
|
"release_artifact": {"archive_sha256": "2" * 64},
|
|
}
|
|
release = {
|
|
"release_sha256": "b" * 64,
|
|
"image": {"digest": f"sha256:{'c' * 64}", "config_digest": f"sha256:{'d' * 64}"},
|
|
}
|
|
rollback = {
|
|
"schema": "livingip.leocleanNoSendRollbackResult.v1",
|
|
"status": "pass",
|
|
"mode": "executed",
|
|
"rolled_back_release_sha256": "b" * 64,
|
|
"restored_release_sha256": None,
|
|
"restored_service_posture": "absent",
|
|
"verification": {"active_state": "active"},
|
|
}
|
|
with pytest.raises(remote.RemoteError, match="absent rollback"):
|
|
remote._rollback_result(
|
|
manifest=manifest,
|
|
release=release,
|
|
rollback=rollback,
|
|
archive_sha256="3" * 64,
|
|
helper_sha256="4" * 64,
|
|
)
|