Coverage for app/venv/lib/python3.14/site-packages/weblate/vcs/ssh.py: 50%
181 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
4from __future__ import annotations
6import hashlib
7import os
8import stat
9import subprocess
10from base64 import b64decode, b64encode
11from typing import TYPE_CHECKING, Literal, TypedDict
13from django.conf import settings
14from django.core.management.utils import find_command
15from django.utils.functional import cached_property
16from django.utils.translation import gettext, pgettext_lazy
18from weblate.trans.util import get_clean_env
19from weblate.utils import messages
20from weblate.utils.data import data_path
21from weblate.utils.files import cleanup_error_message
22from weblate.utils.hash import calculate_checksum
24if TYPE_CHECKING: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true
25 from pathlib import Path
27 from django_stubs_ext import StrOrPromise
29 from weblate.auth.models import AuthenticatedHttpRequest
31# SSH key files
32KNOWN_HOSTS = "known_hosts"
33CONFIG = "config"
36class KeyInfo(TypedDict):
37 private: str
38 public: str
39 name: StrOrPromise
40 keygen: list[str]
43KeyType = Literal["rsa", "ed25519"]
45KEYS: dict[KeyType, KeyInfo] = {
46 "rsa": {
47 "private": "id_rsa",
48 "public": "id_rsa.pub",
49 "name": pgettext_lazy("SSH key type", "RSA"),
50 "keygen": ["-b", "4096", "-t", "rsa"],
51 },
52 "ed25519": {
53 "private": "id_ed25519",
54 "public": "id_ed25519.pub",
55 "name": pgettext_lazy("SSH key type", "Ed25519"),
56 "keygen": ["-t", "ed25519"],
57 },
58}
61def ssh_file(filename: str) -> Path:
62 """Generate full path to SSH configuration file."""
63 return data_path("ssh") / filename
66def is_key_line(key: str) -> bool:
67 """Check whether this line looks like a valid known_hosts line."""
68 if not key:
69 return False
70 # Comment
71 if key[0] == "#":
72 return False
73 # Special entry like @cert-authority
74 if key[0] == "@":
75 return False
76 return (
77 " ssh-rsa " in key or " ecdsa-sha2-nistp256 " in key or " ssh-ed25519 " in key
78 )
81def parse_hosts_line(line: str) -> tuple[str, str, str]:
82 """Parse single hosts line into tuple host, key fingerprint."""
83 host, keytype, key = line.strip().split(None, 3)[:3]
84 digest = hashlib.sha256(b64decode(key)).digest()
85 fingerprint = b64encode(digest).rstrip(b"=").decode()
86 if host.startswith("|1|"):
87 # Translators: placeholder SSH hashed hostname
88 host = gettext("[hostname hashed]")
89 return host, keytype, fingerprint
92def get_host_keys() -> list[tuple[str, str, str]]:
93 """Return list of host keys."""
94 try:
95 result = []
96 with open(ssh_file(KNOWN_HOSTS)) as handle:
97 for line in handle:
98 line = line.strip()
99 if is_key_line(line):
100 result.append(parse_hosts_line(line))
101 except OSError:
102 return []
104 return result
107def get_key_data_raw(
108 key_type: KeyType = "rsa", kind: Literal["public", "private"] = "public"
109) -> tuple[str, str | None]:
110 """Return raw public key data."""
111 # Read key data if it exists
112 filename = KEYS[key_type][kind]
113 key_file = ssh_file(filename)
114 if os.path.exists(key_file):
115 with open(key_file) as handle:
116 return filename, handle.read()
117 return filename, None
120def get_key_data(key_type: KeyType = "rsa") -> dict[str, StrOrPromise | None]:
121 """Parse host key and returns it."""
122 filename, key_data = get_key_data_raw(key_type)
123 if key_data is not None:
124 _key_type_parsed, key_fingerprint, key_id = key_data.strip().split(None, 2)
125 return {
126 "key": key_data,
127 "fingerprint": key_fingerprint,
128 "id": key_id,
129 "filename": filename,
130 "type": key_type,
131 "name": KEYS[key_type]["name"],
132 }
133 return {
134 "key": None,
135 "type": key_type,
136 "name": KEYS[key_type]["name"],
137 }
140def get_all_key_data() -> dict[str, dict[str, StrOrPromise | None]]:
141 """Return all supported SSH keys."""
142 return {key_type: get_key_data(key_type) for key_type in KEYS}
145def ensure_ssh_key():
146 """Ensure SSH key is existing."""
147 result = None
148 for key_type in KEYS:
149 ssh_key = get_key_data(key_type)
150 if not ssh_key["key"]: 150 ↛ 153line 150 didn't jump to line 153 because the condition on line 150 was always true
151 generate_ssh_key(None, key_type)
152 ssh_key = get_key_data()
153 if key_type == "rsa":
154 result = ssh_key
155 return result
158def generate_ssh_key(
159 request: AuthenticatedHttpRequest | None, key_type: KeyType = "rsa"
160) -> None:
161 """Generate SSH key."""
162 key_info = KEYS[key_type]
163 keyfile = ssh_file(key_info["private"])
164 pubkeyfile = ssh_file(key_info["public"])
165 try:
166 # Actually generate the key
167 subprocess.run(
168 [
169 "ssh-keygen",
170 "-q",
171 *key_info["keygen"],
172 "-N",
173 "",
174 "-C",
175 settings.SITE_TITLE,
176 "-f",
177 keyfile,
178 ],
179 text=True,
180 check=True,
181 capture_output=True,
182 env=get_clean_env(),
183 )
184 except (subprocess.CalledProcessError, OSError) as exc:
185 error = getattr(exc, "output", "").strip()
186 if not error:
187 error = str(exc)
188 messages.error(
189 request,
190 gettext("Could not generate key: %s") % cleanup_error_message(error),
191 )
192 return
194 # Fix key permissions
195 os.chmod(keyfile, stat.S_IWUSR | stat.S_IRUSR)
196 os.chmod(pubkeyfile, stat.S_IWUSR | stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
198 messages.success(request, gettext("Created new SSH key."))
201def add_host_key(request: AuthenticatedHttpRequest | None, host, port="") -> None:
202 """Add host key for a host."""
203 if not host:
204 messages.error(request, gettext("Invalid host name given!"))
205 else:
206 cmdline = ["ssh-keyscan"]
207 if port:
208 cmdline.extend(["-p", str(port)])
209 cmdline.append(host)
210 try:
211 result = subprocess.run(
212 cmdline,
213 env=get_clean_env(),
214 check=True,
215 text=True,
216 capture_output=True,
217 )
218 keys = set()
219 for key in result.stdout.splitlines():
220 key = key.strip()
221 if not is_key_line(key):
222 continue
223 keys.add(key)
224 host, keytype, fingerprint = parse_hosts_line(key)
225 messages.warning(
226 request,
227 gettext(
228 "Added host key for %(host)s with fingerprint "
229 "%(fingerprint)s (%(keytype)s), "
230 "please verify that it is correct."
231 )
232 % {"host": host, "fingerprint": fingerprint, "keytype": keytype},
233 )
234 if keys:
235 known_hosts_file = ssh_file(KNOWN_HOSTS)
236 # Remove existing key entries
237 if known_hosts_file.exists():
238 with known_hosts_file.open() as handle:
239 keys.difference_update(line.strip() for line in handle)
240 # Write any new keys
241 if keys:
242 with known_hosts_file.open(mode="a") as handle:
243 for key in keys:
244 handle.write(key)
245 handle.write("\n")
246 else:
247 messages.error(
248 request,
249 gettext("Could not fetch public key for a host: %s") % result.stderr
250 or result.stdout,
251 )
252 except subprocess.CalledProcessError as exc:
253 messages.error(
254 request,
255 gettext("Could not fetch public key for a host: %s")
256 % cleanup_error_message(exc.stderr or exc.stdout),
257 )
258 except OSError as exc:
259 messages.error(request, gettext("Could not get host key: %s") % str(exc))
262GITHUB_RSA_KEY = (
263 "AAAAB3NzaC1yc2EAAAABIwAAAQEAq2A7hRGmdnm9tUDbO9IDSwBK6TbQa+PXYPCPy6rbTrTtw7"
264 "PHkccKrpp0yVhp5HdEIcKr6pLlVDBfOLX9QUsyCOV0wzfjIJNlGEYsdlLJizHhbn2mUjvSAHQq"
265 "ZETYP81eFzLQNnPHt4EVVUh7VfDESU84KezmD5QlWpXLmvU31/yMf+Se8xhHTvKSCZIFImWwoG"
266 "6mbUoWf9nzpIoaSjB+weqqUUmpaaasXVal72J+UX2B+2RPW3RcT0eOzQgqlJL3RKrTJvdsjE3J"
267 "EAvGq3lGHSZXy28G3skua2SmVi/w4yCE6gbODqnTWlg7+wC604ydGXA8VJiS5ap43JXiUFFAaQ=="
268)
271def cleanup_host_keys(*args, **kwargs) -> None:
272 known_hosts_file = ssh_file(KNOWN_HOSTS)
273 if not known_hosts_file.exists(): 273 ↛ 275line 273 didn't jump to line 275 because the condition on line 273 was always true
274 return
275 logger = kwargs.get("logger", print)
276 keys = []
277 with known_hosts_file.open() as handle:
278 for line in handle:
279 # Ignore IP address based RSA keys for GitHub, these
280 # are duplicate to hostname based and cause problems on
281 # migration to ECDSA.
282 # See https://github.com/WeblateOrg/weblate/issues/6830
283 if line[0].isdigit() and GITHUB_RSA_KEY in line:
284 logger(f"Removing deprecated RSA key for GitHub: {line.strip()}")
285 continue
287 # Avoid duplicates
288 if line in keys:
289 logger(f"Skipping duplicate key: {line.strip()}")
290 continue
292 keys.append(line)
294 with known_hosts_file.open(mode="w") as handle:
295 handle.writelines(keys)
298def can_generate_key():
299 """Check whether we can generate key."""
300 return find_command("ssh-keygen") is not None
303SSH_WRAPPER_TEMPLATE = r"""#!/bin/sh
304exec {command} \
305 -o "UserKnownHostsFile={known_hosts}" \
306 -o "IdentityFile={identity_rsa}" \
307 -o "IdentityFile={identity_ed25519}" \
308 -o StrictHostKeyChecking=yes \
309 -o HashKnownHosts=no \
310 -o UpdateHostKeys=yes \
311 -o ConnectTimeout=20 \
312 -o BatchMode=yes \
313 -F {config_file} \
314 {extra_args} \
315 "$@"
316"""
319class SSHWrapper:
320 # Custom ssh wrapper
321 # - use custom location for known hosts and key
322 # - do not hash it
323 # - strict hosk key checking
324 # - force not using system configuration (to avoid evil things as SendEnv)
326 @cached_property
327 def digest(self) -> str:
328 return calculate_checksum(self.get_content())
330 @property
331 def path(self) -> Path:
332 """
333 Calculates unique wrapper path.
335 It is based on template and DATA_DIR settings.
336 """
337 return ssh_file(f"bin-{self.digest}")
339 def get_content(self, command: str = "ssh") -> str:
340 return SSH_WRAPPER_TEMPLATE.format(
341 command=command,
342 known_hosts=ssh_file(KNOWN_HOSTS).as_posix(),
343 config_file=ssh_file(CONFIG).as_posix(),
344 identity_rsa=ssh_file(KEYS["rsa"]["private"]).as_posix(),
345 identity_ed25519=ssh_file(KEYS["ed25519"]["private"]).as_posix(),
346 extra_args=settings.SSH_EXTRA_ARGS,
347 )
349 @property
350 def filename(self) -> Path:
351 """Calculate unique wrapper filename."""
352 return self.path / "ssh"
354 def create(self) -> None:
355 """Create wrapper for SSH to pass custom known hosts and key."""
356 self.path.mkdir(parents=True, exist_ok=True)
358 ssh_config = ssh_file(CONFIG)
359 if not ssh_config.exists():
360 try:
361 with ssh_config.open(mode="x") as handle:
362 handle.write(
363 "# SSH configuration for customising SSH client in Weblate\n"
364 )
365 except OSError:
366 pass
368 for command in ("ssh", "scp"):
369 path = find_command(command)
370 if path is None: 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true
371 continue
372 filename = self.path / command
374 if not filename.exists():
375 filename.write_text(self.get_content(path))
377 if not os.access(filename, os.X_OK):
378 filename.chmod(0o755)
381SSH_WRAPPER = SSHWrapper()