openvpn-endpoint-server/server/app/utils/security.py

76 lines
2.0 KiB
Python

"""Security utilities for password hashing and JWT tokens."""
from datetime import datetime, timedelta
import bcrypt
from jose import jwt, JWTError
from ..config import get_settings
settings = get_settings()
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 get_password_hash(password: str) -> str:
"""Generate password hash."""
return bcrypt.hashpw(
password.encode('utf-8'),
bcrypt.gensalt()
).decode('utf-8')
def create_access_token(
user_id: int,
username: str,
role: str,
tenant_id: int | None = None,
expires_delta: timedelta | None = None
) -> str:
"""Create a JWT access token."""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes)
to_encode = {
"sub": user_id,
"username": username,
"role": role,
"tenant_id": tenant_id,
"exp": expire,
"type": "access"
}
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
def create_refresh_token(
user_id: int,
expires_delta: timedelta | None = None
) -> str:
"""Create a JWT refresh token."""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(days=settings.refresh_token_expire_days)
to_encode = {
"sub": user_id,
"exp": expire,
"type": "refresh"
}
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
def decode_token(token: str) -> dict | None:
"""Decode and validate a JWT token."""
try:
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
return payload
except JWTError:
return None