127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from cryptography import x509
|
|
from pwdlib import PasswordHash
|
|
|
|
|
|
password_hash = PasswordHash.recommended()
|
|
|
|
JWT_SECRET = os.environ["JWT_SECRET"]
|
|
ACCESS_TOKEN_MINUTES = int(os.environ.get("ACCESS_TOKEN_MINUTES", "15"))
|
|
REFRESH_TOKEN_DAYS = int(os.environ.get("REFRESH_TOKEN_DAYS", "30"))
|
|
|
|
JWT_ALGORITHM = "HS256"
|
|
FIREBASE_CERTS_URL = "https://www.googleapis.com/robot/v1/metadata/x509/securetoken@system.gserviceaccount.com"
|
|
_firebase_certs: dict[str, str] = {}
|
|
_firebase_certs_expires_at = 0.0
|
|
|
|
|
|
def normalize_email(email: str) -> str:
|
|
return email.strip().lower()
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return password_hash.hash(password)
|
|
|
|
|
|
def verify_password(password: str, hashed: str) -> bool:
|
|
return password_hash.verify(password, hashed)
|
|
|
|
|
|
def create_access_token(
|
|
user_id: uuid.UUID,
|
|
session_id: uuid.UUID,
|
|
) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"sid": str(session_id),
|
|
"iat": now,
|
|
"exp": now + timedelta(minutes=ACCESS_TOKEN_MINUTES),
|
|
"type": "access",
|
|
}
|
|
|
|
return jwt.encode(
|
|
payload,
|
|
JWT_SECRET,
|
|
algorithm=JWT_ALGORITHM,
|
|
)
|
|
|
|
|
|
def create_refresh_token() -> tuple[str, str]:
|
|
raw = secrets.token_urlsafe(64)
|
|
|
|
digest = hashlib.sha256(
|
|
raw.encode("utf-8")
|
|
).hexdigest()
|
|
|
|
return raw, digest
|
|
|
|
|
|
def hash_refresh_token(token: str) -> str:
|
|
return hashlib.sha256(
|
|
token.encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def create_one_time_token() -> tuple[str, str]:
|
|
raw = secrets.token_urlsafe(48)
|
|
return raw, hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def hash_one_time_token(token: str) -> str:
|
|
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def refresh_expiry() -> datetime:
|
|
return datetime.now(timezone.utc) + timedelta(
|
|
days=REFRESH_TOKEN_DAYS
|
|
)
|
|
|
|
|
|
def decode_access_token(token: str) -> dict:
|
|
return jwt.decode(
|
|
token,
|
|
JWT_SECRET,
|
|
algorithms=[JWT_ALGORITHM],
|
|
)
|
|
|
|
|
|
def _load_firebase_certs() -> dict[str, str]:
|
|
global _firebase_certs, _firebase_certs_expires_at
|
|
now = time.time()
|
|
if _firebase_certs and now < _firebase_certs_expires_at:
|
|
return _firebase_certs
|
|
with urllib.request.urlopen(FIREBASE_CERTS_URL, timeout=5) as response:
|
|
certs = json.loads(response.read())
|
|
cache_control = response.headers.get("Cache-Control", "")
|
|
max_age = next((int(value.split("=", 1)[1]) for value in cache_control.split(",") if value.strip().startswith("max-age=")), 300)
|
|
_firebase_certs = certs
|
|
_firebase_certs_expires_at = now + max_age
|
|
return certs
|
|
|
|
|
|
def decode_firebase_id_token(token: str) -> dict:
|
|
project_id = os.environ.get("FIREBASE_PROJECT_ID", "motoapp-d22e0")
|
|
header = jwt.get_unverified_header(token)
|
|
if header.get("alg") != "RS256" or not header.get("kid"):
|
|
raise jwt.InvalidTokenError("Invalid Firebase token header")
|
|
certificate_pem = _load_firebase_certs().get(header["kid"])
|
|
if certificate_pem is None:
|
|
raise jwt.InvalidTokenError("Unknown Firebase signing key")
|
|
public_key = x509.load_pem_x509_certificate(certificate_pem.encode("ascii")).public_key()
|
|
claims = jwt.decode(token, public_key, algorithms=["RS256"], audience=project_id, issuer=f"https://securetoken.google.com/{project_id}")
|
|
subject = claims.get("sub")
|
|
if not isinstance(subject, str) or not subject or len(subject) > 128:
|
|
raise jwt.InvalidTokenError("Invalid Firebase subject")
|
|
return claims
|