99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from jose import JWTError, jwt
|
|
from passlib.context import CryptContext
|
|
|
|
from core.config import JWT_SECRET_KEY
|
|
|
|
logger = logging.getLogger("app")
|
|
|
|
SECRET_KEY = JWT_SECRET_KEY
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
|
REFRESH_TOKEN_EXPIRE_DAYS = 30
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"])
|
|
|
|
# Verified against when a login names a user that doesn't exist, so the response time
|
|
# is the same either way. Without it, "no such user" returns before bcrypt runs and
|
|
# the timing difference tells an attacker which usernames are real.
|
|
DUMMY_PASSWORD_HASH = pwd_context.hash("not-a-real-password")
|
|
|
|
def hash_password(password: str) -> str:
|
|
return pwd_context.hash(password)
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
return pwd_context.verify(plain, hashed)
|
|
|
|
# All token timestamps are timezone-aware UTC. They used to be naive datetime.now(),
|
|
# which jose serialises as if it were already UTC — with TZ=Europe/Paris that handed
|
|
# out tokens living 1-2 hours longer than intended, drifting again at each DST change.
|
|
#
|
|
# 'iat' is carried so a password reset can invalidate everything issued before it;
|
|
# see token_valid_after on the User model. It is written as a float rather than a
|
|
# datetime: jose truncates datetimes to whole seconds, and at that resolution a token
|
|
# minted just *after* a reset is indistinguishable from one minted just before, so
|
|
# logging in immediately after a password change would be rejected. RFC 7519 permits
|
|
# a non-integer NumericDate.
|
|
|
|
def create_access_token(user_id: str) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
return jwt.encode(
|
|
{
|
|
"sub": user_id,
|
|
"iat": now.timestamp(),
|
|
"exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
"type": "access",
|
|
},
|
|
SECRET_KEY, algorithm=ALGORITHM,
|
|
)
|
|
|
|
def create_refresh_token(user_id: str) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
return jwt.encode(
|
|
{
|
|
"sub": user_id,
|
|
"iat": now.timestamp(),
|
|
"exp": now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
|
|
"type": "refresh",
|
|
},
|
|
SECRET_KEY, algorithm=ALGORITHM,
|
|
)
|
|
|
|
def _decode(token: str, expected_type: str) -> dict | None:
|
|
"""Verified payload, or None if the signature, expiry or type is wrong."""
|
|
try:
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
|
except JWTError:
|
|
return None
|
|
if payload.get("type") != expected_type:
|
|
return None
|
|
return payload
|
|
|
|
def decode_refresh_token_payload(token: str) -> dict | None:
|
|
return _decode(token, "refresh")
|
|
|
|
def decode_access_token_payload(token: str) -> dict | None:
|
|
return _decode(token, "access")
|
|
|
|
def decode_refresh_token(token: str) -> str | None:
|
|
payload = _decode(token, "refresh")
|
|
return payload.get("sub") if payload else None
|
|
|
|
def decode_access_token(token: str) -> str | None:
|
|
payload = _decode(token, "access")
|
|
return payload.get("sub") if payload else None
|
|
|
|
def token_issued_before(payload: dict, cutoff: datetime | None) -> bool:
|
|
"""True if this token predates a revocation cutoff (i.e. should be rejected)."""
|
|
if cutoff is None:
|
|
return False
|
|
iat = payload.get("iat")
|
|
# Tokens minted before 'iat' was introduced have no issue time, so they can't be
|
|
# proven to postdate the cutoff — treat them as revoked rather than trusted.
|
|
if iat is None:
|
|
return True
|
|
if cutoff.tzinfo is None:
|
|
cutoff = cutoff.replace(tzinfo=timezone.utc)
|
|
return datetime.fromtimestamp(iat, timezone.utc) < cutoff |