feat(auth): Iniciando o Oauth2 com JWT

Começando implementaçao de Oauth2 com JWT e sistema de permissions

Squash de commits das branches de @vince e @saciperere

baobaxia/api/v2/endpoints/artigos.py

 #!/usr/bin/env python3
 
-from typing import Optional, Dict
 import re
-from fastapi import (
-    Header,
-    File,
-    UploadFile,
-    HTTPException,
-    Query,
-    APIRouter,
-)
-
-from baobaxia.core.baobaxia_api import bbx
-
-from baobaxia.models.artigo import (
-    ArtigoStatus,
-    Artigo,
-)
-
+from typing import Annotated, Dict, Optional
 
+from baobaxia.api.v2.endpoints.auth import get_current_mocambola
+from baobaxia.constants import ROLE_ACERVO_EDITOR, ROLE_ACERVO_PUBLISHER
+from baobaxia.models.artigo import Artigo, ArtigoStatus
+from baobaxia.root import bbx
+from baobaxia.saberes import Mocambola
 from baobaxia.util import norm
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
 
-from baobaxia.constants import (
-    ROLE_ACERVO_EDITOR,
-    ROLE_ACERVO_PUBLISHER,
-)
-
-router = APIRouter()
+router = APIRouter(tags=["Artigos"])
+CurrentMocambola = Annotated[Mocambola, Depends(get_current_mocambola)]
 
