Cambiato endpoints.py

Aggiornato endpoints.py

Backend/endpoints.py

@@ -1,6 +1,6 @@
-from fastapi import FastAPI, Form, Request, Depends, HTTPException, Body
+from fastapi import FastAPI, Form, Request, HTTPException
 from pydantic import BaseModel
-from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse
+from fastapi.responses import JSONResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
 from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.cors import CORSMiddleware
@@ -8,12 +8,12 @@ from starlette.middleware.sessions import SessionMiddleware
 import secrets
 import uvicorn
 import logging
-import requests  # Import the requests library here
-from typing import Annotated, Dict, List, Optional #import typing
+import requests  
+from typing import Dict, Optional 
 
 
 # --- Keycloak Configuration ---
-KEYCLOAK_SERVER_URL = "http://165.22.75.145:8080"  # Double-check this! Changed to http
+KEYCLOAK_SERVER_URL = "http://165.22.75.145:8080"  
 KEYCLOAK_REALM = "Generation"
 KEYCLOAK_CLIENT_ID = "web-app-pw"
 KEYCLOAK_CLIENT_SECRET = "fQGWt8HSPn65cCKTOzE5FigqZhf8QTYW"
@@ -26,12 +26,6 @@ app = FastAPI()
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
-# --- CORS ---
-# origins = [
-#     "http://localhost:8000/*",
-#     "http://localhost:3000/*",
-# ]
-
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -54,7 +48,7 @@ def get_token_from_keycloak(username, password) -> Dict:
     }
     try:
         response = requests.post(url, headers=headers, data=payload, timeout=10)
-        response.raise_for_status()  # Raise HTTPError for bad responses (4xx or 5xx)
+        response.raise_for_status() 
         return response.json()
     except requests.exceptions.RequestException as e:
         logger.error(f"Error getting token from Keycloak: {e}")
@@ -120,17 +114,171 @@ def logout_keycloak(refresh_token:str):
         logger.error(f"Error logging out user: {e}")
         raise HTTPException(status_code=500, detail=f"Failed to logout user: {e}") from e
 
-class Credenziali(BaseModel):
-    username: str
-    password: str
 
 app.mount("/static", StaticFiles(directory="static"), name="static")
 templates = Jinja2Templates(directory="templates")
 
