From 21d03b7cc9adfeea54af9e8bad95fe67120fab37 Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 15:51:50 +0000 Subject: [PATCH 1/7] feat: preflight git-gate host key population When a git-gate repo entry has no host_key configured, the prepare step now either errors (headless) or fetches the key via ssh-keyscan, shows it to the operator for confirmation, and optionally persists it to the bottle config .md file on disk. Closes #333 Co-Authored-By: Claude Sonnet 4.6 --- bot_bottle/backend/__init__.py | 10 + bot_bottle/cli/start.py | 1 + bot_bottle/git_gate_host_key.py | 279 ++++++++++++++++++ tests/unit/test_git_gate_host_key.py | 404 +++++++++++++++++++++++++++ 4 files changed, 694 insertions(+) create mode 100644 bot_bottle/git_gate_host_key.py create mode 100644 tests/unit/test_git_gate_host_key.py diff --git a/bot_bottle/backend/__init__.py b/bot_bottle/backend/__init__.py index caf8a8f..ed8b894 100644 --- a/bot_bottle/backend/__init__.py +++ b/bot_bottle/backend/__init__.py @@ -75,6 +75,9 @@ class BottleSpec: # Ordered bottle names selected at launch (issue #269). When non-empty # they are merged in order and replace the agent's `bottle:` field. bottle_names: tuple[str, ...] = () + # True when launched via --headless (no TTY, no interactive prompts). + # The git-gate host-key preflight uses this to error rather than prompt. + headless: bool = False @dataclass(frozen=True) @@ -300,6 +303,13 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]): self._preflight() + from ..git_gate_host_key import preflight_host_keys + manifest = preflight_host_keys( + manifest, + headless=spec.headless, + home_md=spec.manifest.home_md, + ) + manifest_bottle = manifest.bottle manifest_agent_provider = manifest_bottle.agent_provider agent_provider = get_provider(manifest_agent_provider.template) diff --git a/bot_bottle/cli/start.py b/bot_bottle/cli/start.py index a71338e..5a22ce0 100644 --- a/bot_bottle/cli/start.py +++ b/bot_bottle/cli/start.py @@ -209,6 +209,7 @@ def _start_headless( label=label, color=args.color or "", bottle_names=bottle_names, + headless=True, ) return _launch_bottle( spec, diff --git a/bot_bottle/git_gate_host_key.py b/bot_bottle/git_gate_host_key.py new file mode 100644 index 0000000..5a0da4f --- /dev/null +++ b/bot_bottle/git_gate_host_key.py @@ -0,0 +1,279 @@ +"""Preflight host-key population for git-gate upstreams (issue #333). + +When a git-gate repo entry lacks a `host_key`, this module either: +- headless: dies with a clear config error. +- interactive: fetches the key via ssh-keyscan, prompts the operator to + confirm, and optionally persists it to the bottle config file on disk. + +Public entry point: `preflight_host_keys(manifest, headless=..., home_md=...)`. +""" + +from __future__ import annotations + +import dataclasses +import re +import subprocess +import sys +from pathlib import Path + +from .log import die, info +from .manifest import Manifest, ManifestBottle +from .manifest_git import ManifestGitEntry + + +def fetch_host_key(host: str, port: str) -> str: + """Return an SSH ed25519 public key for `host`:`port` via ssh-keyscan. + + Returns the key in ` ` format (the host prefix is + stripped so the result can be stored in `host_key` and later formatted + into a known_hosts line by `git_gate_known_hosts_line`). + + Raises `RuntimeError` on subprocess failure, timeout, or no ed25519 + result. Uses only the Python stdlib (subprocess).""" + args = ["ssh-keyscan", "-t", "ed25519"] + if port and port != "22": + args += ["-p", port] + args.append(host) + try: + result = subprocess.run( + args, capture_output=True, text=True, timeout=15, + ) + except OSError as e: + raise RuntimeError( + f"ssh-keyscan: could not launch for {host}:{port}: {e}" + ) from e + except subprocess.TimeoutExpired: + raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") + + for line in result.stdout.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + # known_hosts format: "[host]:port type data" or "host type data" + # Strip the host/port prefix; keep "type data". + parts = line.split(None, 2) + if len(parts) == 3: + return f"{parts[1]} {parts[2]}" + + raise RuntimeError( + f"ssh-keyscan returned no ed25519 key for {host}:{port}." + + (f" stderr: {result.stderr.strip()!r}" if result.stderr.strip() else "") + ) + + +# --------------------------------------------------------------------------- +# Frontmatter text-level editing +# --------------------------------------------------------------------------- + + +def _insert_host_key_in_frontmatter(fm_text: str, repo_name: str, host_key: str) -> str: + """Append `host_key: ` to the named repo entry in raw frontmatter text. + + Locates the `:` block-mapping key, detects its child + indentation from the sibling keys already present, and inserts the + new line after the last child. Returns the original text if the + repo entry is not found or already has a `host_key:` line.""" + lines = fm_text.splitlines(keepends=True) + + repo_pat = re.compile(r"^(\s+)" + re.escape(repo_name) + r"\s*:\s*(\r?\n)?$") + repo_line_idx: int | None = None + repo_indent: int | None = None + for i, line in enumerate(lines): + if repo_pat.match(line): + repo_indent = len(line) - len(line.lstrip()) + repo_line_idx = i + break + + if repo_line_idx is None or repo_indent is None: + return fm_text + + child_indent: int | None = None + last_child_idx = repo_line_idx + + for i in range(repo_line_idx + 1, len(lines)): + raw = lines[i].rstrip("\r\n") + if not raw.strip(): + continue + indent = len(raw) - len(raw.lstrip()) + if indent > repo_indent: + child_indent = indent + last_child_idx = i + else: + break + + if child_indent is None: + child_indent = repo_indent + 2 + + block = "".join(lines[repo_line_idx + 1 : last_child_idx + 1]) + if re.search(r"^\s*host_key\s*:", block, re.MULTILINE): + return fm_text + + new_line = " " * child_indent + f"host_key: {host_key}\n" + lines.insert(last_child_idx + 1, new_line) + return "".join(lines) + + +def _update_frontmatter_in_file(file_text: str, repo_name: str, host_key: str) -> str: + """Apply `_insert_host_key_in_frontmatter` to the frontmatter section of + a Markdown file, leaving the body unchanged. Returns the original text + when no change is needed.""" + lines = file_text.splitlines(keepends=True) + if not lines or lines[0].rstrip("\r\n") != "---": + return file_text + + close_idx: int | None = None + for i in range(1, len(lines)): + if lines[i].rstrip("\r\n") == "---": + close_idx = i + break + if close_idx is None: + return file_text + + fm_text = "".join(lines[1:close_idx]) + body_text = "".join(lines[close_idx:]) + updated_fm = _insert_host_key_in_frontmatter(fm_text, repo_name, host_key) + if updated_fm == fm_text: + return file_text + return "---\n" + updated_fm + body_text + + +def _find_and_update_bottle_file( + bottles_dir: Path, repo_name: str, host_key: str, +) -> bool: + """Scan `bottles_dir/*.md` for the first file that directly declares + `repo_name` without a `host_key`, update it, and return True. + + Walks files in sorted order so the result is deterministic. Does NOT + follow `extends:` chains — only files that explicitly declare the repo + entry are candidates. Returns False when no such file is found.""" + if not bottles_dir.is_dir(): + return False + + from .yaml_subset import YamlSubsetError, parse_frontmatter + + for path in sorted(bottles_dir.glob("*.md")): + try: + text = path.read_text(encoding="utf-8") + fm, _body = parse_frontmatter(text) + except (OSError, YamlSubsetError): + continue + + git_gate = fm.get("git-gate") + if not isinstance(git_gate, dict): + continue + repos = git_gate.get("repos") + if not isinstance(repos, dict): + continue + repo = repos.get(repo_name) + if not isinstance(repo, dict): + continue + if repo.get("host_key"): + continue + + updated = _update_frontmatter_in_file(text, repo_name, host_key) + if updated == text: + continue + path.write_text(updated, encoding="utf-8") + info(f"wrote host_key for {repo_name!r} to {path}") + return True + + return False + + +# --------------------------------------------------------------------------- +# Interactive prompt helper +# --------------------------------------------------------------------------- + + +def _prompt_tty(message: str) -> str: + """Write `message` to stderr and read a line from /dev/tty (or stdin).""" + sys.stderr.write(message) + sys.stderr.flush() + try: + with open("/dev/tty", "r", encoding="utf-8") as tty: + return tty.readline().rstrip("\n") + except OSError: + return sys.stdin.readline().rstrip("\n") + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def preflight_host_keys( + manifest: Manifest, + *, + headless: bool, + home_md: Path | None, +) -> Manifest: + """Ensure every git-gate repo entry has a `host_key` configured. + + For entries whose `KnownHostKey` is empty: + - headless: calls `die()` with a clear message naming the repos. + - interactive: fetches the key via ssh-keyscan, shows it to the + operator, and requests confirmation. If accepted, optionally + persists it to the bottle config file on disk; the key is always + applied in memory for this launch regardless of the persistence + choice. Aborted confirmation calls `die()`. + + Returns a (possibly updated) Manifest. If all entries already have + host keys the original manifest is returned unchanged.""" + bottle = manifest.bottle + missing = [e for e in bottle.git if not e.KnownHostKey] + if not missing: + return manifest + + if headless: + names = ", ".join(repr(e.Name) for e in missing) + die( + f"git-gate: no host_key configured for repo(s) {names}. " + f"Add host_key to each bottle git-gate.repos entry, or run " + f"interactively once to have it fetched and saved automatically." + ) + + bottles_dir = (home_md / "bottles") if home_md is not None else None + updated_entries = list(bottle.git) + + for entry in missing: + host = entry.UpstreamHost + port = entry.UpstreamPort + label = f"git-gate.repos[{entry.Name!r}]" + + info(f"{label}: no host_key configured; fetching from {host}:{port}") + try: + key = fetch_host_key(host, port) + except RuntimeError as e: + die(f"git-gate: {label}: {e}") + + sys.stderr.write(f"\ngit-gate: host key for {label}:\n {key}\n\n") + confirm = _prompt_tty("Is this host key correct? [y/N] ") + if confirm.strip().lower() not in ("y", "yes"): + die(f"git-gate: {label}: host key not confirmed; aborting launch") + + if bottles_dir is not None: + save = _prompt_tty( + f"Save host_key for {entry.Name!r} to the bottle config file? [y/N] " + ) + if save.strip().lower() in ("y", "yes"): + ok = _find_and_update_bottle_file(bottles_dir, entry.Name, key) + if not ok: + sys.stderr.write( + f"git-gate: {label}: bottle file not found or not writable; " + f"host_key kept in memory for this session only\n" + ) + + idx = next(i for i, e in enumerate(updated_entries) if e.Name == entry.Name) + updated_entries[idx] = dataclasses.replace(entry, KnownHostKey=key) + + updated_bottle = dataclasses.replace(bottle, git=tuple(updated_entries)) + return dataclasses.replace(manifest, bottle=updated_bottle) + + +__all__ = [ + "fetch_host_key", + "preflight_host_keys", + "_insert_host_key_in_frontmatter", + "_find_and_update_bottle_file", + "_update_frontmatter_in_file", +] diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py new file mode 100644 index 0000000..ff7118f --- /dev/null +++ b/tests/unit/test_git_gate_host_key.py @@ -0,0 +1,404 @@ +"""Unit tests for git_gate_host_key (issue #333).""" + +from __future__ import annotations + +import dataclasses +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from bot_bottle.git_gate_host_key import ( + _find_and_update_bottle_file, + _insert_host_key_in_frontmatter, + _update_frontmatter_in_file, + fetch_host_key, + preflight_host_keys, +) +from bot_bottle.manifest import Manifest, ManifestIndex + + +# --------------------------------------------------------------------------- +# fetch_host_key +# --------------------------------------------------------------------------- + + +class TestFetchHostKey(unittest.TestCase): + def _run_result(self, stdout: str, returncode: int = 0) -> MagicMock: + r = MagicMock() + r.stdout = stdout + r.stderr = "" + r.returncode = returncode + return r + + def test_returns_type_and_key(self) -> None: + stdout = "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("gitea.example.com", "22") + self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key) + + def test_non_default_port_passes_p_flag(self) -> None: + stdout = "[gitea.example.com]:30009 ssh-ed25519 AAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run: + fetch_host_key("gitea.example.com", "30009") + args = mock_run.call_args[0][0] + self.assertIn("-p", args) + self.assertIn("30009", args) + + def test_default_port_omits_p_flag(self) -> None: + stdout = "github.com ssh-ed25519 AAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)) as mock_run: + fetch_host_key("github.com", "22") + args = mock_run.call_args[0][0] + self.assertNotIn("-p", args) + + def test_skips_comment_lines(self) -> None: + stdout = "# comment\ngithub.com ssh-ed25519 AAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("github.com", "22") + self.assertEqual("ssh-ed25519 AAAA", key) + + def test_raises_on_empty_output(self) -> None: + with patch("subprocess.run", return_value=self._run_result("")): + with self.assertRaises(RuntimeError) as cm: + fetch_host_key("example.com", "22") + self.assertIn("no ed25519 key", str(cm.exception)) + + def test_raises_on_os_error(self) -> None: + with patch("subprocess.run", side_effect=OSError("not found")): + with self.assertRaises(RuntimeError) as cm: + fetch_host_key("example.com", "22") + self.assertIn("could not launch", str(cm.exception)) + + def test_raises_on_timeout(self) -> None: + import subprocess + with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("ssh-keyscan", 15)): + with self.assertRaises(RuntimeError) as cm: + fetch_host_key("example.com", "22") + self.assertIn("timed out", str(cm.exception)) + + +# --------------------------------------------------------------------------- +# _insert_host_key_in_frontmatter +# --------------------------------------------------------------------------- + + +class TestInsertHostKeyInFrontmatter(unittest.TestCase): + _FM = """\ +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +""" + + def test_inserts_host_key_after_last_child(self) -> None: + result = _insert_host_key_in_frontmatter(self._FM, "myrepo", "ssh-ed25519 AAAA") + self.assertIn(" host_key: ssh-ed25519 AAAA", result) + + def test_no_change_when_repo_not_found(self) -> None: + result = _insert_host_key_in_frontmatter(self._FM, "other", "ssh-ed25519 AAAA") + self.assertEqual(self._FM, result) + + def test_no_change_when_host_key_already_present(self) -> None: + fm = self._FM + " host_key: ssh-ed25519 EXISTING\n" + result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 NEW") + self.assertNotIn("NEW", result) + self.assertEqual(fm, result) + + def test_preserves_other_repos(self) -> None: + fm = """\ +git-gate: + repos: + first: + url: ssh://git@host/org/first.git + key: + provider: static + path: /dev/null + second: + url: ssh://git@host/org/second.git + key: + provider: static + path: /dev/null +""" + result = _insert_host_key_in_frontmatter(fm, "first", "ssh-ed25519 AAAA") + self.assertIn("host_key: ssh-ed25519 AAAA", result) + self.assertIn("second:", result) + + def test_detects_child_indent_from_existing_keys(self) -> None: + # 4-space indent (non-standard but valid) + fm = """\ +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +""" + result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") + self.assertIn(" host_key: ssh-ed25519 AAAA", result) + + +# --------------------------------------------------------------------------- +# _update_frontmatter_in_file +# --------------------------------------------------------------------------- + + +class TestUpdateFrontmatterInFile(unittest.TestCase): + _FILE = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +--- +# Body text here +""" + + def test_updates_frontmatter_preserves_body(self) -> None: + result = _update_frontmatter_in_file(self._FILE, "myrepo", "ssh-ed25519 AAAA") + self.assertIn("host_key: ssh-ed25519 AAAA", result) + self.assertIn("# Body text here", result) + + def test_no_change_without_frontmatter(self) -> None: + text = "No frontmatter here\n" + result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA") + self.assertEqual(text, result) + + def test_no_change_when_repo_not_in_file(self) -> None: + result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA") + self.assertEqual(self._FILE, result) + + +# --------------------------------------------------------------------------- +# _find_and_update_bottle_file +# --------------------------------------------------------------------------- + + +class TestFindAndUpdateBottleFile(unittest.TestCase): + def _write_bottle(self, d: str, name: str, content: str) -> Path: + path = Path(d) / f"{name}.md" + path.write_text(content, encoding="utf-8") + return path + + def test_updates_file_with_matching_repo(self) -> None: + with tempfile.TemporaryDirectory() as d: + content = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +--- +""" + path = self._write_bottle(d, "dev", content) + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertTrue(result) + updated = path.read_text() + self.assertIn("host_key: ssh-ed25519 AAAA", updated) + + def test_returns_false_when_no_matching_file(self) -> None: + with tempfile.TemporaryDirectory() as d: + content = """\ +--- +git-gate: + repos: + other: + url: ssh://git@host/org/other.git + key: + provider: static + path: /dev/null +--- +""" + self._write_bottle(d, "dev", content) + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + + def test_skips_file_that_already_has_host_key(self) -> None: + with tempfile.TemporaryDirectory() as d: + content = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null + host_key: ssh-ed25519 EXISTING +--- +""" + path = self._write_bottle(d, "dev", content) + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 NEW") + self.assertFalse(result) + self.assertNotIn("NEW", path.read_text()) + + def test_returns_false_for_nonexistent_dir(self) -> None: + result = _find_and_update_bottle_file( + Path("/nonexistent/dir"), "myrepo", "ssh-ed25519 AAAA" + ) + self.assertFalse(result) + + +# --------------------------------------------------------------------------- +# preflight_host_keys +# --------------------------------------------------------------------------- + + +def _manifest_with_git(host_key: str = "") -> Manifest: + """Build a Manifest with one git entry; host_key defaults to empty.""" + return ManifestIndex.from_json_obj({ + "bottles": { + "dev": { + "git-gate": { + "repos": { + "myrepo": { + "url": "ssh://git@gitea.example.com:30009/org/myrepo.git", + "key": {"provider": "static", "path": "/dev/null"}, + **({"host_key": host_key} if host_key else {}), + }, + }, + }, + }, + }, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }).load_for_agent("demo") + + +class TestPreflightHostKeys(unittest.TestCase): + def test_no_op_when_all_keys_present(self) -> None: + manifest = _manifest_with_git("ssh-ed25519 AAAA") + result = preflight_host_keys(manifest, headless=False, home_md=None) + self.assertIs(result, manifest) + + def test_headless_dies_when_key_missing(self) -> None: + manifest = _manifest_with_git() + from bot_bottle.log import die as real_die + with self.assertRaises(SystemExit): + preflight_host_keys(manifest, headless=True, home_md=None) + + def test_interactive_fetches_and_confirms_key(self) -> None: + manifest = _manifest_with_git() + replies = iter(["y", "n"]) # confirm key, don't persist + + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key._prompt_tty", + side_effect=lambda msg: next(replies), + ), patch("sys.stderr"): + result = preflight_host_keys(manifest, headless=False, home_md=None) + + self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey) + + def test_interactive_dies_when_key_not_confirmed(self) -> None: + manifest = _manifest_with_git() + + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key._prompt_tty", + return_value="n", + ), patch("sys.stderr"), self.assertRaises(SystemExit): + preflight_host_keys(manifest, headless=False, home_md=None) + + def test_interactive_persists_when_requested(self) -> None: + manifest = _manifest_with_git() + replies = iter(["y", "y"]) # confirm key, yes persist + + with tempfile.TemporaryDirectory() as home: + bottles_dir = Path(home) / "bottles" + bottles_dir.mkdir() + bottle_file = bottles_dir / "dev.md" + bottle_file.write_text( + "---\ngit-gate:\n repos:\n myrepo:\n" + " url: ssh://git@gitea.example.com:30009/org/myrepo.git\n" + " key:\n provider: static\n path: /dev/null\n---\n", + encoding="utf-8", + ) + + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key._prompt_tty", + side_effect=lambda msg: next(replies), + ), patch("sys.stderr"): + result = preflight_host_keys( + manifest, headless=False, home_md=Path(home), + ) + + self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey) + written = bottle_file.read_text() + self.assertIn("host_key: ssh-ed25519 FETCHED", written) + + def test_interactive_skips_persist_when_declined(self) -> None: + manifest = _manifest_with_git() + replies = iter(["y", "n"]) # confirm key, no persist + + with tempfile.TemporaryDirectory() as home: + bottles_dir = Path(home) / "bottles" + bottles_dir.mkdir() + bottle_file = bottles_dir / "dev.md" + bottle_file.write_text( + "---\ngit-gate:\n repos:\n myrepo:\n" + " url: ssh://git@gitea.example.com:30009/org/myrepo.git\n" + " key:\n provider: static\n path: /dev/null\n---\n", + encoding="utf-8", + ) + + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key._prompt_tty", + side_effect=lambda msg: next(replies), + ), patch("sys.stderr"): + result = preflight_host_keys( + manifest, headless=False, home_md=Path(home), + ) + + self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey) + written = bottle_file.read_text() + self.assertNotIn("host_key", written) + + def test_headless_names_all_missing_repos_in_error(self) -> None: + manifest = ManifestIndex.from_json_obj({ + "bottles": { + "dev": { + "git-gate": { + "repos": { + "alpha": { + "url": "ssh://git@host/org/alpha.git", + "key": {"provider": "static", "path": "/dev/null"}, + }, + "beta": { + "url": "ssh://git@host/org/beta.git", + "key": {"provider": "static", "path": "/dev/null"}, + }, + }, + }, + }, + }, + "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}}, + }).load_for_agent("demo") + + import io + buf = io.StringIO() + with patch("sys.stderr", buf), self.assertRaises(SystemExit): + preflight_host_keys(manifest, headless=True, home_md=None) + + +if __name__ == "__main__": + unittest.main() -- 2.52.0 From e89dffa899ceeedaf14b77e8a3ae9bf5809c85ee Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 16:03:39 +0000 Subject: [PATCH 2/7] fix: pyright/pylint clean-up and 100% branch coverage - Remove unused ManifestBottle/ManifestGitEntry imports - Add check=False to subprocess.run (pylint subprocess-run-check) - Add `from e` to TimeoutExpired re-raise (raise-missing-from) - Catch UnicodeDecodeError alongside OSError/YamlSubsetError in _find_and_update_bottle_file - Add 14 new test cases covering: malformed ssh-keyscan output lines, stderr in error messages, blank lines inside frontmatter blocks, repo entry with no children (child_indent fallback), missing closing `---` delimiter, files without git-gate section, non-dict repos section, invalid UTF-8 files, unreadable files, update producing no text change, _prompt_tty TTY read and stdin fallback, fetch failure in preflight, and the "persist failed / kept in memory" warning path Co-Authored-By: Claude Sonnet 4.6 --- bot_bottle/git_gate_host_key.py | 11 +- tests/unit/test_git_gate_host_key.py | 156 +++++++++++++++++++++++++++ 2 files changed, 161 insertions(+), 6 deletions(-) diff --git a/bot_bottle/git_gate_host_key.py b/bot_bottle/git_gate_host_key.py index 5a0da4f..922af5e 100644 --- a/bot_bottle/git_gate_host_key.py +++ b/bot_bottle/git_gate_host_key.py @@ -17,8 +17,7 @@ import sys from pathlib import Path from .log import die, info -from .manifest import Manifest, ManifestBottle -from .manifest_git import ManifestGitEntry +from .manifest import Manifest def fetch_host_key(host: str, port: str) -> str: @@ -36,14 +35,14 @@ def fetch_host_key(host: str, port: str) -> str: args.append(host) try: result = subprocess.run( - args, capture_output=True, text=True, timeout=15, + args, capture_output=True, text=True, timeout=15, check=False, ) except OSError as e: raise RuntimeError( f"ssh-keyscan: could not launch for {host}:{port}: {e}" ) from e - except subprocess.TimeoutExpired: - raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") + except subprocess.TimeoutExpired as e: + raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from e for line in result.stdout.splitlines(): line = line.strip() @@ -155,7 +154,7 @@ def _find_and_update_bottle_file( try: text = path.read_text(encoding="utf-8") fm, _body = parse_frontmatter(text) - except (OSError, YamlSubsetError): + except (OSError, UnicodeDecodeError, YamlSubsetError): continue git_gate = fm.get("git-gate") diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index ff7118f..394757e 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -11,6 +11,7 @@ from unittest.mock import MagicMock, patch from bot_bottle.git_gate_host_key import ( _find_and_update_bottle_file, _insert_host_key_in_frontmatter, + _prompt_tty, _update_frontmatter_in_file, fetch_host_key, preflight_host_keys, @@ -58,12 +59,29 @@ class TestFetchHostKey(unittest.TestCase): key = fetch_host_key("github.com", "22") self.assertEqual("ssh-ed25519 AAAA", key) + def test_skips_lines_without_three_parts(self) -> None: + # A line with only two whitespace-separated tokens is not a valid + # known_hosts entry; it should be skipped and the function should + # raise rather than return a partial result. + stdout = "malformed-line\ngithub.com ssh-ed25519 AAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("github.com", "22") + self.assertEqual("ssh-ed25519 AAAA", key) + def test_raises_on_empty_output(self) -> None: with patch("subprocess.run", return_value=self._run_result("")): with self.assertRaises(RuntimeError) as cm: fetch_host_key("example.com", "22") self.assertIn("no ed25519 key", str(cm.exception)) + def test_raises_includes_stderr_when_present(self) -> None: + r = self._run_result("") + r.stderr = "connection refused" + with patch("subprocess.run", return_value=r): + with self.assertRaises(RuntimeError) as cm: + fetch_host_key("example.com", "22") + self.assertIn("connection refused", str(cm.exception)) + def test_raises_on_os_error(self) -> None: with patch("subprocess.run", side_effect=OSError("not found")): with self.assertRaises(RuntimeError) as cm: @@ -78,6 +96,28 @@ class TestFetchHostKey(unittest.TestCase): self.assertIn("timed out", str(cm.exception)) +# --------------------------------------------------------------------------- +# _prompt_tty +# --------------------------------------------------------------------------- + + +class TestPromptTty(unittest.TestCase): + def test_reads_from_tty_when_available(self) -> None: + import io + fake_tty = io.StringIO("yes\n") + with patch("builtins.open", return_value=fake_tty), patch("sys.stderr"): + result = _prompt_tty("question: ") + self.assertEqual("yes", result) + + def test_falls_back_to_stdin_on_os_error(self) -> None: + import io + with patch("builtins.open", side_effect=OSError("no tty")), \ + patch("sys.stdin", io.StringIO("fallback\n")), \ + patch("sys.stderr"): + result = _prompt_tty("question: ") + self.assertEqual("fallback", result) + + # --------------------------------------------------------------------------- # _insert_host_key_in_frontmatter # --------------------------------------------------------------------------- @@ -141,6 +181,27 @@ git-gate: result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") self.assertIn(" host_key: ssh-ed25519 AAAA", result) + def test_blank_lines_inside_block_do_not_end_scan(self) -> None: + fm = """\ +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + + key: + provider: static + path: /dev/null +""" + result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") + self.assertIn("host_key: ssh-ed25519 AAAA", result) + + def test_repo_with_no_children_uses_default_indent(self) -> None: + # Edge case: repo entry has no child keys at all (empty block). + fm = "git-gate:\n repos:\n myrepo:\n" + result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") + # Should fall back to repo_indent+2 = 4 + self.assertIn(" host_key: ssh-ed25519 AAAA", result) + # --------------------------------------------------------------------------- # _update_frontmatter_in_file @@ -175,6 +236,11 @@ git-gate: result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA") self.assertEqual(self._FILE, result) + def test_no_change_when_no_closing_delimiter(self) -> None: + text = "---\ngit-gate:\n repos:\n myrepo:\n url: x\n" + result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA") + self.assertEqual(text, result) + # --------------------------------------------------------------------------- # _find_and_update_bottle_file @@ -248,6 +314,62 @@ git-gate: ) self.assertFalse(result) + def test_skips_file_without_git_gate_section(self) -> None: + with tempfile.TemporaryDirectory() as d: + self._write_bottle(d, "plain", "---\nenv:\n FOO: bar\n---\n") + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + + def test_skips_file_with_non_dict_repos(self) -> None: + with tempfile.TemporaryDirectory() as d: + self._write_bottle(d, "bad", "---\ngit-gate:\n repos:\n - item\n---\n") + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + + def test_skips_invalid_utf8_file(self) -> None: + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "broken.md" + path.write_bytes(b"\xff\xfe") # invalid UTF-8 + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + + def test_skips_unreadable_file(self) -> None: + import os + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "unreadable.md" + path.write_text("---\n---\n") + os.chmod(path, 0o000) + try: + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + finally: + os.chmod(path, 0o644) + + def test_skips_when_update_produces_no_change(self) -> None: + # _update_frontmatter_in_file returns the original text unchanged + # (e.g. because _insert_host_key_in_frontmatter can't locate the + # entry in the raw text). Line 174 must `continue` rather than write. + content = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +--- +""" + with tempfile.TemporaryDirectory() as d: + path = self._write_bottle(d, "dev", content) + with patch( + "bot_bottle.git_gate_host_key._update_frontmatter_in_file", + return_value=content, # no change + ): + result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + self.assertFalse(result) + self.assertNotIn("host_key", path.read_text()) + # --------------------------------------------------------------------------- # preflight_host_keys @@ -373,6 +495,40 @@ class TestPreflightHostKeys(unittest.TestCase): written = bottle_file.read_text() self.assertNotIn("host_key", written) + def test_interactive_dies_when_fetch_fails(self) -> None: + manifest = _manifest_with_git() + + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + side_effect=RuntimeError("connection refused"), + ), patch("sys.stderr"), self.assertRaises(SystemExit): + preflight_host_keys(manifest, headless=False, home_md=None) + + def test_interactive_warns_when_persist_file_not_found(self) -> None: + manifest = _manifest_with_git() + replies = iter(["y", "y"]) # confirm key, yes persist + + with tempfile.TemporaryDirectory() as home: + # bottles/ dir exists but does not contain the repo entry + bottles_dir = Path(home) / "bottles" + bottles_dir.mkdir() + + import io + buf = io.StringIO() + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key._prompt_tty", + side_effect=lambda msg: next(replies), + ), patch("sys.stderr", buf): + result = preflight_host_keys( + manifest, headless=False, home_md=Path(home), + ) + + self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey) + self.assertIn("kept in memory", buf.getvalue()) + def test_headless_names_all_missing_repos_in_error(self) -> None: manifest = ManifestIndex.from_json_obj({ "bottles": { -- 2.52.0 From f9662c88a51fc88164493b98406cca6065aeae33 Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 16:11:14 +0000 Subject: [PATCH 3/7] fix: pyright errors in test file, bump pyright floor to 1.1.411 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove unused `import dataclasses` and stray `die as real_die` import - Replace lambda side_effect callbacks with list side_effect; lambdas with untyped parameters trigger reportUnknownLambdaType in pyright >=1.1.400 — side_effect=[...] is both cleaner and type-safe - Bump pyright floor from >=1.1.300 to >=1.1.411 in requirements-dev.txt so CI installs a version that matches what the lint job runs Co-Authored-By: Claude Sonnet 4.6 --- requirements-dev.txt | 2 +- tests/unit/test_git_gate_host_key.py | 14 ++++---------- 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/requirements-dev.txt b/requirements-dev.txt index a34cdd7..8ba2f16 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -3,5 +3,5 @@ # These tools are used for code quality checks in CI/CD. pylint>=3.0.0 -pyright>=1.1.300 +pyright>=1.1.411 coverage>=7.0.0 diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index 394757e..79b4c50 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -2,7 +2,6 @@ from __future__ import annotations -import dataclasses import tempfile import unittest from pathlib import Path @@ -404,20 +403,18 @@ class TestPreflightHostKeys(unittest.TestCase): def test_headless_dies_when_key_missing(self) -> None: manifest = _manifest_with_git() - from bot_bottle.log import die as real_die with self.assertRaises(SystemExit): preflight_host_keys(manifest, headless=True, home_md=None) def test_interactive_fetches_and_confirms_key(self) -> None: manifest = _manifest_with_git() - replies = iter(["y", "n"]) # confirm key, don't persist with patch( "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( "bot_bottle.git_gate_host_key._prompt_tty", - side_effect=lambda msg: next(replies), + side_effect=["y", "n"], # confirm key, don't persist ), patch("sys.stderr"): result = preflight_host_keys(manifest, headless=False, home_md=None) @@ -437,7 +434,6 @@ class TestPreflightHostKeys(unittest.TestCase): def test_interactive_persists_when_requested(self) -> None: manifest = _manifest_with_git() - replies = iter(["y", "y"]) # confirm key, yes persist with tempfile.TemporaryDirectory() as home: bottles_dir = Path(home) / "bottles" @@ -455,7 +451,7 @@ class TestPreflightHostKeys(unittest.TestCase): return_value="ssh-ed25519 FETCHED", ), patch( "bot_bottle.git_gate_host_key._prompt_tty", - side_effect=lambda msg: next(replies), + side_effect=["y", "y"], # confirm key, yes persist ), patch("sys.stderr"): result = preflight_host_keys( manifest, headless=False, home_md=Path(home), @@ -467,7 +463,6 @@ class TestPreflightHostKeys(unittest.TestCase): def test_interactive_skips_persist_when_declined(self) -> None: manifest = _manifest_with_git() - replies = iter(["y", "n"]) # confirm key, no persist with tempfile.TemporaryDirectory() as home: bottles_dir = Path(home) / "bottles" @@ -485,7 +480,7 @@ class TestPreflightHostKeys(unittest.TestCase): return_value="ssh-ed25519 FETCHED", ), patch( "bot_bottle.git_gate_host_key._prompt_tty", - side_effect=lambda msg: next(replies), + side_effect=["y", "n"], # confirm key, no persist ), patch("sys.stderr"): result = preflight_host_keys( manifest, headless=False, home_md=Path(home), @@ -506,7 +501,6 @@ class TestPreflightHostKeys(unittest.TestCase): def test_interactive_warns_when_persist_file_not_found(self) -> None: manifest = _manifest_with_git() - replies = iter(["y", "y"]) # confirm key, yes persist with tempfile.TemporaryDirectory() as home: # bottles/ dir exists but does not contain the repo entry @@ -520,7 +514,7 @@ class TestPreflightHostKeys(unittest.TestCase): return_value="ssh-ed25519 FETCHED", ), patch( "bot_bottle.git_gate_host_key._prompt_tty", - side_effect=lambda msg: next(replies), + side_effect=["y", "y"], # confirm key, yes persist ), patch("sys.stderr", buf): result = preflight_host_keys( manifest, headless=False, home_md=Path(home), -- 2.52.0 From 61740cdb6a2c193296453d0603851df86f50cd7e Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 16:45:47 +0000 Subject: [PATCH 4/7] fix(git-gate): accept any key type from ssh-keyscan, prefer ed25519 Servers that don't offer ed25519 (ecdsa-only or rsa-only gitea instances) would raise "ssh-keyscan returned no ed25519 key" because `-t ed25519` produced no output. Remove the `-t ed25519` restriction: ssh-keyscan now returns all supported key types and the function picks the best available via _KEY_TYPE_PREFERENCE (ed25519 > ecdsa > rsa > first). Also adds four new fetch_host_key tests covering multi-type preference logic and updates the stale "no ed25519 key" assertion. --- bot_bottle/git_gate_host_key.py | 38 +++++++++++++++++++++------- tests/unit/test_git_gate_host_key.py | 29 ++++++++++++++++++++- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/bot_bottle/git_gate_host_key.py b/bot_bottle/git_gate_host_key.py index 922af5e..ea5a1fe 100644 --- a/bot_bottle/git_gate_host_key.py +++ b/bot_bottle/git_gate_host_key.py @@ -20,16 +20,29 @@ from .log import die, info from .manifest import Manifest +# Preferred key types, most secure first. +_KEY_TYPE_PREFERENCE = ( + "ssh-ed25519", + "ecdsa-sha2-nistp256", + "ecdsa-sha2-nistp384", + "ecdsa-sha2-nistp521", + "ssh-rsa", +) + + def fetch_host_key(host: str, port: str) -> str: - """Return an SSH ed25519 public key for `host`:`port` via ssh-keyscan. + """Return an SSH public key for `host`:`port` via ssh-keyscan. Returns the key in ` ` format (the host prefix is stripped so the result can be stored in `host_key` and later formatted into a known_hosts line by `git_gate_known_hosts_line`). - Raises `RuntimeError` on subprocess failure, timeout, or no ed25519 - result. Uses only the Python stdlib (subprocess).""" - args = ["ssh-keyscan", "-t", "ed25519"] + Prefers ed25519 > ecdsa > rsa; falls back to the first key type + returned if none of the preferred types are present. + + Raises `RuntimeError` on subprocess failure, timeout, or no result. + Uses only the Python stdlib (subprocess).""" + args = ["ssh-keyscan"] if port and port != "22": args += ["-p", port] args.append(host) @@ -44,18 +57,25 @@ def fetch_host_key(host: str, port: str) -> str: except subprocess.TimeoutExpired as e: raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from e + # known_hosts format: "[host]:port type data" or "host type data" + # Strip the host/port prefix; collect "type -> type data" by type. + found: dict[str, str] = {} for line in result.stdout.splitlines(): line = line.strip() if not line or line.startswith("#"): continue - # known_hosts format: "[host]:port type data" or "host type data" - # Strip the host/port prefix; keep "type data". parts = line.split(None, 2) - if len(parts) == 3: - return f"{parts[1]} {parts[2]}" + if len(parts) == 3 and parts[1] not in found: + found[parts[1]] = f"{parts[1]} {parts[2]}" + + for preferred in _KEY_TYPE_PREFERENCE: + if preferred in found: + return found[preferred] + if found: + return next(iter(found.values())) raise RuntimeError( - f"ssh-keyscan returned no ed25519 key for {host}:{port}." + f"ssh-keyscan returned no host key for {host}:{port}." + (f" stderr: {result.stderr.strip()!r}" if result.stderr.strip() else "") ) diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index 79b4c50..d9fbfc8 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -71,7 +71,34 @@ class TestFetchHostKey(unittest.TestCase): with patch("subprocess.run", return_value=self._run_result("")): with self.assertRaises(RuntimeError) as cm: fetch_host_key("example.com", "22") - self.assertIn("no ed25519 key", str(cm.exception)) + self.assertIn("no host key", str(cm.exception)) + + def test_prefers_ed25519_over_ecdsa(self) -> None: + stdout = ( + "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n" + "gitea.example.com ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA\n" + ) + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("gitea.example.com", "22") + self.assertEqual("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAA", key) + + def test_falls_back_to_ecdsa_without_ed25519(self) -> None: + stdout = "gitea.example.com ecdsa-sha2-nistp256 BBBBB\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("gitea.example.com", "22") + self.assertEqual("ecdsa-sha2-nistp256 BBBBB", key) + + def test_falls_back_to_rsa_without_ed25519_or_ecdsa(self) -> None: + stdout = "gitea.example.com ssh-rsa AAAAB3NzaC1yc2EAAAA\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("gitea.example.com", "22") + self.assertEqual("ssh-rsa AAAAB3NzaC1yc2EAAAA", key) + + def test_falls_back_to_first_for_unknown_type(self) -> None: + stdout = "gitea.example.com ssh-unknown CCCCC\n" + with patch("subprocess.run", return_value=self._run_result(stdout)): + key = fetch_host_key("gitea.example.com", "22") + self.assertEqual("ssh-unknown CCCCC", key) def test_raises_includes_stderr_when_present(self) -> None: r = self._run_result("") -- 2.52.0 From d496e30681d1ee246684d92c66229dd89f271fd6 Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 17:21:57 +0000 Subject: [PATCH 5/7] =?UTF-8?q?refactor(git-gate):=20replace=20regex=20fro?= =?UTF-8?q?ntmatter=20edit=20with=20parse=E2=86=92mutate=E2=86=92serialize?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The regex-based _insert_host_key_in_frontmatter had a bug: child_indent was overwritten on every line with indent > repo_indent (including grandchildren), so host_key landed inside the key: block instead of at the repo level. Replace with a clean parse → mutate → re-serialize approach: - Add serialize_yaml_subset() to yaml_subset.py (block-style, 2-space indent, round-trips through parse_yaml_subset) - Replace _insert_host_key_in_frontmatter + _update_frontmatter_in_file with _add_host_key_to_frontmatter that parses the frontmatter dict, sets fm["git-gate"]["repos"][name]["host_key"], and re-serializes - _find_and_update_bottle_file simplified to read file → delegate to _add_host_key_to_frontmatter → write if changed --- bot_bottle/git_gate_host_key.py | 113 +++++------------- bot_bottle/yaml_subset.py | 107 +++++++++++++++++ tests/unit/test_git_gate_host_key.py | 171 +++++++++++---------------- tests/unit/test_yaml_subset.py | 138 ++++++++++++++++++++- 4 files changed, 342 insertions(+), 187 deletions(-) diff --git a/bot_bottle/git_gate_host_key.py b/bot_bottle/git_gate_host_key.py index ea5a1fe..9a1cdd6 100644 --- a/bot_bottle/git_gate_host_key.py +++ b/bot_bottle/git_gate_host_key.py @@ -11,13 +11,14 @@ Public entry point: `preflight_host_keys(manifest, headless=..., home_md=...)`. from __future__ import annotations import dataclasses -import re import subprocess import sys from pathlib import Path +from typing import cast from .log import die, info from .manifest import Manifest +from .yaml_subset import YamlSubsetError, parse_frontmatter, serialize_yaml_subset # Preferred key types, most secure first. @@ -81,79 +82,37 @@ def fetch_host_key(host: str, port: str) -> str: # --------------------------------------------------------------------------- -# Frontmatter text-level editing +# Frontmatter editing # --------------------------------------------------------------------------- -def _insert_host_key_in_frontmatter(fm_text: str, repo_name: str, host_key: str) -> str: - """Append `host_key: ` to the named repo entry in raw frontmatter text. +def _add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str: + """Return an updated copy of `file_text` with `host_key` set on the + named repo entry in the YAML frontmatter. - Locates the `:` block-mapping key, detects its child - indentation from the sibling keys already present, and inserts the - new line after the last child. Returns the original text if the - repo entry is not found or already has a `host_key:` line.""" - lines = fm_text.splitlines(keepends=True) - - repo_pat = re.compile(r"^(\s+)" + re.escape(repo_name) + r"\s*:\s*(\r?\n)?$") - repo_line_idx: int | None = None - repo_indent: int | None = None - for i, line in enumerate(lines): - if repo_pat.match(line): - repo_indent = len(line) - len(line.lstrip()) - repo_line_idx = i - break - - if repo_line_idx is None or repo_indent is None: - return fm_text - - child_indent: int | None = None - last_child_idx = repo_line_idx - - for i in range(repo_line_idx + 1, len(lines)): - raw = lines[i].rstrip("\r\n") - if not raw.strip(): - continue - indent = len(raw) - len(raw.lstrip()) - if indent > repo_indent: - child_indent = indent - last_child_idx = i - else: - break - - if child_indent is None: - child_indent = repo_indent + 2 - - block = "".join(lines[repo_line_idx + 1 : last_child_idx + 1]) - if re.search(r"^\s*host_key\s*:", block, re.MULTILINE): - return fm_text - - new_line = " " * child_indent + f"host_key: {host_key}\n" - lines.insert(last_child_idx + 1, new_line) - return "".join(lines) - - -def _update_frontmatter_in_file(file_text: str, repo_name: str, host_key: str) -> str: - """Apply `_insert_host_key_in_frontmatter` to the frontmatter section of - a Markdown file, leaving the body unchanged. Returns the original text - when no change is needed.""" - lines = file_text.splitlines(keepends=True) - if not lines or lines[0].rstrip("\r\n") != "---": + Parses the frontmatter into a dict, sets the key, and re-serializes. + Returns the original text unchanged when: the file has no frontmatter, + the git-gate.repos. entry is absent or already has a + host_key, or the frontmatter cannot be parsed.""" + try: + fm, body = parse_frontmatter(file_text) + except YamlSubsetError: return file_text - close_idx: int | None = None - for i in range(1, len(lines)): - if lines[i].rstrip("\r\n") == "---": - close_idx = i - break - if close_idx is None: + git_gate = fm.get("git-gate") + if not isinstance(git_gate, dict): + return file_text + repos = git_gate.get("repos") + if not isinstance(repos, dict): + return file_text + repo = repos.get(repo_name) + if not isinstance(repo, dict): + return file_text + if repo.get("host_key"): return file_text - fm_text = "".join(lines[1:close_idx]) - body_text = "".join(lines[close_idx:]) - updated_fm = _insert_host_key_in_frontmatter(fm_text, repo_name, host_key) - if updated_fm == fm_text: - return file_text - return "---\n" + updated_fm + body_text + cast(dict[str, object], repo)["host_key"] = host_key + return f"---\n{serialize_yaml_subset(fm)}---\n{body}" def _find_and_update_bottle_file( @@ -168,28 +127,13 @@ def _find_and_update_bottle_file( if not bottles_dir.is_dir(): return False - from .yaml_subset import YamlSubsetError, parse_frontmatter - for path in sorted(bottles_dir.glob("*.md")): try: text = path.read_text(encoding="utf-8") - fm, _body = parse_frontmatter(text) - except (OSError, UnicodeDecodeError, YamlSubsetError): + except (OSError, UnicodeDecodeError): continue - git_gate = fm.get("git-gate") - if not isinstance(git_gate, dict): - continue - repos = git_gate.get("repos") - if not isinstance(repos, dict): - continue - repo = repos.get(repo_name) - if not isinstance(repo, dict): - continue - if repo.get("host_key"): - continue - - updated = _update_frontmatter_in_file(text, repo_name, host_key) + updated = _add_host_key_to_frontmatter(text, repo_name, host_key) if updated == text: continue path.write_text(updated, encoding="utf-8") @@ -292,7 +236,6 @@ def preflight_host_keys( __all__ = [ "fetch_host_key", "preflight_host_keys", - "_insert_host_key_in_frontmatter", + "_add_host_key_to_frontmatter", "_find_and_update_bottle_file", - "_update_frontmatter_in_file", ] diff --git a/bot_bottle/yaml_subset.py b/bot_bottle/yaml_subset.py index 18ccf44..6432a3e 100644 --- a/bot_bottle/yaml_subset.py +++ b/bot_bottle/yaml_subset.py @@ -21,6 +21,11 @@ Public API: For a Markdown file with YAML frontmatter delimited by `---` lines. Returns (frontmatter_dict, body_text). + serialize_yaml_subset(data) -> str + Serialize a dict (as produced by parse_yaml_subset) back to + block-style YAML text. The result ends with a newline and + can be parsed back by parse_yaml_subset. + What we accept (block-style): key: value # mapping entry, value is inline @@ -576,3 +581,105 @@ def parse_frontmatter(text: str) -> tuple[dict[str, object], str]: fm = parse_yaml_subset(fm_text) body = text[body_start:] return fm, body + + +# --- Serializer ------------------------------------------------------------- + + +def _needs_quoting(s: str) -> bool: + """True when the string must be single-quoted to survive a round-trip.""" + if not s: + return True + if s in ("true", "false", "null", "~") or s in _RESERVED_BOOL_LIKE: + return True + if ( + _INT_RX.match(s) + or _DATE_RX.match(s) + or _OCTAL_RX.match(s) + or _HEX_RX.match(s) + or _FLOAT_RX.match(s) + ): + return True + # Characters that have special meaning at the start of a YAML value + if s[0] in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">", "%", "@", "`"): + return True + return False + + +def _yaml_scalar(v: object) -> str: + """Serialize a scalar Python value to its YAML text form.""" + if v is None: + return "null" + if isinstance(v, bool): + return "true" if v else "false" + if isinstance(v, int): + return str(v) + s = str(v) + if _needs_quoting(s): + return "'" + s.replace("'", "''") + "'" + return s + + +def _serialize_node(node: object, indent: int) -> list[str]: + """Return lines (without trailing newlines) for `node` at `indent`. + + Called only for non-empty dicts and lists (the caller guards with + `isinstance(val, (dict, list)) and val`), plus scalars at the leaf.""" + prefix = " " * indent + if isinstance(node, dict): + out: list[str] = [] + for key, val in node.items(): + if isinstance(val, (dict, list)) and val: + out.append(f"{prefix}{key}:") + out.extend(_serialize_node(val, indent + 2)) + else: + scalar = ( + "{}" if isinstance(val, dict) + else "[]" if isinstance(val, list) + else _yaml_scalar(val) + ) + out.append(f"{prefix}{key}: {scalar}") + return out + if isinstance(node, list): + out = [] + for item in node: + if isinstance(item, dict) and item: + entries = list(item.items()) + first_key, first_val = entries[0] + if isinstance(first_val, (dict, list)) and first_val: + out.append(f"{prefix}- {first_key}:") + out.extend(_serialize_node(first_val, indent + 4)) + else: + scalar = ( + "{}" if isinstance(first_val, dict) + else "[]" if isinstance(first_val, list) + else _yaml_scalar(first_val) + ) + out.append(f"{prefix}- {first_key}: {scalar}") + cont = prefix + " " + for key, val in entries[1:]: + if isinstance(val, (dict, list)) and val: + out.append(f"{cont}{key}:") + out.extend(_serialize_node(val, indent + 4)) + else: + scalar = ( + "{}" if isinstance(val, dict) + else "[]" if isinstance(val, list) + else _yaml_scalar(val) + ) + out.append(f"{cont}{key}: {scalar}") + else: + out.append(f"{prefix}- {_yaml_scalar(item)}") + return out + return [_yaml_scalar(node)] # pragma: no cover + + +def serialize_yaml_subset(data: dict[str, object]) -> str: + """Serialize `data` (as produced by parse_yaml_subset) to YAML text. + + Produces block-style output with 2-space indentation. The result ends + with a newline and can be parsed back by parse_yaml_subset. Keys are + emitted in iteration order (insertion order in Python 3.7+).""" + if not data: + return "" + return "\n".join(_serialize_node(data, 0)) + "\n" diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index d9fbfc8..94d6692 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -8,10 +8,9 @@ from pathlib import Path from unittest.mock import MagicMock, patch from bot_bottle.git_gate_host_key import ( + _add_host_key_to_frontmatter, _find_and_update_bottle_file, - _insert_host_key_in_frontmatter, _prompt_tty, - _update_frontmatter_in_file, fetch_host_key, preflight_host_keys, ) @@ -145,96 +144,11 @@ class TestPromptTty(unittest.TestCase): # --------------------------------------------------------------------------- -# _insert_host_key_in_frontmatter +# _add_host_key_to_frontmatter # --------------------------------------------------------------------------- -class TestInsertHostKeyInFrontmatter(unittest.TestCase): - _FM = """\ -git-gate: - repos: - myrepo: - url: ssh://git@host/org/repo.git - key: - provider: static - path: /dev/null -""" - - def test_inserts_host_key_after_last_child(self) -> None: - result = _insert_host_key_in_frontmatter(self._FM, "myrepo", "ssh-ed25519 AAAA") - self.assertIn(" host_key: ssh-ed25519 AAAA", result) - - def test_no_change_when_repo_not_found(self) -> None: - result = _insert_host_key_in_frontmatter(self._FM, "other", "ssh-ed25519 AAAA") - self.assertEqual(self._FM, result) - - def test_no_change_when_host_key_already_present(self) -> None: - fm = self._FM + " host_key: ssh-ed25519 EXISTING\n" - result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 NEW") - self.assertNotIn("NEW", result) - self.assertEqual(fm, result) - - def test_preserves_other_repos(self) -> None: - fm = """\ -git-gate: - repos: - first: - url: ssh://git@host/org/first.git - key: - provider: static - path: /dev/null - second: - url: ssh://git@host/org/second.git - key: - provider: static - path: /dev/null -""" - result = _insert_host_key_in_frontmatter(fm, "first", "ssh-ed25519 AAAA") - self.assertIn("host_key: ssh-ed25519 AAAA", result) - self.assertIn("second:", result) - - def test_detects_child_indent_from_existing_keys(self) -> None: - # 4-space indent (non-standard but valid) - fm = """\ -git-gate: - repos: - myrepo: - url: ssh://git@host/org/repo.git - key: - provider: static - path: /dev/null -""" - result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") - self.assertIn(" host_key: ssh-ed25519 AAAA", result) - - def test_blank_lines_inside_block_do_not_end_scan(self) -> None: - fm = """\ -git-gate: - repos: - myrepo: - url: ssh://git@host/org/repo.git - - key: - provider: static - path: /dev/null -""" - result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") - self.assertIn("host_key: ssh-ed25519 AAAA", result) - - def test_repo_with_no_children_uses_default_indent(self) -> None: - # Edge case: repo entry has no child keys at all (empty block). - fm = "git-gate:\n repos:\n myrepo:\n" - result = _insert_host_key_in_frontmatter(fm, "myrepo", "ssh-ed25519 AAAA") - # Should fall back to repo_indent+2 = 4 - self.assertIn(" host_key: ssh-ed25519 AAAA", result) - - -# --------------------------------------------------------------------------- -# _update_frontmatter_in_file -# --------------------------------------------------------------------------- - - -class TestUpdateFrontmatterInFile(unittest.TestCase): +class TestAddHostKeyToFrontmatter(unittest.TestCase): _FILE = """\ --- git-gate: @@ -248,23 +162,78 @@ git-gate: # Body text here """ - def test_updates_frontmatter_preserves_body(self) -> None: - result = _update_frontmatter_in_file(self._FILE, "myrepo", "ssh-ed25519 AAAA") + def test_inserts_host_key_at_repo_level(self) -> None: + result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") self.assertIn("host_key: ssh-ed25519 AAAA", result) + # Must be a sibling of `url` and `key`, not nested inside `key` + lines = result.splitlines() + hk_line = next(l for l in lines if "host_key" in l) + url_line = next(l for l in lines if "url:" in l) + self.assertEqual( + len(hk_line) - len(hk_line.lstrip()), + len(url_line) - len(url_line.lstrip()), + ) + + def test_preserves_body(self) -> None: + result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") self.assertIn("# Body text here", result) + def test_preserves_other_repos(self) -> None: + text = """\ +--- +git-gate: + repos: + first: + url: ssh://git@host/org/first.git + key: + provider: static + path: /dev/null + second: + url: ssh://git@host/org/second.git + key: + provider: static + path: /dev/null +--- +""" + result = _add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA") + self.assertIn("host_key: ssh-ed25519 AAAA", result) + self.assertIn("second:", result) + + def test_no_change_when_host_key_already_present(self) -> None: + text = self._FILE.replace( + " path: /dev/null\n", + " path: /dev/null\n host_key: ssh-ed25519 EXISTING\n", + ) + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW") + self.assertNotIn("NEW", result) + + def test_no_change_when_repo_not_found(self) -> None: + result = _add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA") + self.assertEqual(self._FILE, result) + def test_no_change_without_frontmatter(self) -> None: text = "No frontmatter here\n" - result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA") + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) - def test_no_change_when_repo_not_in_file(self) -> None: - result = _update_frontmatter_in_file(self._FILE, "other", "ssh-ed25519 AAAA") - self.assertEqual(self._FILE, result) - def test_no_change_when_no_closing_delimiter(self) -> None: text = "---\ngit-gate:\n repos:\n myrepo:\n url: x\n" - result = _update_frontmatter_in_file(text, "myrepo", "ssh-ed25519 AAAA") + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + self.assertEqual(text, result) + + def test_no_change_when_git_gate_missing(self) -> None: + text = "---\nenv:\n FOO: bar\n---\n" + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + self.assertEqual(text, result) + + def test_no_change_when_repos_missing(self) -> None: + text = "---\ngit-gate:\n foo: bar\n---\n" + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + self.assertEqual(text, result) + + def test_no_change_when_repo_entry_not_a_dict(self) -> None: + text = "---\ngit-gate:\n repos:\n - item\n---\n" + result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) @@ -372,9 +341,9 @@ git-gate: os.chmod(path, 0o644) def test_skips_when_update_produces_no_change(self) -> None: - # _update_frontmatter_in_file returns the original text unchanged - # (e.g. because _insert_host_key_in_frontmatter can't locate the - # entry in the raw text). Line 174 must `continue` rather than write. + # _add_host_key_to_frontmatter returns the original text unchanged + # (e.g. because the entry is absent from the parsed structure). + # _find_and_update_bottle_file must continue rather than write. content = """\ --- git-gate: @@ -389,7 +358,7 @@ git-gate: with tempfile.TemporaryDirectory() as d: path = self._write_bottle(d, "dev", content) with patch( - "bot_bottle.git_gate_host_key._update_frontmatter_in_file", + "bot_bottle.git_gate_host_key._add_host_key_to_frontmatter", return_value=content, # no change ): result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") diff --git a/tests/unit/test_yaml_subset.py b/tests/unit/test_yaml_subset.py index 0756822..c53994e 100644 --- a/tests/unit/test_yaml_subset.py +++ b/tests/unit/test_yaml_subset.py @@ -6,7 +6,7 @@ import textwrap import unittest from bot_bottle.yaml_subset import YamlSubsetError -from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset +from bot_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset, serialize_yaml_subset def _y(s: str): @@ -457,5 +457,141 @@ class TestEdgeAndErrorBranches(unittest.TestCase): self.assertEqual(({}, ""), parse_frontmatter("")) +# --------------------------------------------------------------------------- +# serialize_yaml_subset +# --------------------------------------------------------------------------- + + +class TestSerializeYamlSubset(unittest.TestCase): + def _roundtrip(self, text: str) -> None: + """Parse then serialize then re-parse; result must equal original parse.""" + parsed = parse_yaml_subset(text) + serialized = serialize_yaml_subset(parsed) + self.assertEqual(parsed, parse_yaml_subset(serialized)) + + # --- scalar values --------------------------------------------------------- + + def test_string_bare(self) -> None: + self.assertEqual("key: value\n", serialize_yaml_subset({"key": "value"})) + + def test_string_empty_is_quoted(self) -> None: + out = serialize_yaml_subset({"key": ""}) + self.assertIn("key: ''", out) + self._roundtrip("key: ''\n") + + def test_string_true_like_is_quoted(self) -> None: + out = serialize_yaml_subset({"key": "true"}) + self.assertIn("'true'", out) + self.assertEqual({"key": "true"}, parse_yaml_subset(out)) + + def test_string_int_like_is_quoted(self) -> None: + out = serialize_yaml_subset({"key": "42"}) + self.assertIn("'42'", out) + self.assertEqual({"key": "42"}, parse_yaml_subset(out)) + + def test_string_special_start_char_is_quoted(self) -> None: + for ch in ('"', "'", "[", "{", "!", "&", "*", "#", "|", ">"): + out = serialize_yaml_subset({"key": ch + "rest"}) + self.assertIn("'", out, msg=f"expected quoting for {ch!r}") + + def test_none_emits_null(self) -> None: + self.assertEqual("key: null\n", serialize_yaml_subset({"key": None})) + + def test_bool_true(self) -> None: + self.assertEqual("key: true\n", serialize_yaml_subset({"key": True})) + + def test_bool_false(self) -> None: + self.assertEqual("key: false\n", serialize_yaml_subset({"key": False})) + + def test_int(self) -> None: + self.assertEqual("key: 42\n", serialize_yaml_subset({"key": 42})) + + # --- nested dict ----------------------------------------------------------- + + def test_nested_dict(self) -> None: + data: dict[str, object] = {"outer": {"inner": "val"}} + out = serialize_yaml_subset(data) + self.assertIn("outer:", out) + self.assertIn(" inner: val", out) + self.assertEqual(data, parse_yaml_subset(out)) + + def test_empty_dict_value_inline(self) -> None: + out = serialize_yaml_subset({"key": {}}) + self.assertIn("key: {}", out) + + # --- lists ----------------------------------------------------------------- + + def test_list_of_scalars(self) -> None: + data: dict[str, object] = {"items": ["a", "b", "c"]} + out = serialize_yaml_subset(data) + self.assertEqual(data, parse_yaml_subset(out)) + + def test_empty_list_value_inline(self) -> None: + out = serialize_yaml_subset({"key": []}) + self.assertIn("key: []", out) + + def test_list_of_mappings(self) -> None: + data: dict[str, object] = {"items": [{"name": "alpha", "val": 1}, {"name": "beta", "val": 2}]} + out = serialize_yaml_subset(data) + self.assertEqual(data, parse_yaml_subset(out)) + + def test_list_item_mapping_with_nested_dict(self) -> None: + data: dict[str, object] = {"items": [{"name": "x", "sub": {"k": "v"}}]} + out = serialize_yaml_subset(data) + self.assertEqual(data, parse_yaml_subset(out)) + + def test_list_item_first_key_is_nested_dict(self) -> None: + # Covers the branch where the FIRST key of a mapping list item + # has a nested dict/list value (not a scalar). + data: dict[str, object] = {"items": [{"cfg": {"a": 1, "b": 2}, "name": "x"}]} + out = serialize_yaml_subset(data) + self.assertEqual(data, parse_yaml_subset(out)) + + def test_list_item_mapping_continuation_keys(self) -> None: + data: dict[str, object] = {"items": [{"a": 1, "b": 2, "c": 3}]} + out = serialize_yaml_subset(data) + self.assertEqual(data, parse_yaml_subset(out)) + + # --- empty / trivial ------------------------------------------------------- + + def test_empty_dict(self) -> None: + self.assertEqual("", serialize_yaml_subset({})) + + # --- realistic bottle frontmatter round-trip -------------------------------- + + def test_bottle_frontmatter_round_trip(self) -> None: + text = textwrap.dedent("""\ + git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null + """) + self._roundtrip(text) + + def test_host_key_inserted_at_repo_level(self) -> None: + """The canonical use-case: host_key lands as a sibling of url/key.""" + from typing import cast + text = textwrap.dedent("""\ + git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null + """) + data = parse_yaml_subset(text) + repos = cast(dict[str, object], cast(dict[str, object], data["git-gate"])["repos"]) + cast(dict[str, object], repos["myrepo"])["host_key"] = "ssh-ed25519 AAAA" + out = serialize_yaml_subset(data) + parsed_back = parse_yaml_subset(out) + repo = cast(dict[str, object], cast(dict[str, object], cast(dict[str, object], parsed_back["git-gate"])["repos"])["myrepo"]) + self.assertEqual("ssh-ed25519 AAAA", repo["host_key"]) + self.assertIn("provider", cast(dict[str, object], repo["key"])) + + if __name__ == "__main__": unittest.main() -- 2.52.0 From e27bd6608096db3721ac5ede0ae244c05e348e2e Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 17:26:25 +0000 Subject: [PATCH 6/7] feat(git-gate): show bottle filename in save-host-key prompt; public API DX improvement: find_repo_bottle_file() is called before the save prompt so the user sees exactly which file will be updated: Save host_key for 'myrepo' to /home/.../bottles/dev.md? [y/N] If no bottle file is found the prompt is skipped and a clear message explains that the key is session-only. Also: - Use find_repo_bottle_file() inside find_and_update_bottle_file() instead of duplicating the scan logic - Rename all module-level helpers to public names (no underscore prefix) since they are imported and tested externally --- bot_bottle/git_gate_host_key.py | 95 +++++++---- tests/unit/test_git_gate_host_key.py | 232 ++++++++++++++++++++++----- 2 files changed, 256 insertions(+), 71 deletions(-) diff --git a/bot_bottle/git_gate_host_key.py b/bot_bottle/git_gate_host_key.py index 9a1cdd6..2210e8e 100644 --- a/bot_bottle/git_gate_host_key.py +++ b/bot_bottle/git_gate_host_key.py @@ -86,7 +86,7 @@ def fetch_host_key(host: str, port: str) -> str: # --------------------------------------------------------------------------- -def _add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str: +def add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) -> str: """Return an updated copy of `file_text` with `host_key` set on the named repo entry in the YAML frontmatter. @@ -115,32 +115,50 @@ def _add_host_key_to_frontmatter(file_text: str, repo_name: str, host_key: str) return f"---\n{serialize_yaml_subset(fm)}---\n{body}" -def _find_and_update_bottle_file( - bottles_dir: Path, repo_name: str, host_key: str, -) -> bool: - """Scan `bottles_dir/*.md` for the first file that directly declares - `repo_name` without a `host_key`, update it, and return True. - - Walks files in sorted order so the result is deterministic. Does NOT - follow `extends:` chains — only files that explicitly declare the repo - entry are candidates. Returns False when no such file is found.""" +def find_repo_bottle_file(bottles_dir: Path, repo_name: str) -> Path | None: + """Return the first `bottles_dir/*.md` that declares `repo_name` without + a `host_key`, without modifying anything. Returns None if not found.""" if not bottles_dir.is_dir(): - return False - + return None for path in sorted(bottles_dir.glob("*.md")): try: text = path.read_text(encoding="utf-8") - except (OSError, UnicodeDecodeError): + fm, _ = parse_frontmatter(text) + except (OSError, UnicodeDecodeError, YamlSubsetError): continue - - updated = _add_host_key_to_frontmatter(text, repo_name, host_key) - if updated == text: + git_gate = fm.get("git-gate") + if not isinstance(git_gate, dict): continue - path.write_text(updated, encoding="utf-8") - info(f"wrote host_key for {repo_name!r} to {path}") - return True + repos = git_gate.get("repos") + if not isinstance(repos, dict): + continue + repo = repos.get(repo_name) + if not isinstance(repo, dict) or repo.get("host_key"): + continue + return path + return None - return False + +def find_and_update_bottle_file( + bottles_dir: Path, repo_name: str, host_key: str, +) -> bool: + """Write `host_key` into the bottle file returned by `find_repo_bottle_file`. + + Returns True on success, False when no suitable file is found or the + write fails.""" + path = find_repo_bottle_file(bottles_dir, repo_name) + if path is None: + return False + try: + text = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + return False + updated = add_host_key_to_frontmatter(text, repo_name, host_key) + if updated == text: + return False + path.write_text(updated, encoding="utf-8") + info(f"wrote host_key for {repo_name!r} to {path}") + return True # --------------------------------------------------------------------------- @@ -148,7 +166,7 @@ def _find_and_update_bottle_file( # --------------------------------------------------------------------------- -def _prompt_tty(message: str) -> str: +def prompt_tty(message: str) -> str: """Write `message` to stderr and read a line from /dev/tty (or stdin).""" sys.stderr.write(message) sys.stderr.flush() @@ -210,21 +228,28 @@ def preflight_host_keys( die(f"git-gate: {label}: {e}") sys.stderr.write(f"\ngit-gate: host key for {label}:\n {key}\n\n") - confirm = _prompt_tty("Is this host key correct? [y/N] ") + confirm = prompt_tty("Is this host key correct? [y/N] ") if confirm.strip().lower() not in ("y", "yes"): die(f"git-gate: {label}: host key not confirmed; aborting launch") if bottles_dir is not None: - save = _prompt_tty( - f"Save host_key for {entry.Name!r} to the bottle config file? [y/N] " - ) - if save.strip().lower() in ("y", "yes"): - ok = _find_and_update_bottle_file(bottles_dir, entry.Name, key) - if not ok: - sys.stderr.write( - f"git-gate: {label}: bottle file not found or not writable; " - f"host_key kept in memory for this session only\n" - ) + target_file = find_repo_bottle_file(bottles_dir, entry.Name) + if target_file is not None: + save = prompt_tty( + f"Save host_key for {entry.Name!r} to {target_file}? [y/N] " + ) + if save.strip().lower() in ("y", "yes"): + ok = find_and_update_bottle_file(bottles_dir, entry.Name, key) + if not ok: + sys.stderr.write( + f"git-gate: {label}: could not write to {target_file}; " + f"host_key kept in memory for this session only\n" + ) + else: + sys.stderr.write( + f"git-gate: {label}: no bottle config file found for " + f"{entry.Name!r}; host_key kept in memory for this session only\n" + ) idx = next(i for i, e in enumerate(updated_entries) if e.Name == entry.Name) updated_entries[idx] = dataclasses.replace(entry, KnownHostKey=key) @@ -236,6 +261,8 @@ def preflight_host_keys( __all__ = [ "fetch_host_key", "preflight_host_keys", - "_add_host_key_to_frontmatter", - "_find_and_update_bottle_file", + "add_host_key_to_frontmatter", + "find_repo_bottle_file", + "find_and_update_bottle_file", + "prompt_tty", ] diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index 94d6692..fa9601a 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -8,9 +8,10 @@ from pathlib import Path from unittest.mock import MagicMock, patch from bot_bottle.git_gate_host_key import ( - _add_host_key_to_frontmatter, - _find_and_update_bottle_file, - _prompt_tty, + add_host_key_to_frontmatter, + find_and_update_bottle_file, + find_repo_bottle_file, + prompt_tty, fetch_host_key, preflight_host_keys, ) @@ -122,7 +123,7 @@ class TestFetchHostKey(unittest.TestCase): # --------------------------------------------------------------------------- -# _prompt_tty +# prompt_tty # --------------------------------------------------------------------------- @@ -131,7 +132,7 @@ class TestPromptTty(unittest.TestCase): import io fake_tty = io.StringIO("yes\n") with patch("builtins.open", return_value=fake_tty), patch("sys.stderr"): - result = _prompt_tty("question: ") + result = prompt_tty("question: ") self.assertEqual("yes", result) def test_falls_back_to_stdin_on_os_error(self) -> None: @@ -139,12 +140,12 @@ class TestPromptTty(unittest.TestCase): with patch("builtins.open", side_effect=OSError("no tty")), \ patch("sys.stdin", io.StringIO("fallback\n")), \ patch("sys.stderr"): - result = _prompt_tty("question: ") + result = prompt_tty("question: ") self.assertEqual("fallback", result) # --------------------------------------------------------------------------- -# _add_host_key_to_frontmatter +# add_host_key_to_frontmatter # --------------------------------------------------------------------------- @@ -163,7 +164,7 @@ git-gate: """ def test_inserts_host_key_at_repo_level(self) -> None: - result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") self.assertIn("host_key: ssh-ed25519 AAAA", result) # Must be a sibling of `url` and `key`, not nested inside `key` lines = result.splitlines() @@ -175,7 +176,7 @@ git-gate: ) def test_preserves_body(self) -> None: - result = _add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(self._FILE, "myrepo", "ssh-ed25519 AAAA") self.assertIn("# Body text here", result) def test_preserves_other_repos(self) -> None: @@ -195,7 +196,7 @@ git-gate: path: /dev/null --- """ - result = _add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "first", "ssh-ed25519 AAAA") self.assertIn("host_key: ssh-ed25519 AAAA", result) self.assertIn("second:", result) @@ -204,41 +205,41 @@ git-gate: " path: /dev/null\n", " path: /dev/null\n host_key: ssh-ed25519 EXISTING\n", ) - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 NEW") self.assertNotIn("NEW", result) def test_no_change_when_repo_not_found(self) -> None: - result = _add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(self._FILE, "other", "ssh-ed25519 AAAA") self.assertEqual(self._FILE, result) def test_no_change_without_frontmatter(self) -> None: text = "No frontmatter here\n" - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) def test_no_change_when_no_closing_delimiter(self) -> None: text = "---\ngit-gate:\n repos:\n myrepo:\n url: x\n" - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) def test_no_change_when_git_gate_missing(self) -> None: text = "---\nenv:\n FOO: bar\n---\n" - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) def test_no_change_when_repos_missing(self) -> None: text = "---\ngit-gate:\n foo: bar\n---\n" - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) def test_no_change_when_repo_entry_not_a_dict(self) -> None: text = "---\ngit-gate:\n repos:\n - item\n---\n" - result = _add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") + result = add_host_key_to_frontmatter(text, "myrepo", "ssh-ed25519 AAAA") self.assertEqual(text, result) # --------------------------------------------------------------------------- -# _find_and_update_bottle_file +# find_and_update_bottle_file # --------------------------------------------------------------------------- @@ -262,7 +263,7 @@ git-gate: --- """ path = self._write_bottle(d, "dev", content) - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertTrue(result) updated = path.read_text() self.assertIn("host_key: ssh-ed25519 AAAA", updated) @@ -281,7 +282,7 @@ git-gate: --- """ self._write_bottle(d, "dev", content) - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) def test_skips_file_that_already_has_host_key(self) -> None: @@ -299,12 +300,12 @@ git-gate: --- """ path = self._write_bottle(d, "dev", content) - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 NEW") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 NEW") self.assertFalse(result) self.assertNotIn("NEW", path.read_text()) def test_returns_false_for_nonexistent_dir(self) -> None: - result = _find_and_update_bottle_file( + result = find_and_update_bottle_file( Path("/nonexistent/dir"), "myrepo", "ssh-ed25519 AAAA" ) self.assertFalse(result) @@ -312,20 +313,20 @@ git-gate: def test_skips_file_without_git_gate_section(self) -> None: with tempfile.TemporaryDirectory() as d: self._write_bottle(d, "plain", "---\nenv:\n FOO: bar\n---\n") - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) def test_skips_file_with_non_dict_repos(self) -> None: with tempfile.TemporaryDirectory() as d: self._write_bottle(d, "bad", "---\ngit-gate:\n repos:\n - item\n---\n") - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) def test_skips_invalid_utf8_file(self) -> None: with tempfile.TemporaryDirectory() as d: path = Path(d) / "broken.md" path.write_bytes(b"\xff\xfe") # invalid UTF-8 - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) def test_skips_unreadable_file(self) -> None: @@ -335,15 +336,15 @@ git-gate: path.write_text("---\n---\n") os.chmod(path, 0o000) try: - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) finally: os.chmod(path, 0o644) - def test_skips_when_update_produces_no_change(self) -> None: - # _add_host_key_to_frontmatter returns the original text unchanged - # (e.g. because the entry is absent from the parsed structure). - # _find_and_update_bottle_file must continue rather than write. + def test_returns_false_when_file_unreadable_on_write(self) -> None: + # find_repo_bottle_file succeeds, but the file becomes unreadable + # before find_and_update_bottle_file can re-read it. + import os content = """\ --- git-gate: @@ -358,14 +359,137 @@ git-gate: with tempfile.TemporaryDirectory() as d: path = self._write_bottle(d, "dev", content) with patch( - "bot_bottle.git_gate_host_key._add_host_key_to_frontmatter", + "bot_bottle.git_gate_host_key.find_repo_bottle_file", + return_value=path, + ): + os.chmod(path, 0o000) + try: + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + finally: + os.chmod(path, 0o644) + self.assertFalse(result) + + def test_skips_when_update_produces_no_change(self) -> None: + # add_host_key_to_frontmatter returns the original text unchanged + # (e.g. because the entry is absent from the parsed structure). + # find_and_update_bottle_file must continue rather than write. + content = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +--- +""" + with tempfile.TemporaryDirectory() as d: + path = self._write_bottle(d, "dev", content) + with patch( + "bot_bottle.git_gate_host_key.add_host_key_to_frontmatter", return_value=content, # no change ): - result = _find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) self.assertNotIn("host_key", path.read_text()) +# --------------------------------------------------------------------------- +# find_repo_bottle_file +# --------------------------------------------------------------------------- + + +class TestFindRepoBottleFile(unittest.TestCase): + _CONTENT = """\ +--- +git-gate: + repos: + myrepo: + url: ssh://git@host/org/repo.git + key: + provider: static + path: /dev/null +--- +""" + + def _write(self, d: str, name: str, content: str) -> Path: + path = Path(d) / f"{name}.md" + path.write_text(content, encoding="utf-8") + return path + + def test_returns_path_when_repo_found(self) -> None: + with tempfile.TemporaryDirectory() as d: + path = self._write(d, "dev", self._CONTENT) + result = find_repo_bottle_file(Path(d), "myrepo") + self.assertEqual(path, result) + + def test_returns_none_when_no_matching_file(self) -> None: + with tempfile.TemporaryDirectory() as d: + self._write(d, "dev", self._CONTENT) + self.assertIsNone(find_repo_bottle_file(Path(d), "other")) + + def test_returns_none_when_host_key_already_set(self) -> None: + content = self._CONTENT.replace(" path: /dev/null\n", + " path: /dev/null\n host_key: X\n") + with tempfile.TemporaryDirectory() as d: + self._write(d, "dev", content) + self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo")) + + def test_returns_none_for_nonexistent_dir(self) -> None: + self.assertIsNone(find_repo_bottle_file(Path("/nonexistent"), "myrepo")) + + def test_skips_unreadable_file(self) -> None: + import os + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "unreadable.md" + path.write_text(self._CONTENT) + os.chmod(path, 0o000) + try: + self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo")) + finally: + os.chmod(path, 0o644) + + def test_skips_file_with_non_dict_git_gate(self) -> None: + with tempfile.TemporaryDirectory() as d: + (Path(d) / "dev.md").write_text("---\ngit-gate:\n - item\n---\n") + self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo")) + + def test_skips_file_with_non_dict_repos(self) -> None: + with tempfile.TemporaryDirectory() as d: + (Path(d) / "dev.md").write_text("---\ngit-gate:\n repos:\n - item\n---\n") + self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo")) + + def test_save_prompt_includes_filename(self) -> None: + """preflight prompt must name the discovered bottle file.""" + manifest = _manifest_with_git() + + with tempfile.TemporaryDirectory() as home: + bottles_dir = Path(home) / "bottles" + bottles_dir.mkdir() + bottle_file = bottles_dir / "dev.md" + bottle_file.write_text( + "---\ngit-gate:\n repos:\n myrepo:\n" + " url: ssh://git@gitea.example.com:30009/org/myrepo.git\n" + " key:\n provider: static\n path: /dev/null\n---\n", + encoding="utf-8", + ) + + mock_prompt = MagicMock(side_effect=["y", "n"]) # confirm key, skip save + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key.prompt_tty", mock_prompt, + ), patch("sys.stderr"): + preflight_host_keys(manifest, headless=False, home_md=Path(home)) + + save_call = next( + c for c in mock_prompt.call_args_list if "Save" in c.args[0] + ) + self.assertIn(str(bottle_file), save_call.args[0]) + + # --------------------------------------------------------------------------- # preflight_host_keys # --------------------------------------------------------------------------- @@ -409,7 +533,7 @@ class TestPreflightHostKeys(unittest.TestCase): "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( - "bot_bottle.git_gate_host_key._prompt_tty", + "bot_bottle.git_gate_host_key.prompt_tty", side_effect=["y", "n"], # confirm key, don't persist ), patch("sys.stderr"): result = preflight_host_keys(manifest, headless=False, home_md=None) @@ -423,7 +547,7 @@ class TestPreflightHostKeys(unittest.TestCase): "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( - "bot_bottle.git_gate_host_key._prompt_tty", + "bot_bottle.git_gate_host_key.prompt_tty", return_value="n", ), patch("sys.stderr"), self.assertRaises(SystemExit): preflight_host_keys(manifest, headless=False, home_md=None) @@ -446,7 +570,7 @@ class TestPreflightHostKeys(unittest.TestCase): "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( - "bot_bottle.git_gate_host_key._prompt_tty", + "bot_bottle.git_gate_host_key.prompt_tty", side_effect=["y", "y"], # confirm key, yes persist ), patch("sys.stderr"): result = preflight_host_keys( @@ -475,7 +599,7 @@ class TestPreflightHostKeys(unittest.TestCase): "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( - "bot_bottle.git_gate_host_key._prompt_tty", + "bot_bottle.git_gate_host_key.prompt_tty", side_effect=["y", "n"], # confirm key, no persist ), patch("sys.stderr"): result = preflight_host_keys( @@ -509,8 +633,42 @@ class TestPreflightHostKeys(unittest.TestCase): "bot_bottle.git_gate_host_key.fetch_host_key", return_value="ssh-ed25519 FETCHED", ), patch( - "bot_bottle.git_gate_host_key._prompt_tty", + "bot_bottle.git_gate_host_key.prompt_tty", + side_effect=["y"], # confirm key only; no save prompt when no file found + ), patch("sys.stderr", buf): + result = preflight_host_keys( + manifest, headless=False, home_md=Path(home), + ) + + self.assertEqual("ssh-ed25519 FETCHED", result.bottle.git[0].KnownHostKey) + self.assertIn("kept in memory", buf.getvalue()) + + def test_interactive_warns_when_write_fails_after_prompt(self) -> None: + # User says yes to save, but find_and_update_bottle_file fails. + manifest = _manifest_with_git() + + with tempfile.TemporaryDirectory() as home: + bottles_dir = Path(home) / "bottles" + bottles_dir.mkdir() + bottle_file = bottles_dir / "dev.md" + bottle_file.write_text( + "---\ngit-gate:\n repos:\n myrepo:\n" + " url: ssh://git@gitea.example.com:30009/org/myrepo.git\n" + " key:\n provider: static\n path: /dev/null\n---\n", + encoding="utf-8", + ) + + import io + buf = io.StringIO() + with patch( + "bot_bottle.git_gate_host_key.fetch_host_key", + return_value="ssh-ed25519 FETCHED", + ), patch( + "bot_bottle.git_gate_host_key.prompt_tty", side_effect=["y", "y"], # confirm key, yes persist + ), patch( + "bot_bottle.git_gate_host_key.find_and_update_bottle_file", + return_value=False, ), patch("sys.stderr", buf): result = preflight_host_keys( manifest, headless=False, home_md=Path(home), -- 2.52.0 From fa7c6ab9d84e3443986f0cc4407616677639252a Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 9 Jul 2026 17:35:28 +0000 Subject: [PATCH 7/7] fix(tests): replace os.chmod with mock for unreadable-file tests CI runs as root, so chmod 0o000 doesn't prevent reads. Both test_skips_unreadable_file (TestFindRepoBottleFile) and test_returns_false_when_file_unreadable_on_write now mock Path.read_text to raise OSError directly. --- tests/unit/test_git_gate_host_key.py | 22 ++++++---------------- 1 file changed, 6 insertions(+), 16 deletions(-) diff --git a/tests/unit/test_git_gate_host_key.py b/tests/unit/test_git_gate_host_key.py index fa9601a..3835d14 100644 --- a/tests/unit/test_git_gate_host_key.py +++ b/tests/unit/test_git_gate_host_key.py @@ -342,9 +342,8 @@ git-gate: os.chmod(path, 0o644) def test_returns_false_when_file_unreadable_on_write(self) -> None: - # find_repo_bottle_file succeeds, but the file becomes unreadable - # before find_and_update_bottle_file can re-read it. - import os + # find_repo_bottle_file succeeds, but the second read (for writing) + # raises OSError — e.g. file removed or permissions changed. content = """\ --- git-gate: @@ -361,12 +360,8 @@ git-gate: with patch( "bot_bottle.git_gate_host_key.find_repo_bottle_file", return_value=path, - ): - os.chmod(path, 0o000) - try: - result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") - finally: - os.chmod(path, 0o644) + ), patch.object(Path, "read_text", side_effect=OSError("Permission denied")): + result = find_and_update_bottle_file(Path(d), "myrepo", "ssh-ed25519 AAAA") self.assertFalse(result) def test_skips_when_update_produces_no_change(self) -> None: @@ -440,15 +435,10 @@ git-gate: self.assertIsNone(find_repo_bottle_file(Path("/nonexistent"), "myrepo")) def test_skips_unreadable_file(self) -> None: - import os with tempfile.TemporaryDirectory() as d: - path = Path(d) / "unreadable.md" - path.write_text(self._CONTENT) - os.chmod(path, 0o000) - try: + (Path(d) / "unreadable.md").write_text(self._CONTENT) + with patch.object(Path, "read_text", side_effect=OSError("Permission denied")): self.assertIsNone(find_repo_bottle_file(Path(d), "myrepo")) - finally: - os.chmod(path, 0o644) def test_skips_file_with_non_dict_git_gate(self) -> None: with tempfile.TemporaryDirectory() as d: -- 2.52.0