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),