Disallow users that have been disabled from logging into the application

release-notes.md

@@ -5,6 +5,9 @@
 
 * Add additional explanation about cod_unidade_autorizadora in the
   documentation.
+* Speed up test suite by changing the scope of fixtures
+* Use deep copy in tests to avoid interference in other tests
+* Block users that have been disabled from logging in
 
 
 ## 3.2.0

src/api.py

@@ -43,20 +43,23 @@
 
     description = environment_msg + description
 
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Executa as rotinas de inicialização da API."""
     await create_db_and_tables()
     await crud_auth.init_user_admin()
     yield
 
+
 app = FastAPI(
     title="Plataforma de recebimento de dados do Programa de Gestão - PGD",
     description=description,
     version=os.getenv("TAG_NAME", "dev-build") or "dev-build",
     lifespan=lifespan,
 )
 
+
 @app.middleware("http")
 async def check_user_agent(request: Request, call_next):
     user_agent = request.headers.get("User-Agent", None)
@@ -116,13 +119,19 @@ async def login_for_access_token(
             status.HTTP_422_UNPROCESSABLE_ENTITY, detail=message
         ) from exception
 
-    user = await crud_auth.authenticate_user(db, form_data.username, form_data.password)
-    if not user:
+    try:
+        user = await crud_auth.authenticate_user(
+            db, form_data.username, form_data.password
+        )
+    except (
+        crud_auth.InvalidCredentialsError,
+        crud_auth.DisabledUserError,
+    ) as exception:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Username ou password incorretos",
+            detail=exception.message,
             headers={"WWW-Authenticate": "Bearer"},
-        )
+        ) from exception
     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
     access_token = crud_auth.create_access_token(
         data={"sub": user.email}, expires_delta=access_token_expires

src/crud_auth.py

@@ -20,6 +20,20 @@
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
+# Exceções
+
+
+class InvalidCredentialsError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(self.message)
+
+
+class DisabledUserError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+        super().__init__(self.message)
+
 
 # ## Funções auxiliares
 
@@ -70,14 +84,38 @@ async def get_user(
     return None
 
 
-async def authenticate_user(db, username: str, password: str):
+async def authenticate_user(
+    db: DbContextManager, username: str, password: str
+) -> schemas.UsersSchema:
+    """Acessa o banco de dados e verifica se o usuário existe, não está
+    desabilitado e se as credenciais de acesso estão corretas. Caso tudo
+    esteja certo, retorna os detalhes do usuário.
+
+    Args:
+        db (DbContextManager): Context manager contendo as informações
+            necessárias para acesso ao banco de dados.
+        username (str): Nome do usuário.
+        password (str): Senha do usuário.
+
+    Raises:
+        InvalidCredentialsError: Caso o usuário não exista ou a senha
+            esteja incorreta.
+        DisabledUserError: Caso o usuário esteja desabilitado.
+
+    Returns:
+        schemas.UsersSchema: Detalhes do usuário presentes no banco de
+            dados.
+    """
     user = await get_user(db_session=db, email=username)
 
     if not user:
         return False
 
     if not verify_password(password, user.password):
-        return False
+        raise InvalidCredentialsError("Username ou password incorretos")
+
+    if user.disabled:
+        raise DisabledUserError("Usuário desabilitado")
 
     return user