[TESTING][SECURITY]: Federation security manual test plan (cross-gateway auth, peer validation)

[crivetimihai]

[TESTING][SECURITY]: Upstream gateway authentication manual test plan (OAuth, bearer, discovery)

Goal

Produce a comprehensive manual test plan for validating upstream gateway authentication including OAuth flows, bearer token auth, tool/resource discovery, and visibility RBAC on federated content.

Why Now?

Security testing is critical for GA release:

1. Production Readiness: Gateway authentication must be validated before release
2. Compliance: Required by security standards and audits
3. Defense in Depth: Validates multiple protection layers
4. Attack Mitigation: Prevents unauthorized upstream access
5. User Trust: Security issues erode confidence

---

Architecture Overview

Important: Gateways are registered upstream MCP servers, NOT peer gateways with mutual authentication. The platform authenticates TO upstream gateways using standard client auth methods.

┌─────────────────────────────────────────────────────────────────────┐
│                  Upstream Gateway Architecture                       │
│                                                                      │
│  User ──JWT──> MCP Gateway ──OAuth/Bearer/Basic──> Upstream Gateway │
│                     │                                                │
│                     │ Discovery: tools, resources, prompts          │
│                     │ with federation_source set                    │
│                     │                                                │
│  Authentication TO Upstream:                                         │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │ OAuth client_credentials: App-level tokens                  │    │
│  │ OAuth authorization_code: Per-user tokens via app_user_email│    │
│  │ Bearer token: Static token in auth_token field              │    │
│  │ Basic auth: Username/password in auth_value                 │    │
│  │ Custom headers: Via auth_headers field                      │    │
│  │ Query params: API key in URL (auth_query_param_*)           │    │
│  └─────────────────────────────────────────────────────────────┘    │
│                                                                      │
│  Federated Content:                                                  │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │ Tools, resources, prompts imported with:                    │    │
│  │   federation_source = gateway.name                          │    │
│  │   visibility = gateway's visibility setting                 │    │
│  │   team_id = gateway's team_id                               │    │
│  └─────────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────────┘

---

Test Environment Setup

# Build and start
make docker-prod && make compose-down && make testing-up

# Key settings in .env:
AUTH_REQUIRED=true
JWT_SECRET_KEY=your-secret-key-min-32-chars!!

# Create admin token
export JWT_SECRET_KEY="${JWT_SECRET_KEY:-my-test-key}"
export ADMIN_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username admin@example.com --exp 10080 --secret "$JWT_SECRET_KEY" --admin)

# Regular user token
export USER_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username user@example.com --exp 10080 --secret "$JWT_SECRET_KEY")

# For testing, you can use httpbin or a local MCP server
# Start a local MCP server for testing:
# python -m mcpgateway.translate --stdio "uvx mcp-server-git" --port 9000

---

User Stories

Story 1: Gateway Authentication

As a security administrator
I want gateways to authenticate to upstream servers
So that only authorized connections are made

Acceptance Criteria

Feature: Upstream Gateway Authentication
  Scenario: OAuth client_credentials authentication
    When a gateway is registered with OAuth client_credentials
    Then the gateway obtains tokens automatically
    And tools are discovered using the OAuth token

  Scenario: OAuth authorization_code authentication
    When a gateway is registered with OAuth authorization_code
    Then per-user tokens are required
    And app_user_email is used to retrieve user-specific tokens

  Scenario: Bearer token authentication
    When a gateway is registered with a bearer token
    Then the token is used for all requests

Story 2: Federated Content Access

As a security administrator
I want federated tools protected by visibility
So that access is properly restricted

Acceptance Criteria

Feature: Federated Content RBAC
  Scenario: Public gateway content
    When a public gateway imports tools
    Then all users can see and invoke the tools

  Scenario: Team gateway content
    When a team-scoped gateway imports tools
    Then only team members can access the tools

---

Test Cases

TC-GW-001: Gateway Registration with Bearer Auth

Objective: Verify gateway registration with bearer token authentication

Steps:

# Step 1: Register gateway with bearer auth
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "bearer-test-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "bearer",
    "auth_token": "test-upstream-token",
    "visibility": "public"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Check gateway is created
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bearer-test-gateway")'

