Preserve MCP URL credentials through masked settings roundtrips

Local checkpoint, NOT REVIEWED. Mask complete URL userinfo and restore it only for the same server and exact current masked URL. Reuse the same rule for edited Test candidates, retain existing auth_token behavior and explicit removal. Focused 87 tests and Ruff passed; phase review and paired Cyber integration remain pending.

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:

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

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()

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