From 9795a4c81344a29e5d4fffbfdb2645b140a72d27 Mon Sep 17 00:00:00 2001 From: Rene Luria Date: Mon, 29 Jun 2026 17:35:36 +0200 Subject: [PATCH] fix: proper jwt verification --- myice/webapi.py | 104 ++++++++++++++++++++++++++++++++++-------------- 1 file changed, 74 insertions(+), 30 deletions(-) diff --git a/myice/webapi.py b/myice/webapi.py index 3183dfd..eebf074 100644 --- a/myice/webapi.py +++ b/myice/webapi.py @@ -1,11 +1,13 @@ +import base64 +import hashlib import logging -import requests import os -from typing import Annotated -from fastapi import FastAPI, Header, HTTPException, status, Request +import requests +from fastapi import FastAPI, Header, HTTPException, Request, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, RedirectResponse from pydantic import BaseModel +from typing import Annotated from . import myice # OIDC Configuration @@ -14,6 +16,7 @@ CLIENT_SECRET = os.environ.get("CLIENT_SECRET") if not CLIENT_SECRET: raise RuntimeError("CLIENT_SECRET environment variable must be set") REDIRECT_URI = os.environ.get("REDIRECT_URI", "http://localhost:8000/callback") +SECRET_KEY = os.environ.get("SECRET_KEY", "change_me_in_production") AUTHORIZATION_ENDPOINT = "https://login.infomaniak.com/authorize" TOKEN_ENDPOINT = "https://login.infomaniak.com/token" USERINFO_ENDPOINT = "https://login.infomaniak.com/oauth2/userinfo" @@ -78,26 +81,38 @@ class AuthHeaders(BaseModel): def validate_oidc_token(self) -> bool: try: - import jwt - token = self.token() if not token: return False - # Fetch the JWKS and validate the JWT properly - # This example uses PyJWT with a JWKSet; install with `pip install pyjwt[crypto]` - from jwt import PyJWKClient + # Try JWT validation first (for id_token-style tokens) + if token.count(".") == 2: + import jwt + from jwt import PyJWKClient - jwks_client = PyJWKClient(JWKS_URI) - signing_key = jwks_client.get_signing_key_from_jwt(token) - payload_data = jwt.decode( - token, - signing_key.key, - algorithms=["RS256"], - audience=CLIENT_ID, - issuer="https://login.infomaniak.com/", + jwks_client = PyJWKClient(JWKS_URI) + signing_key = jwks_client.get_signing_key_from_jwt(token) + payload_data = jwt.decode( + token, + signing_key.key, + algorithms=["RS256"], + audience=CLIENT_ID, + issuer="https://login.infomaniak.com/", + ) + user_email = payload_data.get("email") + if ALLOWED_USERS and user_email and user_email not in ALLOWED_USERS: + return False + return True + + # Fallback: validate opaque tokens by calling userinfo endpoint + userinfo_response = requests.get( + USERINFO_ENDPOINT, headers={"Authorization": f"Bearer {token}"} ) - user_email = payload_data.get("email") + if userinfo_response.status_code != 200: + return False + + userinfo = userinfo_response.json() + user_email = userinfo.get("email") if ALLOWED_USERS and user_email and user_email not in ALLOWED_USERS: return False return True @@ -139,12 +154,14 @@ async def login(request: Request): state = secrets.token_urlsafe(32) nonce = secrets.token_urlsafe(32) - # Store in a secure signed session cookie - request.state.oidc_state = state - request.state.oidc_nonce = nonce - # Store state and nonce in session (simplified for this example) - # In production, use proper session management + # Generate PKCE parameters for public clients / Infomaniak requirement + code_verifier = secrets.token_urlsafe(32) + code_challenge = ( + base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode("ascii")).digest()) + .decode("ascii") + .rstrip("=") + ) auth_url = ( f"{AUTHORIZATION_ENDPOINT}" @@ -154,17 +171,30 @@ async def login(request: Request): f"&scope=openid email profile" f"&state={state}" f"&nonce={nonce}" + f"&code_challenge={code_challenge}" + f"&code_challenge_method=S256" ) - return RedirectResponse(auth_url) + response = RedirectResponse(auth_url) + # Store state and nonce in secure http-only cookies for validation on callback + response.set_cookie("oidc_state", state, httponly=True, max_age=300) + response.set_cookie("oidc_nonce", nonce, httponly=True, max_age=300) + response.set_cookie("oidc_verifier", code_verifier, httponly=True, max_age=300) + return response @app.get("/callback") async def callback(request: Request, code: str, state: str): """Handle OIDC callback and exchange code for token""" - expected_state = request.state.oidc_state + expected_state = request.cookies.get("oidc_state") + code_verifier = request.cookies.get("oidc_verifier") + # We should also delete state/nonce cookies once read if not expected_state or state != expected_state: - raise HTTPException(403, detail="Invalid state parameter") + response = RedirectResponse("/?error=Invalid state parameter") + response.delete_cookie("oidc_state") + response.delete_cookie("oidc_nonce") + response.delete_cookie("oidc_verifier") + return response # Exchange authorization code for access token token_data = { "grant_type": "authorization_code", @@ -172,18 +202,24 @@ async def callback(request: Request, code: str, state: str): "client_secret": CLIENT_SECRET, "code": code, "redirect_uri": REDIRECT_URI, + "code_verifier": code_verifier, } - response = requests.post(TOKEN_ENDPOINT, data=token_data) - token_response = response.json() + token_response = requests.post(TOKEN_ENDPOINT, data=token_data).json() if "access_token" not in token_response: + # Log the error for debugging + logging.error("Token exchange failed: %s", token_response) # Redirect back to frontend with error frontend_url = os.environ.get("FRONTEND_URL", "/") error_detail = token_response.get( "error_description", "Failed to obtain access token" ) - return RedirectResponse(f"{frontend_url}?error={error_detail}") + error_response = RedirectResponse(f"{frontend_url}?error={error_detail}") + error_response.delete_cookie("oidc_state") + error_response.delete_cookie("oidc_nonce") + error_response.delete_cookie("oidc_verifier") + return error_response access_token = token_response["access_token"] @@ -199,11 +235,19 @@ async def callback(request: Request, code: str, state: str): if ALLOWED_USERS and user_email not in ALLOWED_USERS: # Redirect back to frontend with error frontend_url = os.environ.get("FRONTEND_URL", "/") - return RedirectResponse(f"{frontend_url}?error=User not authorized") + response = RedirectResponse(f"{frontend_url}?error=User not authorized") + response.delete_cookie("oidc_state") + response.delete_cookie("oidc_nonce") + response.delete_cookie("oidc_verifier") + return response # Redirect back to frontend with access token and user info frontend_url = os.environ.get("FRONTEND_URL", "/") - return RedirectResponse(f"{frontend_url}#access_token={access_token}") + response = RedirectResponse(f"{frontend_url}#access_token={access_token}") + response.delete_cookie("oidc_state") + response.delete_cookie("oidc_nonce") + response.delete_cookie("oidc_verifier") + return response @app.get("/exchange-token")