# Step 3: Verify health check uses auth
# Gateway should show healthy if upstream accepts the token
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bearer-test-gateway") | {name, healthy}'

Expected Results:

• Gateway created successfully
• Auth token stored (encrypted)
• Health check uses configured auth

---

TC-GW-002: Gateway Registration with OAuth Client Credentials

Objective: Verify OAuth client_credentials flow for gateways

Steps:

# Step 1: Register gateway with OAuth
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "oauth-cc-gateway",
    "url": "https://upstream-mcp.example.com/sse",
    "transport": "SSE",
    "auth_type": "oauth",
    "oauth_config": {
      "grant_type": "client_credentials",
      "token_url": "https://auth.example.com/oauth/token",
      "client_id": "my-client-id",
      "client_secret": "my-client-secret",
      "scope": "mcp:read mcp:write"
    },
    "visibility": "public"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Verify OAuth config stored
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "oauth-cc-gateway") | .oauth_config'

Expected Results:

• Gateway created with OAuth config
• Client secret encrypted at rest
• OAuth token obtained automatically when needed

---

TC-GW-003: Gateway with OAuth Authorization Code (Per-User Tokens)

Objective: Verify authorization_code flow uses per-user tokens

Steps:

# Step 1: Register gateway with authorization_code
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "oauth-ac-gateway",
    "url": "https://upstream-mcp.example.com/sse",
    "transport": "SSE",
    "auth_type": "oauth",
    "oauth_config": {
      "grant_type": "authorization_code",
      "authorization_url": "https://auth.example.com/authorize",
      "token_url": "https://auth.example.com/oauth/token",
      "client_id": "my-client-id",
      "client_secret": "my-client-secret",
      "redirect_uri": "http://localhost:4444/oauth/callback",
      "scope": "openid profile"
    },
    "visibility": "public"
  }' \
  http://localhost:4444/gateways | jq .
GATEWAY_ID=$(curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq -r '.[] | select(.name == "oauth-ac-gateway") | .id')

# Step 2: Initiate OAuth flow for user
# User must complete OAuth authorization:
echo "OAuth authorization URL: http://localhost:4444/oauth/authorize/$GATEWAY_ID"

# Step 3: After OAuth completion, tokens are stored per-user
# The gateway will use app_user_email to retrieve user-specific tokens

Expected Results:

• Gateway requires user to complete OAuth flow
• Per-user tokens stored in TokenStorageService
• Requests to upstream use user-specific tokens

---

TC-GW-004: Gateway with Invalid Credentials

Objective: Verify gateway shows unhealthy with wrong credentials

Steps:

# Step 1: Register gateway with wrong token
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "bad-auth-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "bearer",
    "auth_token": "wrong-token-value"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Check health status
# Wait for health check to run
sleep 5
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bad-auth-gateway") | {name, healthy, last_error}'

Expected Results:

• Gateway created but marked unhealthy
• Health check fails due to auth error
• Error message indicates auth failure

---

TC-GW-005: Federated Tools Discovery

Objective: Verify tools are imported from upstream with federation_source

Steps:

# Step 1: Register gateway (assuming upstream has tools)
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "discovery-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "bearer",
    "auth_token": "valid-token"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Trigger tool refresh
GATEWAY_ID=$(curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq -r '.[] | select(.name == "discovery-gateway") | .id')

curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways/$GATEWAY_ID/tools/refresh | jq .

# Step 3: Check federated tools
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/tools | jq '.[] | select(.federation_source == "discovery-gateway")'

Expected Results:

• Tools imported from upstream
• federation_source set to gateway name
• Tools inherit gateway's visibility

---

TC-GW-006: Federated Content Visibility

Objective: Verify federated tools follow gateway visibility

Steps:

# Step 1: Create team-scoped gateway
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "team-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "visibility": "team",
    "team_id": "team-alpha"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Check tools with team token
curl -s -H "Authorization: Bearer $TEAM_TOKEN" \
  http://localhost:4444/tools | jq '.[] | select(.federation_source == "team-gateway")'
# Expected: Tools visible

