Weblate has an argument injection in management console

@@ -81,7 +81,7 @@ jobs:
     - name: Enforced mypy
       # TODO: Enforce mypy to pass on certain modules, this should eventually grow to complete codebase
       env:
-        EXCLUDED_MODULES: accounts/|addons/|api/|auth/|billing/|fonts/|formats/|checks/|lang/|machinery/|metrics/|screenshots/|trans/(admin|autofixes|autotranslate|context_processors|debug|feeds|forms|guide|management|mixins|models/(agreement|pending|suggestion|comment|announcement|category|unit|translation|component)|tests|views|widgets)|utils/|vcs/
+        EXCLUDED_MODULES: accounts/|addons/|api/|auth/|billing/|fonts/|formats/|checks/|lang/|machinery/|metrics/|screenshots/|trans/(admin|autofixes|autotranslate|context_processors|debug|feeds|forms|guide|management|mixins|models/(agreement|pending|suggestion|comment|announcement|category|unit|translation|component)|tests|views|widgets)|utils/|vcs/(gpg|base|git|tests/test_vcs)
       run: |
         echo "::add-matcher::.github/matchers/mypy.json"
         if grep --silent --invert-match --extended-regexp "^weblate/($EXCLUDED_MODULES)" mypy.log ; then

@@ -131,7 +131,7 @@
 from weblate.vcs.base import RepositoryError, RepositorySymlinkError
 from weblate.vcs.git import GitMergeRequestBase, LocalRepository
 from weblate.vcs.models import VCS_REGISTRY
-from weblate.vcs.ssh import add_host_key
+from weblate.vcs.ssh import add_host_key, extract_url_host_port
 
 if TYPE_CHECKING:
     from collections.abc import Iterable
@@ -1695,17 +1695,11 @@ def add_ssh_host_key(self) -> None:
 
         def add(repo) -> None:
             self.log_info("checking for key to add for %s", repo)
-            parsed = urlparse(repo)
-            if not parsed.hostname:
-                parsed = urlparse(f"ssh://{repo}")
-            if not parsed.hostname:
+            hostname, port = extract_url_host_port(repo)
+            if not hostname:
                 return
-            try:
-                port = parsed.port
-            except ValueError:
-                port = ""
-            self.log_info("adding SSH key for %s:%s", parsed.hostname, port)
-            add_host_key(None, parsed.hostname, port)
+            self.log_info("adding SSH key for %s:%s", hostname, port)
+            add_host_key(None, hostname, port)
 
         add(self.repo)
         if self.push:

@@ -1,5 +1,11 @@
 #!/bin/sh
 # Fake key scan program to return SSH key
+if [ "$1" = "1.2.3.4" ]; then
+    exit 1
+elif [ "$1" = "2001:0db8:85a3:0000:0000:8a2e:0370:7334" ]; then
+    echo test error >&2
+    exit 2
+fi
 echo '# github.com SSH-2.0-libssh-0.7.0'
 echo 'github.com ssh-rsa AAAAB3NzaC1yc2EAAAABIwAAAQEAq2A7hRGmdnm9tUDbO9IDSwBK6TbQa+PXYPCPy6rbTrTtw7PHkccKrpp0yVhp5HdEIcKr6pLlVDBfOLX9QUsyCOV0wzfjIJNlGEYsdlLJizHhbn2mUjvSAHQqZETYP81eFzLQNnPHt4EVVUh7VfDESU84KezmD5QlWpXLmvU31/yMf+Se8xhHTvKSCZIFImWwoG6mbUoWf9nzpIoaSjB+weqqUUmpaaasXVal72J+UX2B+2RPW3RcT0eOzQgqlJL3RKrTJvdsjE3JEAvGq3lGHSZXy28G3skua2SmVi/w4yCE6gbODqnTWlg7+wC604ydGXA8VJiS5ap43JXiUFFAaQ=='
 echo '# github.com SSH-2.0-libssh-0.7.0'

