Ինչպես իրականացնել բազմաբնույթ սարքի նույնականացման համակարգը FastAPI-ի, Redis-ի և JWT-ի միջոցով

Շատ խոշոր տեխնոլոգիական ընկերություններ առաջարկում են հիանալի գործառույթ, որը թույլ է տալիս օգտվողներին մուտք գործել բազմաթիվ սարքեր: Օգտատերերը կարող են կառավարել իրենց սարքերը, տեսնել, թե որոնք են մուտք գործել և նույնիսկ դուրս գալ ցանկացած սարքից՝ օգտագործելով իրենց մուտք գործած սարքերից որևէ մեկը: Այսօր ես ուզում եմ ուսումնասիրել, թե ինչպես կարող եմ կիրառել նույնականացման նմանատիպ համակարգ՝ օգտագործելով Redis և JWT:

Այսպիսով, ինչպե՞ս է սա ճիշտ աշխատում:

Դրա համար ես որոշել եմ օգտագործել;

• Fast-API հետին պլանի համար

• Redis՝ քեշավորման համար

  
  

Մենք կօգտագործենք JWT (JSON Web Tokens) հարցումները լիազորելու համար: Քանի որ մեր դիմումը քաղաքացիություն չունեցող է (նշանակում է, որ այն չունի հիշողություն նախորդ հարցումների մասին), մեզ անհրաժեշտ է սեսիայի և օգտատիրոջ տվյալները ուղարկելու միջոց: JWT-ն իդեալական է քաղաքացիություն չունեցող հավելվածներում նույնականացումը կառավարելու համար, և այն հիանալի աշխատանք է կատարում:

Այնուամենայնիվ, JWT-ի բացասական կողմն այն է, որ որքան ավելի շատ ծանրաբեռնվածություն ներառեք նշանի մեջ, այնքան երկար է այն դառնում: Մեր համակարգի համար ես որոշել եմ նշանում ներառել միայն session_id ը և username : Այս տեղեկատվությունը բավարար է հարցումները թույլատրելու համար՝ առանց նշանը չափազանց մեծացնելու: e-ն կօգտագործի JWT (JSON Web Tokens)՝ հարցումները լիազորելու համար: Քանի որ մեր դիմումը քաղաքացիություն չունեցող է (նշանակում է, որ այն չունի հիշողություն նախորդ հարցումների մասին), մեզ անհրաժեշտ է սեսիայի և օգտատիրոջ տվյալները ուղարկելու միջոց: JWT-ն իդեալական է քաղաքացիություն չունեցող հավելվածներում նույնականացումը կառավարելու համար, և այն հիանալի աշխատանք է կատարում այս առումով:

Դուք ասացիք session_id? բայց մենք օգտագործում ենք JWT-ն և մեր հավելվածը քաղաքացիություն չունեցող պայմաններում

Այս համատեքստում «սեսիան» վերաբերում է սարքին կամ միջոցին, որի միջոցով օգտատերը փոխազդում է մեր հավելվածի հետ: Ըստ էության, դա այն սարքն է, որի մեջ օգտվողը մուտք է գործել: Ամեն անգամ, երբ օգտատերը մուտքի հարցում է անում, մենք մեր համակարգում ստեղծում ենք նոր նիստ (սարք), որը պարունակում է սարքի բոլոր համապատասխան տեղեկությունները: Այս տվյալները կպահվեն Redis-ում՝ ապագա հարցումների համար:

Լավ, եկեք կոդավորենք 😁

Առաջին բանը, որ պետք է անել, համոզվելն է, որ ձեր տեղական մեքենայի վրա տեղադրված է Redis-ը: Redis-ը տեղադրելու համար այցելեք https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/ և հետևեք ձեր օպերացիոն համակարգին հատուկ հրահանգներին:

Հաջորդը, մենք տեղադրում ենք python: Դրա համար ես կօգտագործեմ python 3.11-ը (դեռ չեմ տեսել 3.12-ի թարմացման անհրաժեշտություն, անկեղծ ասած, միակ պատճառը, որ ես նույնիսկ օգտագործում եմ 3.11-ը, StrEnum-ի պատճառով է, բացի այն, որ ես դեռ սիրում եմ 3.10-ը):

