Skip to content

adding and validate jwt roles #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ablogo merged 1 commit into from
Jan 15, 2026
Merged

adding and validate jwt roles #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/dependencies.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
load_dotenv()
client = AsyncMongoClient(os.environ["DB_URL"], server_api= ServerApi(version='1', strict=True, deprecation_errors=True))
database = client.get_database(os.environ["DB_NAME"])

async def get_db():
try:
yield database
Expand Down
19 changes: 19 additions & 0 deletions src/middlewares/auth_roles_jwt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
from typing import Dict
from fastapi import HTTPException, Request
from fastapi.security import OAuth2PasswordBearer

from src.services.jwt_service import verify_token_and_roles

class JWTCustom(OAuth2PasswordBearer):

def __init__(self, tokenUrl: str, scheme_name: str | None = None, scopes: Dict[str, str] | None = None, description: str | None = None, auto_error: bool = True):
super().__init__(tokenUrl, scheme_name, scopes, description, auto_error)

async def __call__(self, request: Request):
try:
token = await super().__call__(request)
if token is not None:
email = await verify_token_and_roles(token, ['admin'])
return email
except Exception as e:
raise e
1 change: 1 addition & 0 deletions src/models/user_model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ class User(BaseModel):
email_verified: bool = False
password: str
twofactor_enabled: bool = False
roles: Optional[List[str]] = list()
address: Optional[List[Address]] = list()
online: bool = False
disabled: bool = False
Expand Down
3 changes: 1 addition & 2 deletions src/routers/admin/security_router.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,7 @@
from dependency_injector.wiring import Provide, inject

from src.dependency_injection.containers import Container
from src.logging.mongo_logging import MongoLogger
from src.middlewares.auth_jwt import JWTCustom
from src.middlewares.auth_roles_jwt import JWTCustom
import src.services.totp_service as securitySvc

oauth2_scheme = JWTCustom(tokenUrl="/auth/sign-in")
Expand Down
39 changes: 31 additions & 8 deletions src/services/jwt_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from datetime import datetime, timedelta, timezone
from typing import Optional
from dependency_injector.wiring import Provide, inject
from fastapi import HTTPException, Request
from dotenv import load_dotenv
Expand All @@ -16,7 +17,12 @@
async def create_token(data: dict, expire_time: timedelta = timedelta(minutes=int(str(os.environ["JWT_EXPIRE_MINUTES"]))), crypto = crypto_service, log = log_service):
try:
for item in data:
data[item] = await crypto.encrypt_text(data[item])
if isinstance(data[item], list):
for x in range(len(data[item])):
# To encrypt the roles use the commented code
data[item][x] = data[item][x] #await crypto.encrypt_text(data[item][x])
else:
data[item] = await crypto.encrypt_text(data[item])

expire = datetime.now(timezone.utc) + expire_time
data.update({ "exp": expire })
Expand All @@ -33,8 +39,25 @@ async def verify_token(token: str, crypto = crypto_service, log = log_service):
email = await crypto.decrypt_text(payload.get("sub"))
return email
except Exception as e:
log.logger.error(e)
raise e
log.logger.error(e)
raise e

@inject
async def verify_token_and_roles(token: str, required_roles: Optional[list[str]] = None, crypto = crypto_service, log = log_service):
try:
payload = await verify(token)
email = await crypto.decrypt_text(payload.get("sub"))
if required_roles:
roles = payload.get('roles', [])

if bool(set(required_roles) & set(roles)):
return email
else:
raise HTTPException(status_code=401, detail="Insufficient permissions")

except Exception as e:
log.logger.error(e)
raise e

#async def verify_token(Authorization: str = Header(...)) -> bool:
async def verify_token_from_requests(request: Request):
Expand All @@ -47,8 +70,8 @@ async def verify_token_from_requests(request: Request):
return email

except Exception as e:
#log.logger.error(e)
raise e
#log.logger.error(e)
raise e

@inject
async def verify(request_token: str, log = log_service):
Expand All @@ -62,8 +85,8 @@ async def verify(request_token: str, log = log_service):
log.logger.error(e)
raise HTTPException(status_code=401, detail="Invalid credentials")
except Exception as e:
log.logger.error(e)
raise e
log.logger.error(e)
raise e

@inject
async def get_email(token, crypto = crypto_service, log = log_service):
Expand All @@ -73,6 +96,6 @@ async def get_email(token, crypto = crypto_service, log = log_service):
if email is None:
return None
except Exception as e:
log.logger.error(e)
log.logger.error(e)
else:
return email
6 changes: 3 additions & 3 deletions src/services/login_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,16 @@
logger: MongoLogger = Provide[Container.logging]

@inject
async def login(username, password, db, crypto = crypto_service, log = logger):
async def login(username: str, password: str, db, crypto = crypto_service, log = logger):
try:
user = await get_user(username, db)
if user is not None:
is_password_valid = await crypto.verify_password(password, user.password)
if not user.email_verified:
return True, None
if is_password_valid and not user.disabled:
token = await create_token({ "sub": user.email, "name": user.name })
return True, Token(access_token=token, token_type="bearer")
token = await create_token({ "sub": user.email, "name": user.name, "roles": user.roles })
return True, Token(access_token = token, token_type = "bearer")
except Exception as e:
log.logger.error(e)
return False, None