mirror of
https://github.com/taylanbakircioglu/haproxy-openmanager.git
synced 2026-09-16 15:45:11 +00:00
taylanbakircioglu
56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
import jwt
|
|
import bcrypt
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, Any
|
|
from fastapi import Header, HTTPException
|
|
from config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a password using bcrypt"""
|
|
salt = bcrypt.gensalt()
|
|
hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
|
|
return hashed.decode('utf-8')
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""Verify a password against its hash"""
|
|
return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8'))
|
|
|
|
def create_access_token(data: Dict[Any, Any], expires_delta: Optional[timedelta] = None) -> str:
|
|
"""Create a JWT access token"""
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
return encoded_jwt
|
|
|
|
def decode_access_token(token: str) -> Optional[Dict[str, Any]]:
|
|
"""Decode and verify a JWT access token"""
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
return payload
|
|
except jwt.PyJWTError:
|
|
return None
|
|
|
|
def generate_session_token() -> str:
|
|
"""Generate a secure random session token"""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
async def authenticate_user(authorization: str = Header(None)) -> Dict[str, Any]:
|
|
"""
|
|
Authenticate user from JWT token in Authorization header
|
|
This is a FastAPI dependency that can be used with Depends()
|
|
|
|
Usage:
|
|
@router.post("/endpoint")
|
|
async def endpoint(current_user: dict = Depends(authenticate_user)):
|
|
# current_user contains user data from JWT token
|
|
"""
|
|
# Import here to avoid circular dependency
|
|
from auth_middleware import get_current_user_from_token
|
|
|
|
return await get_current_user_from_token(authorization) |