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   și   î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. session_id username  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   și urmați instrucțiunile specifice sistemului dvs. de operare. https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/  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   care este ceea ce vom folosi pentru acest tutorial. demo_users.json,   { "user124": { "username": "user124", "email": "user124@gmail.com", "password": "1234", "bio": "This is my brief bio" }, "user123": { "username": "user123", "email": "user123@gmail.com", "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   și de a stoca acea sesiune în Redis, împreună cu toate celelalte sesiuni ale acestora. session_id  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   și   . Apoi putem interoga Redis folosind   . session_id username username  Dacă găsim o potrivire, eliminăm sesiunea asociată cu   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. session_id   @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   sau   . Și dacă sunteți fără probleme ca mine și utilizați   , 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 🙂. Kinde Auth0 EdgeDB

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Read My Stories

Acest audio este produs în limba originală a poveștii!

Prea lung; A citi

Make resilience your competitive advantage

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

About Author

COMENTARII

Etichete suspendate

ACEST ARTICOL A FOST PREZENTAT IN

Related Stories

The World is Finally Ready for a New Internet

The Spacecoin Writing Contest by Spacecoin and HackerNoon: Round 1 Results 🎉

Elon Just Made Gambling Define the News - And It's More Honest Than Journalism Ever Was

Round the year with the stars by Garrett Putman Serviss - Table of Links

The World is Finally Ready for a New Internet

The Spacecoin Writing Contest by Spacecoin and HackerNoon: Round 1 Results 🎉

Elon Just Made Gambling Define the News - And It's More Honest Than Journalism Ever Was

Round the year with the stars by Garrett Putman Serviss - Table of Links

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps