copilot-swe-agent[bot]
f439f887d0
Backend fixes: - Fix JWT sub claim: encode as str(user.id), decode with int() cast (python-jose requirement) - Replace all deprecated datetime.utcnow() with datetime.now(timezone.utc) - Replace deprecated FastAPI @app.on_event() with modern lifespan context manager - Replace deprecated Pydantic class Config with model_config = ConfigDict(...) - Replace deprecated Pydantic .dict() with .model_dump() - Fix overly broad except (GmailInjectionError, Exception) → except Exception - Remove unused GmailInjectionError import - Fix TokenPayload schema sub field type from int to str Frontend: - Create frontend/src/lib/api.ts — API client module with auth, user, mail accounts, processing runs APIs - Add !frontend/src/lib/ to .gitignore negation Tests: - Add 3 new JWT tests (sub string encoding, access token type, refresh token type) - Update test_token_payload_schema for string sub claim - All 128 tests pass Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> Agent-Logs-Url: https://github.com/christianlouis/pop_puller_to_gmail/sessions/e0b13eb0-8de7-4f02-81e4-e202cbba4608
138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
"""
|
|
Security utilities for encryption, hashing, and token generation.
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional, Dict, Any
|
|
import bcrypt
|
|
from jose import JWTError, jwt
|
|
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
|
|
|
|
|
|
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(
|
|
data: Dict[str, Any], expires_delta: Optional[timedelta] = None
|
|
) -> str:
|
|
"""Create JWT access token"""
|
|
to_encode = data.copy()
|
|
|
|
if expires_delta:
|
|
expire = datetime.now(timezone.utc) + expires_delta
|
|
else:
|
|
expire = datetime.now(timezone.utc) + 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.now(timezone.utc) + 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)
|