import hashlib import importlib.util import io import json import os import shutil import stat import subprocess import tempfile import unittest from contextlib import redirect_stderr, redirect_stdout from tests.fakes import ROOT spec = importlib.util.spec_from_file_location("host_secret", os.path.join(ROOT, "tools", "host_secret.py")) host_secret = importlib.util.module_from_spec(spec) spec.loader.exec_module(host_secret) OLD = "old-token-0123456789abcdef" NEW = "new-token-fedcba9876543210" OTHER = "other-key-0123456789abcdef" def sha(value): return hashlib.sha256(value.encode()).hexdigest() def yml(token, key_quote=""): return f"""builds: ai-review: werkdock: env: GITEA_BOT_TOKEN: {key_quote}{token}{key_quote} OPENROUTER_API_KEY: "{OTHER}" WERKREVISOR_CONFIG_TOML: | [model] id = "m" """ ENV = f"GITEA_CLAUDE_CODE_TOKEN={OTHER}\nGITEA_BOT_TOKEN={OLD}\n" class HostSecretTest(unittest.TestCase): def setUp(self): self.home = tempfile.mkdtemp(prefix="werkrevisor-test-") self.ymls = {} for repo, content in (("alpha", yml(OLD)), ("beta", yml(OLD, '"')), ("gamma", "builds: {}\n")): directory = os.path.join(self.home, repo, ".git", "werkator") os.makedirs(directory) self.ymls[repo] = os.path.join(directory, ".werkator.yml") with open(self.ymls[repo], "w") as f: f.write(content) os.chmod(self.ymls[repo], 0o600) self.env_file = os.path.join(self.home, "secrets.env") with open(self.env_file, "w") as f: f.write(ENV) os.chmod(self.env_file, 0o600) def tearDown(self): shutil.rmtree(self.home) def run_tool(self, *argv, payload=None, now="20261020120000"): out, err = io.StringIO(), io.StringIO() with redirect_stdout(out), redirect_stderr(err): code = host_secret.main(list(argv), io.StringIO(json.dumps(payload or {})), self.home, now) self.assertNotIn(OLD, out.getvalue() + err.getvalue()) self.assertNotIn(NEW, out.getvalue() + err.getvalue()) return code, out.getvalue(), err.getvalue() def read(self, path): with open(path) as f: return f.read() def test_check_names_each_repository_with_the_key(self): code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)}) self.assertEqual((code, out), (0, "alpha: equal; 0 backups hold that value\n" "beta: equal; 0 backups hold that value\n")) code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(NEW)}) self.assertEqual((code, out), (1, "alpha: differs; 0 backups hold that value\n" "beta: differs; 0 backups hold that value\n")) def test_set_replaces_only_the_value_and_keeps_quotes_and_mode(self): code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}) self.assertEqual((code, out), (0, "alpha: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-20261020120000\n" "beta: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-20261020120000\n" f"old value: sha256 {sha(OLD)}\n")) self.assertEqual(self.read(self.ymls["alpha"]), yml(NEW)) self.assertEqual(self.read(self.ymls["beta"]), yml(NEW, '"')) self.assertEqual(self.read(self.ymls["gamma"]), "builds: {}\n") self.assertEqual(self.read(self.ymls["alpha"] + ".bak-secret-GITEA_BOT_TOKEN-20261020120000"), yml(OLD)) for path in (self.ymls["alpha"], self.ymls["alpha"] + ".bak-secret-GITEA_BOT_TOKEN-20261020120000"): self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600) self.assertEqual(self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)})[1], "alpha: differs; 1 backups hold that value\nbeta: differs; 1 backups hold that value\n") self.assertEqual(self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW})[1], "alpha: equal\nbeta: equal\n") def test_set_in_an_env_file(self): code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", self.env_file, payload={"value": NEW}, now="1") self.assertEqual((code, out), (0, "secrets.env: replaced; backup secrets.env.bak-secret-GITEA_BOT_TOKEN-1\n" f"old value: sha256 {sha(OLD)}\n")) self.assertEqual(self.read(self.env_file), ENV.replace(OLD, NEW)) self.assertEqual(stat.S_IMODE(os.stat(self.env_file).st_mode), 0o600) def test_set_adds_a_missing_key_to_an_env_file_only(self): code, out, _ = self.run_tool("set", "WERKREVISOR_STATS_TOKEN", self.env_file, payload={"value": NEW}, now="1") self.assertEqual((code, out), (0, "secrets.env: added; backup secrets.env.bak-secret-WERKREVISOR_STATS_TOKEN-1\n")) self.assertEqual(self.read(self.env_file), ENV + f"WERKREVISOR_STATS_TOKEN={NEW}\n") code, _, err = self.run_tool("set", "WERKREVISOR_STATS_TOKEN", payload={"value": NEW}) self.assertEqual((code, err), (2, "error: no file sets WERKREVISOR_STATS_TOKEN\n")) def test_set_refuses_a_third_value(self): with open(self.ymls["beta"], "w") as f: f.write(yml(OTHER)) code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}) self.assertEqual((code, err), (2, "error: the files hold 2 different old values of GITEA_BOT_TOKEN\n")) code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW, "old_sha256": sha(OLD)}) self.assertEqual((code, err), (2, "error: a file holds another value of GITEA_BOT_TOKEN than the old one\n")) self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD)) def test_set_continues_a_rotation_that_stopped_half_way(self): with open(self.ymls["beta"], "w") as f: f.write(yml(NEW)) code, out, _ = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW, "old_sha256": sha(OLD)}, now="1") self.assertEqual((code, out), (0, "alpha: replaced; backup .werkator.yml.bak-secret-GITEA_BOT_TOKEN-1\n" f"beta: equal\nold value: sha256 {sha(OLD)}\n")) def test_set_refuses_a_value_that_would_need_quoting(self): for value in ("", "short", "with space 0123456789", 'quote"0123456789abc', "line\nbreak0123456789"): code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": value}) self.assertEqual((code, err.startswith("error: the new value")), (2, True), value) self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD)) def test_a_file_with_two_lines_for_the_key_stops_set_before_any_change(self): with open(self.ymls["beta"], "a") as f: f.write(f" GITEA_BOT_TOKEN: {OLD}\n") code, out, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}) self.assertEqual((code, out, err), (2, "", "error: 2 lines set GITEA_BOT_TOKEN\n")) self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD)) self.assertEqual(os.listdir(os.path.dirname(self.ymls["alpha"])), [".werkator.yml"]) def test_a_line_with_a_comment_is_not_changed(self): with open(self.ymls["alpha"], "w") as f: f.write(yml(OLD).replace(OLD, OLD + " # rotated")) code, _, err = self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}) self.assertEqual((code, "a form this tool does not change" in err), (2, True)) def test_purge_removes_the_backups_with_the_old_value_only(self): directory = os.path.dirname(self.ymls["alpha"]) for name, content in ((".werkator.yml.bak-before-0.54.1", yml(OLD)), (".werkator.yml.pre-0.21.0", yml(OTHER + "x")), (".werkator.yml.bak-other", "no key\n")): with open(os.path.join(directory, name), "w") as f: f.write(content) code, _, err = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)}) self.assertEqual((code, err), (2, "error: alpha still holds that value; set the new value first\n")) self.assertEqual(len(os.listdir(directory)), 4) self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}, now="2") code, out, _ = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)}) self.assertEqual((code, out), (0, "alpha: removed .werkator.yml.bak-before-0.54.1, .werkator.yml.bak-secret-GITEA_BOT_TOKEN-2\n" "beta: removed .werkator.yml.bak-secret-GITEA_BOT_TOKEN-2\n")) self.assertEqual(sorted(os.listdir(directory)), [".werkator.yml", ".werkator.yml.bak-other", ".werkator.yml.pre-0.21.0"]) def test_purge_keeps_a_file_that_only_shares_the_prefix(self): production = self.env_file + ".production" with open(production, "w") as f: f.write(ENV) with open(self.env_file + ".bak-secret-GITEA_BOT_TOKEN-1", "w") as f: f.write(ENV) self.run_tool("set", "GITEA_BOT_TOKEN", self.env_file, payload={"value": NEW}, now="2") code, out, _ = self.run_tool("purge", "GITEA_BOT_TOKEN", self.env_file, payload={"sha256": sha(OLD)}) self.assertEqual((code, out), (0, "secrets.env: removed secrets.env.bak-secret-GITEA_BOT_TOKEN-1, " "secrets.env.bak-secret-GITEA_BOT_TOKEN-2\n")) self.assertEqual(self.read(production), ENV) def test_an_unreadable_machine_config_is_an_error(self): with open(self.ymls["beta"], "wb") as f: f.write(b"\xff\xfe broken") for mode, payload in (("check", {"sha256": sha(OLD)}), ("purge", {"sha256": sha(NEW)}), ("set", {"value": NEW})): code, _, err = self.run_tool(mode, "GITEA_BOT_TOKEN", payload=payload) self.assertEqual((code, err), (2, "error: cannot read .werkator.yml: UnicodeDecodeError\n"), mode) self.assertEqual(self.read(self.ymls["alpha"]), yml(OLD)) def test_a_backup_it_cannot_parse_may_hold_the_value(self): directory = os.path.dirname(self.ymls["alpha"]) with open(os.path.join(directory, ".werkator.yml.bak-broken"), "w") as f: f.write(yml(OLD).replace(OLD, OLD + " # comment")) code, out, _ = self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)}) self.assertEqual((code, out.splitlines()[0]), (2, "alpha: equal; 0 backups hold that value; cannot read or parse .werkator.yml.bak-broken")) self.run_tool("set", "GITEA_BOT_TOKEN", payload={"value": NEW}, now="2") code, _, err = self.run_tool("purge", "GITEA_BOT_TOKEN", payload={"sha256": sha(OLD)}) self.assertEqual((code, err), (2, "error: alpha: cannot read or parse .werkator.yml.bak-broken; " "check or remove them by hand\n")) self.assertEqual(len(os.listdir(directory)), 3) def test_export_writes_the_one_value_as_json_only_into_a_pipe(self): out = io.StringIO() with redirect_stdout(out): code = host_secret.main(["export", "GITEA_BOT_TOKEN"], io.StringIO(""), self.home) self.assertEqual((code, json.loads(out.getvalue())), (0, {"value": OLD})) class Terminal(io.StringIO): def isatty(self): return True out, err = Terminal(), io.StringIO() with redirect_stdout(out), redirect_stderr(err): code = host_secret.main(["export", "GITEA_BOT_TOKEN"], io.StringIO(""), self.home) self.assertEqual((code, out.getvalue(), err.getvalue()), (2, "", "error: export writes a value; it refuses a terminal\n")) with open(self.ymls["beta"], "w") as f: f.write(yml(NEW)) self.assertEqual(self.run_tool("export", "GITEA_BOT_TOKEN")[2], "error: the files hold 2 different values of GITEA_BOT_TOKEN\n") def test_invalid_input_prints_no_payload(self): out, err = io.StringIO(), io.StringIO() with redirect_stdout(out), redirect_stderr(err): code = host_secret.main(["set", "GITEA_BOT_TOKEN"], io.StringIO(f"{{\"value\": \"{NEW}\""), self.home) self.assertEqual((code, err.getvalue()), (2, "error: the input is no JSON object with the needed field\n")) self.assertEqual(self.run_tool("check", "GITEA_BOT_TOKEN", payload={"sha256": "abc"})[2], "error: a SHA-256 must be 64 lower-case hex digits\n") self.assertEqual(self.run_tool("check", "lower")[0], 2) # Runs the command that ssh would run on the host here, with HOME as the home of the host. FAKE_SSH = """#!/usr/bin/env bash echo "$@" >> "$SSH_LOG" exec bash -c "${@: -1}" """ class RotateSecretScriptTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="werkrevisor-test-") self.home = os.path.join(self.tmp, "home") directory = os.path.join(self.home, "repo", ".git", "werkator") os.makedirs(directory) self.yml = os.path.join(directory, ".werkator.yml") with open(self.yml, "w") as f: f.write(yml(OLD)) self.env_file = os.path.join(self.tmp, "secrets.env") with open(self.env_file, "w") as f: f.write(ENV) bin_dir = os.path.join(self.tmp, "bin") os.makedirs(bin_dir) with open(os.path.join(bin_dir, "ssh"), "w") as f: f.write(FAKE_SSH) os.chmod(os.path.join(bin_dir, "ssh"), 0o755) self.env = {**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}", "HOME": self.home, "SSH_LOG": os.path.join(self.tmp, "ssh.log"), "WERKREVISOR_SECRETS_ENV": self.env_file} def tearDown(self): shutil.rmtree(self.tmp) def run_script(self, *args, stdin=""): result = subprocess.run([os.path.join(ROOT, "tools", "rotate-secret"), *args], cwd=self.tmp, env=self.env, input=stdin, capture_output=True, text=True) self.assertNotIn(OLD, result.stdout + result.stderr) self.assertNotIn(NEW, result.stdout + result.stderr) return result def ssh_calls(self): with open(self.env["SSH_LOG"]) as f: return f.read().splitlines() def test_set_then_check_then_purge(self): check = self.run_script("check", "user@host", "GITEA_BOT_TOKEN") self.assertEqual((check.returncode, check.stdout), (0, "repo: equal; 0 backups hold that value\n" "secrets.env: equal; 0 backups hold that value\n")) result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW + "\n") self.assertEqual(result.returncode, 0, result.stderr) lines = result.stdout.splitlines() self.assertRegex(lines[0], r"^repo: replaced; backup \.werkator\.yml\.bak-secret-GITEA_BOT_TOKEN-\d{14}$") self.assertEqual(lines[1], f"old value: sha256 {sha(OLD)}") self.assertRegex(lines[2], r"^secrets\.env: replaced; backup secrets\.env\.bak-secret-GITEA_BOT_TOKEN-\d{14}$") self.assertEqual(lines[3:], ["repo: equal; 0 backups hold that value", "secrets.env: equal; 0 backups hold that value", f"after the old value is revoked: tools/rotate-secret purge user@host GITEA_BOT_TOKEN {sha(OLD)}"]) with open(self.yml) as f: self.assertEqual(f.read(), yml(NEW)) with open(self.env_file) as f: self.assertEqual(f.read(), ENV.replace(OLD, NEW)) purge = self.run_script("purge", "user@host", "GITEA_BOT_TOKEN", sha(OLD)) self.assertEqual(purge.returncode, 0, purge.stderr) self.assertEqual(sorted(os.listdir(os.path.dirname(self.yml))), [".werkator.yml"]) self.assertEqual(sorted(os.listdir(self.tmp)), ["bin", "home", "secrets.env", "ssh.log"]) calls = self.ssh_calls() self.assertEqual(len(calls), 4) self.assertTrue(all(call.startswith("-o BatchMode=yes user@host python3 -c ") for call in calls)) self.assertTrue(all(OLD not in call and NEW not in call for call in calls)) def test_set_takes_the_old_value_from_the_host_if_the_secrets_file_lacks_the_key(self): with open(self.env_file, "w") as f: f.write(ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", "")) result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn(f"secrets.env: added; ", result.stdout) self.assertTrue(result.stdout.endswith(f"purge user@host GITEA_BOT_TOKEN {sha(OLD)}\n")) with open(self.env_file) as f: self.assertEqual(f.read(), ENV.replace(OLD, NEW)) def test_fetch_adds_the_value_of_the_host_to_the_secrets_file(self): without = ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", "") with open(self.env_file, "w") as f: f.write(without) result = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN") self.assertEqual(result.returncode, 0, result.stderr) self.assertRegex(result.stdout, r"^secrets\.env: added; backup secrets\.env\.bak-secret-GITEA_BOT_TOKEN-\d{14}\n$") with open(self.env_file) as f: self.assertEqual(f.read(), without + f"GITEA_BOT_TOKEN={OLD}\n") again = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN") self.assertEqual((again.returncode, again.stderr), (2, "secrets.env sets GITEA_BOT_TOKEN already\n")) def test_fetch_stops_if_the_host_holds_no_single_value(self): without = ENV.replace(f"GITEA_BOT_TOKEN={OLD}\n", "") with open(self.env_file, "w") as f: f.write(without) other = os.path.join(self.home, "other", ".git", "werkator") os.makedirs(other) with open(os.path.join(other, ".werkator.yml"), "w") as f: f.write(yml(NEW)) result = self.run_script("fetch", "user@host", "GITEA_BOT_TOKEN") self.assertEqual((result.returncode, result.stderr), (2, "error: the files hold 2 different values of GITEA_BOT_TOKEN\n")) with open(self.env_file) as f: self.assertEqual(f.read(), without) def test_a_secrets_file_without_the_key_is_an_error_with_2(self): self.assertEqual(self.run_script("probe", "WERKREVISOR_STATS_TOKEN").returncode, 2) self.assertEqual(self.run_script("digest", "WERKREVISOR_STATS_TOKEN").returncode, 2) def test_set_refuses_if_the_host_holds_another_value_than_the_secrets_file(self): with open(self.yml, "w") as f: f.write(yml(OTHER)) result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW) self.assertEqual((result.returncode, result.stderr), (2, "error: a file holds another value of GITEA_BOT_TOKEN than the old one\n")) with open(self.env_file) as f: self.assertEqual(f.read(), ENV) def test_set_stops_before_the_host_if_the_secrets_file_cannot_be_parsed(self): with open(self.env_file, "w") as f: f.write(ENV + f"GITEA_BOT_TOKEN={OLD}\n") result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=NEW) self.assertEqual((result.returncode, result.stderr), (2, "2 lines set GITEA_BOT_TOKEN\n")) self.assertFalse(os.path.exists(self.env["SSH_LOG"])) with open(self.yml) as f: self.assertEqual(f.read(), yml(OLD)) def test_set_refuses_the_value_of_the_secrets_file(self): result = self.run_script("set", "user@host", "GITEA_BOT_TOKEN", stdin=OLD) self.assertEqual((result.returncode, result.stderr), (2, "the secrets file holds this value already\n")) self.assertFalse(os.path.exists(self.env["SSH_LOG"])) def test_check_reports_another_value(self): result = self.run_script("check", "user@host", "GITEA_BOT_TOKEN", sha(NEW)) self.assertEqual((result.returncode, result.stdout), (1, "repo: differs; 0 backups hold that value\n" "secrets.env: differs; 0 backups hold that value\n")) def test_digest(self): self.assertEqual(self.run_script("digest", "GITEA_BOT_TOKEN").stdout, sha(OLD)) def test_an_unreachable_host_is_an_error_with_2(self): bin_dir = os.path.join(self.tmp, "bin") for name, script in (("ssh", "exit 255"), ("sleep", "exit 0")): with open(os.path.join(bin_dir, name), "w") as f: f.write(f"#!/usr/bin/env bash\n{script}\n") os.chmod(os.path.join(bin_dir, name), 0o755) result = self.run_script("check", "user@host", "GITEA_BOT_TOKEN") self.assertEqual(result.returncode, 2) self.assertTrue(result.stderr.endswith("the host refused six connections\n"), result.stderr) def test_usage(self): self.assertEqual(self.run_script("--help").returncode, 0) for args in (("set", "user@host"), ("probe", "GITEA_BOT_TOKEN", "x"), ("check", "user@host", "lower"), ("rotate",)): self.assertEqual(self.run_script(*args).returncode, 2, args) self.assertFalse(os.path.exists(self.env["SSH_LOG"])) if __name__ == "__main__": unittest.main()