diff --git a/ouroboros/gateway/mcp.py b/ouroboros/gateway/mcp.py index 9266f2e40..14fc38708 100644 --- a/ouroboros/gateway/mcp.py +++ b/ouroboros/gateway/mcp.py @@ -16,7 +16,7 @@ from ouroboros.mcp_client import ( get_manager, reconfigure_from_settings, ) -from ouroboros.secret_masking import looks_masked_mcp_secret +from ouroboros.secret_masking import looks_masked_mcp_secret, rehydrate_mcp_url log = logging.getLogger(__name__) @@ -58,7 +58,7 @@ async def api_mcp_refresh(request: Request) -> JSONResponse: async def api_mcp_test(request: Request) -> JSONResponse: - """Probe unsaved or edited MCP config; rehydrate masked saved auth token.""" + """Probe the edited candidate with the same URL rehydration as Settings.""" try: body: Dict[str, Any] = await request_json_or(request, {}) await asyncio.to_thread(_ensure_configured) @@ -82,8 +82,10 @@ async def api_mcp_test(request: Request) -> JSONResponse: # Use the edited candidate, but rehydrate masked token # values from the saved config. The caller can also omit # auth_token entirely to intentionally test without auth. - probe = dict(candidate) - if looks_masked_mcp_secret(probe.get("auth_token")): + from ouroboros.gateway.settings import _rehydrate_mcp_servers_payload + + probe = _rehydrate_mcp_servers_payload([candidate], [target])[0] + if looks_masked_mcp_secret(candidate.get("auth_token")): probe["auth_token"] = str(target.get("auth_token") or "") target = probe outcome = await asyncio.to_thread(manager.test_server, target, settings=settings) @@ -94,6 +96,10 @@ async def api_mcp_test(request: Request) -> JSONResponse: {"ok": False, "error": "request body must include `server` (object) or `server_id` (string)"}, status_code=400, ) + # Without a selected saved server, URL masks cannot identify credentials. + candidate = dict(candidate) + if "url" in candidate: + candidate["url"] = rehydrate_mcp_url(candidate["url"], "") outcome = await asyncio.to_thread(manager.test_server, candidate, settings=settings) return JSONResponse(outcome) except Exception as exc: diff --git a/ouroboros/gateway/settings.py b/ouroboros/gateway/settings.py index 72906b1cb..e46948886 100644 --- a/ouroboros/gateway/settings.py +++ b/ouroboros/gateway/settings.py @@ -45,6 +45,8 @@ from ouroboros.secret_masking import ( looks_masked_mcp_secret, looks_masked_settings_secret, mask_prefixed_secret, + mask_mcp_url, + rehydrate_mcp_url, mask_settings_secret, ) from ouroboros.server_runtime import ( @@ -142,6 +144,8 @@ def _mask_mcp_servers_payload(servers: Any) -> list: clone = dict(entry) if clone.get("id"): clone["id"] = _mcp_canonical_id(clone.get("id")) + if "url" in clone: + clone["url"] = mask_mcp_url(clone["url"]) token = str(clone.get("auth_token") or "") if token: clone["auth_token"] = mask_prefixed_secret(token, visible_chars=8) @@ -226,9 +230,11 @@ def _rehydrate_mcp_servers_payload(incoming: Any, current: Any) -> list: clone = {key: value for key, value in entry.items() if key not in MCP_RESPONSE_ONLY_FIELDS} if clone.get("id"): clone["id"] = _mcp_canonical_id(clone.get("id")) + existing = current_by_id.get(_mcp_canonical_id(clone.get("id"))) or {} + if "url" in clone: + clone["url"] = rehydrate_mcp_url(clone["url"], existing.get("url")) token = str(clone.get("auth_token") or "") if looks_masked_mcp_secret(token): - existing = current_by_id.get(_mcp_canonical_id(clone.get("id"))) clone["auth_token"] = str((existing or {}).get("auth_token") or "") out.append(clone) return out diff --git a/ouroboros/secret_masking.py b/ouroboros/secret_masking.py index 16c9d2d4f..7682a28ca 100644 --- a/ouroboros/secret_masking.py +++ b/ouroboros/secret_masking.py @@ -130,6 +130,44 @@ def looks_masked_mcp_secret(value: Any) -> bool: return text == "***" or _looks_prefixed_mask(text, visible_chars=4) or _looks_prefixed_mask(text, visible_chars=8) + +def mask_mcp_url(value: Any) -> str: + """Mask URL userinfo for the editable Settings surface, preserving its address.""" + from urllib.parse import urlsplit + + text = str(value or "") + try: + authority = urlsplit(text).netloc + except ValueError: + # A malformed saved address remains editable without exposing credentials. + return CONFIGURED_SECRET_PLACEHOLDER if "@" in text else text + _userinfo, separator, host = authority.rpartition("@") + return text.replace(authority, "***@" + host, 1) if separator else text + + +def rehydrate_mcp_url(value: Any, current_value: Any) -> str: + """Restore only the exact mask of this server's current URL. + + A changed address with a placeholder has no newly supplied credentials; + remove the placeholder, never carry the old userinfo to a different target. + A real new userinfo or an explicit URL without it is kept as supplied. + """ + from urllib.parse import urlsplit + + text, current = str(value or ""), str(current_value or "") + if text == CONFIGURED_SECRET_PLACEHOLDER: + return current if current and mask_mcp_url(current) == text else "" + try: + authority = urlsplit(text).netloc + except ValueError: + return text # a new invalid address is the ordinary config validator's input + userinfo, separator, host = authority.rpartition("@") + if not separator or userinfo != "***": + return text + if current and mask_mcp_url(current) == text: + return current + return text.replace(authority, host, 1) + def looks_masked_secret(value: Any) -> bool: """Compatibility union of the exact placeholder shapes this module emits.""" text = str(value or "").strip() diff --git a/tests/test_mcp_url_settings.py b/tests/test_mcp_url_settings.py new file mode 100644 index 000000000..645cbfc39 --- /dev/null +++ b/tests/test_mcp_url_settings.py @@ -0,0 +1,147 @@ +"""Editable MCP URL secrets survive Settings and Test without new credentials.""" +from __future__ import annotations + +import copy +from types import SimpleNamespace + +import pytest +from starlette.applications import Starlette +from starlette.routing import Route +from starlette.testclient import TestClient + +from ouroboros.gateway.settings import _mask_mcp_servers_payload, _rehydrate_mcp_servers_payload +from ouroboros.secret_masking import mask_mcp_url, rehydrate_mcp_url +from tests.test_settings_secret_mask import settings_client # noqa: F401 + +URL = "https://fixture-user:fixture-password@service.example:9443/mcp?route=one#fragment" + + +def server(**changes): + return {"id": "Demo Server!", "name": "Demo", "transport": "streamable_http", "enabled": True, + "url": URL, "auth_token": "Bearer fixture-header-token", **changes} + + +@pytest.mark.parametrize("url", [URL, "https://user-only@service.example/mcp", + "https://u%40name:p%3Ass@[::1]:9443/a?b=#c", "HTTPS://u:p@EXAMPLE.test/mcp?"]) +def test_mask_roundtrip_preserves_exact_original_url(url): + masked = mask_mcp_url(url) + assert "***@" in masked and masked != url + assert rehydrate_mcp_url(masked, url) == url + assert mask_mcp_url(masked) == masked + + +def test_masked_settings_roundtrip_changes_name_without_losing_url_or_header(): + current = [server()] + before = copy.deepcopy(current) + incoming = _mask_mcp_servers_payload(current) + assert "fixture-user" not in str(incoming) and "fixture-password" not in str(incoming) + incoming[0]["name"] = "Edited label" + restored = _rehydrate_mcp_servers_payload(incoming, current)[0] + assert restored["url"] == URL and restored["auth_token"] == current[0]["auth_token"] + assert restored["name"] == "Edited label" and "auth_configured" not in restored + assert current == before + + +@pytest.mark.parametrize("change", [ + {"url": "https://***@different.example:9443/mcp?route=one#fragment"}, + {"url": "https://***@service.example:9443/changed?route=one#fragment"}, + {"url": "http://***@service.example:9443/mcp?route=one#fragment"}, + {"url": "https://***@service.example:9443/mcp?route=two#fragment"}, + {"url": "https://***@service.example:9443/mcp?route=one#new"}, + {"id": "another-server"}, +]) +def test_changed_server_or_target_never_receives_saved_userinfo(change): + masked = _mask_mcp_servers_payload([server()])[0] + masked.update(change) + restored = _rehydrate_mcp_servers_payload([masked], [server()])[0] + assert "@" not in restored["url"] + assert "fixture-user" not in restored["url"] and "fixture-password" not in restored["url"] + assert "***" not in restored["url"] + + +@pytest.mark.parametrize("replacement", ["https://service.example:9443/mcp?route=one#fragment", + "https://new-user:new-password@other.example/mcp", ""]) +def test_explicit_removal_replacement_and_clear_are_kept(replacement): + incoming = _mask_mcp_servers_payload([server()])[0] + incoming["url"] = replacement + restored = _rehydrate_mcp_servers_payload([incoming], [server()])[0] + assert restored["url"] == replacement + assert restored["auth_token"] == server()["auth_token"] + + +def test_mask_without_current_server_cannot_be_saved_as_credentials(): + incoming = _mask_mcp_servers_payload([server()]) + restored = _rehydrate_mcp_servers_payload(incoming, [])[0] + assert restored["url"] == URL.split("@", 1)[-1].join(["https://", ""]) + assert "***" not in restored["url"] + + +def test_invalid_saved_authority_masks_secrets_and_can_still_roundtrip_or_be_replaced(): + malformed = "https://private-user:private-password@[broken/mcp" + masked = mask_mcp_url(malformed) + assert "private" not in masked + assert rehydrate_mcp_url(masked, malformed) == malformed + assert rehydrate_mcp_url("https://fixed.example/mcp", malformed) == "https://fixed.example/mcp" + + +def test_real_settings_get_post_uses_url_roundtrip(settings_client, monkeypatch): # noqa: F811 + from ouroboros import mcp_client + monkeypatch.setattr(mcp_client, "refresh_all_background", lambda **kwargs: None) + client, on_disk, saved = settings_client + on_disk.update(MCP_ENABLED=False, MCP_SERVERS=[server()]) + received = client.get("/api/settings") + assert received.status_code == 200 + assert "fixture-user" not in received.text and "fixture-password" not in received.text + candidate = received.json()["MCP_SERVERS"][0] + candidate["name"] = "Changed title" + response = client.post("/api/settings", json={"MCP_SERVERS": [candidate]}) + assert response.status_code == 200, response.text + assert saved["MCP_SERVERS"][0]["url"] == URL + assert saved["MCP_SERVERS"][0]["name"] == "Changed title" + assert "fixture-user" not in response.text and "fixture-password" not in response.text + + +@pytest.fixture +def test_candidate_client(monkeypatch): + from ouroboros.gateway import mcp + current = server() + seen = [] + def test(candidate, *, settings): + seen.append(copy.deepcopy(candidate)) + return {"ok": True, "tool_count": 0} + monkeypatch.setattr(mcp, "_ensure_configured", lambda: None) + monkeypatch.setattr(mcp, "load_settings", lambda: {"MCP_SERVERS": [copy.deepcopy(current)]}) + monkeypatch.setattr(mcp, "get_manager", lambda: SimpleNamespace(test_server=test)) + with TestClient(Starlette(routes=[Route("/test", mcp.api_mcp_test, methods=["POST"])])) as client: + yield client, current, seen + + +@pytest.mark.parametrize("change", [{}, {"name": "Edited"}, {"url": "https://***@different.example/mcp"}, + {"id": "other"}, {"url": "https://service.example/no-auth"}]) +def test_saved_test_candidate_uses_same_scoped_url_rehydration(test_candidate_client, change): + client, current, seen = test_candidate_client + candidate = _mask_mcp_servers_payload([current])[0] + candidate.update(change) + response = client.post("/test", json={"server_id": current["id"], "server": candidate}) + assert response.status_code == 200 and response.json()["ok"] + expected = _rehydrate_mcp_servers_payload([candidate], [current])[0] + assert seen[-1]["url"] == expected["url"] + # auth_token keeps its older selected-server behavior, independent of URL userinfo. + assert seen[-1]["auth_token"] == current["auth_token"] + assert "fixture-user" not in response.text and "fixture-password" not in response.text + + +def test_test_candidate_can_explicitly_omit_header_and_url_credentials(test_candidate_client): + client, current, seen = test_candidate_client + candidate = {"id": current["id"], "transport": "streamable_http", "url": "https://service.example/mcp"} + assert client.post("/test", json={"server_id": current["id"], "server": candidate}).json()["ok"] + assert "auth_token" not in seen[-1] and seen[-1]["url"] == candidate["url"] + + +def test_unsaved_test_candidate_cannot_recover_another_saved_url(test_candidate_client): + client, current, seen = test_candidate_client + candidate = _mask_mcp_servers_payload([current])[0] + response = client.post("/test", json={"server": candidate}) + assert response.json()["ok"] + assert "@" not in seen[-1]["url"] + assert seen[-1]["auth_token"] == candidate["auth_token"] # existing inline-token semantics