copilot-swe-agent[bot]
bcbef88803
- Add backend/conftest.py that inserts the backend directory into sys.path, fixing ModuleNotFoundError when pytest runs from the backend/ directory (as CI does with `cd backend && pytest tests/`) - Run black formatter on all 28 backend files that needed reformatting - All 53 tests pass with both `pytest tests/` and `python -m pytest tests/` Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> Agent-Logs-Url: https://github.com/christianlouis/pop_puller_to_gmail/sessions/beb47db2-da25-416a-8fb0-c1452a3b22a7
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
"""
|
|
Security utilities for encryption, hashing, and token generation.
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, Any
|
|
from jose import JWTError, jwt
|
|
from passlib.context import CryptContext
|
|
from cryptography.fernet import Fernet
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
import base64
|
|
|
|
from app.core.config import settings
|
|
|
|
# Password hashing context
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""Verify a password against its hash"""
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
|
def get_password_hash(password: str) -> str:
|
|
"""Generate password hash"""
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def create_access_token(
|
|
data: Dict[str, Any], expires_delta: Optional[timedelta] = None
|
|
) -> str:
|
|
"""Create JWT access token"""
|
|
to_encode = data.copy()
|
|
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(
|
|
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
|
|
)
|
|
|
|
to_encode.update({"exp": expire, "type": "access"})
|
|
encoded_jwt = jwt.encode(
|
|
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
|
|
)
|
|
return encoded_jwt
|
|
|
|
|
|
def create_refresh_token(data: Dict[str, Any]) -> str:
|
|
"""Create JWT refresh token"""
|
|
to_encode = data.copy()
|
|
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
|
|
to_encode.update({"exp": expire, "type": "refresh"})
|
|
encoded_jwt = jwt.encode(
|
|
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
|
|
)
|
|
return encoded_jwt
|
|
|
|
|
|
def decode_token(token: str) -> Optional[Dict[str, Any]]:
|
|
"""Decode and validate JWT token"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
def generate_random_token(length: int = 32) -> str:
|
|
"""Generate a secure random token"""
|
|
return secrets.token_urlsafe(length)
|
|
|
|
|
|
class CredentialEncryption:
|
|
"""Handles encryption/decryption of sensitive credentials (POP3/IMAP passwords)"""
|
|
|
|
def __init__(self, key: Optional[str] = None, user_id: Optional[int] = None):
|
|
"""
|
|
Initialize encryption with a key.
|
|
If no key provided, uses the one from settings.
|
|
In production, use a unique salt per user for enhanced security.
|
|
|
|
Args:
|
|
key: Encryption key (defaults to settings.ENCRYPTION_KEY)
|
|
user_id: Optional user ID for per-user salt generation
|
|
"""
|
|
if key is None:
|
|
key = settings.ENCRYPTION_KEY
|
|
|
|
# Generate salt - unique per user for enhanced security
|
|
if user_id is not None:
|
|
salt = hashlib.sha256(f"pop3fwd_usr_{user_id}".encode()).digest()[:16]
|
|
else:
|
|
# Default salt for system-wide operations (use with caution)
|
|
salt = b"pop3_forwarder_0"
|
|
|
|
# Derive a proper Fernet key from the provided key
|
|
kdf = PBKDF2HMAC(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=salt,
|
|
iterations=100000,
|
|
)
|
|
key_bytes = key.encode("utf-8")
|
|
derived_key = base64.urlsafe_b64encode(kdf.derive(key_bytes))
|
|
self.fernet = Fernet(derived_key)
|
|
|
|
def encrypt(self, plain_text: str) -> str:
|
|
"""Encrypt a string and return base64-encoded ciphertext"""
|
|
encrypted = self.fernet.encrypt(plain_text.encode("utf-8"))
|
|
return base64.b64encode(encrypted).decode("utf-8")
|
|
|
|
def decrypt(self, encrypted_text: str) -> str:
|
|
"""Decrypt a base64-encoded ciphertext"""
|
|
encrypted_bytes = base64.b64decode(encrypted_text.encode("utf-8"))
|
|
decrypted = self.fernet.decrypt(encrypted_bytes)
|
|
return decrypted.decode("utf-8")
|
|
|
|
|
|
# Global encryption instance
|
|
credential_encryptor = CredentialEncryption()
|
|
|
|
|
|
def encrypt_credential(credential: str) -> str:
|
|
"""Convenience function to encrypt a credential"""
|
|
return credential_encryptor.encrypt(credential)
|
|
|
|
|
|
def decrypt_credential(encrypted_credential: str) -> str:
|
|
"""Convenience function to decrypt a credential"""
|
|
return credential_encryptor.decrypt(encrypted_credential)
|