238 lines
6.0 KiB
Python
238 lines
6.0 KiB
Python
"""
|
|
Security Utilities
|
|
|
|
Provides password hashing, JWT token management, and other security functions
|
|
following the Python Quick Start Guide best practices.
|
|
"""
|
|
|
|
import bcrypt
|
|
import jwt
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Optional, Any
|
|
|
|
|
|
# ==========================================
|
|
# Password Hashing (bcrypt)
|
|
# ==========================================
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""
|
|
Hash a password using bcrypt.
|
|
|
|
Args:
|
|
password: Plain text password
|
|
|
|
Returns:
|
|
Hashed password as string
|
|
|
|
Example:
|
|
hashed = hash_password("mypassword123")
|
|
"""
|
|
password_bytes = password.encode('utf-8')
|
|
salt = bcrypt.gensalt()
|
|
hashed = bcrypt.hashpw(password_bytes, salt)
|
|
return hashed.decode('utf-8')
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""
|
|
Verify a password against a hash.
|
|
|
|
Args:
|
|
plain_password: Plain text password to check
|
|
hashed_password: Hashed password to compare against
|
|
|
|
Returns:
|
|
True if password matches, False otherwise
|
|
|
|
Example:
|
|
if verify_password("mypassword123", user.password_hash):
|
|
# Password is correct
|
|
"""
|
|
password_bytes = plain_password.encode('utf-8')
|
|
hash_bytes = hashed_password.encode('utf-8')
|
|
return bcrypt.checkpw(password_bytes, hash_bytes)
|
|
|
|
|
|
# ==========================================
|
|
# JWT Token Management
|
|
# ==========================================
|
|
|
|
def create_jwt_token(
|
|
payload: Dict[str, Any],
|
|
secret_key: str,
|
|
algorithm: str = 'HS256',
|
|
expires_in: int = 3600
|
|
) -> str:
|
|
"""
|
|
Create a JWT token with expiration.
|
|
|
|
Args:
|
|
payload: Token payload data
|
|
secret_key: Secret key for signing
|
|
algorithm: JWT algorithm (HS256, RS256, etc.)
|
|
expires_in: Token lifetime in seconds
|
|
|
|
Returns:
|
|
Encoded JWT token string
|
|
|
|
Example:
|
|
token = create_jwt_token(
|
|
payload={'user_id': 123, 'role': 'admin'},
|
|
secret_key=app.config['SECRET_KEY'],
|
|
expires_in=3600
|
|
)
|
|
"""
|
|
payload = payload.copy()
|
|
expire = datetime.utcnow() + timedelta(seconds=expires_in)
|
|
payload.update({'exp': expire, 'iat': datetime.utcnow()})
|
|
|
|
return jwt.encode(payload, secret_key, algorithm=algorithm)
|
|
|
|
|
|
def decode_jwt_token(
|
|
token: str,
|
|
secret_key: str,
|
|
algorithm: str = 'HS256'
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Decode and verify a JWT token.
|
|
|
|
Args:
|
|
token: JWT token string
|
|
secret_key: Secret key for verification
|
|
algorithm: JWT algorithm used
|
|
|
|
Returns:
|
|
Decoded payload dict, or None if invalid
|
|
|
|
Example:
|
|
payload = decode_jwt_token(token, app.config['SECRET_KEY'])
|
|
if payload:
|
|
user_id = payload['user_id']
|
|
"""
|
|
try:
|
|
payload = jwt.decode(token, secret_key, algorithms=[algorithm])
|
|
return payload
|
|
except jwt.ExpiredSignatureError:
|
|
# Token has expired
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
# Token is invalid
|
|
return None
|
|
|
|
|
|
def create_id_token(
|
|
user_data: Dict[str, Any],
|
|
client_id: str,
|
|
issuer: str,
|
|
private_key: str,
|
|
algorithm: str = 'RS256',
|
|
expires_in: int = 3600
|
|
) -> str:
|
|
"""
|
|
Create an OIDC ID Token (JWT).
|
|
|
|
Args:
|
|
user_data: User information (sub, email, name, etc.)
|
|
client_id: OAuth client ID (aud claim)
|
|
issuer: OIDC issuer URL (iss claim)
|
|
private_key: Private key for RS256 signing
|
|
algorithm: JWT algorithm (should be RS256 for OIDC)
|
|
expires_in: Token lifetime in seconds
|
|
|
|
Returns:
|
|
Encoded ID token string
|
|
|
|
Example:
|
|
id_token = create_id_token(
|
|
user_data={'sub': 'user-123', 'email': 'user@example.com'},
|
|
client_id='my-app',
|
|
issuer='https://auth.example.com',
|
|
private_key=app.config['OIDC_JWT_PRIVATE_KEY']
|
|
)
|
|
"""
|
|
now = datetime.utcnow()
|
|
payload = {
|
|
'iss': issuer,
|
|
'sub': user_data.get('sub'),
|
|
'aud': client_id,
|
|
'exp': now + timedelta(seconds=expires_in),
|
|
'iat': now,
|
|
**user_data # Include all user claims
|
|
}
|
|
|
|
return jwt.encode(payload, private_key, algorithm=algorithm)
|
|
|
|
|
|
# ==========================================
|
|
# Token Generation
|
|
# ==========================================
|
|
|
|
def generate_secure_token(length: int = 32) -> str:
|
|
"""
|
|
Generate a cryptographically secure random token.
|
|
|
|
Args:
|
|
length: Token length in bytes (default 32)
|
|
|
|
Returns:
|
|
URL-safe token string
|
|
|
|
Example:
|
|
auth_code = generate_secure_token(32)
|
|
access_token = generate_secure_token(64)
|
|
"""
|
|
return secrets.token_urlsafe(length)
|
|
|
|
|
|
def generate_client_secret() -> str:
|
|
"""
|
|
Generate a secure client secret for OIDC clients.
|
|
|
|
Returns:
|
|
URL-safe client secret string
|
|
|
|
Example:
|
|
client_secret = generate_client_secret()
|
|
"""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
# ==========================================
|
|
# Password Validation
|
|
# ==========================================
|
|
|
|
def validate_password_strength(password: str) -> tuple[bool, Optional[str]]:
|
|
"""
|
|
Validate password strength.
|
|
|
|
Args:
|
|
password: Password to validate
|
|
|
|
Returns:
|
|
Tuple of (is_valid, error_message)
|
|
|
|
Rules:
|
|
- Minimum 8 characters
|
|
- At least one digit (optional but recommended)
|
|
|
|
Example:
|
|
is_valid, error = validate_password_strength("password123")
|
|
if not is_valid:
|
|
raise ValueError(error)
|
|
"""
|
|
if len(password) < 8:
|
|
return False, "Password must be at least 8 characters long"
|
|
|
|
# Optional: Check for digit
|
|
# if not any(char.isdigit() for char in password):
|
|
# return False, "Password must contain at least one digit"
|
|
|
|
# Optional: Check for uppercase
|
|
# if not any(char.isupper() for char in password):
|
|
# return False, "Password must contain at least one uppercase letter"
|
|
|
|
return True, None
|