-base_path = bbx.baobaxia.config.data_path / bbx.baobaxia.datastore.get_cached_path(
-    smid=bbx.baobaxia.config.default_mucua,
-    balaio_smid=bbx.baobaxia.config.default_balaio,
+base_path = bbx.config.data_path / bbx.datastore.get_cached_path(
+    smid=bbx.config.default_mucua,
+    balaio_smid=bbx.config.default_balaio,
 )
 
 blog_path = base_path / "blog"
 
 saberes_patterns_artigo = ["blog/*/"]
 
-bbx.baobaxia.discover_saberes(
+bbx.discover_saberes(
     model=Artigo, patterns=saberes_patterns_artigo, indexes_names=["tags"]
 )
 
-bbx.add_saberes_api(
-    Artigo,
-    router,
-    url_path="blog/artigo",
-    # skip_put_method=True,
-    get_summary="Retornar informações do artigo",
-    tags=["Artigos"],
+
+@router.post(
+    "/{balaio_slug_smid}/{mucua_slug_smid}/blog/artigo", summary="Cria um artigo"
 )
+@router.put(
+    "/{balaio_slug_smid}/{mucua_slug_smid}/blog/artigo", summary="Atualiza um artigo"
+)
+async def post_put_artigo(
+    *,
+    balaio_slug_smid: str,
+    mucua_slug_smid: str,
+    saber: Artigo,
+    mocambola: CurrentMocambola
+) -> Dict:
+    """Cria ou atualiza um artigo"""
+    balaio_smid = bbx.extract_smid(balaio_slug_smid)
+    mucua_smid = bbx.extract_smid(mucua_slug_smid)
 
+    print("Artigo, mocambola %s" % mocambola)
+    bbx.put_artigo(balaio_smid, mucua_smid, saber, mocambola)
+    return {"detail": "success"}
 
+
+@router.post("/{balaio_slug_smid}/{mucua_slug_smid}/blog/upload/{slug_smid:str}")
 async def upload_artigo(
-    *,
     balaio_slug_smid: str,
     mucua_slug_smid: str,
     slug_smid: str,
+    mocambola: CurrentMocambola,
     arquivo: UploadFile = File(...),
-    token: str = Header(...)
-):
+) -> Dict:
+    """Enviar artigo já existente"""
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     smid = bbx.extract_smid(slug_smid)
-    mocambola = bbx.baobaxia.get_session(token).mocambola
     if ROLE_ACERVO_EDITOR not in mocambola.roles:
         raise HTTPException(status_code=401, detail="Mocambola não é um editor")
-    saber = bbx.baobaxia.get_artigo(balaio_smid, mucua_smid, smid, token=token)
+    saber = bbx.get_artigo(balaio_smid, mucua_smid, smid, mocambola)
     if (
         saber.status == ArtigoStatus.published
         and ROLE_ACERVO_PUBLISHER not in mocambola.roles
     else:
         saber.content[0] = arquivo.filename
     with (
-        bbx.baobaxia.get_balaio_mucua_path(balaio_smid, mucua_smid)
+        bbx.get_balaio_mucua_path(balaio_smid, mucua_smid)
         / saber.path
         / saber.content[0]
     ).open("wb") as arquivo_saber:
         arquivo_saber.write(arquivo.file.read())
         arquivo_saber.close()
-        bbx.baobaxia.put_artigo(balaio_smid, mucua_smid, saber, token)
+        bbx.put_artigo(balaio_smid, mucua_smid, saber, mocambola)
         saber.is_local = True
         saber_path = (
-            bbx.baobaxia.get_balaio_mucua_path(balaio_smid, mucua_smid)
+            bbx.get_balaio_mucua_path(balaio_smid, mucua_smid)
             / saber.path
-            / bbx.baobaxia.config.saber_file_ext
+            / bbx.config.saber_file_ext
         )
         saber_path.with_suffix(".local").open("w").write(saber.json())
     return {"detail": "success"}
 
 
-router.add_api_route(
-    "/{balaio_slug_smid}/{mucua_slug_smid}/blog/upload/{slug_smid:str}",
-    upload_artigo,
-    response_model=dict,
-    methods=["POST"],
-    summary="Enviar o arquivo de um artigo já existente",
-    tags=["Artigos"],
-)
-
-
+@router.get("/blog/find")
 async def find_artigos(
-    *,
     balaio_slug_smid: Optional[str] = Query(None),
     mucua_slug_smid: Optional[str] = Query(None),
+    mocambola: Optional[CurrentMocambola] = None,
     keywords: Optional[str] = Query(None),
     hashtags: Optional[str] = Query(None),
     status: Optional[str] = Query(None),
     ordem_decrescente: bool = Query(False),
     pag_tamanho: int = 12,
     pag_atual: int = 1,
-    token: Optional[str] = Header(None)
-):
+) -> Dict:
+    """Busca artigos de acordo com os parâmetros fornecidos."""
     if balaio_slug_smid is not None and balaio_slug_smid:
         balaio_smid = bbx.extract_smid(balaio_slug_smid)
     else:
         else:
             return 0
 
-    return bbx.baobaxia.find_artigos(
-        token,
+    return bbx.find_artigos(
         balaio_smid,
         mucua_smid,
+        mocambola,
         filter_function=filter_function,
         sorted_function=sorted_function,
         sorted_reverse=ordem_decrescente,
     )
 
 
-router.add_api_route(
-    "/blog/find",
-    find_artigos,
-    response_model=Dict,
-    methods=["GET"],
-    summary="Busca artigos de acordo com os parâmetros fornecidos",
-    tags=["Artigos"],
-)
+@router.delete("/{balaio_slug_smid}/{mucua_slug_smid}/blog/artigo/{slug_smid:str}")
+async def delete_artigo(
+    balaio_slug_smid: str,
+    mucua_slug_smid: str,
+    slug_smid: str,
+    mocambola: CurrentMocambola,
+) -> Dict:
+    """Remove um saber artigo"""
+    balaio_smid = bbx.extract_smid(balaio_slug_smid)
+    mucua_smid = bbx.extract_smid(mucua_slug_smid)
+    smid = bbx.extract_smid(slug_smid)
+    bbx.del_artigo(balaio_smid, mucua_smid, smid, mocambola)
+    return {"detail": "success"}

baobaxia/api/v2/endpoints/auth.py

 #!/usr/bin/env python3
-
 """Grupo de rotas para autenticação."""
-from typing import Annotated
-
-from fastapi import APIRouter, Body, Depends, Form
-
-from baobaxia.core.baobaxia_api import bbx as bbx_default
+from datetime import timedelta
+from typing import Annotated, List
 
-# from loguru import logger
+from baobaxia.constants import ACCESS_TOKEN_EXPIRE_MINUTES, ALGORITHM, SECRET_KEY
+from baobaxia.models.auth import Token, TokenData
+from baobaxia.permissions.base import ModelPermission
+from baobaxia.permissions.roles import get_role_permissions
+from baobaxia.root import Baobaxia, bbx
+from baobaxia.saberes import Mocambola
+from baobaxia.utils.auth import create_access_token, oauth2_scheme, verify_password
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from jose import JWTError, jwt
+from loguru import logger
 
 router = APIRouter(tags=["Autenticação"])
 
 
-@router.post("/auth", response_model=dict, summary="Autenticar mocambola")
-async def authenticate(
-    *,
-    username: str = Body(...),
-    password: str = Body(...),
-    bbx: Annotated[str, Depends(bbx_default)],
+async def get_current_mocambola(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    bbx: Annotated[Baobaxia, Depends(bbx)],
 ):
-    """Autenticar mocambola."""
-    return bbx.baobaxia.authenticate(username, password)
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    mocambola = bbx.get_full_mocambola(username=token_data.username)
+    if mocambola is None:
+        raise credentials_exception
+    return mocambola
 
 
-@router.get("/session", summary="Sessão ativa de mocambola")
-async def get_session(
-    *,
-    token: str,
-    bbx: Annotated[str, Depends(bbx_default)],
-):
-    """Retorna a sessão ativa de mocambola."""
-    return bbx.baobaxia.get_session(token)
+@router.post("/token")
+async def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    bbx: Annotated[Baobaxia, Depends(bbx)],
+) -> Token:
+    """Autentica e retorna o token JWT"""
+    logger.debug(bbx.config)
+    mocambola = bbx.get_full_mocambola(form_data.username)
+    if not verify_password(form_data.password, mocambola.password_hash):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": mocambola.username}, expires_delta=access_token_expires
+    )
+    return Token(access_token=access_token, token_type="bearer")
 
 
-@router.post("/recover", response_model=dict, summary="Recuperar acesso de mocambola")
-async def recover(
-    *,
-    username: str = Form(...),
-    recovery_answer: str = Form(...),
-    bbx: Annotated[str, Depends(bbx_default)],
+@router.get("/mocambola/logged", response_model=Mocambola)
+async def mocambola_logged(
+    current_mocambola: Annotated[Mocambola, Depends(get_current_mocambola)]
 ):
-    """Recuperar mocambola."""
-    return bbx.baobaxia.authenticate(username, recovery_answer=recovery_answer)
+    return current_mocambola
+
+
+class PermissionChecker:
+    def __init__(self, permissions_required: List[ModelPermission]):
+        self.permissions_required = permissions_required
+
+    def __call__(self, mocambola: Mocambola = Depends(get_current_mocambola)):
+        for permission_required in self.permissions_required:
+            print("PermissionChecker, permission_required: %s" % permission_required)
+            for p in [*get_role_permissions(mocambola.role), *mocambola.permissions]:
+                print("permissions: %s" % p)
+            if permission_required not in [
+                *get_role_permissions(mocambola.role),
+                *mocambola.permissions,
+            ]:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail="Not enough permissions to access this resource",
+                )
+        return True
+
+
+# FIX.. REIMPLEMENTAR RESGATE DA SENHA
+# @router.post("/recover", response_model=dict, summary="Recuperar acesso de mocambola")
+# async def recover(*, username: str = Form(...), recovery_answer: str = Form(...)):
+#    """Recuperar mocambola."""
+#    return baobaxia.authenticate(username, recovery_answer=recovery_answer)

baobaxia/api/v2/endpoints/balaios.py

 #!/usr/bin/env python3
 
 """Grupo de rotas para balaios."""
-from typing import Optional, List
+from typing import Annotated, List
 
-from fastapi import APIRouter, Form, Header
-
-from baobaxia.core.baobaxia_api import bbx
-from baobaxia.saberes import Balaio
+from baobaxia.root import bbx
+from baobaxia.saberes import Balaio, Mocambola
+from baobaxia.utils.mocambola import get_current_mocambola
+from fastapi import APIRouter, Depends
 
 router = APIRouter(tags=["Balaios"])
 
+CurrentMocambola = Annotated[Mocambola, Depends(get_current_mocambola)]
+
 
-@router.get(
-    "/balaio/default", response_model=Balaio, summary="Retornar o balaio padrão"
-)
-async def get_default_balaio(token: Optional[str] = Header(None)):
+@router.get("/balaio/default")
+async def get_default_balaio() -> Balaio:
     """Retornar o balaio padrão."""
-    return bbx.baobaxia.get_default_balaio(token=token)
+    return bbx.get_default_balaio()
 
 
-@router.get("/balaio", response_model=List[Balaio], summary="Listar balaios")
-async def list_balaios(token: Optional[str] = Header(None)):
+@router.get("/balaio")
+async def list_balaios() -> List[Balaio]:
     """Listar balaios."""
-    return bbx.baobaxia.list_balaios(token=token)
+    return bbx.list_balaios()
 
 
-@router.post("/balaio", response_model=Balaio, summary="Criar um novo balaio")
+@router.post("/balaio", summary="Cria um novo balaio")
+@router.put("/balaio", summary="Atualiza um balaio")
 async def post_balaio(
-    name: str = Form(...), default_mucua: str = Form(...), token: str = Header(...)
-):
+    balaio: Balaio,
+    mocambola: CurrentMocambola,
+) -> Balaio:
     """Criar um novo balaio."""
-    return bbx.baobaxia.put_balaio(name=name, default_mucua=default_mucua, token=token)
+    return bbx.put_balaio(balaio=balaio, mocambola=mocambola)
 
 
-@router.get(
-    "/balaio/{balaio_slug_smid}", response_model=Balaio, summary="Retornar um balaio"
-)
-async def get_balaio(balaio_slug_smid: str, token: Optional[str] = Header(None)):
+@router.get("/balaio/{balaio_slug_smid}")
+async def get_balaio(balaio_slug_smid: str) -> Balaio:
     """Retornar um balaio."""
-    # TODO: self => bbx.extract...
-    balaio_smid = bbx.extract_smid(balaio_slug_smid)
-    return bbx.baobaxia.get_balaio(balaio_smid, token=token)
-
-
-@router.put(
-    "/balaio/{balaio_slug_smid}", response_model=Balaio, summary="Atualizar um balaio"
-)
-async def put_balaio(
-    balaio_slug_smid: str,
-    name: str = Form(...),
-    default_mucua: str = Form(...),
-    token: str = Header(...),
-):
-    """Atualizar um balaio."""
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
-    return bbx.baobaxia.put_balaio(
-        smid=balaio_smid, name=name, default_mucua=default_mucua, token=token
-    )
+    return bbx.get_balaio(balaio_smid)
 
 
-@router.delete("/balaio/{balaio_slug_smid}", summary="Deletar um balaio")
-async def del_balaio(balaio_slug_smid: str, token: str = Header(...)):
+@router.delete("/balaio/{balaio_slug_smid}")
+async def del_balaio(balaio_slug_smid: str, mocambola: CurrentMocambola):
     """Deletar um balaio."""
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
-    bbx.baobaxia.del_balaio(balaio_smid, token=token)
+    bbx.del_balaio(balaio_smid, mocambola)
     return {"detail": "success"}

baobaxia/api/v2/endpoints/mocambolas.py

 #!/usr/bin/env python3
 
+from pathlib import Path
+
 """Grupo de rotas para autenticação."""
-from fastapi import APIRouter, Header, Form
+from fastapi import APIRouter, Header, Form, Depends
 from fastapi.exceptions import HTTPException
-from typing import Optional, List
+from typing import Annotated, Dict, List, Optional
 from baobaxia.saberes import Mocambola
-from baobaxia.core.baobaxia_api import bbx
+from baobaxia.root import Baobaxia
 from baobaxia.constants import ROLE_MUCUA_PAJE
 
-router = APIRouter(tags=["Mocambolas"])
+from baobaxia.api.v2.endpoints.auth import (
+    get_current_mocambola, PermissionChecker )
+from baobaxia.permissions.model_permissions import *
 
+CurrentMocambola = Annotated[Mocambola, Depends(get_current_mocambola)]
+router = APIRouter(tags=["Mocambolas"])
+bbx = Baobaxia()
 
 @router.get(
     "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola",
     summary="Listar mocambolas",
 )
 async def list_mocambolas(
-    *, balaio_slug_smid: str, mucua_slug_smid: str, token: Optional[str] = Header(None)
-):
+        balaio_slug_smid: str, mucua_slug_smid: str,
+        token: Optional[str] = Header(None)
+) -> List[Mocambola]:
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     return bbx.baobaxia.list_mocambolas(balaio_smid, mucua_smid, token)
 
