fix(machinery): limit allowed URLs

Allow only public URLs by default to reduce risk of SSRF.

Author: nijel

docs/admin/config.rst

@@ -81,6 +81,26 @@ This is currently used in the following places:
 
    * :setting:`ALLOWED_ASSET_SIZE`
 
+.. setting:: ALLOWED_MACHINERY_DOMAINS
+
+ALLOWED_MACHINERY_DOMAINS
+-------------------------
+
+Configures which custom machinery domains are explicitly allowed in project-level
+machine translation configuration.
+
+This setting applies only to machinery services and does not affect
+:setting:`ALLOWED_ASSET_DOMAINS`.
+
+It expects a list of host/domain names. You can use fully qualified names or
+prepend with a period as a wildcard to match all subdomains.
+
+Defaults to ``[]``.
+
+The allowlist only affects configuration-time validation for project-managed
+machinery endpoints. Runtime checks still reject destinations that resolve to
+private or otherwise non-public addresses.
+
 .. setting:: ALLOWED_ASSET_SIZE
 
 ALLOWED_ASSET_SIZE

docs/changes.rst

@@ -20,6 +20,7 @@ Weblate 5.17
 
 .. rubric:: Bug fixes
 
+* Hardened project-level machine translation against SSRF by blocking private-network targets for untrusted endpoints and hiding untrusted remote error details.
 * Prevented removing the last team from a project token.
 * Batch automatic translation now uses project-level machinery configuration instead of only site-wide settings.
 * Fixed sorting by the **Unreviewed** column in listings.

weblate/api/tests.py

@@ -3467,6 +3467,24 @@ def test_install_machinery(self) -> None:
 
         self.assertEqual(new_config, response.data)
 
+    def test_install_machinery_blocks_private_project_target(self) -> None:
+        self.component.project.add_user(self.user, "Administration")
+
+        response = self.do_request(
+            "api:project-machinery-settings",
+            self.project_kwargs,
+            method="post",
+            code=400,
+            superuser=False,
+            request={
+                "service": "deepl",
+                "configuration": {"key": "x", "url": "http://127.0.0.1:11434/"},
+            },
+            format="json",
+        )
+
+        self.assertIn("URL domain is not allowed.", str(response.data))
+
 
 class ComponentAPITest(APIBaseTest):
     def setUp(self) -> None:

weblate/api/views.py

@@ -1515,7 +1515,9 @@ def machinery_settings(self, request: Request, **kwargs):
                 raise ValidationError({"service": "Missing service name"}) from error
 
             service, configuration, errors = validate_service_configuration(
-                service_name, request.data.get("configuration", "{}")
+                service_name,
+                request.data.get("configuration", "{}"),
+                allow_private_targets=False,
             )
 
             if service is None or errors:
@@ -1554,7 +1556,9 @@ def machinery_settings(self, request: Request, **kwargs):
             valid_configurations: dict[str, dict] = {}
             for service_name, configuration in request.data.items():
                 service, configuration, errors = validate_service_configuration(
-                    service_name, configuration
+                    service_name,
+                    configuration,
+                    allow_private_targets=False,
                 )
 
                 if service is None or errors:

weblate/machinery/anthropic.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+from typing import ClassVar
 from urllib.parse import urljoin
 
 from .base import MachineryRateLimitError
