import json import pytest from httpx import ASGITransport, AsyncClient from mcp.client.session import ClientSession from mcp.client.streamable_http import streamable_http_client from gnexus_creds.models import ApiToken, AuditEvent from gnexus_creds.schemas import ApiTokenCreate, Scope, SecretCreate, SecretFieldIn from gnexus_creds.services import Actor, create_api_token, create_secret def test_mcp_protocol_server_has_instructions(): from gnexus_creds.mcp_protocol import create_mcp_protocol_server server = create_mcp_protocol_server() assert "personal secret storage" in server.instructions assert "Use reveal_secret only when the user explicitly needs" in server.instructions @pytest.mark.anyio async def test_mcp_requires_mcp_scope(app, actor): actor.channel = "rest" actor.api_token = ApiToken( user_id=actor.user.id, public_id="pub", name="test", token_hash="hash", scopes=["read"], ) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post("/mcp/tools/search_secrets", json={"arguments": {}}) assert response.status_code == 403 @pytest.mark.anyio async def test_mcp_update_requires_secret_allow_mcp(app, db_session, actor): actor.channel = "mcp" actor.api_token = ApiToken( user_id=actor.user.id, public_id="mcp", name="mcp", token_hash="hash", scopes=["mcp", "read", "write"], ) secret = create_secret( db_session, Actor(user=actor.user, channel="ui"), SecretCreate( title="UI only", allow_mcp=False, fields=[SecretFieldIn(name="username", value="demo", encrypted=False)], ), ) db_session.commit() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post( "/mcp/tools/update_secret", json={"arguments": {"secret_id": str(secret.id), "title": "Changed"}}, ) assert response.status_code == 404 @pytest.mark.anyio async def test_mcp_search_excludes_archived_and_reveal_audits(app, db_session, actor): actor.channel = "mcp" actor.api_token = ApiToken( user_id=actor.user.id, public_id="mcp-full", name="mcp-full", token_hash="hash", scopes=["mcp", "read", "reveal"], ) visible = create_secret( db_session, Actor(user=actor.user, channel="ui"), SecretCreate( title="Visible MCP", allow_mcp=True, fields=[SecretFieldIn(name="password", value="visible", encrypted=True)], ), ) create_secret( db_session, Actor(user=actor.user, channel="ui"), SecretCreate( title="Archived MCP", archived=True, allow_mcp=True, fields=[SecretFieldIn(name="password", value="archived", encrypted=True)], ), ) db_session.commit() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: search_response = await client.post( "/mcp/tools/search_secrets", json={"arguments": {"q": "MCP"}}, ) reveal_response = await client.post( "/mcp/tools/reveal_secret", json={"arguments": {"secret_id": str(visible.id)}}, ) assert search_response.status_code == 200 assert search_response.json()["total"] == 1 assert search_response.json()["items"][0]["title"] == "Visible MCP" assert reveal_response.status_code == 200 event = ( db_session.query(AuditEvent) .filter(AuditEvent.action == "secret.revealed", AuditEvent.secret_id == visible.id) .one() ) assert event.channel == "mcp" @pytest.mark.anyio async def test_mcp_http_adapter_tool_variants(app, db_session, actor): actor.channel = "mcp" actor.api_token = ApiToken( user_id=actor.user.id, public_id="mcp-write", name="mcp-write", token_hash="hash", scopes=["mcp", "read", "reveal", "write"], ) secret = create_secret( db_session, Actor(user=actor.user, channel="ui"), SecretCreate( title="MCP variants", allow_mcp=True, fields=[SecretFieldIn(name="username", value="variant", encrypted=False)], ), ) db_session.commit() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: sse_response = await client.get("/mcp/sse") get_response = await client.post( "/mcp/tools/get_secret", json={"arguments": {"secret_id": str(secret.id)}}, ) create_response = await client.post( "/mcp/tools/create_secret", json={ "arguments": { "title": "MCP adapter created", "allow_mcp": True, "fields": [{"name": "token", "value": "adapter", "encrypted": True}], } }, ) status_response = await client.post( "/mcp/tools/set_secret_status", json={"arguments": {"secret_id": str(secret.id), "status": "outdated"}}, ) archive_response = await client.post( "/mcp/tools/archive_secret", json={"arguments": {"secret_id": str(secret.id)}}, ) missing_response = await client.post( "/mcp/tools/nope", json={"arguments": {}}, ) assert sse_response.status_code == 200 discovery = json.loads(sse_response.text.split("data: ", 1)[1]) assert "personal secret storage" in discovery["instructions"] assert discovery["tools"][0]["name"] == "search_secrets" assert "Does not decrypt encrypted values" in discovery["tools"][0]["description"] assert get_response.status_code == 200 assert get_response.json()["title"] == "MCP variants" assert create_response.status_code == 200 assert create_response.json()["title"] == "MCP adapter created" assert status_response.status_code == 200 assert status_response.json()["status"] == "outdated" assert archive_response.status_code == 200 assert archive_response.json()["archived"] is True assert missing_response.status_code == 404 @pytest.mark.anyio async def test_mcp_protocol_streamable_http_tools(session_factory, db_session, user, monkeypatch): monkeypatch.setattr("gnexus_creds.mcp_protocol.SessionLocal", session_factory) api_token_row, token = create_api_token( db_session, Actor(user=user, channel="ui"), ApiTokenCreate(scopes=[Scope.read, Scope.reveal, Scope.write, Scope.mcp], name="MCP SDK"), ) secret = create_secret( db_session, Actor(user=user, channel="ui"), SecretCreate( title="Protocol secret", allow_mcp=True, tags=["protocol"], fields=[ SecretFieldIn(name="username", value="protocol-user", encrypted=False), SecretFieldIn(name="password", value="protocol-pass", encrypted=True, masked=True), ], ), ) db_session.commit() assert api_token_row.public_id from gnexus_creds.main import create_app app = create_app() http_client = AsyncClient( transport=ASGITransport(app=app), base_url="http://127.0.0.1:8000", headers={"Authorization": f"Bearer {token}"}, ) async with app.router.lifespan_context(app): async with http_client: async with streamable_http_client( "http://127.0.0.1:8000/mcp-protocol/", http_client=http_client, terminate_on_close=False, ) as (read_stream, write_stream, _): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tools = await session.list_tools() tool_descriptions = {tool.name: tool.description for tool in tools.tools} assert "search_secrets" in tool_descriptions assert "Does not decrypt" in tool_descriptions["search_secrets"] assert "audit event" in tool_descriptions["reveal_secret"] search = await session.call_tool("search_secrets", {"q": "protocol"}) assert not search.isError assert search.structuredContent["total"] == 1 reveal = await session.call_tool("reveal_secret", {"secret_id": str(secret.id)}) assert not reveal.isError fields = { field["name"]: field["value"] for field in reveal.structuredContent["fields"] } assert fields["password"] == "protocol-pass" created = await session.call_tool( "create_secret", { "title": "Protocol created", "allow_mcp": True, "fields": [ { "name": "token", "value": "created-through-protocol", "encrypted": True, "masked": True, } ], }, ) assert not created.isError assert created.structuredContent["title"] == "Protocol created"