-
-@router.post(
-    "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola",
-    response_model=Mocambola,
-    summary="Criar um novo mocambola",
-)
+@router.post("/{balaio_slug_smid}/{mucua_slug_smid}/mocambola")
 async def post_mocambola(
-    *,
     balaio_slug_smid: str,
     mucua_slug_smid: str,
-    mocambola: Mocambola,
-    token: str = Header(...)
-):
+    mocambola: Mocambola
+) -> Mocambola:
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     return bbx.baobaxia.put_mocambola(
-        balaio_smid, mucua_smid, mocambola.username, mocambola, None, token=token
-    )
-
+        balaio_smid, mucua_smid, mocambola.username, mocambola)
 
 @router.get(
     "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/{username}",
     summary="Retornar um mocambola",
 )
 async def get_mocambola(
-    *,
     balaio_slug_smid: str,
     mucua_slug_smid: str,
-    username: str,
-    token: Optional[str] = Header(None)
+    username: str
 ):
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
-    return bbx.baobaxia.get_mocambola(balaio_smid, mucua_smid, username, token)
-
+    return bbx.baobaxia.get_mocambola(balaio_smid, mucua_smid, username)
 
 @router.put(
     "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/{username}",
     summary="Atualizar um mocambola",
 )
 async def put_mocambola(
-    *,
     balaio_slug_smid: str,
     mucua_slug_smid: str,
     username: str,
     mocambola: Mocambola,
-    token: str = Header(...)
 ):
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     return bbx.baobaxia.put_mocambola(
-        balaio_smid, mucua_smid, username, mocambola, None, token=token
+        balaio_smid, mucua_smid, username, mocambola
     )
 
