Como implementar um sistema de autenticação multidispositivo com FastAPI, Redis e JWT

Muitas grandes empresas de tecnologia oferecem um recurso interessante que permite que os usuários façam login em vários dispositivos. Os usuários podem gerenciar seus dispositivos, visualizar quais estão conectados e até mesmo sair de qualquer dispositivo usando qualquer um dos seus dispositivos conectados. Hoje, quero explorar como posso implementar um sistema de autenticação semelhante usando Redis e JWT.

Então, como isso funciona exatamente?

Para isso, decidi usar;

• API rápida para o back-end
• Redis para cache

  
  

Usaremos JWT (JSON Web Tokens) para autorizar solicitações. Como nosso aplicativo é stateless (o que significa que não tem memória de solicitações anteriores), precisamos de uma maneira de enviar dados de sessão e usuário. O JWT é ideal para gerenciar autenticação em aplicativos stateless e faz um excelente trabalho.

No entanto, uma desvantagem do JWT é que quanto mais carga útil você inclui em um token, mais longo ele se torna. Para nosso sistema, decidi incluir apenas o session_id e o username no token. Essas informações são suficientes para autorizar solicitações sem tornar o token excessivamente grande. Usaremos JWT (JSON Web Tokens) para autorizar solicitações. Como nosso aplicativo é sem estado (o que significa que não tem memória de solicitações anteriores), precisamos de uma maneira de enviar dados de sessão e usuário. O JWT é ideal para gerenciar autenticação em aplicativos sem estado e faz um excelente trabalho nesse sentido.

Você disse session_id? mas estamos usando JWT e nosso aplicativo em stateless

Neste contexto, "sessão" se refere ao dispositivo ou meio pelo qual um usuário interage com nosso aplicativo. Essencialmente, é o dispositivo no qual o usuário está conectado. Sempre que um usuário faz uma solicitação de login, criamos uma nova sessão (dispositivo) em nosso sistema que contém todas as informações relevantes do dispositivo. Esses dados serão armazenados no Redis para solicitações futuras.

Tudo bem, vamos codificar 😁

A primeira coisa a fazer é certificar-se de que você tem o Redis instalado na sua máquina local. Para instalar o Redis, vá para e siga as instruções específicas para o seu sistema operacional.

Em seguida, instalamos o python. Para isso, usarei o python 3.11 (ainda não vi necessidade de atualizar para o 3.12, para ser franco, a única razão pela qual uso o 3.11 é por causa do StrEnum, além disso, ainda amo o 3.10)

Em seguida, precisamos instalar o poetry, este é o gerenciador de pacotes que eu uso

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

Isso está resolvido, então vá em frente e clone o repositório

 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

Configurando nossa conexão 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))

Configuração do banco de dados

Criei um banco de dados de demonstração em demo_users.json , que é o que usaremos neste 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" } }

Agora, precisamos adicionar nossos esquemas e funções auxiliares para nosso banco de dados. Para ser breve, não colocarei todo o código aqui.

 @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

Configuração do servidor

Estamos usando FastAPI para executar nosso aplicativo, então vamos configurá-lo para

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

Tudo bem, isso é bom, nossa aplicação parece estar indo bem

Entrar / iniciar sessão

Cada vez que um usuário faz login no sistema, precisamos de uma maneira de gerar um session_id e armazenar essa sessão no Redis, junto com todas as outras sessões.

Quando um usuário faz login, primeiro autenticamos a solicitação para garantir que ela seja válida. Uma vez validada, podemos recuperar todas as informações do dispositivo da solicitação. Depois disso, armazenaremos essas informações no Redis, geraremos um novo token e retornaremos esse token ao usuário.

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

Sair/sair

Esta é a parte mais fácil. Cada vez que um usuário faz uma solicitação ao nosso aplicativo, decodificamos o token Bearer para recuperar o session_id e username . Podemos então consultar o Redis usando o username .

Se encontrarmos uma correspondência, removemos a sessão associada ao session_id do token decodificado. Por exemplo, se a sessão não existir, simplesmente retornamos uma mensagem ao usuário. Isso indica que o usuário já fez logout daquele dispositivo de um dispositivo diferente, ou que o token é inválido.

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

Então, sim, não foi tão difícil, foi? Tenho esse projeto na cabeça há algumas semanas e queria testá-lo. Embora esse sistema não seja completamente perfeito (quero dizer, nenhum sistema está livre de falhas), obviamente podemos torná-lo melhor. Por exemplo, como gerenciamos solicitações de um lugar como o Curl ou um aplicativo de console ou até mesmo o carteiro? Várias solicitações dessas fontes podem levar a muitas sessões, preenchendo nosso banco de dados com dados desnecessários. Sim, poderíamos verificar de onde a solicitação está vindo e criar nossa lógica para lidar com isso, mas, para ser honesto, isso daria muito trabalho. É por isso que não recomendo construir sistemas de autorização e autenticação para aplicativos de produção, a menos que você seja um verdadeiro "agba" (engenheiro sênior). Prefiro usar o OAuth 2.0 (Google ou Apple) ou um provedor externo como ou . E se você estiver quebrado como eu e estiver usando , ele vem com um sistema de autenticação pronto para uso. Dessa forma, se algo acontecer, você terá outra pessoa para culpar e não apenas o estagiário 🙂.