Skip to content
Snippets Groups Projects

improve front

Merged Antoine Gaudron-Desjardins requested to merge collab_front into main
4 files
+ 152
16
Compare changes
  • Side-by-side
  • Inline

Files

+ 146
0
 
from pydantic import BaseModel
 
from fastapi import HTTPException, APIRouter, Response, Depends
 
from uuid import UUID, uuid4
 
from fastapi_sessions.backends.implementations import InMemoryBackend
 
from fastapi_sessions.session_verifier import SessionVerifier
 
from fastapi_sessions.frontends.implementations import SessionCookie, CookieParameters
 
from fastapi.responses import RedirectResponse
 
from urllib.parse import urlencode
 
from dotenv import load_dotenv
 
from sqlalchemy.orm import Session
 
from requests import get, post
 
import os
 
 
 
load_dotenv()
 
 
 
class SessionData(BaseModel):
 
username: str
 
 
 
cookie_params = CookieParameters()
 
 
# Uses UUID
 
cookie = SessionCookie(
 
cookie_name="cookie",
 
identifier="general_verifier",
 
auto_error=True,
 
secret_key="DONOTUSE",
 
cookie_params=cookie_params,
 
)
 
backend = InMemoryBackend[UUID, SessionData]()
 
 
 
class BasicVerifier(SessionVerifier[UUID, SessionData]):
 
def __init__(
 
self,
 
*,
 
identifier: str,
 
auto_error: bool,
 
backend: InMemoryBackend[UUID, SessionData],
 
auth_http_exception: HTTPException,
 
):
 
self._identifier = identifier
 
self._auto_error = auto_error
 
self._backend = backend
 
self._auth_http_exception = auth_http_exception
 
 
@property
 
def identifier(self):
 
return self._identifier
 
 
@property
 
def backend(self):
 
return self._backend
 
 
@property
 
def auto_error(self):
 
return self._auto_error
 
 
@property
 
def auth_http_exception(self):
 
return self._auth_http_exception
 
 
def verify_session(self, model: SessionData) -> bool:
 
"""If the session exists, it is valid"""
 
return True
 
 
 
verifier = BasicVerifier(
 
identifier="general_verifier",
 
auto_error=True,
 
backend=backend,
 
auth_http_exception=HTTPException(status_code=403, detail="invalid session"),
 
)
 
 
 
router = APIRouter(prefix="/api", tags=["auth"])
 
 
 
@router.post("/create_session/{name}")
 
async def create_session(name: str, response: Response):
 
 
session = uuid4()
 
data = SessionData(username=name)
 
 
await backend.create(session, data)
 
cookie.attach_to_response(response, session)
 
 
return f"created session for {name}"
 
 
 
@router.get("/whoami", dependencies=[Depends(cookie)])
 
async def whoami(session_data: SessionData = Depends(verifier)):
 
return session_data
 
 
 
@router.post("/delete_session")
 
async def del_session(response: Response, session_id: UUID = Depends(cookie)):
 
await backend.delete(session_id)
 
cookie.delete_from_response(response)
 
return "deleted session"
 
 
 
@router.get("/login")
 
async def login(code):
 
 
if not code:
 
new_state = crud.create_state(db)
 
params = urlencode({
 
"client_id": os.getenv("CLIENT_ID"),
 
"redirect_uri": f"{os.getenv('API_ROOT')}/login",
 
"response_type": "code",
 
"state": new_state.state,
 
"scope": "default linkcs-event:read"})
 
return RedirectResponse(f"{os.getenv('AUTH_ROOT')}/oauth/authorize?{params}")
 
 
stored_states = crud.get_states(db)
 
 
if state not in map(lambda state: state.state, stored_states):
 
raise HTTPException(status_code=403, detail="State Invalid")
 
crud.delete_state(state, db)
 
 
headers = {"content-type": "application/x-www-form-urlencoded"}
 
data = {
 
"grant_type": "authorization_code",
 
"code": code,
 
"redirect_uri": f"{os.getenv('API_ROOT')}/login",
 
"client_id": os.getenv("CLIENT_ID"),
 
"client_secret": os.getenv("CLIENT_SECRET"),
 
}
 
 
token_response = post(
 
f"{os.getenv('AUTH_ROOT')}/oauth/token",
 
data=data,
 
headers=headers
 
)
 
 
access_token = token_response.json()["access_token"]
 
user_info = get(
 
f"{os.getenv('AUTH_ROOT')}/api/user/show/me",
 
headers={"Authorization": f"Bearer {access_token}"}
 
)
 
 
return crud.create_user(user_info.json(), db)
 
Loading