# Step 3: Check tools with non-team token
curl -s -H "Authorization: Bearer $USER_TOKEN" \
  http://localhost:4444/tools | jq '.[] | select(.federation_source == "team-gateway")'
# Expected: Tools NOT visible (empty result)

Expected Results:

• Federated tools inherit gateway visibility
• Team members see team-gateway tools
• Non-members don't see team-gateway tools

---

TC-GW-007: Gateway CRUD Permission Enforcement

Objective: Verify gateways.* permissions are enforced

Steps:

# Step 1: Create token with only gateways.read
READ_ONLY_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username reader@example.com --exp 10080 --secret "$JWT_SECRET_KEY" \
  --scopes gateways.read)

# Step 2: List gateways - should work
curl -s -w "\nStatus: %{http_code}\n" \
  -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  http://localhost:4444/gateways
# Expected: 200 OK

# Step 3: Try to create gateway - should fail
curl -s -w "\nStatus: %{http_code}\n" \
  -X POST -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"test","url":"http://example.com"}' \
  http://localhost:4444/gateways
# Expected: 403 Forbidden

Expected Results:

• gateways.read: Can list and view gateways
• gateways.create: Required to create gateways
• gateways.update: Required to update gateways
• gateways.delete: Required to delete gateways

---

TC-GW-008: Gateway with Query Parameter Auth

Objective: Verify query parameter authentication (for APIs requiring key in URL)

Steps:

# Step 1: Register gateway with query param auth
# Note: This is insecure (CWE-598) but needed for some APIs
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "query-param-gateway",
    "url": "https://api.example.com/mcp/sse",
    "transport": "SSE",
    "auth_type": "query_param",
    "auth_query_param_key": "api_key",
    "auth_query_param_value": "secret-api-key-value"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Verify auth is stored (encrypted)
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "query-param-gateway") | .auth_type'

Expected Results:

• API key stored encrypted
• Query param added to URL when calling upstream
• Key not logged in plain text

---

TC-GW-009: Audit Logging

Objective: Verify gateway operations are logged

Steps:

# Step 1: Perform gateway operations
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"audit-test-gw","url":"http://example.com"}' \
  http://localhost:4444/gateways

# Step 2: Check logs for gateway activities
docker logs mcpgateway 2>&1 | grep -i "gateway" | tail -10

# Expected log fields:
# - user_email
# - resource_type: "gateway"
# - resource_id
# - action: "create", "update", "delete"

Expected Results:

• Gateway CRUD operations logged
• User identity captured
• Resource IDs included

---

Test Matrix

Test Case	Auth Type	OAuth	Discovery	Visibility	Permissions
TC-GW-001	✓ Bearer
TC-GW-002		✓ CC
TC-GW-003		✓ AC
TC-GW-004	✓ Invalid
TC-GW-005			✓
TC-GW-006				✓
TC-GW-007					✓
TC-GW-008	✓ Query
TC-GW-009

---

Success Criteria

• All 9 test cases pass
• Bearer token authentication works
• OAuth client_credentials flow obtains tokens
• OAuth authorization_code uses per-user tokens
• Invalid credentials result in unhealthy status
• Tools imported with federation_source
• Federated content follows gateway visibility
• Permission checks enforced
• Query parameter auth supported (encrypted)

---

Features NOT Implemented

The following features from the original issue do NOT exist:

