fix(studio): enforce security boundaries

tests/auth/test_oauth2_redirect.py

@@ -0,0 +1,60 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from starlette.requests import Request
+
+from veadk.auth.middleware.oauth2_auth import _resolve_redirect_after_auth
+
+
+def _request() -> Request:
+    return Request(
+        {
+            "type": "http",
+            "method": "GET",
+            "scheme": "https",
+            "server": ("studio.example.com", 443),
+            "path": "/oauth2/login",
+            "query_string": b"",
+            "headers": [(b"host", b"studio.example.com")],
+        }
+    )
+
+
+@pytest.mark.parametrize(
+    "redirect",
+    [
+        "//evil.example/phish",
+        r"/\\evil.example/phish",
+        "/safe\nlocation",
+        "https://evil.example/phish",
+    ],
+)
+def test_redirect_rejects_external_or_ambiguous_targets(redirect: str) -> None:
+    assert _resolve_redirect_after_auth(_request(), redirect) == "/"
+
+
+@pytest.mark.parametrize(
+    ("redirect", "expected"),
+    [
+        ("/agents?tab=mine", "/agents?tab=mine"),
+        ("agents", "/agents"),
+        (
+            "https://studio.example.com/agents",
+            "https://studio.example.com/agents",
+        ),
+    ],
+)
+def test_redirect_keeps_local_targets(redirect: str, expected: str) -> None:
+    assert _resolve_redirect_after_auth(_request(), redirect) == expected

tests/cli/test_generated_agent_backend_codegen.py

@@ -105,6 +105,45 @@ def test_project_policy_allows_mcp_stdio_but_debug_rejects_it() -> None:
         validate_debug_policy(draft, allow_local_runtime_resources=True)
 
 
