Cum să implementați sistemul de autentificare multi-dispozitiv cu FastAPI, Redis și JWT

Multe companii mari de tehnologie oferă o caracteristică grozavă care permite utilizatorilor să se conecteze pe mai multe dispozitive. Utilizatorii își pot gestiona dispozitivele, pot vedea care sunt conectate și chiar se pot deconecta de pe orice dispozitiv folosind oricare dintre dispozitivele lor conectate. Astăzi, vreau să explorez modul în care pot implementa un sistem de autentificare similar folosind Redis și JWT.

Deci, cum funcționează asta exact?

Pentru aceasta, am decis să folosesc;

• Fast-API pentru back-end

• Redis pentru stocarea în cache

  
  

Vom folosi JWT (JSON Web Tokens) pentru a autoriza solicitările. Deoarece aplicația noastră este apatridă (înseamnă că nu are memorie pentru solicitările anterioare), avem nevoie de o modalitate de a trimite date despre sesiune și utilizator. JWT este ideal pentru gestionarea autentificării în aplicațiile fără stat și face o treabă excelentă.

Cu toate acestea, un dezavantaj al JWT este că, cu cât includeți mai multă sarcină utilă într-un token, cu atât acesta devine mai lung. Pentru sistemul nostru, am decis să includ doar session_id și username în token. Aceste informații sunt suficiente pentru autorizarea cererilor fără a face tokenul excesiv de mare. E va folosi JWT (JSON Web Tokens) pentru a autoriza cereri. Deoarece aplicația noastră este apatridă (înseamnă că nu are memorie pentru solicitările anterioare), avem nevoie de o modalitate de a trimite date despre sesiune și utilizator. JWT este ideal pentru gestionarea autentificării în aplicațiile fără stat și face o treabă excelentă în acest sens.

Ai spus session_id? dar folosim JWT și aplicația noastră în apatrid

În acest context, „sesiune” se referă la dispozitivul sau mijlocul prin care un utilizator interacționează cu aplicația noastră. În esență, este dispozitivul la care este conectat utilizatorul. Ori de câte ori un utilizator face o cerere de conectare, creăm o nouă sesiune (dispozitiv) în sistemul nostru care conține toate informațiile relevante despre dispozitiv. Aceste date vor fi stocate în Redis pentru solicitări viitoare.

Bine, haideți să codificăm 😁

Primul lucru de făcut este să vă asigurați că aveți Redis instalat pe mașina dvs. locală. Pentru a instala Redis, accesați https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/ și urmați instrucțiunile specifice sistemului dvs. de operare.

Apoi, instalăm python. Pentru aceasta, voi folosi python 3.11 (nu am văzut încă nevoia să fac upgrade la 3.12, să fiu sincer, singurul motiv pentru care chiar folosesc 3.11 este din cauza StrEnum, în afară de faptul că încă iubesc 3.10)

Apoi, trebuie să instalăm poetry, acesta este managerul de pachete pe care îl folosesc

 pip install poetry # or python3.11 -m install poetry

S-a rezolvat, așa că mergeți mai departe și clonați repo-ul

 git clone https://github.com/emperorsixpacks/multi_device_sign_in_with_redis.git && cd server poetry shell && poetry install # create a new virtual environment and install all dependanceies

Configurarea conexiunii noastre Redis

 import os from redis import Redis load_dotenv(".env") REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = os.getenv("REDIS_PORT", "6379") redis_client = Redis(REDIS_HOST, int(REDIS_PORT))

Configurarea bazei de date

Am creat o bază de date demonstrativă în demo_users.json, care este ceea ce vom folosi pentru acest tutorial.

 { "user124": { "username": "user124", "email": "[email protected]", "password": "1234", "bio": "This is my brief bio" }, "user123": { "username": "user123", "email": "[email protected]", "password": "1234", "bio": "This is my brief bio" } }

Acum, trebuie să adăugăm schemele și funcțiile de ajutor pentru baza noastră de date. Pentru concizie, nu voi pune tot codul aici.

 @dataclass class Session: """ A class to represent a user's session. Attributes: session_id (str): A unique id for the session. device_name (str): The name of the device used for the session. ip_address (str): The ip address of the device used for the session. device_id (str): A unique id for the device. date_created (datetime): The date and time the session was created. """ session_id: str = field(default_factory=create_new_session_id) device_name: str = field(default=None) ip_address: str = field(default=None) device_id: str = field(default_factory=generate_new_device_id) date_created: datetime = field(default_factory=now_date_time_to_str) @dataclass class User: """ A class to represent a user. Attributes: username (str): The username of the user. email (str): The email of the user. password (str): The password of the user. bio (str): The bio of the user. sessions (List[Session] | None): A list of Session objects representing the user's sessions. """ username: str = field(default=None) email: str = field(default=None) password: str = field(default=None) bio: str = field(default=None) sessions: List[Session] | None = None @property def __dict__(self): """ Returns a dictionary representing the user. Returns: Dict[str, Any]: A dictionary representing the user """ return { "username": self.username, "email": self.email, "password": self.password, "bio": self.bio, "sessions": self.return_session_dict(), } def return_session_dict(self): """ Returns a list of dictionaries representing the user's sessions. If the sessions field is a list of Session objects, returns a list of dictionaries where each dictionary is the __dict__ of a Session object. If the sessions field is a list of dictionaries, returns the list as is. Returns: List[Dict[str, Any]]: A list of dictionaries representing the user's sessions """ try: return [session.__dict__ for session in self.sessions] except AttributeError: return [session for session in self.sessions] # Utiliy finctions def return_user_from_db(username) -> User | None: """ Retrieves a user from the database by their username. Args: username (str): The username of the user to be retrieved Returns: User | None: The user if found, None otherwise """ with open("demo_users.json", "r", encoding="utf-8") as file: user = json.load(file).get(str(username), None) return User(**user) or None

