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   e o   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. session_id username  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. https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/  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 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 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   , que é o que usaremos neste 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" } }  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   e armazenar essa sessão no Redis, junto com todas as outras sessões. session_id  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   e   . Podemos então consultar o Redis usando o   . session_id username username  Se encontrarmos uma correspondência, removemos a sessão associada ao   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. 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"})  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 🙂. Kinde Auth0 o 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 áudio é produzido no idioma original da história!

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

About Author

COMENTARIOS

Rótulos

ESTE ARTIGO FOI APRESENTADO EM

Related Stories

Como melhorar seu fluxo de trabalho em 10 vezes: 17 aplicativos essenciais

Aumente sua produtividade com estas 18 ferramentas para desenvolvedores 🚀🔥

Digital Nomads Ouçam: O que você precisa saber sobre o novo visto DTV da Tailândia

Crescimento de Criptomoedas: Criando Personas de Usuários Eficazes

Como melhorar seu fluxo de trabalho em 10 vezes: 17 aplicativos essenciais

Aumente sua produtividade com estas 18 ferramentas para desenvolvedores 🚀🔥

Digital Nomads Ouçam: O que você precisa saber sobre o novo visto DTV da Tailândia

Crescimento de Criptomoedas: Criando Personas de Usuários Eficazes

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps