"""Security helpers: JWT issue/verify, password hashing. No credential encryption here - target-system credentials live exclusively on the satellites, never in the central database. """ from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any import bcrypt from jose import JWTError, jwt from app.core.config import get_settings from app.core.exceptions import InvalidTokenError settings = get_settings() def hash_password(password: str) -> str: # bcrypt hard limit: 72 bytes return bcrypt.hashpw(password.encode()[:72], bcrypt.gensalt()).decode() def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(plain.encode()[:72], hashed.encode()) except ValueError: return False def _read_key(path: str) -> str | None: p = Path(path) return p.read_text() if p.exists() else None def create_access_token(subject: str, extra_claims: dict[str, Any] | None = None) -> str: expire = datetime.now(UTC) + timedelta(minutes=settings.jwt_access_token_expire_minutes) claims: dict[str, Any] = {"sub": subject, "exp": expire, "type": "access"} if extra_claims: claims.update(extra_claims) private_key = _read_key(settings.jwt_private_key_path) if private_key and settings.jwt_algorithm == "RS256": return jwt.encode(claims, private_key, algorithm="RS256") # Dev fallback when no key pair exists yet return jwt.encode(claims, settings.secret_key, algorithm="HS256") def decode_token(token: str) -> dict[str, Any]: """Decode and validate a JWT. Raises InvalidTokenError on failure.""" try: public_key = _read_key(settings.jwt_public_key_path) if public_key and settings.jwt_algorithm == "RS256": return jwt.decode(token, public_key, algorithms=["RS256"]) # type: ignore[no-any-return] return jwt.decode(token, settings.secret_key, algorithms=["HS256"]) # type: ignore[no-any-return] except JWTError as exc: raise InvalidTokenError("Token is invalid or expired") from exc