feat: implement OpenID Connect authentication with Infomaniak

myice/webapi.py

@@ -1,12 +1,31 @@
 import logging
 import requests
+import os
 from typing import Annotated
-from fastapi import FastAPI, Header, HTTPException, status
+from fastapi import FastAPI, Header, HTTPException, status, Request
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, RedirectResponse
 from pydantic import BaseModel
 from . import myice
 
+# OIDC Configuration
+CLIENT_ID = os.environ.get("CLIENT_ID", "8ea04fbb-4237-4b1d-a895-0b3575a3af3f")
+CLIENT_SECRET = os.environ.get(
+    "CLIENT_SECRET", "5iXycu7aU6o9e17NNTOUeetQkRkBqQlomoD2hTyLGLTAcwj0dwkkLH3mz1IrZSmJ"
+)
+REDIRECT_URI = os.environ.get("REDIRECT_URI", "http://localhost:8000/callback")
+AUTHORIZATION_ENDPOINT = "https://login.infomaniak.com/authorize"
+TOKEN_ENDPOINT = "https://login.infomaniak.com/token"
+USERINFO_ENDPOINT = "https://login.infomaniak.com/oauth2/userinfo"
+JWKS_URI = "https://login.infomaniak.com/oauth2/jwks"
+
+# Allowed users (based on email claim)
+ALLOWED_USERS = (
+    os.environ.get("ALLOWED_USERS", "").split(",")
+    if os.environ.get("ALLOWED_USERS")
+    else []
+)
+
 origins = ["*"]
 
 app = FastAPI()
@@ -41,10 +60,43 @@ class AuthHeaders(BaseModel):
         return self.Authorization.split(" ")[1]
 
     def authorized(self) -> bool:
+        # First check if it's a valid OIDC token
+        if self.validate_oidc_token():
+            return True
+        # Fallback to the old static key for compatibility
         if self.token() == "abc":
             return True
         return False
 
+    def validate_oidc_token(self) -> bool:
+        try:
+            token = self.token()
+            # Basic validation - in production, you would validate the JWT signature
+            # and check issuer, audience, expiration, etc.
+            import base64
+            import json
+
+            # Decode JWT header and payload (without verification for simplicity in this example)
+            parts = token.split(".")
+            if len(parts) != 3:
+                return False
+
+            # Decode payload (part 1)
+            payload = parts[1]
+            # Add padding if needed
+            payload += "=" * (4 - len(payload) % 4)
+            decoded_payload = base64.urlsafe_b64decode(payload)
+            payload_data = json.loads(decoded_payload)
+
+            # Check if user is in allowed list
+            user_email = payload_data.get("email")
+            if ALLOWED_USERS and user_email not in ALLOWED_USERS:
+                return False
+
+            return True
+        except Exception:
+            return False
+
 
 class HealthCheck(BaseModel):
     """Response model to validate and return when performing a health check."""
@@ -71,6 +123,79 @@ async def favico():
     return FileResponse("favicon.ico")
 
 
+@app.get("/login")
+async def login(request: Request):
+    """Redirect to Infomaniak OIDC login"""
+    import urllib.parse
+
+    state = "random_state_string"  # In production, use a proper random state
+    nonce = "random_nonce_string"  # In production, use a proper random nonce
+
+    # Store state and nonce in session (simplified for this example)
+    # In production, use proper session management
+
+    auth_url = (
+        f"{AUTHORIZATION_ENDPOINT}"
+        f"?client_id={CLIENT_ID}"
+        f"&redirect_uri={urllib.parse.quote(REDIRECT_URI)}"
+        f"&response_type=code"
+        f"&scope=openid email profile"
+        f"&state={state}"
+        f"&nonce={nonce}"
+    )
+
+    return RedirectResponse(auth_url)
+
+
+@app.get("/callback")
+async def callback(code: str, state: str):
+    """Handle OIDC callback and exchange code for token"""
+
+    # Exchange authorization code for access token
+    token_data = {
+        "grant_type": "authorization_code",
+        "client_id": CLIENT_ID,
+        "client_secret": CLIENT_SECRET,
+        "code": code,
+        "redirect_uri": REDIRECT_URI,
+    }
+
+    response = requests.post(TOKEN_ENDPOINT, data=token_data)
+    token_response = response.json()
+
+    if "access_token" not in token_response:
+        raise HTTPException(status_code=400, detail="Failed to obtain access token")
+
+    access_token = token_response["access_token"]
+
+    # Get user info
+    userinfo_response = requests.get(
+        USERINFO_ENDPOINT, headers={"Authorization": f"Bearer {access_token}"}
+    )
+
+    userinfo = userinfo_response.json()
+
+    # Check if user is in allowed list
+    user_email = userinfo.get("email")
+    if ALLOWED_USERS and user_email not in ALLOWED_USERS:
+        raise HTTPException(status_code=403, detail="User not authorized")
+
+    # Return token to frontend
+    return {"access_token": access_token, "user": userinfo}
+
+
+@app.get("/userinfo")
+async def userinfo(headers: Annotated[AuthHeaders, Header()]):
+    """Get user info from access token"""
+    access_token = headers.token()
+
+    userinfo_response = requests.get(
+        USERINFO_ENDPOINT, headers={"Authorization": f"Bearer {access_token}"}
+    )
+
+    return userinfo_response.json()
+
+
 @app.get("/schedule")
 async def schedule(
     headers: Annotated[AuthHeaders, Header()],