@@ -20,6 +21,7 @@ class AnthropicTranslation(BaseLLMTranslation):
     """
 
     name = "Anthropic"
+    trusted_error_hosts: ClassVar[set[str]] = {"api.anthropic.com"}
     end_point = "/v1/messages"
     settings_form = AnthropicMachineryForm
     version_added = "5.16"

weblate/machinery/base.py

@@ -15,8 +15,9 @@
 from html import escape, unescape
 from itertools import chain
 from typing import TYPE_CHECKING, ClassVar
-from urllib.parse import quote
+from urllib.parse import quote, urlparse
 
+from django.conf import settings
 from django.core.cache import cache
 from django.core.exceptions import ValidationError
 from django.utils.functional import cached_property
@@ -28,8 +29,10 @@
 from weblate.machinery.forms import BaseMachineryForm
 from weblate.utils.docs import DocVersionsMixin
 from weblate.utils.errors import report_error
+from weblate.utils.forms import WeblateServiceURLField
 from weblate.utils.hash import calculate_dict_hash, calculate_hash, hash_to_checksum
-from weblate.utils.requests import http_request
+from weblate.utils.outbound import is_allowlisted_hostname
+from weblate.utils.requests import http_request, validate_request_url
 from weblate.utils.similarity import Comparer
 from weblate.utils.site import get_site_url
 
@@ -108,20 +111,21 @@ class BatchMachineTranslation(DocVersionsMixin):
 
     validate_source_language = "en"
     validate_target_language = "de"
+    trusted_error_hosts: ClassVar[set[str]] = set()
 
     @classmethod
     def get_rank(cls):
         return cls.max_score + cls.rank_boost
 
-    def __init__(self, settings: SettingsDict) -> None:
+    def __init__(self, configuration: SettingsDict) -> None:
         """Create new machine translation object."""
         self.mtid = self.get_identifier()
         self.rate_limit_cache = f"{self.mtid}-rate-limit"
         self.languages_cache = f"{self.mtid}-languages"
         self.comparer = Comparer()
         self.supported_languages_error: Exception | None = None
         self.supported_languages_error_age: float = 0
-        self.settings = settings
+        self.settings = configuration
 
     def delete_cache(self) -> None:
         cache.delete_many([self.rate_limit_cache, self.languages_cache])
@@ -187,29 +191,108 @@ def check_failure(self, response: Response) -> None:
         try:
             response.raise_for_status()
         except HTTPError as error:
-            detail = response.text
-            try:
-                payload = response.json()
-            except JSONDecodeError:
-                pass
-            else:
-                if isinstance(payload, dict) and payload:
-                    if detail_error := payload.get("error"):
-                        if isinstance(detail_error, str):
-                            detail = detail_error
-                        elif isinstance(detail_error, dict):
-                            if "message" in detail_error:
-                                detail = detail_error["message"]
-                            else:
-                                detail = str(detail_error)
-                    else:
-                        detail = str(payload)
-
-            if detail:
+            if detail := self.get_error_detail(response):
                 message = f"{error.args[0]}: {detail[:200]}"
                 raise HTTPError(message, response=response) from error
             raise
 
+    @property
+    def allow_private_targets(self) -> bool:
+        return "_project" not in self.settings
+
+    def validate_runtime_url(self, url: str) -> None:
+        validate_request_url(url, allow_private_targets=self.allow_private_targets)
+
+    @staticmethod
+    def get_host_from_setting(value: object) -> str | None:
+        if not isinstance(value, str):
+            return None
+        if "://" in value:
+            return urlparse(value).hostname
+        return value or None
+
+    def get_trusted_error_hosts(self) -> set[str]:
+        hosts = set(settings.ALLOWED_MACHINERY_DOMAINS)
+        hosts.update(self.trusted_error_hosts)
+        if self.allow_private_targets or self.settings_form is None:
+            return hosts
+
+        form = self.settings_form(self.__class__)
+        for field_name, field in form.fields.items():
+            values: set[str] = set()
+            if initial := getattr(field, "initial", None):
+                values.add(initial)
+            values.update(value for value, _label in getattr(field, "choices", ()))
+
+            current_value = self.settings.get(field_name)
+            if current_value in values and (
+                host := self.get_host_from_setting(current_value)
+            ):
+                hosts.add(host)
+
+            if isinstance(field, WeblateServiceURLField):
+                for value in values:
+                    if host := self.get_host_from_setting(value):
+                        hosts.add(host)
+        return hosts
+
+    @classmethod
+    def has_configurable_outbound_target(cls) -> bool:
+        if cls.settings_form is None:
+            return False
+
+        form = cls.settings_form(cls)
+        for field_name, field in form.fields.items():
+            if isinstance(field, WeblateServiceURLField):
+                return True
+            if field_name in form.network_host_fields:
+                return True
+        return False
+
+    def can_display_error_detail(self, response: Response) -> bool:
+        if self.allow_private_targets:
+            return True
+        return self.is_trusted_error_host(response)
+
+    def is_trusted_error_host(self, response: Response) -> bool:
+        if (
+            self.settings_form is not None
+            and not self.has_configurable_outbound_target()
+        ):
+            return True
+        hostname = urlparse(response.url).hostname or ""
+        return is_allowlisted_hostname(hostname, list(self.get_trusted_error_hosts()))
+
+    def get_error_detail(self, response: Response) -> str | None:
+        if not self.can_display_error_detail(response):
+            return None
+        trusted_host = self.is_trusted_error_host(response)
+
+        try:
+            payload = response.json()
+        except JSONDecodeError:
+            if trusted_host:
+                return response.text or None
+            return None
+
+        if isinstance(payload, dict):
+            if (message := payload.get("message")) and isinstance(message, str):
+                return message
+            if (detail := payload.get("detail")) and isinstance(detail, str):
+                return detail
+            if error := payload.get("error"):
+                if isinstance(error, str):
+                    return error
+                if isinstance(error, dict):
+                    detail_message = error.get("message")
+                    if isinstance(detail_message, str):
+                        return detail_message
+        elif isinstance(payload, str) and trusted_host:
+            return payload
+        if trusted_host:
+            return response.text or None
+        return None
+
     def request(self, method, url, skip_auth=False, **kwargs):
         """Perform JSON request."""
         # Create custom headers
@@ -231,6 +314,8 @@ def request(self, method, url, skip_auth=False, **kwargs):
             timeout=self.request_timeout,
             auth=self.get_auth(),
             raise_for_status=False,
+            validate_url=True,
+            allow_private_targets=self.allow_private_targets,
             **kwargs,
         )
 

weblate/machinery/deepl.py

@@ -46,6 +46,10 @@ class DeepLTranslation(
     target_language_map: ClassVar[dict[str, str]] = {
         "PT": "PT-PT",
     }
+    trusted_error_hosts: ClassVar[set[str]] = {
+        "api.deepl.com",
+        "api-free.deepl.com",
+    }
     highlight_syntax = True
     settings_form = DeepLMachineryForm
     glossary_count_limit = 1000

weblate/machinery/forms.py

@@ -12,11 +12,14 @@
 from django.utils.translation import gettext, gettext_lazy, pgettext_lazy
 
 from weblate.utils.forms import WeblateServiceURLField
+from weblate.utils.validators import validate_machinery_hostname, validate_machinery_url
 
 from .types import SourceLanguageChoices
 
 
 class BaseMachineryForm(forms.Form):
+    network_host_fields = frozenset({"base_url", "endpoint_url"})
+
     source_language = forms.ChoiceField(
         label=pgettext_lazy(
             "Automatic suggestion service configuration", "Source language selection"
@@ -26,8 +29,11 @@ class BaseMachineryForm(forms.Form):
         required=False,
     )
 
-    def __init__(self, machinery, *args, **kwargs) -> None:
+    def __init__(
+        self, machinery, *args, allow_private_targets: bool = True, **kwargs
+    ) -> None:
         self.machinery = machinery
+        self.allow_private_targets = allow_private_targets
         super().__init__(*args, **kwargs)
 
     def serialize_form(self):
@@ -40,9 +46,25 @@ def clean(self) -> None:
                 continue
             if field not in settings:
                 return
+        self.validate_endpoint_fields(settings)
+        if not self.allow_private_targets:
+            settings = {**settings, "_project": object()}
         machinery = self.machinery(settings)
         machinery.validate_settings()
 
+    def validate_endpoint_fields(self, settings) -> None:
+        for field_name, field in self.fields.items():
+            if (value := settings.get(field_name)) in {"", None}:
+                continue
+            if isinstance(field, WeblateServiceURLField):
+                validate_machinery_url(
+                    value, allow_private_targets=self.allow_private_targets
+                )
+            elif field_name in self.network_host_fields and isinstance(value, str):
+                validate_machinery_hostname(
+                    value, allow_private_targets=self.allow_private_targets
+                )
+
 
 class KeyMachineryForm(BaseMachineryForm):
     key = forms.CharField(

weblate/machinery/libretranslate.py

@@ -6,7 +6,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 from .base import BatchMachineTranslation
 from .forms import LibreTranslateMachineryForm
@@ -22,6 +22,7 @@ class LibreTranslateTranslation(BatchMachineTranslation):
 
     name = "LibreTranslate"
     max_score = 89
+    trusted_error_hosts: ClassVar[set[str]] = {"libretranslate.com"}
     version_added = "4.7.1"
     settings_form = LibreTranslateMachineryForm
     request_timeout = 20

weblate/machinery/microsoft.py

@@ -6,6 +6,7 @@
 
 from datetime import timedelta
 from typing import TYPE_CHECKING, ClassVar
+from urllib.parse import urlparse
 
 from django.utils import timezone
 
@@ -57,6 +58,12 @@ class MicrosoftCognitiveTranslation(XMLMachineTranslationMixin, MachineTranslati
     def get_identifier(cls) -> str:
         return "microsoft-translator"
 
+    @classmethod
+    def get_application_hosts(cls) -> set[str]:
+        return {
+            value for value, _label in cls.settings_form.base_fields["base_url"].choices
+        }
+
     def __init__(self, settings: SettingsDict) -> None:
         """Check configuration."""
         super().__init__(settings)
@@ -108,10 +115,8 @@ def check_failure(self, response) -> None:
         # Microsoft tends to use utf-8-sig instead of plain utf-8
         response.encoding = response.apparent_encoding
         super().check_failure(response)
-        if (
-            response.url.startswith("https://api.cognitive.microsofttranslator.com/")
-            and response.status_code == 200
-        ):
+        hostname = urlparse(response.url).hostname
+        if response.status_code == 200 and hostname in self.get_application_hosts():
             payload = response.json()
 
             # We should get an object, string usually means an error