#!/usr/bin/env python3 """Closed-lab HTTP servers for ToolHive DCR discovery SSRF variant testing.""" import argparse import http.server import json import socketserver import time class ReuseTCPServer(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def log_line(path, obj): obj = dict(obj) obj.setdefault("ts", time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(obj, sort_keys=True) + "\n") f.flush() def send_json(handler, body, status=200): data = json.dumps(body).encode("utf-8") handler.send_response(status) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(data))) handler.end_headers() handler.wfile.write(data) class RemoteHandler(http.server.BaseHTTPRequestHandler): def _challenge(self): log_line(self.server.log_path, { "server": "remote_mcp", "event": "auth_challenge", "method": self.command, "path": self.path, "headers": dict(self.headers), "client": self.client_address[0], "resource_metadata": self.server.resource_metadata_url, }) value = f'Bearer realm="toolhive-variant", resource_metadata="{self.server.resource_metadata_url}"' body = b'{"error":"authentication required","variant":"dcr discovery refetch"}\n' self.send_response(401) self.send_header("WWW-Authenticate", value) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): self._challenge() def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length).decode("utf-8", "replace") if length else "" log_line(self.server.log_path, {"server": "remote_mcp", "event": "post_body", "body": body}) self._challenge() def log_message(self, fmt, *args): log_line(self.server.log_path, {"server": "remote_mcp_http", "message": fmt % args}) class AuthServerHandler(http.server.BaseHTTPRequestHandler): def _metadata_doc(self): return { "issuer": self.server.issuer_url, "authorization_endpoint": self.server.issuer_url + "/authorize", "token_endpoint": self.server.issuer_url + "/token", "jwks_uri": self.server.issuer_url + "/jwks", "registration_endpoint": self.server.issuer_url + "/register", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code", "refresh_token"], "token_endpoint_auth_methods_supported": ["none"], "code_challenge_methods_supported": ["S256"], "scopes_supported": ["openid", "profile"], } def do_GET(self): log_line(self.server.log_path, { "server": "auth_server", "event": "get", "method": "GET", "path": self.path, "headers": dict(self.headers), "client": self.client_address[0], }) if self.path.startswith("/resource-metadata"): send_json(self, { "resource": self.server.remote_url, "authorization_servers": [self.server.issuer_url], "scopes_supported": ["openid", "profile"], }) return if self.path.startswith("/.well-known/openid-configuration"): send_json(self, self._metadata_doc()) return if self.path.startswith("/.well-known/oauth-authorization-server"): # This is the second metadata discovery performed by the DCR resolver. # A guarded client should refuse this cross-port redirect; ToolHive v0.31.0 # and current latest still use an unguarded oauthproto client here. log_line(self.server.log_path, { "server": "auth_server", "event": "dcr_metadata_redirect", "location": self.server.canary_metadata_url, }) self.send_response(302) self.send_header("Location", self.server.canary_metadata_url) self.send_header("Content-Length", "0") self.end_headers() return if self.path.startswith("/authorize"): send_json(self, {"authorization": "not completed in headless test"}, status=200) return if self.path.startswith("/jwks"): send_json(self, {"keys": []}) return send_json(self, {"error": "not found", "path": self.path}, status=404) def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length).decode("utf-8", "replace") if length else "" log_line(self.server.log_path, { "server": "auth_server", "event": "post", "path": self.path, "headers": dict(self.headers), "body": body, "client": self.client_address[0], }) if self.path.startswith("/register"): send_json(self, { "client_id": "toolhive-variant-client", "redirect_uris": ["http://localhost:8666/callback"], "token_endpoint_auth_method": "none", "grant_types": ["authorization_code", "refresh_token"], "response_types": ["code"], "scope": "openid profile", }, status=201) return send_json(self, {"error": "not found", "path": self.path}, status=404) def log_message(self, fmt, *args): log_line(self.server.log_path, {"server": "auth_server_http", "message": fmt % args}) class CanaryHandler(http.server.BaseHTTPRequestHandler): def _metadata_doc(self): return { "issuer": self.server.issuer_url, "authorization_endpoint": self.server.issuer_url + "/authorize", "token_endpoint": self.server.issuer_url + "/token", "jwks_uri": self.server.issuer_url + "/jwks", "registration_endpoint": self.server.issuer_url + "/register", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code", "refresh_token"], "token_endpoint_auth_methods_supported": ["none"], "code_challenge_methods_supported": ["S256"], "scopes_supported": ["openid", "profile"], } def do_GET(self): log_line(self.server.log_path, { "server": "internal_canary", "event": "hit", "method": "GET", "path": self.path, "headers": dict(self.headers), "client": self.client_address[0], }) send_json(self, self._metadata_doc()) def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length).decode("utf-8", "replace") if length else "" log_line(self.server.log_path, { "server": "internal_canary", "event": "hit", "method": "POST", "path": self.path, "headers": dict(self.headers), "body": body, "client": self.client_address[0], }) send_json(self, self._metadata_doc()) def log_message(self, fmt, *args): log_line(self.server.log_path, {"server": "internal_canary_http", "message": fmt % args}) ROLE_MAP = { "remote": RemoteHandler, "auth": AuthServerHandler, "canary": CanaryHandler, } def main(): parser = argparse.ArgumentParser() parser.add_argument("--role", choices=sorted(ROLE_MAP), required=True) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, required=True) parser.add_argument("--log", required=True) parser.add_argument("--remote-url", default="") parser.add_argument("--resource-metadata-url", default="") parser.add_argument("--issuer-url", default="") parser.add_argument("--canary-metadata-url", default="") args = parser.parse_args() handler = ROLE_MAP[args.role] httpd = ReuseTCPServer((args.host, args.port), handler) httpd.log_path = args.log httpd.remote_url = args.remote_url httpd.resource_metadata_url = args.resource_metadata_url httpd.issuer_url = args.issuer_url httpd.canary_metadata_url = args.canary_metadata_url log_line(args.log, {"server": args.role, "event": "started", "host": args.host, "port": args.port}) print(f"READY {args.role} {args.host}:{args.port}", flush=True) httpd.serve_forever() if __name__ == "__main__": main()