כיצד ליישם מערכת אימות מרובת מכשירים עם 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 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 שלנו בנתונים מיותרים. כן, נוכל לבדוק מאיפה מגיעה הבקשה וליצור את ההיגיון שלנו לטפל בזה, אבל למען האמת, זו תהיה עבודה רבה. לכן אני לא ממליץ לבנות מערכות הרשאות ואימות עבור אפליקציות ייצור, אלא שאתה "אגבה" אמיתי (מהנדס בכיר). אני מעדיף להשתמש ב-OAuth 2.0 (גוגל או אפל) או בספק חיצוני כמו Kinde או Auth0 . ואם אתה שבור כמוני ומשתמש ב- EdgeDB , הוא מגיע עם מערכת אישור מוכנה לשימוש מהקופסה. בדרך זו, אם משהו קורה, יש לך מישהו אחר להאשים ולא רק את המתמחה 🙂.