"""tgit-runner against a fake docker: argument validation, archive checks, sandbox flags, rollback.""" import io import json import shutil import subprocess import tarfile from collections.abc import Callable from pathlib import Path import pytest from tests.conftest import ROOT RUNNER = ROOT / "infra" / "runner" / "tgit-runner" FAKE = Path(__file__).with_name("fake_docker.py") pytestmark = pytest.mark.skipif(shutil.which("bash") is None, reason="needs bash") def tar_of(files: dict[str, str | bytes], extra: Callable[[tarfile.TarFile], None] | None = None) -> bytes: buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: for name, content in files.items(): data = content.encode() if isinstance(content, str) else content info = tarfile.TarInfo(name) info.size = len(data) tar.addfile(info, io.BytesIO(data)) if extra: extra(tar) return buf.getvalue() HELLO = tar_of({"main.py": "print('hi')\n"}) class Runner: def __init__(self, tmp: Path) -> None: self.tmp = tmp bin_dir = tmp / "bin" bin_dir.mkdir() (bin_dir / "docker").symlink_to(FAKE) self.env = { "PATH": f"{bin_dir}:/usr/bin:/bin", "FAKE_DOCKER_STATE": str(tmp / "state.json"), "FAKE_DOCKER_LOG": str(tmp / "calls.log"), "TGIT_RUNNER_STATE": str(tmp / "runner"), "TGIT_HEALTH_TIMEOUT": "4", "TGIT_SETTLE": "0", } def __call__( self, *args: str, stdin: bytes = b"", ssh: str | None = None, **env: str ) -> subprocess.CompletedProcess[str]: e = {**self.env, **env} if ssh is not None: e["SSH_ORIGINAL_COMMAND"] = ssh return subprocess.run( ["bash", str(RUNNER), *args], input=stdin.decode("latin-1"), env=e, capture_output=True, text=True, timeout=30, encoding="latin-1", ) def calls(self) -> list[list[str]]: path = self.tmp / "calls.log" return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def state(self) -> dict[str, object]: return json.loads((self.tmp / "state.json").read_text()) @pytest.fixture def runner(tmp_path: Path) -> Runner: return Runner(tmp_path) def last(out: subprocess.CompletedProcess[str]) -> str: return out.stdout.strip().splitlines()[-1] @pytest.mark.parametrize( ("ssh", "reason"), [ ("", "usage"), ("rm -rf /", "unknown command"), ("deploy hello; rm -rf /", "wrong number of arguments"), ("deploy Hello abc1234", "bad service name"), ("deploy ../etc abc1234", "bad service name"), ("deploy hello $(id)", "bad sha"), ("deploy hello abc1234 extra", "wrong number of arguments"), ("restart *", "bad service name"), ("list now", "list takes no arguments"), ], ) def test_bad_requests_rejected_before_docker(runner: Runner, ssh: str, reason: str) -> None: out = runner(ssh=ssh) assert out.returncode == 2 assert reason in last(out) assert runner.calls() == [] def test_deploy_runs_in_sandbox(runner: Runner) -> None: out = runner(ssh="deploy hello abc1234", stdin=tar_of({"main.py": "x", "tgit.yaml": "network: true\n"})) assert out.returncode == 0, out.stdout + out.stderr assert last(out) == "OK deployed hello abc1234 network=none healthcheck=no" # network not allowed by admin [run] = [c for c in runner.calls() if c[0] == "run"] flags = " ".join(run) for flag in [ "--user 10001:10001", "--read-only", "--cap-drop ALL", "--security-opt no-new-privileges", "--memory 128m --memory-swap 128m", "--cpus 0.25", "--pids-limit 64", "--network none", "--label tgit.managed=true", "--label tgit.sha=abc1234", ]: assert flag in flags assert "/var/run/docker.sock" not in flags assert list(runner.state()["containers"]) == ["tgit-svc-hello"] # -next renamed into place def test_egress_only_when_admin_allows(runner: Runner) -> None: (runner.tmp / "runner").mkdir() (runner.tmp / "runner" / "egress-allow").write_text("weather\n") out = runner(ssh="deploy weather abc1234", stdin=tar_of({"main.py": "x", "tgit.yaml": "network: true\n"})) assert "network=tgit-svc-egress" in last(out) def test_healthcheck_from_tgit_yaml_without_a_shell(runner: Runner) -> None: yaml = "health_url: http://localhost:8080/health # probe\n" out = runner(ssh="deploy web abc1234", stdin=tar_of({"main.py": "x", "tgit.yaml": yaml})) assert out.returncode == 0 dockerfile = runner.state()["dockerfiles"]["tgit-svc-web:abc1234"] # type: ignore[index] assert 'CMD ["python", "-c", "import sys, urllib.request' in dockerfile assert '"http://localhost:8080/health"]' in dockerfile @pytest.mark.parametrize( "url", ["http://evil.com/x", "http://localhost:80/$(id)", 'http://localhost:80/"]; RUN id #', "file:///etc/passwd"] ) def test_health_url_must_be_local(runner: Runner, url: str) -> None: out = runner(ssh="deploy web abc1234", stdin=tar_of({"main.py": "x", "tgit.yaml": f"health_url: {url}\n"})) assert out.returncode == 2 and "health_url" in last(out) assert not [c for c in runner.calls() if c[0] == "build"] def symlink_to_etc(tar: tarfile.TarFile) -> None: info = tarfile.TarInfo("link") info.type = tarfile.SYMTYPE info.linkname = "/etc/passwd" tar.addfile(info) @pytest.mark.parametrize( ("archive", "reason"), [ (b"not a tar", "not a tar archive"), (tar_of({"main.py": "x"}, symlink_to_etc), "only files and directories"), (tar_of({"main.py": "x", "../escape.py": "x"}), "unsafe path"), (tar_of({"lib.py": "x"}), "main.py is missing"), ], ) def test_bad_archives(runner: Runner, archive: bytes, reason: str) -> None: out = runner(ssh="deploy hello abc1234", stdin=archive) assert out.returncode == 2 and reason in last(out) def test_archive_size_limit(runner: Runner) -> None: out = runner(ssh="deploy hello abc1234", stdin=HELLO, TGIT_MAX_TAR="100") assert out.returncode == 2 and "larger than" in last(out) def test_unhealthy_new_version_keeps_the_old_one(runner: Runner) -> None: assert runner(ssh="deploy hello abc1234", stdin=HELLO).returncode == 0 out = runner(ssh="deploy hello def5678", stdin=HELLO, FAKE_HEALTH="unhealthy", FAKE_STATUS="exited") assert out.returncode == 3 assert last(out) == "FAIL not healthy within 4s, previous version kept" assert "Traceback: boom" in out.stdout # the new version's log tail for the user assert list(runner.state()["containers"]) == ["tgit-svc-hello"] assert runner.state()["containers"]["tgit-svc-hello"]["image"] == "tgit-svc-hello:abc1234" # type: ignore[index] def test_restarting_without_healthcheck_is_not_ready(runner: Runner) -> None: out = runner(ssh="deploy hello abc1234", stdin=HELLO, FAKE_RESTARTS="2") assert out.returncode == 3 and last(out) == "FAIL not healthy within 4s" def test_build_failure(runner: Runner) -> None: out = runner(ssh="deploy hello abc1234", stdin=HELLO, FAKE_BUILD_FAIL="1") assert out.returncode == 5 and "could not install requirements" in out.stdout def test_service_limit(runner: Runner) -> None: for i in range(3): assert runner(ssh=f"deploy s{i}x abc1234", stdin=HELLO, TGIT_MAX_SERVICES="3").returncode == 0 out = runner(ssh="deploy extra abc1234", stdin=HELLO, TGIT_MAX_SERVICES="3") redeploy = runner(ssh="deploy s0x def5678", stdin=HELLO, TGIT_MAX_SERVICES="3") assert out.returncode == 4 and "limit 3" in last(out) assert redeploy.returncode == 0 # an existing service may always be updated def test_old_images_pruned(runner: Runner) -> None: for sha in ("aaa1111", "bbb2222", "ccc3333"): assert runner(ssh=f"deploy hello {sha}", stdin=HELLO).returncode == 0 assert sorted(runner.state()["images"]) == ["tgit-svc-hello:bbb2222", "tgit-svc-hello:ccc3333"] # type: ignore[type-var] def test_restart_remove_list(runner: Runner) -> None: assert last(runner(ssh="restart hello")) == "FAIL no such service" runner(ssh="deploy hello abc1234", stdin=HELLO) assert last(runner(ssh="restart hello")) == "OK restarted hello" assert last(runner(ssh="list")) == "OK list" assert last(runner(ssh="remove hello")) == "OK removed hello" assert runner.state()["containers"] == {} assert runner.state()["images"] == [] def test_direct_invocation_without_ssh(runner: Runner) -> None: out = runner("list") assert out.returncode == 0 def test_glob_in_ssh_command_is_not_expanded(runner: Runner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) (tmp_path / "abc1234").write_text("") out = runner(ssh="deploy hello abc12*") assert out.returncode == 2 and "bad sha" in last(out)