Configurare server

Folosim FastAPI pentru a rula aplicația noastră, așa că haideți să o setăm

 # Setting up server from fastapi import FastAPI from fastapi.responses import JSONResponse app = FastAPI( name="Multi device sign in with Redis", description="Multi device sign in with Redis in stateless applications", ) @app.get("/") def index_route(): return JSONResponse(content={"Message": "hello, this seems to be working :)"}) if __name__ == "__main__": import uvicorn uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=True, use_colors=True)

În regulă, este bine că aplicația noastră pare să se îmbine bine

Autentificare/conectare

De fiecare dată când un utilizator se conectează în sistem, avem nevoie de o modalitate de a genera un session_id și de a stoca acea sesiune în Redis, împreună cu toate celelalte sesiuni ale acestora.

Când un utilizator se conectează, vom autentifica mai întâi cererea pentru a ne asigura că este validă. Odată validate, putem prelua toate informațiile despre dispozitiv din cerere. După aceea, vom stoca aceste informații în Redis, vom genera un nou token și vom returna acel token utilizatorului.

 @app.post("/login") def login_route( form: Annotated[LoginForm, Depends()], request: Request ) -> JSONResponse: """ Handles a login request. Args: form (Annotated[LoginForm, Depends()]): The form data containing the username and password request (Request): The request containing the User-Agent header and client host Returns: JSONResponse: A JSON response containing a JWT token if the login is successful, otherwise a JSONResponse with a 404 status code and a message indicating that the username or password is invalid """ username = form.username password = form.password # Authenticate the user user = authenticate_user(username, password) if user is None: return JSONResponse( status_code=404, content={"message": "Invalid username or password"} ) # Create a new session session = Session( device_name=request.headers.get("User-Agent"), ip_address=request.client.host ) # Get the user from the cache user_from_cache = get_user_from_cache(username) if user_from_cache is None: return JSONResponse(content={"message": "one minute"}, status_code=404) # Get the user's sessions user_sessions = get_sessions(userid=username) # Add the new session to the user's sessions try: user_sessions.append(session) except AttributeError: user_sessions = [session] # Update the user in the cache user_from_cache.sessions = user_sessions update_user_cache(userid=username, new_data=user_from_cache) # Create a JWT token token = create_token(Token(user=username, session_id=session.session_id)) # Return the JWT token return JSONResponse(content={"message": "logged in", "token": token})

Deconectare/ deconectare

Aceasta este partea mai ușoară. De fiecare dată când un utilizator face o solicitare către aplicația noastră, decodificăm simbolul Bearer pentru a prelua session_id și username . Apoi putem interoga Redis folosind username .

Dacă găsim o potrivire, eliminăm sesiunea asociată cu session_id din jetonul decodat. De exemplu, dacă sesiunea nu există, pur și simplu returnăm un mesaj utilizatorului. Aceasta indică faptul că utilizatorul s-a deconectat deja de pe acel dispozitiv de pe un alt dispozitiv sau că simbolul este invalid.

 @app.post("/logout") def logout_route(request: Request): """ Handles a request to log out the user. This endpoint will delete the user's session from the cache and return a JSON response with a message indicating that the user has been logged out. Args: request (Request): The request containing the Authorization header with the JWT token Returns: JSONResponse: A JSON response containing the message "logged out" if the token is valid, otherwise a JSONResponse with a 404 status code and a message indicating that the token is invalid """ # Get the JWT token from the Authorization header _, token = get_authorization_scheme_param(request.headers.get("Authorization")) # Decode the JWT token payload = decode_token(token) # Check if the token is invalid if payload is None: return JSONResponse(content={"message": "Invalid token"}, status_code=404) # Check if the user or session does not exist if get_single_session(userid=payload.user, session_id=payload.session_id) is None or get_user_from_cache( userid=payload.user) is None: return JSONResponse(content={"message": "Invalid token"}, status_code=404) # Delete the session from the cache delete_session(payload.user, payload.session_id) # Return a JSON response with a message indicating that the user has been logged out return JSONResponse(content={"message": "logged out"})

Deci da, nu a fost atât de greu, nu-i așa? Am acest proiect în cap de câteva săptămâni și am vrut să-l testez. Deși acest sistem nu este complet perfect (adică, niciun sistem nu este lipsit de defecte), evident îl putem îmbunătăți pe acesta. De exemplu, cum gestionăm solicitările dintr-un loc precum Curl sau o aplicație de consolă sau chiar poștaș? Solicitările multiple din aceste surse ar putea duce la o mulțime de sesiuni, prin urmare populând baza noastră de date cu date inutile. Da, am putea verifica să vedem de unde vine cererea și să creăm logica noastră pentru a gestiona asta, dar, să fiu sincer, ar fi multă muncă. De aceea nu recomand construirea de sisteme de autorizare și autentificare pentru aplicațiile de producție, cu excepția faptului că ești un adevărat „agba” (inginer superior). Prefer să folosesc OAuth 2.0 (Google sau Apple) sau un furnizor extern precum Kinde sau Auth0 . Și dacă sunteți fără probleme ca mine și utilizați EdgeDB , acesta vine cu un sistem de autentificare gata de utilizat din cutie. Astfel, dacă se întâmplă ceva, ai pe altcineva de vină și nu doar pe stagiar 🙂.