Fix CI failures: add backend/conftest.py for module resolution and run black formatting
- 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
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
"""
|
||||
Security utilities for encryption, hashing, and token generation.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import secrets
|
||||
from datetime import datetime, timedelta
|
||||
@@ -14,7 +15,6 @@ import base64
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
# Password hashing context
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
@@ -29,17 +29,23 @@ def get_password_hash(password: str) -> str:
|
||||
return pwd_context.hash(password)
|
||||
|
||||
|
||||
def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
|
||||
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)
|
||||
|
||||
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)
|
||||
encoded_jwt = jwt.encode(
|
||||
to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
|
||||
)
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
@@ -48,14 +54,18 @@ def create_refresh_token(data: Dict[str, Any]) -> str:
|
||||
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)
|
||||
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])
|
||||
payload = jwt.decode(
|
||||
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
|
||||
)
|
||||
return payload
|
||||
except JWTError:
|
||||
return None
|
||||
@@ -68,27 +78,27 @@ def generate_random_token(length: int = 32) -> str:
|
||||
|
||||
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]
|
||||
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'
|
||||
|
||||
salt = b"pop3_forwarder_0"
|
||||
|
||||
# Derive a proper Fernet key from the provided key
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
@@ -96,20 +106,20 @@ class CredentialEncryption:
|
||||
salt=salt,
|
||||
iterations=100000,
|
||||
)
|
||||
key_bytes = key.encode('utf-8')
|
||||
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')
|
||||
|
||||
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'))
|
||||
encrypted_bytes = base64.b64decode(encrypted_text.encode("utf-8"))
|
||||
decrypted = self.fernet.decrypt(encrypted_bytes)
|
||||
return decrypted.decode('utf-8')
|
||||
return decrypted.decode("utf-8")
|
||||
|
||||
|
||||
# Global encryption instance
|
||||
|
||||
Reference in New Issue
Block a user