Muchas grandes empresas tecnológicas ofrecen una función interesante que permite a los usuarios iniciar sesión en varios dispositivos. Los usuarios pueden administrar sus dispositivos, ver cuáles han iniciado sesión e incluso cerrar sesión en cualquier dispositivo utilizando cualquiera de los dispositivos en los que hayan iniciado sesión. Hoy quiero explorar cómo puedo implementar un sistema de autenticación similar utilizando Redis y JWT.  Entonces, ¿cómo funciona esto exactamente?  Para esto, he decidido utilizar;  API rápida para el back-end  Redis para almacenamiento en caché  Usaremos JWT (JSON Web Tokens) para autorizar solicitudes. Dado que nuestra aplicación no tiene estado (es decir, no tiene memoria de solicitudes anteriores), necesitamos una forma de enviar datos de sesión y de usuario. JWT es ideal para administrar la autenticación en aplicaciones sin estado y hace un excelente trabajo.  Sin embargo, una desventaja de JWT es que cuanto más carga útil se incluye en un token, más largo se vuelve. Para nuestro sistema, he decidido incluir solo el   y el   en el token. Esta información es suficiente para autorizar solicitudes sin hacer que el token sea excesivamente grande. Usaremos JWT (JSON Web Tokens) para autorizar solicitudes. Dado que nuestra aplicación no tiene estado (lo que significa que no tiene memoria de solicitudes anteriores), necesitamos una forma de enviar datos de sesión y de usuario. JWT es ideal para administrar la autenticación en aplicaciones sin estado y hace un excelente trabajo en este sentido. session_id username  Dijiste session_id? pero estamos usando JWT y nuestra aplicación no tiene estado  En este contexto, "sesión" se refiere al dispositivo o medio a través del cual un usuario interactúa con nuestra aplicación. Básicamente, es el dispositivo en el que el usuario ha iniciado sesión. Cada vez que un usuario realiza una solicitud de inicio de sesión, creamos una nueva sesión (dispositivo) en nuestro sistema que contiene toda la información relevante del dispositivo. Estos datos se almacenarán en Redis para futuras solicitudes.  Muy bien, comencemos a codificar 😁  Lo primero que debes hacer es asegurarte de tener Redis instalado en tu máquina local. Para instalar Redis, dirígete a   y sigue las instrucciones específicas para tu sistema operativo. https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/  A continuación, instalamos Python. Para ello, utilizaré Python 3.11 (aún no he visto la necesidad de actualizar a 3.12; para ser sincero, la única razón por la que utilizo 3.11 es por StrEnum; aparte de eso, todavía me encanta 3.10).  A continuación, necesitamos instalar poetry, este es el administrador de paquetes que uso.   pip install poetry # or python3.11 -m install poetry  Eso está resuelto, así que sigue adelante y clona el repositorio.   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  Configurando nuestra conexión 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))  Configuración de la base de datos  He creado una base de datos de demostración en   , que es lo que usaremos para este tutorial. demo_users.json   { "user124": { "username": "user124", "email": "user124@gmail.com", "password": "1234", "bio": "This is my brief bio" }, "user123": { "username": "user123", "email": "user123@gmail.com", "password": "1234", "bio": "This is my brief bio" } }  Ahora, necesitamos agregar nuestros esquemas y funciones auxiliares para nuestra base de datos. Para abreviar, no incluiré todo el código aquí.   @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  Configuración del servidor  Estamos usando FastAPI para ejecutar nuestra aplicación, así que configurémosla.   # 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)  Está bien, esto es bueno, nuestra aplicación parece estar tomando forma muy bien.  Iniciar sesión / iniciar sesión  Cada vez que un usuario inicia sesión en el sistema, necesitamos una forma de generar un   y almacenar esa sesión en Redis, junto con todas sus otras sesiones. session_id  Cuando un usuario inicia sesión, primero autenticamos la solicitud para asegurarnos de que sea válida. Una vez validada, podemos recuperar toda la información del dispositivo a partir de la solicitud. Después de eso, almacenaremos esta información en Redis, generaremos un nuevo token y se lo devolveremos al usuario.   @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})  Cerrar sesión/salir  Esta es la parte más fácil. Cada vez que un usuario realiza una solicitud a nuestra aplicación, decodificamos el token de Bearer para recuperar el   y   . Luego podemos consultar Redis usando el   . session_id username username  Si encontramos una coincidencia, eliminamos la sesión asociada con el   del token decodificado. Por ejemplo, si la sesión no existe, simplemente devolvemos un mensaje al usuario. Esto indica que el usuario ya ha cerrado sesión en ese dispositivo desde un dispositivo diferente o que el token no es válido. 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"})  Entonces sí, eso no fue tan difícil, ¿verdad? He tenido este proyecto en mi cabeza durante un par de semanas y quería probarlo. Aunque este sistema no es completamente perfecto (quiero decir, ningún sistema está libre de fallas), obviamente podemos mejorarlo. Por ejemplo, ¿cómo gestionamos las solicitudes de un lugar como Curl o una aplicación de consola o incluso Postman? Múltiples solicitudes de estas fuentes podrían generar muchas sesiones, por lo tanto, llenando nuestra base de datos con datos innecesarios. Sí, podríamos verificar de dónde proviene la solicitud y crear nuestra lógica para manejar eso, pero para ser honesto, eso sería mucho trabajo. Es por eso que no recomiendo construir sistemas de autorización y autenticación para aplicaciones de producción, a menos que seas un verdadero "agba" (ingeniero senior). Prefiero usar OAuth 2.0 (Google o Apple) o un proveedor externo como   o   . Y si estás en la ruina como yo y estás usando   , viene con un sistema de autenticación listo para usar. De esta manera, si algo sucede, tienes a alguien más a quien culpar y no sólo al pasante 🙂. Kinde Auth0 EdgeDB

Walkthroughs, tutorials, guides, and tips. This story will teach you how to do something new or how to do something better.

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

Read My Stories

Este audio es producido en el idioma original de la historia!

Cómo implementar un sistema de autenticación multidispositivo con FastAPI, Redis y JWT

About Author

COMENTARIOS

ETIQUETAS

ESTE ARTÍCULO FUE PRESENTADO EN

Related Stories

Las capas invisibles: por qué las entrevistas con los usuarios son un activo irremplazable

Una guía del arquitecto para crear una arquitectura de referencia para un lago de datos de IA/ML

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

Telegram: el puente de Crypto Island hacia el continente

Las capas invisibles: por qué las entrevistas con los usuarios son un activo irremplazable

Una guía del arquitecto para crear una arquitectura de referencia para un lago de datos de IA/ML

Toque para ganar: Telegram puede incorporar a los próximos 10 mil millones de usuarios de criptomonedas antes de Solana

Telegram: el puente de Crypto Island hacia el continente

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps