add project base

backend/src/auth/auth.py

@@ -0,0 +1,80 @@
+from fastapi import APIRouter, Security, HTTPException
+from fastapi.responses import RedirectResponse
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from src.secrets import CLIENT_ID, REDIRECT_URI, AUTH_URL, CLIENT_SECRET, TOKEN_URL, JWKS_URL, ISSUER
+import secrets
+import jwt
+from jwt import PyJWKClient
+import requests
+
+router = APIRouter(prefix="/auth")
+
+jwk_client = PyJWKClient(JWKS_URL)
+security = HTTPBearer()
+
+@router.get('/login')
+def login():
+    state = secrets.token_urlsafe(16)
+
+    params = {
+        "client_id": CLIENT_ID,
+        "response_type": "code",
+        "scope": "openid",
+        "redirect_uri": REDIRECT_URI,
+        "state": state,
+    }
+
+    request_url = requests.Request('GET', AUTH_URL, params=params).prepare().url
+    return RedirectResponse(request_url)
+
+@router.get("/callback")
+def callback(code: str):
+    data = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": REDIRECT_URI,
+        "client_id": CLIENT_ID,
+        "client_secret": CLIENT_SECRET,
+    }
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded"
+    }
+    response = requests.post(TOKEN_URL, data=data, headers=headers)
+    if response.status_code != 200:
+        return JSONResponse(
+            {"error": "Failed to get token"},
+            status_code=400
+        )
+    token_data = response.json()
+    return {
+        "access_token": token_data["access_token"],
+        "id_token": token_data["id_token"],
+        "refresh_token": token_data["refresh_token"],
+    }
+
+def verify_token(token: str):
+    try:
+        signing_key = jwk_client.get_signing_key_from_jwt(token)
+        decoded = jwt.decode(token, options={"verify_signature": False})
+        print(decoded, ISSUER)
+        print(decoded["exp"])
+        payload = jwt.decode(
+            token,
+            signing_key.key,
+            algorithms=["RS256"],
+            audience=CLIENT_ID,
+            issuer=ISSUER,
+        )
+        return payload
+
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=401, detail="Token expired")
+
+    except jwt.InvalidTokenError:
+        raise HTTPException(status_code=401, detail="Invalid token")
+
+
+def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Security(security)
+):
+    return verify_token(credentials.credentials)