chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(opax-mcp): add IAP identity token auth for opax.vauco.no calls

Browse Source 
Add _iap_token() that fetches identity token from GCP metadata server
using IAP client ID as audience. Update _opax_get and _opax_post to
send Authorization: Bearer <token> header on all calls to opax.vauco.no.

Fixes 401 Unauthorized from IAP-protected opax.vauco.no backend.
Related: LEARNING-013
This commit is contained in:
Chris Christiansen 2026-06-17 16:57:54 +00:00
parent afabe8d2a9
commit eb351b5e85
1 changed files with 16 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
opax-mcp/server.py
View File

@ -12,6 +12,16 @@ from typing import Optional, Any
app = FastAPI(title="opax-mcp", version="2.0.0") app = FastAPI(title="opax-mcp", version="2.0.0")
OPAX_BASE_URL = os.environ.get("OPAX_BASE_URL", "https://opax.vauco.no") OPAX_BASE_URL = os.environ.get("OPAX_BASE_URL", "https://opax.vauco.no")
IAP_CLIENT_ID = os.environ.get("IAP_CLIENT_ID", "357036551735-ka7t2fv9ue2jp01bs826hpdctlvispuo.apps.googleusercontent.com")
async def _iap_token() -> str:
"""Fetch IAP identity token from GCP metadata server."""
url = (f"http://metadata.google.internal/computeMetadata/v1/instance/"
f"service-accounts/default/identity?audience={IAP_CLIENT_ID}&format=full")
async with httpx.AsyncClient(timeout=10) as c:
r = await c.get(url, headers={"Metadata-Flavor": "Google"})
r.raise_for_status()
return r.text.strip()
MCP_SECRET = os.environ.get("MCP_SECRET", "") MCP_SECRET = os.environ.get("MCP_SECRET", "")
# Gitea # Gitea
@ -69,15 +79,19 @@ async def list_tools(x_mcp_secret: Optional[str] = Header(default=None)):
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
async def _opax_get(path: str) -> Any: async def _opax_get(path: str) -> Any:
token = await _iap_token()
async with httpx.AsyncClient(timeout=30) as client: async with httpx.AsyncClient(timeout=30) as client:
r = await client.get(f"{OPAX_BASE_URL}{path}") r = await client.get(f"{OPAX_BASE_URL}{path}",
headers={"Authorization": f"Bearer {token}"})
r.raise_for_status() r.raise_for_status()
return r.json() return r.json()
async def _opax_post(path: str, body: dict) -> Any: async def _opax_post(path: str, body: dict) -> Any:
token = await _iap_token()
async with httpx.AsyncClient(timeout=30) as client: async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{OPAX_BASE_URL}{path}", json=body) r = await client.post(f"{OPAX_BASE_URL}{path}", json=body,
headers={"Authorization": f"Bearer {token}"})
r.raise_for_status() r.raise_for_status()
return r.json() return r.json()