+def test_debug_policy_rejects_custom_model_api_base() -> None:
+    draft = AgentDraft(
+        name="demo",
+        modelApiBase="https://attacker.example/api/v3",
+    )
+
+    validate_project_policy(draft)
+    with pytest.raises(DebugPolicyError, match="Custom modelApiBase"):
+        validate_debug_policy(draft)
+
+
+@pytest.mark.parametrize(
+    "model_api_base",
+    [
+        "https://ark.cn-beijing.volces.com/api/v3/",
+        "https://ark.ap-southeast.bytepluses.com/api/v3",
+    ],
+)
+def test_debug_policy_allows_builtin_model_api_bases(model_api_base: str) -> None:
+    validate_debug_policy(AgentDraft(name="demo", modelApiBase=model_api_base))
+
+
+def test_debug_policy_requires_https_for_external_tools(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        socket,
+        "getaddrinfo",
+        lambda *_args, **_kwargs: [(socket.AF_INET, 0, 0, "", ("203.0.113.10", 443))],
+    )
+    draft = AgentDraft(
+        name="demo",
+        mcpTools=[{"transport": "http", "url": "http://tools.example/mcp"}],
+    )
+
+    with pytest.raises(DebugPolicyError, match="must use https"):
+        validate_debug_policy(draft)
+
+
 def test_security_rejects_enabled_a2a_registry_without_space_id() -> None:
     draft = AgentDraft(
         name="demo",

tests/cli/test_studio_rbac.py

@@ -728,6 +728,104 @@ def test_gateway_role_uses_jwt_and_ignores_local_identity_header(
     assert response.json()["role"] == "admin"
 
 
+def test_user_cannot_enumerate_server_credential_resources(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    app = _create_studio_app(
+        monkeypatch,
+        tmp_path,
+        auth_mode="gateway",
+        admins="admin",
+        developers="developer",
+    )
+    headers = {"Authorization": f"Bearer {_unsigned_jwt({'sub': 'reader'})}"}
+
+    with TestClient(app) as client:
+        responses = [
+            client.get("/web/a2a-spaces", headers=headers),
+            client.get("/web/viking-knowledgebases", headers=headers),
+            client.get("/web/skill-spaces", headers=headers),
+            client.get("/web/skill-spaces/space-1/skills", headers=headers),
+            client.get("/web/skill-spaces/space-1/skills/skill-1", headers=headers),
+        ]
+
+    assert [response.status_code for response in responses] == [403] * 5
+
+
+def test_user_cannot_access_another_users_local_sessions_or_media(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    app = _create_studio_app(
+        monkeypatch,
+        tmp_path,
+        auth_mode="gateway",
+        admins="admin",
+        developers="developer",
+    )
+    headers = {"Authorization": f"Bearer {_unsigned_jwt({'sub': 'reader'})}"}
+
+    async def _echo_run_body(request: Request) -> dict[str, Any]:
+        return await request.json()
+
+    app.add_api_route("/run_sse", _echo_run_body, methods=["POST"])
+    app.router.routes.insert(0, app.router.routes.pop())
+
+    with TestClient(app) as client:
+        own_upload = client.post(
+            "/web/media",
+            headers=headers,
+            data={
+                "app_name": "demo",
+                "user_id": "reader",
+                "session_id": "session",
+            },
+            files={"file": ("canary.txt", b"owner canary", "text/plain")},
+        )
+        cross_upload = client.post(
+            "/web/media",
+            headers=headers,
+            data={
+                "app_name": "demo",
+                "user_id": "owner",
+                "session_id": "session",
+            },
+            files={"file": ("canary.txt", b"cross user", "text/plain")},
+        )
+        cross_session = client.get(
+            "/apps/site-packages/users/owner/sessions/session",
+            headers=headers,
+        )
+        cross_delete = client.delete(
+            "/apps/site-packages/users/owner/sessions/session",
+            headers=headers,
+        )
+        own_run_payload = {
+            "appName": "site-packages",
+            "userId": "reader",
+            "sessionId": "session",
+        }
+        own_run = client.post(
+            "/run_sse",
+            headers=headers,
+            json=own_run_payload,
+        )
+        cross_run = client.post(
+            "/run_sse",
+            headers=headers,
+            json={"appName": "site-packages", "userId": "owner"},
+        )
+
+    assert own_upload.status_code == 200
+    assert cross_upload.status_code == 403
+    assert cross_session.status_code == 403
+    assert cross_delete.status_code == 403
+    assert own_run.status_code == 200
+    assert own_run.json() == own_run_payload
+    assert cross_run.status_code == 403
+
+
 def test_non_admin_runtime_list_uses_one_owner_filtered_request(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,

tests/multimodal/test_api.py

@@ -15,7 +15,7 @@
 
 from pathlib import Path
 
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.testclient import TestClient
 
 from veadk.multimodal.api import mount_media_routes
@@ -75,3 +75,43 @@ def test_upload_rejects_unsupported_file(tmp_path: Path) -> None:
 
     assert response.status_code == 400
     assert "Unsupported media type" in response.json()["detail"]
+
+
+def test_media_routes_authorize_every_user_scoped_operation(tmp_path: Path) -> None:
+    app = FastAPI()
+    authorized: list[str] = []
+
+    def authorize(_request: Request, user_id: str) -> None:
+        authorized.append(user_id)
+        if user_id != "owner":
+            raise HTTPException(status_code=403, detail="forbidden")
+
+    mount_media_routes(
+        app,
+        MediaService(LocalMediaStorage(tmp_path)),
+        authorize=authorize,
+    )
+    client = TestClient(app)
+
+    denied_upload = client.post(
+        "/web/media",
+        data={"app_name": "demo", "user_id": "other", "session_id": "session"},
+        files={"file": ("notes.txt", b"secret", "text/plain")},
+    )
+    owner_upload = client.post(
+        "/web/media",
+        data={"app_name": "demo", "user_id": "owner", "session_id": "session"},
+        files={"file": ("notes.txt", b"secret", "text/plain")},
+    )
+    media_id = owner_upload.json()["id"]
+
+    assert denied_upload.status_code == 403
+    assert owner_upload.status_code == 200
+    assert (
+        client.get(f"/web/media/demo/other/session/{media_id}/content").status_code
+        == 403
+    )
+    assert client.get(f"/web/media/demo/other/session/{media_id}").status_code == 403
+    assert client.delete(f"/web/media/demo/other/session/{media_id}").status_code == 403
+    assert client.delete("/web/media/demo/other/session").status_code == 403
+    assert authorized == ["other", "owner", "other", "other", "other", "other"]

tests/test_ve_apig.py

@@ -0,0 +1,73 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from types import SimpleNamespace
+from typing import Any
+
+from veadk.integrations.ve_apig.ve_apig import APIGateway
+
+
+def test_disable_route_cors_turns_off_credentials_and_origin_reflection() -> None:
+    captured: dict[str, Any] = {}
+
+    class _Result:
+        def __init__(self, value: Any) -> None:
+            self.value = value
+
+        def get(self) -> Any:
+            return self.value
+
+    class _Client:
+        def get_route(self, request: Any, *, async_req: bool) -> _Result:
+            captured["get_request"] = request
+            return _Result(
+                SimpleNamespace(
+                    route=SimpleNamespace(
+                        name="studio-route",
+                        enable=True,
+                        fallback_setting="fallback",
+                        match_rule="match",
+                        priority=10,
+                        upstream_list="upstream",
+                        advanced_setting=SimpleNamespace(
+                            header_operations="headers",
+                            mirror_policies="mirrors",
+                            retry_policy_setting="retry",
+                            timeout_setting="timeout",
+                            url_rewrite_setting="rewrite",
+                        ),
+                    )
+                )
+            )
+
+        def update_route(self, request: Any, *, async_req: bool) -> _Result:
+            captured["request"] = request
+            captured["async_req"] = async_req
+            return _Result(None)
+
+    gateway = object.__new__(APIGateway)
+    gateway.apig_20221112_client = _Client()
+
+    gateway.disable_route_cors("route-1")
+
+    request = captured["request"]
+    cors = request.advanced_setting.cors_policy_setting
+    assert request.id == "route-1"
+    assert request.name == "studio-route"
+    assert request.match_rule == "match"
+    assert request.upstream_list == "upstream"
+    assert request.advanced_setting.timeout_setting == "timeout"
+    assert cors.enable is False
+    assert cors.allow_credentials is False
+    assert captured["async_req"] is True

veadk/auth/middleware/oauth2_auth.py

@@ -1324,15 +1324,21 @@ def _resolve_redirect_after_auth(request: Request, redirect: Optional[str]) -> s
         return "/"
 
     redirect = redirect.strip()
-    if redirect.startswith("/"):
-        return redirect
-
+    if not redirect or "\\" in redirect or any(ord(char) < 32 for char in redirect):
+        logger.warning("Unsafe redirect ignored: %s", redirect)
+        return "/"
     parsed = urllib.parse.urlparse(redirect)
     if not parsed.scheme and not parsed.netloc:
+        if redirect.startswith("/") and not redirect.startswith("//"):
+            return redirect
         return f"/{redirect.lstrip('/')}"
 
     current = urllib.parse.urlparse(str(request.url))
-    if parsed.scheme == current.scheme and parsed.netloc == current.netloc:
+    if (
+        parsed.scheme in {"http", "https"}
+        and parsed.scheme == current.scheme
+        and parsed.netloc == current.netloc
+    ):
         return redirect
 
     logger.warning("Unsafe redirect ignored: %s", redirect)