Հաջորդը, մենք պետք է տեղադրենք պոեզիա, սա փաթեթի կառավարիչն է, որը ես օգտագործում եմ

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

Դա լուծված է, այնպես որ առաջ գնացեք և կլոնավորեք ռեպո

 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

Մեր 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))

Տվյալների բազայի կարգավորում

Ես ստեղծել եմ ցուցադրական տվյալների բազա demo_users.json- ում, որը մենք կօգտագործենք այս ձեռնարկի համար:

 { "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" } }

Այժմ մենք պետք է ավելացնենք մեր սխեմաները և օգնական գործառույթները մեր տվյալների բազայի համար: Հակիրճ լինելու համար ես այստեղ չեմ դնի ամբողջ ծածկագիրը։

 @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

Սերվերի կարգավորում

Մենք օգտագործում ենք FastAPI-ն մեր հավելվածը գործարկելու համար, այնպես որ եկեք գնանք և կարգավորենք դա

 # 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)

Լավ, լավ է, որ մեր դիմումը կարծես լավ է հավաքվում

Մուտք / մուտք

Ամեն անգամ, երբ օգտատերը մուտք է գործում համակարգ, մեզ անհրաժեշտ է միջոց՝ ստեղծելու session_id և այդ նիստը պահել Redis-ում՝ իր բոլոր մյուս նիստերի հետ միասին:

Երբ օգտատերը մուտք է գործում, մենք նախ կհաստատենք հարցումը՝ համոզվելու համար, որ այն վավեր է: Հաստատվելուց հետո մենք կարող ենք ստանալ սարքի բոլոր տեղեկությունները հարցումից: Դրանից հետո մենք կպահենք այս տեղեկատվությունը Redis-ում, կստեղծենք նոր նշան և կվերադարձնենք այդ նշանն օգտագործողին:

 @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})

Դուրս գալ/դուրս գալ

Սա ավելի հեշտ մասն է: Ամեն անգամ, երբ օգտատերը հարցում է անում մեր հավելվածին, մենք վերծանում ենք «Bearer» նշանը, որպեսզի առբերենք session_id ը և username : Այնուհետև մենք կարող ենք հարցումներ կատարել Redis-ին՝ օգտագործելով username :

Եթե համընկնում ենք, ապա վերծանված նշանից հեռացնում ենք 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"})

Այսպիսով, այո, դա այնքան էլ դժվար չէր, չէ՞: Արդեն մի քանի շաբաթ է, ինչ այս նախագիծը իմ գլխում է, և ես ուզում էի փորձարկել այն: Չնայած այս համակարգը լիովին կատարյալ չէ (նկատի ունեմ, որ ոչ մի համակարգ առանց իր թերությունների չէ), մենք ակնհայտորեն կարող ենք այս մեկը ավելի լավը դարձնել: Օրինակ, ինչպե՞ս ենք մենք կառավարում հարցումները այնպիսի վայրից, ինչպիսին է Curl-ը կամ վահանակի հավելվածը կամ նույնիսկ փոստատարը: Այս աղբյուրներից բազմաթիվ հարցումները կարող են հանգեցնել բազմաթիվ նիստերի, հետևաբար մեր db-ն ավելորդ տվյալներով համալրելով: Այո, մենք կարող էինք ստուգել՝ տեսնելու համար, թե որտեղից է գալիս խնդրանքը և ստեղծել մեր տրամաբանությունը՝ դա կարգավորելու համար, բայց եթե անկեղծ լինենք, դա շատ աշխատանք կլինի: Այդ իսկ պատճառով ես խորհուրդ չեմ տալիս արտադրական հավելվածների համար թույլտվության և վավերացման համակարգեր կառուցել, բացառությամբ, որ դուք իսկական «agba» եք (ավագ ինժեներ): Ես նախընտրում եմ օգտագործել OAuth 2.0 (Google կամ Apple) կամ արտաքին մատակարար, ինչպիսիք են Kinde-ը կամ Auth0-ը : Եվ եթե դուք ինձ նման կոտրված եք և օգտագործում եք EdgeDB , այն գալիս է վավերացման համակարգով, որը պատրաստ է օգտագործել առանց տուփի: Այս կերպ, եթե ինչ-որ բան պատահի, դու ուրիշին ունես մեղավոր և ոչ միայն պրակտիկանտին 🙂: