Authorization Request
Scope: {html.escape(scope)}
Please try connecting again.
", status_code=400) redirect_uri = pending["redirect_uri"] state = pending["state"] if action != "approve": qs = urlencode({"error": "access_denied", "state": state}) return RedirectResponse(f"{redirect_uri}?{qs}", status_code=302) if not GATEWAY_PASSWORD: return HTMLResponse("OAUTH_PASSWORD not set.
", status_code=500) if password != GATEWAY_PASSWORD: PENDING_AUTH[internal_state] = pending client = REGISTERED_CLIENTS.get(pending["client_id"], {}) client_name = client.get("client_name", "Unknown Client") return HTMLResponse(render_consent_page(client_name, pending["scope"], internal_state, "Invalid password. Please try again.")) code = secrets.token_urlsafe(48) AUTH_CODES[code] = { "client_id": pending["client_id"], "redirect_uri": redirect_uri, "code_challenge": pending["code_challenge"], "code_challenge_method": pending["code_challenge_method"], "scope": pending["scope"], "expires_at": time.time() + AUTH_CODE_TTL, "user": "owner", } qs = urlencode({"code": code, "state": state}) logger.info(f"OAuth: issued auth code for client {pending['client_id']}") return RedirectResponse(f"{redirect_uri}?{qs}", status_code=302) return JSONResponse({"error": "method_not_allowed"}, status_code=405) async def oauth_token(request: Request) -> JSONResponse: _clean_expired() content_type = request.headers.get("content-type", "") if "application/json" in content_type: try: body = await request.json() except Exception: body = {} else: form = await request.form() body = dict(form) grant_type = body.get("grant_type", "") if grant_type == "authorization_code": code = body.get("code", "") client_id = body.get("client_id", "") code_verifier = body.get("code_verifier", "") code_info = AUTH_CODES.pop(code, None) if not code_info: return JSONResponse({"error": "invalid_grant", "error_description": "Code invalid or expired."}, status_code=400) if code_info["expires_at"] < time.time(): return JSONResponse({"error": "invalid_grant", "error_description": "Code expired."}, status_code=400) if code_info["client_id"] != client_id: return JSONResponse({"error": "invalid_grant", "error_description": "Client mismatch."}, status_code=400) if code_info.get("code_challenge"): if not code_verifier: return JSONResponse({"error": "invalid_grant", "error_description": "code_verifier required."}, status_code=400) expected = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).rstrip(b"=").decode() if expected != code_info["code_challenge"]: return JSONResponse({"error": "invalid_grant", "error_description": "PKCE verification failed."}, status_code=400) access_token = secrets.token_urlsafe(48) refresh_token = secrets.token_urlsafe(48) access_hash = _hash(access_token) refresh_hash = _hash(refresh_token) ACCESS_TOKENS[access_hash] = { "client_id": client_id, "scope": code_info["scope"], "expires_at": time.time() + ACCESS_TOKEN_TTL, "user": code_info["user"], } REFRESH_TOKENS[refresh_hash] = { "client_id": client_id, "scope": code_info["scope"], "expires_at": time.time() + REFRESH_TOKEN_TTL, "user": code_info["user"], "access_token_hash": access_hash, } logger.info(f"OAuth: issued access token for client {client_id}") return JSONResponse({ "access_token": access_token, "token_type": "Bearer", "expires_in": ACCESS_TOKEN_TTL, "refresh_token": refresh_token, "scope": code_info["scope"], }) elif grant_type == "refresh_token": refresh_token_val = body.get("refresh_token", "") client_id = body.get("client_id", "") refresh_hash = _hash(refresh_token_val) refresh_info = REFRESH_TOKENS.get(refresh_hash) if not refresh_info: return JSONResponse({"error": "invalid_grant", "error_description": "Refresh token invalid."}, status_code=400) if refresh_info["expires_at"] < time.time(): del REFRESH_TOKENS[refresh_hash] return JSONResponse({"error": "invalid_grant", "error_description": "Refresh token expired."}, status_code=400) if refresh_info["client_id"] != client_id: return JSONResponse({"error": "invalid_grant", "error_description": "Client mismatch."}, status_code=400) old_access_hash = refresh_info.get("access_token_hash") if old_access_hash and old_access_hash in ACCESS_TOKENS: del ACCESS_TOKENS[old_access_hash] del REFRESH_TOKENS[refresh_hash] new_access_token = secrets.token_urlsafe(48) new_refresh_token = secrets.token_urlsafe(48) new_access_hash = _hash(new_access_token) new_refresh_hash = _hash(new_refresh_token) ACCESS_TOKENS[new_access_hash] = { "client_id": client_id, "scope": refresh_info["scope"], "expires_at": time.time() + ACCESS_TOKEN_TTL, "user": refresh_info["user"], } REFRESH_TOKENS[new_refresh_hash] = { "client_id": client_id, "scope": refresh_info["scope"], "expires_at": time.time() + REFRESH_TOKEN_TTL, "user": refresh_info["user"], "access_token_hash": new_access_hash, } logger.info(f"OAuth: refreshed token for client {client_id}") return JSONResponse({ "access_token": new_access_token, "token_type": "Bearer", "expires_in": ACCESS_TOKEN_TTL, "refresh_token": new_refresh_token, "scope": refresh_info["scope"], }) return JSONResponse({"error": "unsupported_grant_type"}, status_code=400) # --------------------------------------------------------------------------- # MCP Endpoint (OAuth-protected) # --------------------------------------------------------------------------- async def handle_mcp(request: Request) -> Response: if request.method == "HEAD": return Response(status_code=200, headers={"MCP-Protocol-Version": "2024-11-05"}) if request.method == "GET": token_info = validate_bearer_token(request) if not token_info: return Response( status_code=401, headers={ "WWW-Authenticate": f'Bearer resource_metadata="{ISSUER_URL}/.well-known/oauth-protected-resource"', "Content-Type": "application/json", }, media_type="application/json", content=json.dumps({"error": "unauthorized", "message": "Bearer token required."}), ) # Authenticated GET — return empty SSE stream (mcpo uses this for server-sent events) async def empty_sse(): yield b": ping\n\n" return StreamingResponse(empty_sse(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}) if request.method == "DELETE": session_id = request.headers.get("Mcp-Session-Id") if session_id and session_id in GATEWAY_SESSIONS: del GATEWAY_SESSIONS[session_id] return Response(status_code=200) # POST — require Bearer token token_info = validate_bearer_token(request) if not token_info: return Response( status_code=401, headers={ "WWW-Authenticate": f'Bearer resource_metadata="{ISSUER_URL}/.well-known/oauth-protected-resource"', "Content-Type": "application/json", }, media_type="application/json", content=json.dumps({"error": "unauthorized", "message": "Valid Bearer token required."}), ) try: body = await request.json() except Exception: return JSONResponse( {"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}, status_code=400, ) method = body.get("method", "") params = body.get("params", {}) req_id = body.get("id") if method == "initialize": session_id = str(uuid.uuid4()) GATEWAY_SESSIONS[session_id] = True response = JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "mcp-gateway-proxy", "version": "1.0.0"}, }, }) response.headers["Mcp-Session-Id"] = session_id return response if method == "notifications/initialized": return Response(status_code=202) if method == "tools/list": tools = list(TOOL_DEFINITIONS.values()) response = JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": {"tools": tools}, }) session_id = request.headers.get("Mcp-Session-Id") if session_id: response.headers["Mcp-Session-Id"] = session_id return response if method == "tools/call": tool_name = params.get("name", "") arguments = params.get("arguments", {}) backend_name = TOOL_REGISTRY.get(tool_name) if not backend_name: return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"}, }) try: result = await forward_tool_call(backend_name, tool_name, arguments, req_id) if isinstance(result, dict) and "jsonrpc" in result: response = JSONResponse(result) elif isinstance(result, dict) and "result" in result: response = JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": result["result"], }) elif isinstance(result, dict) and "error" in result: response = JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": result["error"], }) else: response = JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": result, }) session_id = request.headers.get("Mcp-Session-Id") if session_id: response.headers["Mcp-Session-Id"] = session_id return response except Exception as e: logger.error(f"Tool call failed: {tool_name} - {e}") return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(e)}, }) if method == "ping": return JSONResponse({"jsonrpc": "2.0", "id": req_id, "result": {}}) return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, }) # --------------------------------------------------------------------------- # Health / Status # --------------------------------------------------------------------------- async def health(request: Request) -> JSONResponse: return JSONResponse({ "status": "healthy", "oauth": "enabled", }) async def status(request: Request) -> JSONResponse: token_info = validate_bearer_token(request) if not token_info: return JSONResponse({"error": "unauthorized"}, status_code=401) return JSONResponse({ "backends": { name: { "url": url, "tools": len([t for t, b in TOOL_REGISTRY.items() if b == name]), "session": BACKEND_SESSIONS.get(name), } for name, url in BACKENDS.items() }, "total_tools": len(TOOL_DEFINITIONS), "tool_list": sorted(TOOL_DEFINITIONS.keys()), }) # --------------------------------------------------------------------------- # Dashboard # --------------------------------------------------------------------------- DISPLAY_NAMES = { "erpnext": "ERPNext", "truenas": "TrueNAS", "homeassistant": "Home Assistant", "wave": "Wave Finance", "linkedin": "LinkedIn", } async def probe_backend(name: str, url: str) -> dict: """Live-probe a backend: initialize it and count its tools in real time.""" start = time.time() try: result = await mcp_request(url, "initialize", { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "mcp-gateway-dashboard", "version": "1.0.0"}, }) session_id = result.get("session_id") tools_result = await mcp_request(url, "tools/list", {}, request_id=2, session_id=session_id) tools_data = tools_result.get("result", {}) if isinstance(tools_data, dict): if "result" in tools_data: tools_data = tools_data["result"] tools = tools_data.get("tools", []) else: tools = [] elapsed = round((time.time() - start) * 1000) return { "status": "healthy", "toolCount": len(tools), "responseTime": elapsed, } except Exception as e: elapsed = round((time.time() - start) * 1000) logger.warning(f"Dashboard probe failed for {name}: {e}") # Fall back to cached registry values if live probe fails cached_tools = len([t for t, b in TOOL_REGISTRY.items() if b == name]) return { "status": "healthy" if cached_tools > 0 else "unhealthy", "toolCount": cached_tools, "responseTime": elapsed, "note": "cached", } async def dashboard_status(request: Request) -> JSONResponse: """Auth-protected endpoint — live-probes all backends.""" token_info = validate_bearer_token(request) if not token_info: return JSONResponse({"error": "unauthorized"}, status_code=401) probes = await asyncio.gather( *[probe_backend(name, url) for name, url in BACKENDS.items()], return_exceptions=True ) services = [] for (name, url), probe in zip(BACKENDS.items(), probes): if isinstance(probe, Exception): probe = {"status": "unhealthy", "toolCount": 0, "responseTime": None} services.append({ "name": DISPLAY_NAMES.get(name, name.capitalize()), "key": name, "url": url, "status": probe["status"], "toolCount": probe["toolCount"], "responseTime": probe.get("responseTime"), "lastCheck": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), }) return JSONResponse({ "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "services": services, "summary": { "total": len(services), "healthy": len([s for s in services if s["status"] == "healthy"]), "unhealthy": len([s for s in services if s["status"] == "unhealthy"]), "totalTools": sum(s["toolCount"] for s in services), } }) DASHBOARD_HTML = """Service Status Dashboard
Total Services
—
Total Tools
—
Healthy
—