fix(security): validate workspace_id and region parameters to prevent hostname injection attacks

Author: agentscope-ai-java

dashscope/__init__.py

@@ -40,6 +40,7 @@
     base_compatible_api_url,
     base_http_api_url,
     base_websocket_api_url,
+    validate_workspace_id,
 )
 from dashscope.finetune.deployments import Deployments
 from dashscope.finetune.finetunes import FineTunes
@@ -103,8 +104,8 @@ def set_region(region: str, workspace_id: str = None):
             of the MaaS endpoint.
 
     Raises:
-        ValueError: If region is not supported or workspace_id is
-            empty.
+        ValueError: If region is not supported, workspace_id is
+            empty, or workspace_id is not a valid hostname label.
     """
     if region not in _MAAS_REGIONS:
         raise ValueError(
@@ -115,6 +116,8 @@ def set_region(region: str, workspace_id: str = None):
     if not workspace_id:
         raise ValueError("workspace_id is required")
 
+    validate_workspace_id(workspace_id)
+
     global base_http_api_url, base_compatible_api_url
     global base_websocket_api_url
 

dashscope/agentstudio/client.py

@@ -37,6 +37,7 @@
 )
 from dashscope.agentstudio.transport import SyncTransport, AsyncTransport
 from dashscope.common.api_key import get_default_api_key
+from dashscope.common.env import validate_region, validate_workspace_id
 
 
 def _resolve_base_url(
@@ -50,6 +51,10 @@ def _resolve_base_url(
     1. explicit base_url parameter (full URL)
     2. DASHSCOPE_AGENTSTUDIO_URL / AGENTSTUDIO_URL env
     3. Build from workspace + region template
+
+    Raises:
+        ValueError: If workspace or region is missing, or is not a valid
+            hostname label.
     """
     if explicit_url:
         return explicit_url
@@ -67,6 +72,11 @@ def _resolve_base_url(
             "or use base_url=... to override)",
         )
     rgn = region or AGENTSTUDIO_DEFAULT_REGION