@@ -97,4 +97,6 @@ def is_excluded(path: str) -> bool:
 
 def cleanup_error_message(text: str) -> str:
     """Remove absolute paths from the text."""
-    return text.replace(settings.CACHE_DIR, "...").replace(settings.DATA_DIR, "...")
+    return text.replace(settings.CACHE_DIR or "NONEXISTING_CACHE", "...").replace(
+        settings.DATA_DIR, "..."
+    )

@@ -10,8 +10,10 @@
 from base64 import b64decode, b64encode
 from contextlib import suppress
 from typing import TYPE_CHECKING, Literal, TypedDict
+from urllib.parse import urlparse
 
 from django.conf import settings
+from django.core.exceptions import ValidationError
 from django.core.management.utils import find_command
 from django.utils.functional import cached_property
 from django.utils.translation import gettext, pgettext_lazy
@@ -21,6 +23,7 @@
 from weblate.utils.data import data_path
 from weblate.utils.files import cleanup_error_message
 from weblate.utils.hash import calculate_checksum
+from weblate.utils.validators import DomainOrIPValidator
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -199,7 +202,43 @@ def generate_ssh_key(
     messages.success(request, gettext("Created new SSH key."))
 
 
-def add_host_key(request: AuthenticatedHttpRequest | None, host, port="") -> None:
+def extract_url_host_port(repo: str) -> tuple[str | None, int | None]:
+    """Extract hostname and port from repository URL."""
+    parsed = urlparse(repo)
+    if not parsed.hostname:
+        parsed = urlparse(f"ssh://{repo}")
+    if not parsed.hostname:
+        # Could not parse URL
+        return None, None
+    validator = DomainOrIPValidator()
+    try:
+        validator(parsed.hostname)
+    except ValidationError:
+        # Not a valid hostname
+        return None, None
+    port: int | None
+    try:
+        port = parsed.port
+    except ValueError:
+        port = None
+    return parsed.hostname, port
+
+
+def add_host_key_error(
+    request: AuthenticatedHttpRequest | None, error_text: str
+) -> None:
+    error_text = cleanup_error_message(error_text.strip())
+    if error_text:
+        messages.error(
+            request, gettext("Could not fetch public key for a host: %s") % error_text
+        )
+    else:
+        messages.error(request, gettext("Could not fetch public key for a host."))
+
+
+def add_host_key(
+    request: AuthenticatedHttpRequest | None, host: str, port: int | None = None
+) -> None:
     """Add host key for a host."""
     if not host:
         messages.error(request, gettext("Invalid host name given!"))
@@ -245,19 +284,11 @@ def add_host_key(request: AuthenticatedHttpRequest | None, host, port="") -> Non
                             handle.write(key)
                             handle.write("\n")
             else:
-                messages.error(
-                    request,
-                    gettext("Could not fetch public key for a host: %s") % result.stderr
-                    or result.stdout,
-                )
+                add_host_key_error(request, result.stderr or result.stdout)
         except subprocess.CalledProcessError as exc:
-            messages.error(
-                request,
-                gettext("Could not fetch public key for a host: %s")
-                % cleanup_error_message(exc.stderr or exc.stdout),
-            )
+            add_host_key_error(request, exc.stderr or exc.stdout)
         except OSError as exc:
-            messages.error(request, gettext("Could not get host key: %s") % str(exc))
+            add_host_key_error(request, str(exc))
 
 
 GITHUB_RSA_KEY = (

@@ -12,7 +12,7 @@
 from weblate.trans.tests.utils import get_test_file
 from weblate.utils.apps import check_data_writable
 from weblate.utils.unittest import tempdir_setting
-from weblate.vcs.ssh import SSHWrapper, get_host_keys, ssh_file
+from weblate.vcs.ssh import SSHWrapper, extract_url_host_port, get_host_keys, ssh_file
 
 TEST_HOSTS = get_test_file("known_hosts")
 
@@ -56,3 +56,14 @@ def test_ssh_args(self) -> None:
         timestamp = os.stat(filename).st_mtime
         wrapper.create()
         self.assertEqual(timestamp, os.stat(filename).st_mtime)
+
+    def test_extract_url_host_port(self) -> None:
+        self.assertEqual((None, None), extract_url_host_port(""))
+        self.assertEqual((None, None), extract_url_host_port("http://"))
+        self.assertEqual((None, None), extract_url_host_port("http:// invalid/url"))
+        self.assertEqual(
+            ("github.com", None), extract_url_host_port("git@github.com:repo")
+        )
+        self.assertEqual(
+            ("github.com", 1234), extract_url_host_port("git://github.com:1234/repo")
+        )

@@ -8,6 +8,7 @@
 from django.utils.translation import gettext_lazy
 
 from weblate.accounts.forms import EmailField
+from weblate.utils.validators import DomainOrIPValidator
 from weblate.wladmin.models import BackupService
 
 
@@ -24,7 +25,10 @@ class ActivateForm(forms.Form):
 
 class SSHAddForm(forms.Form):
     host = forms.CharField(
-        label=gettext_lazy("Hostname"), required=True, max_length=400
+        label=gettext_lazy("Hostname"),
+        required=True,
+        max_length=400,
+        validators=[DomainOrIPValidator()],
     )
     port = forms.IntegerField(
         label=gettext_lazy("Port"), required=False, min_value=1, max_value=65535
@@ -34,8 +38,6 @@ def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
         self.helper = FormHelper(self)
         self.helper.form_tag = False
-        self.helper.form_class = "form-inline"
-        self.helper.field_template = "bootstrap5/layout/inline_field.html"
 
 
 class TestMailForm(forms.Form):

@@ -81,22 +81,50 @@ def test_ssh_generate(self) -> None:
     def test_ssh_add(self) -> None:
         self.assertEqual(check_data_writable(app_configs=None, databases=None), [])
         oldpath = os.environ["PATH"]
+        hostsfile = data_path("ssh") / "known_hosts"
         try:
             os.environ["PATH"] = f"{get_test_file('')}:{os.environ['PATH']}"
             # Verify there is button for adding
             response = self.client.get(reverse("manage-ssh"))
             self.assertContains(response, "Add host key")
 
+            # Invalid parameters
+            response = self.client.post(
+                reverse("manage-ssh"), {"action": "add-host", "host": "-github.com"}
+            )
+            self.assertContains(response, "Enter a valid domain name or IP address.")
+            self.assertFalse(hostsfile.exists())
+
+            # Non-responding host
+            response = self.client.post(
+                reverse("manage-ssh"), {"action": "add-host", "host": "1.2.3.4"}
+            )
+            self.assertContains(response, "Could not fetch public key for a host.")
+            self.assertFalse(hostsfile.exists())
+
+            # Error response
+            response = self.client.post(
+                reverse("manage-ssh"),
+                {
+                    "action": "add-host",
+                    "host": "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
+                },
+            )
+            self.assertContains(
+                response, "Could not fetch public key for a host: test error"
+            )
+            self.assertFalse(hostsfile.exists())
+
             # Add the key
             response = self.client.post(
                 reverse("manage-ssh"), {"action": "add-host", "host": "github.com"}
             )
             self.assertContains(response, "Added host key for github.com")
+            self.assertTrue(hostsfile.exists())
         finally:
             os.environ["PATH"] = oldpath
 
         # Check the file contains it
-        hostsfile = data_path("ssh") / "known_hosts"
         self.assertIn("github.com", hostsfile.read_text())
 
     @tempdir_setting("BACKUP_DIR")

@@ -442,7 +442,7 @@ def ssh(request: AuthenticatedHttpRequest) -> HttpResponse:
     if action == "add-host":
         form = SSHAddForm(request.POST)
         if form.is_valid():
-            add_host_key(request, **form.cleaned_data)
+            add_host_key(request, form.cleaned_data["host"], form.cleaned_data["port"])
 
     context = {
         "public_ssh_keys": keys,