-
 @router.delete(
-    "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/{username}",
-    summary="Deletar um mocambola",
+    "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/{username}"
 )
 async def del_mocambola(
-    *,
-    balaio_slug_smid: str,
-    mucua_slug_smid: str,
-    username: str,
-    token: str = Header(...)
-):
+        balaio_slug_smid: str,
+        mucua_slug_smid: str,
+        username: str,
+        mocambola: CurrentMocambola
+) -> Dict:
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
-    bbx.baobaxia.del_mocambola(balaio_smid, mucua_smid, username, token)
+    bbx.baobaxia.del_mocambola(balaio_smid, mucua_smid, username, mocambola)
     return {"detail": "success"}
 
-
 @router.post(
     "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/add_balaio",
-    summary="Adicionar mocambola no balaio",
 )
 async def add_mocambola_balaio(
-    *,
     balaio_slug_smid: str,
     mucua_slug_smid: str,
     novo_balaio_slug_smid: str,
     username: str,
-    token: str = Header(...)
+    mocambola: CurrentMocambola
 ):
-    mocambola = bbx.baobaxia.get_session(token).mocambola
+    """Adicionar mocambola no balaio"""
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     novo_balaio_smid = bbx.extract_smid(novo_balaio_slug_smid)
     if ROLE_MUCUA_PAJE not in mocambola.live_roles[balaio_smid]:
         raise HTTPException(status_code=401, detail="Mocambola não tem permissão")
     return bbx.baobaxia.add_mocambola_balaio(
-        balaio_smid, mucua_smid, novo_balaio_smid, username, token
+        balaio_smid, mucua_smid, novo_balaio_smid, username, mocambola
     )
 
-
+#FIX
 @router.post(
     "/{balaio_slug_smid}/{mucua_slug_smid}/mocambola/password",
-    summary="Atualiza a senha do mocambola autenticado",
 )
 async def set_password(
-    *,
-    balaio_slug_smid: str,
-    mucua_slug_smid: str,
-    new_password: str = Form(...),
-    password: Optional[str] = Form(None),
-    recovery_answer: Optional[str] = Form(None),
-    token: str = Header(...)
+        balaio_slug_smid: str,
+        mucua_slug_smid: str,
+        mocambola: CurrentMocambola,
+        new_password: str = Form(...),
+        password: Optional[str] = Form(None),
+        recovery_answer: Optional[str] = Form(None),
 ):
+    """Atualiza a senha do mocambola autenticado."""
     balaio_smid = bbx.extract_smid(balaio_slug_smid)
     mucua_smid = bbx.extract_smid(mucua_slug_smid)
     return bbx.baobaxia.set_password(
         new_password=new_password,
         password=password,
         recovery_answer=recovery_answer,
-        token=token,
+        mocambola=mocambola
     )