herel/myice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: proper jwt verification

Browse Source
This commit is contained in:
Rene Luria
2026-06-29 17:35:36 +02:00
parent 8f9aec7631
commit c3098cb415
1 changed files with 74 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
myice/webapi.py
+74 -30
View File
@@ -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")