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.
This commit is contained in:
@@ -20,16 +20,29 @@ from .log import die, info
|
|||||||
from .manifest import Manifest
|
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:
|
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 `<type> <base64-data>` format (the host prefix is
|
Returns the key in `<type> <base64-data>` format (the host prefix is
|
||||||
stripped so the result can be stored in `host_key` and later formatted
|
stripped so the result can be stored in `host_key` and later formatted
|
||||||
into a known_hosts line by `git_gate_known_hosts_line`).
|
into a known_hosts line by `git_gate_known_hosts_line`).
|
||||||
|
|
||||||
Raises `RuntimeError` on subprocess failure, timeout, or no ed25519
|
Prefers ed25519 > ecdsa > rsa; falls back to the first key type
|
||||||
result. Uses only the Python stdlib (subprocess)."""
|
returned if none of the preferred types are present.
|
||||||
args = ["ssh-keyscan", "-t", "ed25519"]
|
|
||||||
|
Raises `RuntimeError` on subprocess failure, timeout, or no result.
|
||||||
|
Uses only the Python stdlib (subprocess)."""
|
||||||
|
args = ["ssh-keyscan"]
|
||||||
if port and port != "22":
|
if port and port != "22":
|
||||||
args += ["-p", port]
|
args += ["-p", port]
|
||||||
args.append(host)
|
args.append(host)
|
||||||
@@ -44,18 +57,25 @@ def fetch_host_key(host: str, port: str) -> str:
|
|||||||
except subprocess.TimeoutExpired as e:
|
except subprocess.TimeoutExpired as e:
|
||||||
raise RuntimeError(f"ssh-keyscan timed out for {host}:{port}") from 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():
|
for line in result.stdout.splitlines():
|
||||||
line = line.strip()
|
line = line.strip()
|
||||||
if not line or line.startswith("#"):
|
if not line or line.startswith("#"):
|
||||||
continue
|
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)
|
parts = line.split(None, 2)
|
||||||
if len(parts) == 3:
|
if len(parts) == 3 and parts[1] not in found:
|
||||||
return f"{parts[1]} {parts[2]}"
|
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(
|
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 "")
|
+ (f" stderr: {result.stderr.strip()!r}" if result.stderr.strip() else "")
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -71,7 +71,34 @@ class TestFetchHostKey(unittest.TestCase):
|
|||||||
with patch("subprocess.run", return_value=self._run_result("")):
|
with patch("subprocess.run", return_value=self._run_result("")):
|
||||||
with self.assertRaises(RuntimeError) as cm:
|
with self.assertRaises(RuntimeError) as cm:
|
||||||
fetch_host_key("example.com", "22")
|
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:
|
def test_raises_includes_stderr_when_present(self) -> None:
|
||||||
r = self._run_result("")
|
r = self._run_result("")
|
||||||
|
|||||||
Reference in New Issue
Block a user