import secrets import hashlib from datetime import datetime, timedelta, timezone from typing import Optional, Any import jwt import bcrypt from app.core.config import settings def get_password_hash(password: str) -> str: """Hashes a password with bcrypt.""" salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8") def verify_password(plain_password: str, hashed_password: str) -> bool: """Verifies a plain password against a bcrypt hash.""" try: return bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8")) except Exception: return False def create_access_token(subject: Any, role: str = "ADMIN", expires_delta: Optional[timedelta] = None) -> str: """Creates a JWT access token for authentication.""" 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 = { "exp": expire, "sub": str(subject), "role": role, "iat": datetime.now(timezone.utc) } encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def decode_access_token(token: str) -> Optional[dict]: """Decodes and validates a JWT token.""" try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) return payload except Exception: return None def generate_registration_code() -> str: """Generates an agent registration code formatted as OED-XXXX-XXXX.""" part1 = secrets.token_hex(2).upper() part2 = secrets.token_hex(2).upper() return f"OED-{part1}-{part2}" def generate_device_token() -> str: """Generates a high-entropy secret token for an agent device.""" return f"oed_sec_{secrets.token_urlsafe(32)}" def hash_token(token: str) -> str: """Returns the SHA-256 hex digest of a token for secure database lookup.""" return hashlib.sha256(token.encode("utf-8")).hexdigest()