Autenticação e Autorização com JWT
Objetivos da Aula:
- Um entendimento básico sobre JWT
- Implementar autenticação de usuários com JWT.
- Adicionar lógica de autorização aos endpoints de atualização e deleção.
- Utilizar a biblioteca Bcrypt para encriptar as senhas dos usuários.
Caso prefira ver a aula em vídeo
Esse aula ainda não está disponível em formato de vídeo, somente em texto!
Introdução
Nesta aula, abordaremos dois aspectos cruciais de qualquer aplicação web: a autenticação e a autorização. Até agora, nossos usuários podem criar, ler, atualizar e deletar suas contas, mas qualquer pessoa pode fazer essas ações. Não queremos que qualquer usuário possa deletar ou modificar a conta de outro usuário. Para evitar isso, vamos implementar autenticação e autorização em nossa aplicação.
A autenticação é o processo de verificar quem um usuário é, enquanto a autorização é o processo de verificar o que ele tem permissão para fazer. Usaremos o JSON Web Token (JWT) para implementar a autenticação, e adicionaremos lógica de autorização aos nossos endpoints.
Além disso, até agora, estamos armazenando as senhas dos usuários como texto puro no banco de dados, o que é uma prática insegura. Corrigiremos isso utilizando a biblioteca Bcrypt para encriptar as senhas.
O que é um JWT
O JWT é um padrão (RFC 7519) que define uma maneira compacta e autônoma de transmitir informações entre as partes de maneira segura. Essas informações são transmitidas como um objeto JSON que é digitalmente assinado usando um segredo (com o algoritmo HMAC) ou um par de chaves pública/privada usando RSA, ou ECDSA.
Um JWT consiste em três partes:
-
Header: O cabeçalho do JWT consiste tipicamente em dois componentes: o tipo de token, que é JWT neste caso, e o algoritmo de assinatura, como HMAC SHA256 ou RSA. Essas informações são codificadas em Base64Url e formam a primeira parte do JWT.
-
Payload: O payload de um JWT é onde as reivindicações (ou declarações) são armazenadas. As reivindicações são informações que queremos transmitir e que são relevantes para a interação entre o cliente e o servidor. As reivindicações são codificadas em Base64Url e formam a segunda parte do JWT.
-
Signature: A assinatura é utilizada para verificar que o remetente do JWT é quem afirma ser e para garantir que a mensagem não foi alterada ao longo do caminho. Para criar a assinatura, você precisa codificar o cabeçalho, o payload, e um segredo utilizando o algoritmo especificado no cabeçalho. A assinatura é a terceira parte do JWT. Uma assinatura de JWT pode ser criada como se segue:
Essas três partes são separadas por pontos (.) e juntas formam um token JWT.
Formando a estrutura: HEADER.PAYLOAD.SIGNATURE que formam um token parecido com
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ0ZXN0ZUB0ZXN0LmNvbSIsImV4cCI6MTY5MDI1ODE1M30.Nx0P_ornVwJBH_LLLVrlJoh6RmJeXR-Nr7YJ_mlGY04
É importante ressaltar que, apesar de a informação em um JWT estar codificada, ela não está criptografada. Isso significa que qualquer pessoa com acesso ao token pode decodificar e ler as informações nele. No entanto, sem o segredo usado para assinar o token, eles não podem alterar as informações ou forjar um novo token. Portanto, não devemos incluir informações sensíveis ou confidenciais no payload do JWT.
Se quisermos ver o header, o payload e a assinatura contidas nesse token podemos acessar o debuger do jwt e checar quais as informações que estão nesse token:
Como funciona o JWT
Em uma aplicação web, o processo de autenticação geralmente funciona da seguinte maneira:
- O usuário envia suas credenciais (e-mail e senha) para o servidor em um endpoint de geração de token (
/tokenpor exemplo); - O servidor verifica as credenciais e, se estiverem corretas, gera um token JWT e o envia de volta ao cliente;
- Nas solicitações subsequentes, o cliente deve incluir esse token no cabeçalho de autorização de suas solicitações. Como, por exemplo:
Authorization: Bearer <token>; - Quando o servidor recebe uma solicitação com um token JWT, ele pode verificar a assinatura e se o token é válido e não expirou, ele processa a solicitação.
sequenceDiagram
participant Cliente as Cliente
participant Servidor as Servidor
Cliente->>Servidor: Envia credenciais (e-mail e senha)
Servidor->>Cliente: Verifica as credenciais
Servidor->>Cliente: Envia token JWT
Cliente->>Servidor: Envia solicitação com token JWT no cabeçalho de autorização
Servidor->>Cliente: Verifica o token JWT e processa a solicitaçãoNos próximos tópicos, vamos detalhar como podemos gerar e verificar tokens JWT em nossa aplicação FastAPI, bem como adicionar autenticação e autorização aos nossos endpoints.
Gerando tokens JWT
Para gerar tokens JWT, precisamos de duas bibliotecas extras: pyjwt e passlib. A primeira será usada para a geração do token, enquanto a segunda será usada para criptografar as senhas dos usuários. Para instalá-las, execute o seguinte comando no terminal:
Agora, criaremos uma função para gerar nossos tokens JWT. Criaremos um novo arquivo para gerenciar a segurança: security.py. Nesse arquivo iniciaremos a geração dos tokens:
A função create_access_token é responsável por criar um novo token JWT que será usado para autenticar o usuário. Ela recebe um dicionário de dados, adiciona um tempo de expiração ao token (baseado na constante ACCESS_TOKEN_EXPIRE_MINUTES). Esses dados, em conjunto, formam o payload do JWT. Em seguida, usa a biblioteca pyjwt para codificar essas informações em um token JWT, que é então retornado.
Note que a constante SECRET_KEY é usada para assinar o token, e o algoritmo HS256 é usado para a codificação. Em um cenário de produção, você deve manter a SECRET_KEY em um local seguro e não expô-la em seu código.
Testando a geração de tokens
Embora esse código será coberto no futuro com a utilização do token, é interessante criarmos um teste para essa função com uma finalidade puramente didática. De forma que vejamos os tokens gerados pelo pyjwt e interagirmos com ele.
Com isso criaremos um arquivo chamado tests/test_security.py para efetuar esse teste:
Na próxima seção, veremos como podemos usar a biblioteca passlib para tratar as senhas dos usuários.
Hashing de Senhas
Armazenar senhas em texto puro é uma prática de segurança extremamente perigosa. Em vez disso, é uma prática padrão criptografar ("hash") as senhas antes de armazená-las. Quando um usuário tenta se autenticar, a senha inserida é criptografada novamente e comparada com a versão criptografada armazenada no banco de dados. Se as duas correspondem, o usuário é autenticado.
Implementaremos essa funcionalidade usando a biblioteca passlib. Criaremos duas funções: uma para criar o hash da senha e outra para verificar se uma senha inserida corresponde ao hash armazenado. Adicione o seguinte código ao arquivo security.py:
| fast_zero/security.py | |
|---|---|
A função get_password_hash recebe uma senha em texto puro como argumento e retorna uma versão criptografada dessa senha. A função verify_password recebe uma senha em texto puro e uma senha criptografada como argumentos, e verifica se a senha em texto puro, quando criptografada, corresponde à senha criptografada. Ambas as funções utilizam o objeto pwd_context, que definimos anteriormente usando a biblioteca passlib.
Agora, quando um usuário se registra em nossa aplicação, devemos usar a função get_password_hash para armazenar uma versão criptografada da senha. Quando um usuário tenta se autenticar, devemos usar a função verify_password para verificar se a senha inserida corresponde à senha armazenada.
Na próxima seção, modificaremos nossos endpoints para fazer uso dessas funções.
Modificando o endpoint de POST para encriptar a senha
Com as funções de criação de hash de senha e verificação de senha em vigor, agora podemos atualizar nossos endpoints para usar essa nova funcionalidade de encriptação.
Primeiro, modificaremos a função create_user para criar um hash da senha antes de armazená-la no banco de dados. Para fazer isso precisamos importar a função de geração de hash get_password_hash e no momento da criação do registro na tabela a senha deve ser passada com o hash gerado:
from fast_zero.security import get_password_hash
# ...
@app.post('/users/', status_code=HTTPStatus.CREATED, response_model=UserPublic)
def create_user(user: UserSchema, session: Session = Depends(get_session)):
db_user = session.scalar(
select(User).where(
or_(User.username == user.username, User.email == user.email)
)
)
if db_user:
if db_user.username == user.username:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail='Username already exists',
)
elif db_user.email == user.email:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail='Email already exists',
)
hashed_password = get_password_hash(user.password)
db_user = User(
email=user.email,
username=user.username,
password=hashed_password,
)
session.add(db_user)
session.commit()
session.refresh(db_user)
return db_user
Desta forma, a senha não será mais criada em texto plano no objeto User. Fazendo com que caso exista algum problema relacionado a vazamento de dados, as senhas das pessoas nunca sejam expostas.
Sobre o teste da POST /users/
Por não validar o password, usando o retorno UserPublic, o teste já escrito deve passar normalmente:
task test
# ...
tests/test_app.py::test_root_deve_retornar_ok_e_ola_mundo PASSED
tests/test_app.py::test_create_user PASSED
Modificando o endpoint de atualização de usuários
É igualmente importante modificar a função update_user para também criar um hash da senha antes de atualizar User no banco de dados. Caso contrário, a senha em texto puro seria armazenada no banco de dados no momento da atualização.
@app.put('/users/{user_id}', response_model=UserPublic)
def update_user(
user_id: int,
user: UserSchema,
session: Session = Depends(get_session),
):
db_user = session.scalar(select(User).where(User.id == user_id))
if db_user is None:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail='User not found'
)
db_user.username = user.username
db_user.password = get_password_hash(user.password)
db_user.email = user.email
session.commit()
session.refresh(db_user)
return db_user
Assim, a atualização de um User, via método PUT, também criará o hash da senha no momento da atualização. Pois, nesse caso em específico, existe a possibilidade de alterar qualquer coluna da tabela, inclusive o campo password.
Sobre os testes da PUT /users/{user_id}
Assim como no teste da rota de criação, os testes também passam normalmente por não validarem o campo password.
task test
# ...
tests/test_app.py::test_root_deve_retornar_ok_e_ola_mundo PASSED
tests/test_app.py::test_create_user PASSED
tests/test_app.py::test_read_users PASSED
tests/test_app.py::test_read_users_with_users PASSED
tests/test_app.py::test_update_user PASSED
Criando um endpoint de geração do token
Antes de criar o endpoint, precisamos criar um schema para o nosso token. Em um contexto JWT, access_token é o próprio token que representa a sessão do usuário e contém informações sobre o usuário, enquanto token_type é um tipo de autenticação que será incluído no cabeçalho de autorização de cada solicitação. Em geral, o token_type para JWT é "bearer".
Utilizando OAuth2PasswordRequestForm
A classe OAuth2PasswordRequestForm é uma classe especial do FastAPI que gera automaticamente um formulário para solicitar o username (email neste caso) e a senha. Este formulário será apresentado automaticamente no Swagger UI e Redoc, facilitando a realização de testes de autenticação.
Para usar os formulários no FastAPI, precisamos instalar o python-multipart:
Criando um endpoint de geração do token
Agora criaremos o endpoint que irá autenticar o usuário e fornecer um token de acesso JWT. Este endpoint irá receber as informações de login do usuário, verificar se as credenciais são válidas e, em caso afirmativo, retornar um token de acesso JWT.
from fastapi.security import OAuth2PasswordRequestForm
from fast_zero.schemas import Message, Token, UserList, UserPublic, UserSchema
from fast_zero.security import (
create_access_token,
get_password_hash,
verify_password,
)
# ...
@app.post('/token', response_model=Token)
def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
session: Session = Depends(get_session),
):
user = session.scalar(select(User).where(User.email == form_data.username))
if not user:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail='Incorrect email or password'
)
if not verify_password(form_data.password, user.password):
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST,
detail='Incorrect email or password'
)
access_token = create_access_token(data={'sub': user.email})
return {'access_token': access_token, 'token_type': 'bearer'}
Esse endpoint recebe os dados do formulário através do form_data (que são injetados automaticamente graças ao Depends()) e tenta recuperar um usuário com o email fornecido. Se o usuário não for encontrado ou a senha não corresponder ao hash armazenado no banco de dados, uma exceção é lançada. Caso contrário, um token de acesso é criado usando o create_access_token() que criamos anteriormente e retornado como uma resposta.
Testando /token
Agora escreveremos um teste para verificar se o nosso novo endpoint está funcionando corretamente.
| tests/test_app.py | |
|---|---|
Nesse teste, nós enviamos uma requisição POST para o endpoint "/token" com um username e uma senha válidos. Então, nós verificamos que a resposta contém um "access_token" e um "token_type", que são os campos que esperamos de um JWT válido.
No entanto, há um problema. Agora que a senha está sendo criptografada, nosso teste falhará:
Para corrigir isso, precisamos garantir que a senha esteja sendo criptografada na fixture antes de ser salva:
from fast_zero.security import get_password_hash
# ...
@pytest.fixture()
def user(session):
user = User(
username='Teste',
email='teste@test.com',
password=get_password_hash('testtest'),
)
session.add(user)
session.commit()
session.refresh(user)
return user
Rodaremos o teste novamente. No entanto, ainda teremos um problema. Agora só temos a versão criptografada da senha, que não é útil para fazer o login, já que o login exige a senha em texto puro:
Para resolver isso, faremos uma modificação no objeto user (um monkey patch) para adicionar a senha em texto puro:
| tests/conftest.py | |
|---|---|
Monkey patching é uma técnica em que modificamos ou estendemos o código em tempo de execução. Neste caso, estamos adicionando um novo atributo clean_password ao objeto user para armazenar a senha em texto puro.
Agora, podemos alterar o teste para usar clean_password:
| tests/test_app.py | |
|---|---|
E agora todos os testes devem passar normalmente:
task test
# ...
tests/test_app.py::test_root_deve_retornar_ok_e_ola_mundo PASSED
tests/test_app.py::test_create_user PASSED
tests/test_app.py::test_read_users PASSED
tests/test_app.py::test_read_users_with_users PASSED
tests/test_app.py::test_update_user PASSED
tests/test_app.py::test_delete_user PASSED
tests/test_app.py::test_get_token PASSED
tests/test_db.py::test_create_user PASSED
Isso conclui a parte de autenticação de nossa API. No próximo passo, implementaremos a autorização nos endpoints.
Protegendo os Endpoints
Agora que temos uma forma de autenticar nossos usuários e emitir tokens JWT, é hora de usar essa infraestrutura para proteger nossos endpoints. Neste passo, adicionaremos autenticação aos endpoints PUT e DELETE.
Para garantir que as informações do usuário sejam extraídas corretamente do token JWT, precisamos de um schema especial, o TokenData. Esse schema será utilizado para tipificar os dados extraídos do token JWT e garantir que temos um campo username que será usado para identificar o usuário.
Nesse ponto, criaremos uma a função get_current_user que será responsável por extrair o token JWT do header Authorization da requisição, decodificar esse token, extrair as informações do usuário e obter finalmente o usuário do banco de dados. Se qualquer um desses passos falhar, uma exceção será lançada e a requisição será negada. A adicionaremos ao security.py:
from datetime import datetime, timedelta
from http import HTTPStatus
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from jwt import DecodeError, decode, encode
from passlib.context import CryptContext
from sqlalchemy import select
from sqlalchemy.orm import Session
from fast_zero.database import get_session
from fast_zero.models import User
from fast_zero.schemas import TokenData
# ...
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
session: Session = Depends(get_session),
token: str = Depends(oauth2_scheme),
):
credentials_exception = HTTPException(
status_code=HTTPStatus.UNAUTHORIZED,
detail='Could not validate credentials',
headers={'WWW-Authenticate': 'Bearer'},
)
try:
payload = decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get('sub')
if not username:
raise credentials_exception
token_data = TokenData(username=username)
except DecodeError:
raise credentials_exception
user = session.scalar(
select(User).where(User.email == token_data.username)
)
if user is None:
raise credentials_exception
return user
Aqui, a função get_current_user é definida como assíncrona, indicando que ela pode realizar operações de IO (como consultar um banco de dados) de forma não bloqueante. Esta função aceita dois argumentos: session e token. O session é obtido através da função get_session (não mostrada aqui), que deve retornar uma sessão de banco de dados ativa. O token é obtido do header de autorização da requisição, que é esperado ser do tipo Bearer (indicado pelo esquema OAuth2).
A variável credentials_exception é definida como uma exceção HTTP que será lançada sempre que houver um problema com as credenciais fornecidas pelo usuário. O status 401 indica que a autenticação falhou e a mensagem "Could not validate credentials" é retornada ao cliente. Além disso, um cabeçalho 'WWW-Authenticate' é incluído na resposta, indicando que o cliente deve fornecer autenticação.
No bloco try, tentamos decodificar o token JWT usando a chave secreta e o algoritmo especificado. O token decodificado é armazenado na variável payload. Extraímos o campo 'sub' (normalmente usado para armazenar o identificador do usuário no token JWT) e verificamos se ele existe. Se não, lançamos a exceção credentials_exception. Em seguida, criamos um objeto TokenData com o username.
Por fim, realizamos uma consulta ao banco de dados para encontrar o usuário com o e-mail correspondente ao username contido no token. session.scalar é usado para retornar a primeira coluna do primeiro resultado da consulta. Se nenhum usuário for encontrado, lançamos a exceção credentials_exception. Se um usuário for encontrado, retornamos esse usuário.
Aplicação da proteção ao endpoint
Primeiro, aplicaremos a autenticação no endpoint PUT. Se o user_id da rota não corresponder ao id do usuário autenticado, retornaremos um erro 400. Se tudo estiver correto, o usuário será atualizado normalmente.
from fast_zero.security import (
create_access_token,
get_current_user,
get_password_hash,
verify_password,
)
# ...
@app.put('/users/{user_id}', response_model=UserPublic)
def update_user(
user_id: int,
user: UserSchema,
session: Session = Depends(get_session),
current_user: User = Depends(get_current_user),
):
if current_user.id != user_id:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST, detail='Not enough permissions'
)
current_user.username = user.username
current_user.password = get_password_hash(user.password)
current_user.email = user.email
session.commit()
session.refresh(current_user)
return current_user
Com isso, podemos remover a query feita no endpoint para encontrar o User, pois ela já está sendo feita no get_current_user, simplificando ainda mais nosso endpoint.
Agora, aplicaremos a autenticação no endpoint DELETE. Semelhante ao PUT, se o user_id da rota não corresponder ao id do usuário autenticado, retornaremos um erro 400. Se tudo estiver correto, o usuário será deletado.
Com essa nova dependência, o FastAPI automaticamente garantirá que um token de autenticação válido seja fornecido antes de permitir o acesso a esses endpoints. Se o token não for válido, ou se o usuário tentar modificar ou deletar um usuário diferente, um erro será retornado.
Atualizando os Testes
Os testes precisam ser atualizados para refletir essas mudanças. Primeiro, precisamos criar uma nova fixture que gere um token para um usuário de teste.
| tests/conftest.py | |
|---|---|
Agora, podemos atualizar os testes para o endpoint PUT e DELETE para incluir a autenticação.
Finalmente, podemos rodar todos os testes para garantir que tudo esteja funcionando corretamente.
task test
# ...
tests/test_app.py::test_root_deve_retornar_ok_e_ola_mundo PASSED
tests/test_app.py::test_create_user PASSED
tests/test_app.py::test_read_users PASSED
tests/test_app.py::test_read_users_with_users PASSED
tests/test_app.py::test_update_user PASSED
tests/test_app.py::test_delete_user PASSED
tests/test_app.py::test_get_token PASSED
tests/test_db.py::test_create_user PASSED
Com essas alterações, nossos endpoints agora estão seguramente protegidos pela autenticação. Apenas os usuários autenticados podem alterar ou deletar seus próprios dados. Isso traz uma camada adicional de segurança e integridade para o nosso aplicativo.
Exercício
- Faça um teste para cobrir o cenário que levanta exception
credentials_exceptionna autenticação. Ao olhar a cobertura desecurity.pyvocê vai notar que esse contexto não está coberto.
Commit
Depois de finalizar a proteção dos endpoints e atualizar os testes, é hora de fazer commit das alterações. Não se esqueça de revisar as alterações antes de fazer o commit.
git status
git add .
git commit -m "Protege os endpoints PUT e DELETE com autenticação"
Conclusão
Nesta aula, demos um passo importante para aumentar a segurança da nossa API. Implementamos a autenticação e a autorização para os endpoints PUT e DELETE, garantindo que apenas usuários autenticados possam alterar ou excluir seus próprios dados. Também atualizamos os testes para incluir a autenticação. Na próxima aula, continuaremos a expandir a funcionalidade da nossa API. Até lá!