91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from passlib.context import CryptContext
|
|
import hashlib
|
|
import jwt
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Any
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
import contextvars
|
|
|
|
# Password hashing
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a plain-text password using PBKDF2-SHA256.
|
|
|
|
This supports arbitrary password lengths and avoids backend problems.
|
|
"""
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
return pwd_context.verify(plain, hashed)
|
|
# return True
|
|
|
|
|
|
# JWT configuration
|
|
JWT_SECRET = os.getenv("JWT_SECRET", "your-secret-key-change-this-in-production")
|
|
JWT_ALGORITHM = "HS256"
|
|
JWT_EXPIRATION_HOURS = int(os.getenv("JWT_EXPIRATION_HOURS", 24))
|
|
|
|
|
|
def create_access_token(data: Dict[str, Any], expires_delta: timedelta = None) -> str:
|
|
"""Create a JWT access token.
|
|
|
|
Args:
|
|
data: Dictionary of claims to encode
|
|
expires_delta: Optional timedelta for token expiration; defaults to JWT_EXPIRATION_HOURS
|
|
|
|
Returns:
|
|
JWT token string
|
|
"""
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(hours=JWT_EXPIRATION_HOURS)
|
|
to_encode.update({"exp": expire, "jti": uuid.uuid4().hex, "iat": datetime.utcnow()})
|
|
encoded_jwt = jwt.encode(to_encode, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
return encoded_jwt
|
|
|
|
|
|
def decode_access_token(token: str) -> Dict[str, Any]:
|
|
"""Decode and verify a JWT access token.
|
|
|
|
Args:
|
|
token: JWT token string
|
|
|
|
Returns:
|
|
Decoded payload dictionary
|
|
|
|
Raises:
|
|
jwt.InvalidTokenError: If token is invalid or expired
|
|
"""
|
|
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
|
|
|
|
def audit_log_path() -> str:
|
|
base = os.getenv("AUDIT_LOG_DIR", "audit_logs")
|
|
p = Path(base).resolve()
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
now = datetime.utcnow()
|
|
month_dir = p / now.strftime("%Y%m")
|
|
month_dir.mkdir(parents=True, exist_ok=True)
|
|
return str(month_dir / f"{now.strftime('%Y-%m-%d')}.txt")
|
|
|
|
# Per-request user context for audit logging
|
|
CURRENT_USER_CTX: contextvars.ContextVar = contextvars.ContextVar("CURRENT_USER_CTX", default=None)
|
|
|
|
def set_current_user_ctx(user: dict | None) -> None:
|
|
try:
|
|
CURRENT_USER_CTX.set(user)
|
|
except Exception:
|
|
pass
|
|
|
|
def get_current_user_ctx() -> dict | None:
|
|
try:
|
|
return CURRENT_USER_CTX.get()
|
|
except Exception:
|
|
return None
|