+@app.get("/mappa_login")
+async def mappa_login(request: Request):
+    request.session.clear()
+    return templates.TemplateResponse("mappa_login.html", {"request": request})
+
+@app.get("/tfo")
+async def tfo(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("TFO.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/tfo?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
+@app.get("/instructions")
+async def instructions(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("instructions.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/instructions?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
+@app.get("/buildings")
+async def buildings(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("buildings.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/buildings?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
+@app.get("/faq")
+async def faq(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("faq.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/faq?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
+@app.get("/techsup")
+async def techsup(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("techsup.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/techsup?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
 
 @app.get("/callback")
 async def callback(request: Request):
-    return templates.TemplateResponse("mappa_logout.html", context={"request": request, "title": "Mappa"})
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return templates.TemplateResponse("mappa_logout.html", {"request": request})
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/callback?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
+@app.get("/admin")
+async def admin(request: Request):
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            if get_user_info_from_keycloak(str(request.session.get("access_token")))["ruolo"] == "amministratore":
+                return templates.TemplateResponse("ADMIN.html", {"request": request})
+            return RedirectResponse(url="/callback", status_code=303)
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/admin?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
+
+        except Exception as e:
+            request.session.clear()
+            return RedirectResponse(url="/access", status_code=303)
+    request.session.clear()
+    return RedirectResponse(url="/access", status_code=303)
+
 
 @app.get("/access")
 @app.post("/access")
@@ -138,27 +286,21 @@ async def login(request: Request, username: Optional[str] = Form(None), password
     error = None
     if request.method == "POST":
         try:
-            request.session["user_info"] = {}
-            #print(request.session.get("tokens"))
-            
-            tokens = get_token_from_keycloak(username, password)#request.body.json()["username"], request.json()["password"])
-            
-            # Save tokens and information to the session
-
-            request.session["access_token"] = tokens["access_token"]
-            request.session["refresh_token"] = tokens["refresh_token"]
+            request.session.clear()
 
-            #print(tokens["access_token"])
+            tokens = get_token_from_keycloak(username, password)
 
-            # Get user info and save it to the session too
+            request.session.update(
+                {"access_token": tokens["access_token"], "refresh_token": tokens["refresh_token"]}
+            )
 
             user_info = get_user_info_from_keycloak(tokens["access_token"])
 
-            request.session["user_info"] = user_info
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
 
-
-
-            return RedirectResponse(url=f"/callback?access_token={tokens["access_token"]}&refresh_token={tokens["refresh_token"]}", status_code=303)
+            return RedirectResponse(url=f"/callback?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
 
         except HTTPException as e:
 
@@ -170,24 +312,108 @@ async def login(request: Request, username: Optional[str] = Form(None), password
 
             return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
     
-    if request.session.get("access_token") is not None:
-                #print(request.session.get("access_token"))
-                if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
-                    #print(3)
-                    return RedirectResponse(url="/callback", status_code=303)
-                try:
-                    tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+    if "access_token" in request.session.keys():
+        if introspect_keycloak_token_request(str(request.session.get("access_token")))["active"] == True:
+            return RedirectResponse(url="/callback", status_code=303)
+        try:
+            tokens = refresh_token_from_keycloak(str(request.session.get("refresh_token")))
+
+            request.session.update({"access_token" : tokens["access_token"], "refresh_token" : tokens["refresh_token"]})
+            user_info = get_user_info_from_keycloak(tokens["access_token"])
+
+            request.session.update(
+                {"ruolo": user_info["ruolo"], "codice_fiscale": user_info["CF"]}
+            )
+            return RedirectResponse(url=f"/callback?ruolo={user_info["ruolo"]}&codice_fiscale={user_info["CF"]}", status_code=303)
 
-                    #localStorage.setItem('jwtToken', tuoJWT);
-                    request.session["access_token"] = tokens["access_token"]
-                    request.session["refresh_token"] = tokens["refresh_token"]
+        except Exception as e:
+            request.session.clear()
+            return templates.TemplateResponse("login.html", {"request": request, "error": error})
+    
+    request.session.clear()
+    return templates.TemplateResponse("login.html", {"request": request, "error": error})
 
-                    return RedirectResponse(url=f"/callback?access_token={tokens["access_token"]}&refresh_token={tokens["refresh_token"]}", status_code=303)
+@app.get("/refresh")
+async def refresh(request: Request):
+    """Refreshes the access token using the refresh token in the session."""
+    refresh_token = request.session.get("refresh_token")
+    if not refresh_token:
+        raise HTTPException(status_code=401, detail="Refresh token not found in session")
+
+    try:
+        new_tokens = refresh_token_from_keycloak(refresh_token)
+        request.session["access_token"] = new_tokens["access_token"]
+        request.session["refresh_token"] = new_tokens["refresh_token"]
+        new_user_info = get_user_info_from_keycloak(new_tokens["access_token"])
+        request.session["user_info"] = new_user_info
+        return {"message": "Token refreshed successfully", "access_token": new_tokens["access_token"], "user_info": new_user_info}
+    except HTTPException as e:
+        return JSONResponse(content={"detail": e.detail}, status_code=e.status_code)
+    except Exception as e:
+        logger.error(f"An unexpected error occurred during token refresh: {e}")
+        return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
+
+
+@app.get("/introspect")
+async def introspect(request: Request):
+    """Introspects the access token in the session."""
+    access_token = request.session.get("access_token")
+    if not access_token:
+        raise HTTPException(status_code=401, detail="Access token not found in session")
+
+    try:
+        introspect_data = introspect_keycloak_token_request(access_token)
+        return {"message": "Token introspection successful", "introspect_data": introspect_data}
+    except HTTPException as e:
+        return JSONResponse(content={"detail": e.detail}, status_code=e.status_code)
+    except Exception as e:
+        logger.error(f"An unexpected error occurred during token introspection: {e}")
+        return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
+
+@app.get("/userinfo")
+async def user_info(request: Request):
+    """Retrieves and returns user information stored in the session."""
+    user_info = request.session.get("user_info")
+    if not user_info:
+        raise HTTPException(status_code=401, detail="User info not found in session")
+    return {"message": "User information retrieved", "user_info": user_info}
+
+@app.get("/logout_keycloak")
+async def logout_user(request: Request):
+    """Logs out a user by revoking the refresh token."""
+    refresh_token = request.session.get("refresh_token")
+    if not refresh_token:
+        raise HTTPException(status_code=401, detail="Refresh token not found in session")
+    try:
+        logout_keycloak(refresh_token)
+        request.session.clear()
+        return {"message": "Logout successful"}
+    except HTTPException as e:
+        return JSONResponse(content={"detail": e.detail}, status_code=e.status_code)
+    except Exception as e:
+        logger.error(f"An unexpected error occurred during logout: {e}")
+        return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
+
+@app.get("/protected")
+async def protected_endpoint(request: Request):
+    """A protected endpoint that requires a valid access token."""
+    access_token = request.session.get("access_token")
+    if not access_token:
+        raise HTTPException(status_code=401, detail="Access token not found in session")
+    try:
+        introspect_data = introspect_keycloak_token_request(access_token)
+        if not introspect_data.get("active"):
+            raise HTTPException(status_code=401, detail="Access token is not active")
+        return JSONResponse({"message": f"Hello, world! (Protected)", "introspect": introspect_data})
+    except HTTPException as e:
+        return JSONResponse(content={"detail": e.detail}, status_code=e.status_code)
+    except Exception as e:
+        logger.error(f"An unexpected error occurred during token introspection: {e}")
+        return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
+    
 
-                except Exception as e:
+# --- Run the App ---
+if __name__ == "__main__":
+    uvicorn.run(app, host="localhost", port=8000)
 
-                    logger.error(f"An unexpected error occurred during login: {e}")
 
-                    return JSONResponse(content={"detail": "An unexpected error occurred"}, status_code=500)
-                
-    return templates.TemplateResponse("login.html", {"request": request, "error": error})