• ❌ Peer-to-peer gateway authentication (gateways are upstream servers, not peers)
• ❌ /api/federation/* endpoints (no dedicated federation API)
• ❌ Mutual TLS for federation (TLS exists but not mTLS verification)
• ❌ Bell-LaPadula clearance levels (only visibility RBAC)
• ❌ Gateway credential rotation endpoint (manual update required)
• ❌ Rate limiting on gateway routes (no @rate_limit decorator)
• ❌ User identity propagation to upstream (except via authorization_code per-user tokens)

---

Related Files

• mcpgateway/services/gateway_service.py - Gateway service implementation
• mcpgateway/services/oauth_manager.py - OAuth token management
• mcpgateway/main.py:1678 - Gateway router (/gateways prefix)
• mcpgateway/main.py:4639-4905 - Gateway endpoint handlers
• mcpgateway/schemas.py:2464-2550 - GatewayCreate schema

---

Related Issues

• None identified

[marekdano]

Test Environment Setup

# Key settings in .env:
AUTH_REQUIRED=true
JWT_SECRET_KEY=your-secret-key-min-32-chars!!

# disable SSRF for testing on localhost
export SSRF_ALLOW_LOCALHOST=true

# Build and start
make serve

# Create admin token
export JWT_SECRET_KEY="${JWT_SECRET_KEY:-my-test-key}"
export ADMIN_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username admin@example.com --exp 10080 --secret "$JWT_SECRET_KEY" --admin)

# Regular user token
export USER_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username user@example.com --exp 10080 --secret "$JWT_SECRET_KEY")

# Start a local MCP server for testing:
python -m mcpgateway.translate --stdio "uvx mcp-server-git" --port 9000

Test Cases

TC-GW-001: Gateway Registration with Bearer Auth

Objective: Verify gateway registration with bearer token authentication

Steps:

# Step 1: Register gateway with bearer auth
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "bearer-test-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "bearer",
    "auth_token": "test-upstream-token",
    "visibility": "public"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Check gateway is created
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bearer-test-gateway")'

# Step 3: Verify health check uses auth
# Gateway should show healthy if upstream accepts the token
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bearer-test-gateway") |  {name, reachable, lastSeen}'

# Returns

{
  "name": "bearer-test-gateway",
  "reachable": true,
  "lastSeen": "2026-02-24T13:35:52.029039"
}

Expected results returns:

• Gateway created successfully
• Auth token stored (encrypted) - field "authToken": "*****"
• Health check uses configured auth - fields "reachable": true, "lastSeen": "2026-02-24T13:35:52.029039"

---

TC-GW-002: Gateway Registration with OAuth Client Credentials

Objective: Verify OAuth client_credentials flow for gateways

Steps:

# Step 1: Mock OAuth server with a simple Python script
# Create a tiny FastAPI server that acts as a `token` endpoint

# mock_oauth.py file
from fastapi import FastAPI, Form
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/token")
async def token(grant_type: str = Form(...), client_id: str = Form(""), client_secret: str = Form(""), scope: str = Form("")):
   if grant_type == "client_credentials" and client_id == "test-id" and client_secret == "test-secret":
      return {"access_token": "mock-token-12345", "token_type": "bearer", "expires_in": 3600}
   return JSONResponse(status_code=401, content={"error": "invalid_client"})
# end - mock_oauth.py file
 
#Step 2: Run the  mock oauth server
uvicorn mock_oauth:app --port 9090

#Step 3: Register gateway with OAuth
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "name": "oauth-cc-gateway",
      "url": "http://localhost:9000/sse",
      "transport": "SSE",
      "auth_type": "oauth",
      "oauth_config": {
        "grant_type": "client_credentials",
        "token_url": "http://localhost:9090/token",
        "client_id": "test-id",
        "client_secret": "test-secret",
        "scope": "mcp:read mcp:write"
      },
      "visibility": "public"
    }' \
    http://localhost:4444/gateways | jq .

# Step 4: Verify OAuth config stored
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "oauth-cc-gateway") | .oauthConfig'

Returns:

{
  "grant_type": "client_credentials",
  "token_url": "http://localhost:9090/token",
  "client_id": "test-id",
  "client_secret": "*****",
  "scope": "mcp:read mcp:write"
}

Expected results returned:

• Gateway created with OAuth config
• Client secret encrypted at rest ("client_secret": "*****")
• OAuth token obtained automatically when needed

---

TC-GW-003: Gateway with OAuth Authorization Code (Per-User Tokens)

Objective: Verify authorization_code flow uses per-user tokens

Steps:

# Step 1: Mock OAuth server with a simple Python script
# Create a tiny FastAPI server that acts as a `token` and `authorize` endpoints

# mock_oauth_authorize.py
from fastapi import FastAPI, Form, Query
from fastapi.responses import JSONResponse, RedirectResponse

app = FastAPI()

# Client Credentials flow 
@app.post("/token")
async def token(
    grant_type: str = Form(...),
    client_id: str = Form(""),
    client_secret: str = Form(""),
    scope: str = Form(""),
    code: str = Form(""),
    redirect_uri: str = Form(""),
):
    # Client credentials flow
    if grant_type == "client_credentials" and client_id == "test-id" and client_secret == "test-secret":
        return {"access_token": "mock-cc-token-12345", "token_type": "bearer", "expires_in": 3600}

    # Authorization code exchange
    if grant_type == "authorization_code" and code == "mock-auth-code":
        if client_id == "test-id" and client_secret == "test-secret":
            return {
                "access_token": "mock-user-token-67890",
                "token_type": "bearer",
                "expires_in": 3600,
                "refresh_token": "mock-refresh-token",
                "scope": scope,
            }

    return JSONResponse(status_code=401, content={"error": "invalid_client"})


# --- Authorization Code flow: authorize endpoint ---
@app.get("/authorize")
async def authorize(
    response_type: str = Query("code"),
    client_id: str = Query(""),
    redirect_uri: str = Query(""),
    scope: str = Query(""),
    state: str = Query(""),
):
    # Simulate user approval — immediately redirect back with a code
    return RedirectResponse(url=f"{redirect_uri}?code=mock-auth-code&state={state}")
# end - mock_oauth_authorize.py

# Step 2: Run the  mock oauth server
uvicorn mock_oauth_authorize:app --port 9091

# Step 3: Register gateway with authorization_code
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "name": "oauth-ac-gateway",
      "url": "http://localhost:9000/sse",
      "transport": "SSE",
      "auth_type": "oauth",
      "oauth_config": {
        "grant_type": "authorization_code",
        "authorization_url": "http://localhost:9091/authorize",
        "token_url": "http://localhost:9091/token",
        "client_id": "test-id",
        "client_secret": "test-secret",
        "redirect_uri": "http://localhost:4444/oauth/callback",
        "scope": "openid profile"
      },
      "visibility": "public"
    }' \
    http://localhost:4444/gateways | jq .

GATEWAY_ID=$(curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq -r '.[] | select(.name == "oauth-ac-gateway") | .id')

# Step 2: Initiate OAuth flow for user
# User must complete OAuth authorization:
echo "OAuth authorization URL: http://localhost:4444/oauth/authorize/$GATEWAY_ID"
# call the url in the browser to finish authorization, e.g.
http://localhost:4444/oauth/authorize/1655cee55cd04b72b336f71aeee744b6

# Step 3: Authorize another user
curl -sv -L -H "Authorization: Bearer $USER_TOKEN" \
    http://localhost:4444/oauth/authorize/$GATEWAY_ID

# Step 4: Verify the two records has been created in the `oauth_tokens` table
sqlite3 mcp.db "SELECT gateway_id, user_id, app_user_email, created_at FROM oauth_tokens;"

# Result
1655cee55cd04b72b336f71aeee744b6|test-id|admin@example.com|2026-02-25 08:24:20.422385
1655cee55cd04b72b336f71aeee744b6|test-id|user@example.com|2026-02-25 08:51:23.798971

Expected results returned:

• Gateway requires user to complete OAuth flow
• Per-user tokens stored in TokenStorageService

---

TC-GW-004: Gateway with Invalid Credentials

Objective: Verify gateway shows unhealthy with wrong credentials

Steps:

# Step 1: Create a mock MCP SSE server file

# mock_mcp_server.py - start
import asyncio
import json
import logging
import uuid

from fastapi import FastAPI, Header, HTTPException, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

VALID_TOKEN = "valid-token-123"
app = FastAPI()

sessions: dict[str, asyncio.Queue] = {}

def check_auth(authorization: str | None):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing token")
    if authorization[7:] != VALID_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid token")


@app.get("/sse")
async def sse_endpoint(request: Request, authorization: str | None = Header(None)):
    logger.info(f"SSE Authorization: {authorization}")
    check_auth(authorization)

    session_id = uuid.uuid4().hex
    queue: asyncio.Queue = asyncio.Queue()
    sessions[session_id] = queue

    async def stream():
        try:
            yield f"event: endpoint\ndata: /message?session_id={session_id}\n\n"
            while True:
                try:
                    msg = await asyncio.wait_for(queue.get(), timeout=15.0)
                    yield f"event: message\ndata: {json.dumps(msg)}\n\n"
                except asyncio.TimeoutError:
                    yield ": ping\n\n"
        finally:
            sessions.pop(session_id, None)

    return StreamingResponse(stream(), media_type="text/event-stream")


@app.post("/message")
async def message_endpoint(
    request: Request,
    session_id: str = Query(...),
    authorization: str | None = Header(None),
):
    check_auth(authorization)

    if session_id not in sessions:
        raise HTTPException(status_code=404, detail="Session not found")

    body = await request.json()
    method = body.get("method", "")
    msg_id = body.get("id")

    if method == "initialize":
        response = {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {"name": "mock-auth-mcp-server", "version": "1.0.0"},
            },
        }
    elif method == "notifications/initialized":
        return JSONResponse(content={"ok": True})
    elif method == "tools/list":
        response = {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "tools": [
                    {
                        "name": "echo",
                        "description": "Echoes the input",
                        "inputSchema": {
                            "type": "object",
                            "properties": {"message": {"type": "string"}},
                            "required": ["message"],
                        },
                    }
                ]
            },
        }
    else:
        response = {
            "jsonrpc": "2.0",
            "id": msg_id,
            "error": {"code": -32601, "message": f"Method not found: {method}"},
        }

    await sessions[session_id].put(response)
    return JSONResponse(content={"ok": True})
# mock_mcp_server.py - END

# Step 2: Run the mock server
uvicorn mock_mcp_server:app --port 9100

# Step 3: Run the app
export HEALTH_CHECK_INTERVAL=5
export UNHEALTHY_THRESHOLD=1
make serve

Step 4: Register gateway with the correct credentials
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
     "name": "bad-auth-gateway",
      "url": "http://localhost:9100/sse",
      "transport": "SSE",
      "auth_type": "bearer",
      "auth_token": "valid-token-123",
      "visibility": "public"
  }' \
  http://localhost:4444/gateways | jq .

# Step 4: Stop the mock server on port 9100 (Ctrl+C)

# Step 5: Wait for 1 health check cycle
sleep 10

curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "bad-auth-gateway") | {name, reachable}'

Expected results returned:

• Gateway created but marked unhealthy
• Health check fails due to auth error

---

TC-GW-005: Federated Tools Discovery

Objective: Verify tools are imported from upstream with federation_source

Steps:

# Step 1: Register gateway (assuming upstream has tools)
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "discovery-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "bearer",
    "auth_token": "valid-token"
  }' \
  http://localhost:4444/gateways | jq .

# Step 2: Trigger tool refresh
GATEWAY_ID=$(curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq -r '.[] | select(.name == "discovery-gateway") | .id')

curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways/$GATEWAY_ID/tools/refresh | jq .

# Step 3: Check federated tools
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/tools | jq '.[] | select(.federationSource == "discovery-gateway")'

Expected results returned:

• Tools imported from upstream
• federationSource set to gateway name
• Tools inherit gateway's visibility

---

TC-GW-006: Federated Content Visibility

Objective: Verify federated tools follow gateway visibility

Steps:

# Step 1: Generate TEAM_TOKEN from team `team-alpha` with `team_id=605ccb9a451b4090b3b27f63fd7b0745`
export TEAM_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
    --username user@example.com --exp 10080 --secret "$JWT_SECRET_KEY" \
    --teams "605ccb9a451b4090b3b27f63fd7b0745")

# Step 2: Create team-scoped gateway
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "team-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "visibility": "team",
    "team_id": "605ccb9a451b4090b3b27f63fd7b0745"
  }' \
  http://localhost:4444/gateways | jq .

# Step 3: Check tools with team token
curl -s -H "Authorization: Bearer $TEAM_TOKEN" \
  http://localhost:4444/tools/ | jq '.[] | select(.federationSource == "team-gateway")'

# Expected: Tools visible!!!

# Step 3: Check tools with non-team token
curl -s -H "Authorization: Bearer $USER_TOKEN" \
  http://localhost:4444/tools/ | jq '.[] | select(.federationSource == "team-gateway")'

# Expected: Tools NOT visible (empty result)!!!

Expected results returned:

• Federated tools inherit gateway visibility
• Team members see team-gateway tools
• Non-members don't see team-gateway tools

---

TC-GW-007: Gateway CRUD Permission Enforcement

Objective: Verify gateways.* permissions are enforced

Steps:

# Step 1: Create token with only gateways.read
READ_ONLY_TOKEN=$(python -m mcpgateway.utils.create_jwt_token \
  --username reader@example.com --exp 10080 --secret "$JWT_SECRET_KEY" \
  --scopes '{"permissions": ["gateways.read"]}')

# Step 2: List gateways - should work
curl -s -w "\nStatus: %{http_code}\n" \
  -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  http://localhost:4444/gateways
# Expected: 200 OK

# Step 3: Try to create gateway - should fail
curl -s -w "\nStatus: %{http_code}\n" \
  -X POST -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"test","url":"http://example.com"}' \
  http://localhost:4444/gateways

# Step 4: Try to update gateway - should fail
curl -s -w "\nStatus: %{http_code}\n" \
  -X PATCH -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"test","url":"http://example.com"}' \
  http://localhost:4444/gateways/6135b4bb78bf46f1a582fdf8fbfa5246

# Step 5: Try to delete gateway - should fail
curl -s -w "\nStatus: %{http_code}\n" \
  -X DELETE -H "Authorization: Bearer $READ_ONLY_TOKEN" \
  -H "Content-Type: application/json" \
  http://localhost:4444/gateways/6135b4bb78bf46f1a582fdf8fbfa5246

# Expected: 403 Forbidden

Expected results returned:

• gateways.read: Can list and view gateways
• gateways.create: Required to create gateways
• gateways.update: Required to update gateways
• gateways.delete: Required to delete gateways

---

TC-GW-008: Gateway with Query Parameter Auth

Objective: Verify query parameter authentication (for APIs requiring key in URL)

Steps:

# Step 1: enable `INSECURE_ALLOW_QUERYPARAM_AUTH`
export INSECURE_ALLOW_QUERYPARAM_AUTH=true
make serve

# Step 2: Register gateway with query param auth
# Note: This is insecure (CWE-598) but needed for some APIs
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "query-param-gateway",
    "url": "http://localhost:9000/sse",
    "transport": "SSE",
    "auth_type": "query_param",
    "auth_query_param_key": "api_key",
    "auth_query_param_value": "secret-api-key-value"
  }' \
  http://localhost:4444/gateways | jq .

# Step 3: Verify auth is stored (encrypted)
curl -s -H "Authorization: Bearer $ADMIN_TOKEN" \
  http://localhost:4444/gateways | jq '.[] | select(.name == "query-param-gateway") | .authType'

# Expected: "query_param"

Expected Results returned:

• API key stored encrypted

sqlite3 mcp.db "SELECT name, auth_query_params FROM gateways WHERE name='query-param-gateway';"
# returns
query-param-gateway|{"api_key": <api_key>} 

• Query param added to URL when calling upstream (e.g. HTTP Request: GET http://localhost:9000/sse?api_key=secret-api-key-value)

---

TC-GW-009: Audit Logging

Objective: Verify gateway operations are logged

Steps:

# Step 1: enable logging
export LOG_LEVEL=DEBUG
export LOG_TO_FILE=true
export STRUCTURED_LOGGING_DATABASE_ENABLED=true
make serve

# Step 2: Perform gateway operations
curl -s -X POST -H "Authorization: Bearer $ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"audit-test-gw","url":"http://localhost:9000/see"}' \
  http://localhost:4444/gateways/

# Step 2: Check logs for gateway activities
docker logs mcpgateway 2>&1 | grep -i "gateway" | tail -10

# Expected log fields:
# - user_email
# - resource_type: "gateway"
# - resource_id
# - action: "create", "update", "delete"

Expected results returned:

• Gateway CRUD operations logged
• User identity captured
• Resource IDs included