FastAPI, Redis और JWT के साथ मल्टी-डिवाइस प्रमाणीकरण प्रणाली को कैसे लागू करें

कई बड़ी टेक कंपनियाँ एक शानदार सुविधा प्रदान करती हैं जो उपयोगकर्ताओं को कई डिवाइस पर साइन इन करने की अनुमति देती है। उपयोगकर्ता अपने डिवाइस प्रबंधित कर सकते हैं, देख सकते हैं कि कौन से डिवाइस साइन इन हैं, और यहां तक कि अपने किसी भी साइन-इन डिवाइस का उपयोग करके किसी भी डिवाइस से साइन आउट भी कर सकते हैं। आज, मैं यह पता लगाना चाहता हूँ कि मैं Redis और JWT का उपयोग करके एक समान प्रमाणीकरण प्रणाली कैसे लागू कर सकता हूँ।

तो फिर यह वास्तव में कैसे काम करता है?

इसके लिए, मैंने उपयोग करने का निर्णय लिया है;

• बैक-एंड के लिए फास्ट-एपीआई

• कैशिंग के लिए Redis

  
  

हम अनुरोधों को अधिकृत करने के लिए JWT (JSON वेब टोकन) का उपयोग करेंगे। चूँकि हमारा एप्लिकेशन स्टेटलेस है (जिसका अर्थ है कि इसमें पिछले अनुरोधों की कोई मेमोरी नहीं है), हमें सत्र और उपयोगकर्ता डेटा भेजने का एक तरीका चाहिए। JWT स्टेटलेस एप्लिकेशन में प्रमाणीकरण के प्रबंधन के लिए आदर्श है, और यह एक बेहतरीन काम करता है।

हालाँकि, JWT का एक नुकसान यह है कि आप टोकन में जितना अधिक पेलोड शामिल करते हैं, वह उतना ही लंबा होता जाता है। हमारे सिस्टम के लिए, मैंने टोकन में केवल session_id और username शामिल करने का निर्णय लिया है। यह जानकारी टोकन को अत्यधिक बड़ा किए बिना अनुरोधों को अधिकृत करने के लिए पर्याप्त है। हम अनुरोधों को अधिकृत करने के लिए JWT (JSON वेब टोकन) का उपयोग करेंगे। चूँकि हमारा एप्लिकेशन स्टेटलेस है (जिसका अर्थ है कि इसमें पिछले अनुरोधों की कोई मेमोरी नहीं है), हमें सत्र और उपयोगकर्ता डेटा भेजने का एक तरीका चाहिए। 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})

लॉग आउट/ साइन आउट करना

यह सबसे आसान हिस्सा है। जब भी कोई उपयोगकर्ता हमारे ऐप से अनुरोध करता है, तो हम session_id और username प्राप्त करने के लिए Bearer टोकन को डिकोड करते हैं। फिर हम username उपयोग करके Redis को क्वेरी कर सकते हैं।

अगर हमें कोई मिलान मिलता है, तो हम डिकोड किए गए टोकन से 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"})

तो हाँ, यह इतना मुश्किल नहीं था, है न? मेरे दिमाग में यह प्रोजेक्ट पिछले कुछ हफ़्तों से है, और मैं इसे परखना चाहता था। हालाँकि यह सिस्टम पूरी तरह से सही नहीं है (मेरा मतलब है, कोई भी सिस्टम अपनी खामियों के बिना नहीं होता) लेकिन हम इसे बेहतर बना सकते हैं। उदाहरण के लिए, हम कर्ल या कंसोल ऐप या यहाँ तक कि पोस्टमैन जैसी जगह से अनुरोधों को कैसे प्रबंधित करते हैं? इन स्रोतों से कई अनुरोधों के कारण बहुत सारे सत्र हो सकते हैं, इसलिए हमारे डेटाबेस में अनावश्यक डेटा भर सकता है। हाँ, हम यह देख सकते हैं कि अनुरोध कहाँ से आ रहा है और इसे संभालने के लिए अपना तर्क बना सकते हैं, लेकिन ईमानदारी से कहूँ तो यह बहुत काम होगा। यही कारण है कि मैं उत्पादन ऐप्स के लिए प्राधिकरण और प्रमाणीकरण प्रणाली बनाने की अनुशंसा नहीं करता, सिवाय इसके कि आप एक वास्तविक "अग्बा" (वरिष्ठ इंजीनियर) हों। मैं OAuth 2.0 (Google या Apple) या Kinde या Auth0 जैसे बाहरी प्रदाता का उपयोग करना पसंद करूँगा। और अगर आप मेरी तरह कंगाल हैं और EdgeDB का उपयोग कर रहे हैं, तो यह बॉक्स से बाहर उपयोग करने के लिए तैयार प्रमाणीकरण प्रणाली के साथ आता है। इस तरह, यदि कुछ घटित होता है, तो आपके पास केवल इंटर्न को नहीं, बल्कि किसी और को दोषी ठहराने का मौका होगा 🙂।