+    # Both are interpolated into the host below; unvalidated values such as
+    # "a@evil.com/#" would move the effective host off *.maas.aliyuncs.com
+    # while the transport still sends Authorization: Bearer <api_key>.
+    validate_workspace_id(ws)
+    validate_region(rgn)
     return AGENTSTUDIO_BASE_URL_TEMPLATE.format(
         workspace=ws,
         region=rgn,

dashscope/common/env.py

@@ -2,6 +2,7 @@
 # Copyright (c) Alibaba, Inc. and its affiliates.
 
 import os
+import re
 
 from dashscope.common.constants import (
     DASHSCOPE_API_KEY_ENV,
@@ -11,6 +12,55 @@
 )
 from dashscope.common.error import InputRequired
 
+# workspace_id and region become DNS labels of the MaaS endpoint host, so
+# they must not contain characters (/, ?, #, @, :, whitespace, ...) that
+# could break out of the host and redirect requests elsewhere.
+# \A...\Z rather than ^...$: $ also matches before a trailing newline.
+_WORKSPACE_ID_PATTERN = re.compile(r"\A[A-Za-z0-9][A-Za-z0-9_-]{0,63}\Z")
+# RFC 1123 hostname label: 1-63 chars, no leading/trailing hyphen. Upper
+# case is allowed because DNS is case-insensitive and httpx lowercases the
+# host, so rejecting it would break working callers for no security gain.
+_REGION_PATTERN = re.compile(
+    r"\A[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\Z",
+)
+
+
+def validate_workspace_id(workspace_id):
+    """Validate that workspace_id is safe to use as a URL subdomain.
+
+    Raises:
+        ValueError: If workspace_id contains characters outside
+            [A-Za-z0-9_-], is empty, or exceeds 64 characters.
+    """
+    if not workspace_id or not _WORKSPACE_ID_PATTERN.fullmatch(
+        workspace_id,
+    ):
+        raise ValueError(
+            f"Invalid workspace_id {workspace_id!r}: must match "
+            "^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$",
+        )
+    return workspace_id
+
+
+def validate_region(region):
+    """Validate that region is safe to use as a URL subdomain.
+
+    Deliberately a hostname-label check and not a MAAS_REGIONS whitelist:
+    rejecting an unknown-but-well-formed region would break callers as soon
+    as a subpackage supports a region this list has not caught up with.
+
+    Raises:
+        ValueError: If region is not a valid hostname label.
+    """
+    if not region or not _REGION_PATTERN.fullmatch(region):
+        raise ValueError(
+            f"Invalid region {region!r}: must be a hostname label "
+            "(1-63 chars of [A-Za-z0-9-], not starting or ending "
+            "with '-')",
+        )
+    return region
+
+
 # MaaS regions: region -> URL subdomain identifier
 # cn-beijing is deliberately excluded: it is the default
 # DASHSCOPE_API_REGION and must keep using the legacy
@@ -36,6 +86,8 @@
 # define api base url, ensure end /
 if api_region in MAAS_REGIONS:
     _maas_region_id = MAAS_REGIONS[api_region]
+    if _default_workspace_id:
+        validate_workspace_id(_default_workspace_id)
     _ws = _default_workspace_id if _default_workspace_id else "{workspace_id}"
     _maas_host = f"{_ws}.{_maas_region_id}.maas.aliyuncs.com"
     base_http_api_url = os.environ.get(
@@ -83,6 +135,7 @@ def resolve_base_url(url, workspace_id=None):
         return url
     ws = workspace_id or _default_workspace_id
     if ws:
+        validate_workspace_id(ws)
         return url.replace("{workspace_id}", ws)
     raise InputRequired(
         "The base URL contains '{workspace_id}' but no workspace id "

tests/unit/test_agentstudio_base_url.py

@@ -0,0 +1,119 @@
+# -*- coding: utf-8 -*-
+"""AgentStudio base URL resolution.
+
+``workspace`` and ``region`` are interpolated into the MaaS host
+template ``https://{workspace}.{region}.maas.aliyuncs.com``. Both reach
+:func:`_resolve_base_url` from a public ``Client`` keyword argument or
+from the ``DASHSCOPE_WORKSPACE`` env var, so both are validated as DNS
+labels: an unvalidated ``"a@evil.com/#"`` makes httpx resolve the host
+to ``evil.com`` while the transport still attaches
+``Authorization: Bearer <api_key>``.
+"""
+
+import httpx
+import pytest
+
+from dashscope.agentstudio.client import _resolve_base_url
+
+MAAS_SUFFIX = ".maas.aliyuncs.com"
+
+# Values that break out of the host component when interpolated raw.
+HOST_BREAKING = [
+    "a@evil.com/#",
+    "evil.com/x",
+    "evil.com:443",
+    "evil.com?x",
+    "evil.com#frag",
+    "white space",
+    "abc\n",
+]
+
+
+@pytest.fixture(autouse=True)
+def _clean_env(monkeypatch):
+    """Keep ambient env vars from changing resolution precedence."""
+    for var in [
+        "DASHSCOPE_WORKSPACE",
+        "DASHSCOPE_AGENTSTUDIO_URL",
+        "AGENTSTUDIO_URL",
+    ]:
+        monkeypatch.delenv(var, raising=False)
+
+
+def test_builds_url_from_workspace_and_region():
+    url = _resolve_base_url(None, "ws-demo", "cn-beijing")
+    assert url == (
+        "https://ws-demo.cn-beijing.maas.aliyuncs.com/api/v1/agentstudio"
+    )
+
+
+def test_region_defaults_when_omitted():
+    url = _resolve_base_url(None, "ws-demo", None)
+    assert httpx.URL(url).host == f"ws-demo.cn-beijing{MAAS_SUFFIX}"
+
+
+@pytest.mark.parametrize("bad", HOST_BREAKING)
+def test_rejects_host_breaking_workspace(bad):
+    with pytest.raises(ValueError, match="workspace_id"):
+        _resolve_base_url(None, bad, "cn-beijing")
+
+
+@pytest.mark.parametrize("bad", HOST_BREAKING + ["-dash", "a" * 64])
+def test_rejects_host_breaking_region(bad):
+    with pytest.raises(ValueError, match="region"):
+        _resolve_base_url(None, "ws-demo", bad)
+
+
+def test_uppercase_region_still_resolves():
+    """Compat guard: httpx lowercases the host, so case must not be rejected."""
+    url = _resolve_base_url(None, "ws-demo", "CN-Beijing")
+    assert httpx.URL(url).host == f"ws-demo.cn-beijing{MAAS_SUFFIX}"
+
+
+def test_rejects_workspace_from_env(monkeypatch):
+    """The env-sourced workspace is validated on the same path."""
+    monkeypatch.setenv("DASHSCOPE_WORKSPACE", "a@evil.com/#")
+    with pytest.raises(ValueError, match="workspace_id"):
+        _resolve_base_url(None, None, "cn-beijing")
+
+
+def test_resolved_host_stays_under_maas_domain():
+    """Positive control for the property the checks above protect."""
+    url = _resolve_base_url(None, "llm-abc_def", "ap-southeast-1")
+    assert httpx.URL(url).host.endswith(MAAS_SUFFIX)
+
+
+def test_missing_workspace_error_is_actionable():
+    with pytest.raises(ValueError, match="workspace is required"):
+        _resolve_base_url(None, None, "cn-beijing")
+
+
+def test_explicit_base_url_overrides_without_validation():
+    """A deliberate full-URL override is trusted, per documented priority."""
+    assert _resolve_base_url("https://my.host/api", None, None) == (
+        "https://my.host/api"
+    )
+
+
+def test_env_base_url_overrides_without_validation(monkeypatch):
+    monkeypatch.setenv("DASHSCOPE_AGENTSTUDIO_URL", "https://my.host/api")
+    assert _resolve_base_url(None, None, None) == "https://my.host/api"
+
+
+def test_sync_client_rejects_injected_workspace():
+    """The attack surface is the public constructor, not just the helper."""
+    from dashscope.agentstudio import Client
+
+    with pytest.raises(ValueError, match="workspace_id"):
+        Client(api_key="test-key", workspace="a@evil.com/#")
+
+
+def test_async_client_rejects_injected_region():
+    from dashscope.agentstudio import AsyncClient
+
+    with pytest.raises(ValueError, match="region"):
+        AsyncClient(
+            api_key="test-key",
+            workspace="ws-demo",
+            region="x@evil.com/#",
+        )

tests/unit/test_maas_region.py

@@ -51,6 +51,84 @@ def test_resolve_all_regions(self):
                 f"https://my-ws.{region_id}.maas.aliyuncs.com/api/v1",
             )
 
+    def test_resolve_rejects_invalid_workspace(self):
+        url = "https://{workspace_id}.us-east-1.maas.aliyuncs.com/api/v1"
+        for bad in ["evil.com/x", "a@evil.com", "evil.com#f", "evil com"]:
+            with self.assertRaises(ValueError):
+                resolve_base_url(url, bad)
+
+    def test_resolve_rejects_invalid_env_default_workspace(self):
+        import dashscope.common.env as env_mod
+
+        url = "https://{workspace_id}.us-east-1.maas.aliyuncs.com/api/v1"
+        with patch.object(env_mod, "_default_workspace_id", "evil.com/x"):
+            with self.assertRaises(ValueError):
+                resolve_base_url(url, None)
+
+
+class TestValidateWorkspaceId(unittest.TestCase):
+    """Test validate_workspace_id character restrictions."""
+
+    def test_valid_ids(self):
+        from dashscope.common.env import validate_workspace_id
+
+        for valid in ["ws-123", "llm-abc_def", "A0", "a" * 64]:
+            self.assertEqual(validate_workspace_id(valid), valid)
+
+    def test_invalid_ids(self):
+        from dashscope.common.env import validate_workspace_id
+
+        for invalid in [
+            "",
+            None,
+            "-leading-dash",
+            "evil.com",
+            "evil.com/x",
+            "evil.com?x",
+            "evil.com#f",
+            "a@evil.com",
+            "evil.com:443",
+            "white space",
+            "a" * 65,
+            # $ also matches before a trailing newline; \Z does not
+            "abc\n",
+        ]:
+            with self.assertRaises(ValueError):
+                validate_workspace_id(invalid)
+
+
+class TestValidateRegion(unittest.TestCase):
+    """Test validate_region character restrictions."""
+
+    def test_valid_regions(self):
+        from dashscope.common.env import validate_region
+
+        # Upper case is accepted: DNS is case-insensitive and httpx
+        # lowercases the host, so rejecting it broke working callers.
+        for valid in [*MAAS_REGIONS, "cn-beijing", "CN-Beijing", "a"]:
+            self.assertEqual(validate_region(valid), valid)
+
+    def test_invalid_regions(self):
+        from dashscope.common.env import validate_region
+
+        for invalid in [
+            "",
+            None,
+            # would move the effective host off *.maas.aliyuncs.com
+            "x@evil.com/#",
+            "evil.com/x",
+            "evil.com:443",
+            "evil.com?x",
+            "white space",
+            "-leading-dash",
+            "trailing-dash-",
+            "abc\n",
+            # RFC 1123 caps a label at 63 chars
+            "a" * 64,
+        ]:
+            with self.assertRaises(ValueError):
+                validate_region(invalid)
+
 
 class TestMaasRegions(unittest.TestCase):
     """Test MAAS_REGIONS constant."""
@@ -128,6 +206,16 @@ def test_maas_region_with_workspace(self):
             "https://ws-test-123.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1",
         )
 
+    def test_maas_region_with_invalid_workspace_raises(self):
+        """Invalid DASHSCOPE_WORKSPACE_ID must fail fast at env load."""
+        with self.assertRaises(ValueError):
+            self._reload_env(
+                {
+                    "DASHSCOPE_API_REGION": "ap-southeast-1",
+                    "DASHSCOPE_WORKSPACE_ID": "evil.com/x",
+                },
+            )
+
     def test_maas_region_without_workspace(self):
         env = self._reload_env(
             {"DASHSCOPE_API_REGION": "us-east-1"},

tests/unit/test_set_region.py

@@ -139,6 +139,29 @@ def test_set_region_none_workspace_id(self):
 
         assert "workspace_id is required" in str(exc_info.value)
 
+    @pytest.mark.parametrize(
+        "bad_workspace_id",
+        [
+            "evil.com/x",
+            "evil.com?x=1",
+            "evil.com#frag",
+            "a@evil.com",
+            "evil.com:443",
+            "evil com",
+            "evil.com",
+            "-evil",
+            "a" * 65,
+        ],
+    )
+    def test_set_region_invalid_workspace_id(self, bad_workspace_id):
+        """workspace_id with URL-breaking characters must be rejected."""
+        with pytest.raises(ValueError) as exc_info:
+            dashscope.set_region("ap-southeast-1", bad_workspace_id)
+
+        assert "Invalid workspace_id" in str(exc_info.value)
+        # Globals must not be modified on rejection.
+        assert bad_workspace_id not in dashscope.base_http_api_url
+
     def test_set_region_multiple_times(self):
         """Test switching between multiple regions."""
         # First region