FastAPI, Redis және JWT көмегімен көп құрылғының аутентификация жүйесін қалай енгізу керек

Көптеген ірі технологиялық компаниялар пайдаланушыларға бірнеше құрылғыда кіруге мүмкіндік беретін керемет мүмкіндікті ұсынады. Пайдаланушылар өз құрылғыларын басқара алады, қайсысы кіргенін көре алады және тіпті жүйеге кірген кез келген құрылғыны пайдаланып кез келген құрылғыдан шыға алады. Бүгін мен Redis және JWT көмегімен ұқсас аутентификация жүйесін қалай енгізуге болатынын зерттегім келеді.

Сонымен, бұл қалай жұмыс істейді?

Бұл үшін мен пайдалануды шештім;

• Back-end үшін 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})

Шығу/шығу

Бұл оңай бөлігі. Пайдаланушы қолданбамызға сұрау салған сайын, 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"})

Иә, бұл соншалықты қиын емес еді, солай ма? Менің бұл жоба бірнеше аптадан бері ойымда бар, мен оны сынап көргім келді. Бұл жүйе толығымен мінсіз болмаса да (яғни, ешбір жүйе кемшіліктерсіз болмайды) біз оны жақсартуға болатыны анық. Мысалы, Curl немесе консоль қолданбасы немесе тіпті пошташы сияқты жерден сұрауларды қалай басқарамыз? Осы көздерден бірнеше сұраулар көптеген сеанстарға әкелуі мүмкін, сондықтан біздің ДБ-ны қажетсіз деректермен толтырады. Иә, біз сұраудың қайдан келгенін тексеріп, оны шешу үшін логиканы жасай аламыз, бірақ шынын айтқанда, бұл көп жұмыс болар еді. Сондықтан мен өндірістік қолданбалар үшін авторизация және аутентификация жүйелерін құруды ұсынбаймын, тек сіз нағыз «агба» (аға инженер) болсаңыз. Мен OAuth 2.0 (Google немесе Apple) немесе Kinde немесе Auth0 сияқты сыртқы провайдерді пайдаланғым келеді. Егер сіз мен сияқты бұзылған болсаңыз және EdgeDB қолдансаңыз, ол қораптан тыс пайдалануға дайын аутентификация жүйесімен бірге келеді. Осылайша, бірдеңе болып қалса, сізде тек интерн емес, басқа біреу кінәлі болады 🙂.