Sådan implementeres Multi-Device Authentication System med FastAPI, Redis og JWT

Mange store teknologivirksomheder tilbyder en cool funktion, der giver brugerne mulighed for at logge ind på flere enheder. Brugere kan administrere deres enheder, se, hvilke der er logget ind og endda logge ud fra enhver enhed ved hjælp af en hvilken som helst af deres enheder, der er logget ind. I dag vil jeg undersøge, hvordan jeg kan implementere et lignende autentificeringssystem ved hjælp af Redis og JWT.

Så hvordan fungerer dette præcist?

Til dette har jeg besluttet at bruge;

• Fast-API til back-end
• Redis til caching

  
  

Vi vil bruge JWT (JSON Web Tokens) til at godkende anmodninger. Da vores applikation er statsløs (hvilket betyder, at den ikke har nogen hukommelse af tidligere anmodninger), har vi brug for en måde at sende sessions- og brugerdata på. JWT er ideel til at administrere godkendelse i statsløse applikationer, og det gør et fremragende stykke arbejde.

En ulempe ved JWT er dog, at jo mere nyttelast du inkluderer i et token, jo længere bliver det. For vores system har jeg besluttet kun at inkludere session_id og username i tokenet. Disse oplysninger er tilstrækkelige til at godkende anmodninger uden at gøre tokenet for stort.e vil bruge JWT (JSON Web Tokens) til at godkende anmodninger. Da vores applikation er statsløs (hvilket betyder, at den ikke har nogen hukommelse af tidligere anmodninger), har vi brug for en måde at sende sessions- og brugerdata på. JWT er ideel til at administrere godkendelse i statsløse applikationer, og det gør et fremragende stykke arbejde i denne henseende.

Du sagde session_id? men vi bruger JWT og vores app i stateless

I denne sammenhæng refererer "session" til den enhed eller midler, hvorigennem en bruger interagerer med vores app. I bund og grund er det den enhed, brugeren er logget på. Når en bruger fremsætter en login-anmodning, opretter vi en ny session (enhed) i vores system, der indeholder alle relevante enhedsoplysninger. Disse data vil blive gemt i Redis til fremtidige anmodninger.

Okay, lad os komme i gang med kodning 😁

Den første ting at gøre er at sikre, at du har Redis installeret på din lokale maskine. For at installere Redis skal du gå over til og følge instruktionerne, der er specifikke for dit operativsystem.

Dernæst installerer vi python. Til dette vil jeg bruge python 3.11 (jeg har ikke set behovet for at opgradere til 3.12 endnu, for at være ærlig er den eneste grund til at jeg overhovedet bruger 3.11 på grund af StrEnum, bortset fra at jeg stadig elsker 3.10)

Dernæst skal vi installere poesi, dette er pakkehåndteringen, som jeg bruger

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

Det er afgjort, så gå videre og klon repoen

 git clone //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

Opsætning af vores Redis-forbindelse

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

Database opsætning

Jeg har oprettet en demodatabase i demo_users.json , som er det, vi skal bruge til denne tutorial.

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

Nu skal vi tilføje vores skemaer og hjælpefunktioner til vores database. For kortheds skyld vil jeg ikke lægge al koden her.

 @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

Server opsætning

Vi bruger FastAPI til at køre vores app, så lad os gå og indstille det til

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

Okay, det er godt, vores applikation ser ud til at komme godt sammen

Log ind / log ind

Hver gang en bruger logger ind på systemet, har vi brug for en måde at generere et session_id og gemme den session i Redis sammen med alle deres andre sessioner.

Når en bruger logger på, vil vi først godkende anmodningen for at sikre, at den er gyldig. Når den er valideret, kan vi hente alle enhedsoplysninger fra anmodningen. Derefter gemmer vi disse oplysninger i Redis, genererer et nyt token og returnerer det token til brugeren.

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

Log ud/log ud

Dette er den nemmere del. Hver gang en bruger fremsætter en anmodning til vores app, afkoder vi bærer-tokenet for at hente session_id og username . Vi kan derefter forespørge Redis ved hjælp af username .

Hvis vi finder et match, fjerner vi sessionen forbundet med session_id fra det afkodede token. For eksempel, hvis sessionen ikke eksisterer, returnerer vi blot en besked til brugeren. Dette indikerer, at brugeren allerede har logget ud af den pågældende enhed fra en anden enhed, eller at tokenet er ugyldigt.

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

Så ja, det var ikke så svært, vel? Jeg har haft dette projekt i mit hoved i et par uger nu, og jeg ville gerne teste det af. Selvom dette system ikke er helt perfekt (jeg mener, intet system er uden sine fejl), kan vi naturligvis gøre dette bedre. Hvordan administrerer vi for eksempel anmodninger fra et sted som Curl eller en konsol-app eller endda postbud? Flere anmodninger fra disse kilder kan føre til mange sessioner, og derfor fylde vores database med unødvendige data. Ja, vi kunne tjekke for at se, hvor anmodningen kommer fra og skabe vores logik til at håndtere det, men for at være ærlig, ville det være meget arbejde. Derfor anbefaler jeg ikke at bygge autorisations- og autentificeringssystemer til produktionsapps, bortset fra at du er en rigtig "agba" (senioringeniør). Jeg vil hellere bruge OAuth 2.0 (Google eller Apple) eller en ekstern udbyder som eller . Og hvis du er brok som mig og bruger , kommer det med et godkendelsessystem klar til brug ud af æsken. På denne måde, hvis der sker noget, har du en anden at skyde skylden på og ikke kun praktikanten 🙂.