Ki jan yo aplike sistèm otantifikasyon milti-aparèy ak FastAPI, Redis, ak JWT

Anpil gwo konpayi teknoloji ofri yon karakteristik fre ki pèmèt itilizatè yo konekte sou plizyè aparèy. Itilizatè yo ka jere aparèy yo, wè kiyès ki konekte yo, e menm soti nan nenpòt aparèy lè l sèvi avèk nenpòt nan aparèy ki konekte yo. Jodi a, mwen vle eksplore ki jan mwen ka aplike yon sistèm otantifikasyon menm jan an lè l sèvi avèk Redis ak JWT.

Se konsa, ki jan sa a travay egzakteman?

Pou sa, mwen te deside sèvi ak;

• Fast-API pou back-end la

• Redis pou kachèt

  
  

Nou pral sèvi ak JWT (JSON Web Tokens) pou otorize demann yo. Depi aplikasyon nou an apatrid (sa vle di li pa gen okenn memwa sou demann anvan yo), nou bezwen yon fason pou voye done sesyon ak itilizatè. JWT se ideyal pou jere otantifikasyon nan aplikasyon apatrid, epi li fè yon travay ekselan.

Sepandan, yon dezavantaj nan JWT se ke plis chaj ou enkli nan yon siy, se pi long li vin. Pou sistèm nou an, mwen te deside mete sèlman session_id la ak username a nan siy la. Enfòmasyon sa a ase pou otorize demann san yo pa fè siy la twò gwo. e pral sèvi ak JWT (JSON Web Tokens) pou otorize demann yo. Depi aplikasyon nou an apatrid (sa vle di li pa gen okenn memwa sou demann anvan yo), nou bezwen yon fason pou voye done sesyon ak itilizatè. JWT se ideyal pou jere otantifikasyon nan aplikasyon apatrid, epi li fè yon travay ekselan nan sans sa a.

Ou te di session_id? men nou ap itilize JWT ak aplikasyon nou an nan apatrid

Nan kontèks sa a, "sesyon" refere a aparèy oswa mwayen kote yon itilizatè kominike avèk aplikasyon nou an. Esansyèlman, li se aparèy itilizatè a konekte nan. Chak fwa yon itilizatè fè yon demann konekte, nou kreye yon nouvo sesyon (aparèy) nan sistèm nou an ki gen tout enfòmasyon ki enpòtan sou aparèy la. Done sa yo pral estoke nan Redis pou demann nan lavni.

Oke, ann jwenn kodaj 😁

Premye bagay ou dwe fè se asire w ke ou gen Redis enstale sou machin lokal ou a. Pou enstale Redis, ale nan https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/ epi swiv enstriksyon yo espesifik nan sistèm opere ou a.

Apre sa, nou enstale python. Pou sa, mwen pral sèvi ak python 3.11 la (mwen pa te wè bezwen an ajou nan 3.12 ankò, yo dwe fran sèl rezon ki fè mwen menm sèvi ak 3.11 se paske nan StrEnum, lòt pase ke mwen toujou renmen 3.10)

Apre sa, nou bezwen enstale pwezi, sa a se manadjè pake ke mwen itilize

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

Sa rezoud, kidonk ale pi devan epi klonaj repo a

 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

Mete kanpe koneksyon Redis nou an

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

Konfigirasyon baz done

Mwen te kreye yon baz done Demo nan demo_users.json ki se sa nou pral itilize pou leson patikilye sa a.

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

Koulye a, nou bezwen ajoute chema nou yo ak fonksyon asistan pou baz done nou an. Pou konsizyon, mwen pa pral mete tout kòd la isit la.

 @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

Konfigirasyon sèvè

Nou ap itilize FastAPI pou kouri aplikasyon nou an, kidonk kite nou ale epi mete sa jiska

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

Oke sa a bon aplikasyon nou an sanble ap vini ansanm joliman

Konekte / konekte

Chak fwa yon itilizatè konekte nan sistèm nan, nou bezwen yon fason pou jenere yon session_id epi estoke sesyon sa a nan Redis, ansanm ak tout lòt sesyon yo.

Lè yon itilizatè konekte, nou pral premye otantifye demann lan pou asire li valab. Yon fwa valide, nou ka rekipere tout enfòmasyon sou aparèy nan demann lan. Apre sa, nou pral estoke enfòmasyon sa yo nan Redis, jenere yon nouvo siy, epi retounen siy sa a bay itilizatè a.

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

Dekonekte / dekonekte

Sa a se pati ki pi fasil. Chak fwa yon itilizatè fè yon demann nan aplikasyon nou an, nou dekode siy Bearer la pou rekipere session_id ak username a. Lè sa a, nou ka mande Redis lè l sèvi avèk username a.

Si nou jwenn yon match, nou retire sesyon ki asosye ak session_id nan siy dekode a. Pou egzanp, si sesyon an pa egziste, nou tou senpleman retounen yon mesaj bay itilizatè a. Sa a endike ke itilizatè a te deja dekonekte aparèy sa a soti nan yon lòt aparèy, oswa ke siy la pa valab.

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

Se konsa, wi, sa pa t tèlman difisil, pa vre? Mwen te gen pwojè sa a nan tèt mwen pou yon koup de semèn kounye a, epi mwen te vle teste li soti. Malgre ke sistèm sa a pa konplètman pafè (mwen vle di, pa gen okenn sistèm ki san defo li yo) nou ka evidamman fè yon sèl sa a pi byen. Pou egzanp, ki jan nou jere demann ki soti nan yon kote tankou Curl oswa yon app konsole oswa menm postman? Plizyè demann ki soti nan sous sa yo ka mennen nan yon anpil nan sesyon, kidonk peple db nou an ak done ki pa nesesè. Wi, nou ta ka tcheke pou wè ki kote demann lan soti epi kreye lojik nou pou okipe sa, men yo dwe onèt, sa ta pral anpil travay. Se poutèt sa mwen pa rekòmande bati otorizasyon ak sistèm otantifikasyon pou aplikasyon pwodiksyon, eksepte ou se yon reyèl "agba" (enjenyè segondè). Mwen ta pito itilize OAuth 2.0 (Google oswa Apple) oswa yon founisè ekstèn tankou Kinde oswa Auth0 . Men, si w ap kraze tankou m epi w ap itilize EdgeDB , li vini ak yon sistèm otorizasyon pare pou itilize soti nan bwat la. Nan fason sa a, si yon bagay rive, ou gen yon lòt moun blame epi li pa sèlman estajyè a 🙂.