"""Password hashing + JWT create/verify. Vendored from OWC `auth/security.py` (snapshot 2026-04-20). YPMS additions: `create_access_token()` bakes in YPMS-specific claims — `uid`, `roles`, `iam.assurance_level`, `mfa_verified` — so downstream authorization can gate high-risk actions above base RBAC (see ARCHITECTURE.md §4.5, §8.2). `iam.assurance_level` values are defined in `ypms.users.models` and follow NIST 800-63 IAL/AAL semantics: `pre_application` < `standard` < `payroll_authorized` < `staff` < `admin`. Payroll-touching endpoints will require `payroll_authorized` or higher (Phase 7+). pbkdf2_sha256 not bcrypt — see ARCHITECTURE.md §4.6. JWT library: PyJWT, not OWC's python-jose (milestone 0.8). python-jose hard- depends on `ecdsa`, which carries an unfixable advisory (PYSEC-2026-1325) that failed the pip-audit gate. YPMS only signs HS256. Flag to OWC. """ from __future__ import annotations from datetime import UTC, datetime, timedelta from typing import Any import jwt from passlib.context import CryptContext from ypms.settings import get_settings _pwd = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto") def hash_password(pw: str) -> str: h: str = _pwd.hash(pw) return h def verify_password(pw: str, hashed: str) -> bool: return bool(_pwd.verify(pw, hashed)) def create_access_token( *, sub: str, uid: int, roles: list[str], assurance_level: str, mfa_verified: bool, ) -> str: s = get_settings() now = datetime.now(UTC) payload: dict[str, Any] = { "sub": sub, "uid": uid, "roles": roles, "iam.assurance_level": assurance_level, "mfa_verified": mfa_verified, "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=s.access_token_ttl_minutes)).timestamp()), "type": "access", } token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg) return token def create_refresh_token(*, sub: str, uid: int) -> str: s = get_settings() now = datetime.now(UTC) payload: dict[str, Any] = { "sub": sub, "uid": uid, "iat": int(now.timestamp()), "exp": int((now + timedelta(days=s.refresh_token_ttl_days)).timestamp()), "type": "refresh", } token: str = jwt.encode(payload, s.jwt_secret, algorithm=s.jwt_alg) return token def decode_token(token: str) -> dict[str, Any]: s = get_settings() try: payload: dict[str, Any] = jwt.decode(token, s.jwt_secret, algorithms=[s.jwt_alg]) except jwt.PyJWTError as e: raise ValueError("invalid token") from e return payload