| import base64 |
| import bcrypt |
| import datetime |
| import hashlib |
| import hmac |
| import jwt |
| import secrets |
| import struct |
| import time as _time |
| from typing import Dict, Optional, Tuple |
|
|
| ALGORITHM = "HS256" |
| ACCESS_TOKEN_EXPIRE_HOURS = 24 * 7 |
|
|
|
|
| def _secret() -> str: |
| """Lazy-import to avoid circular import with config.""" |
| from .config import config |
|
|
| return config.JWT_SECRET_KEY |
|
|
|
|
| |
| |
| |
|
|
|
|
| def get_password_hash(password: str) -> str: |
| salt = bcrypt.gensalt() |
| return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8") |
|
|
|
|
| def verify_password(plain: str, hashed: str) -> bool: |
| try: |
| return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) |
| except Exception: |
| return False |
|
|
|
|
| |
| |
| |
|
|
|
|
| def create_access_token( |
| data: dict, expires_delta: Optional[datetime.timedelta] = None |
| ) -> str: |
| payload = data.copy() |
| expire = datetime.datetime.utcnow() + ( |
| expires_delta or datetime.timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS) |
| ) |
| payload["exp"] = expire |
| return jwt.encode(payload, _secret(), algorithm=ALGORITHM) |
|
|
|
|
| def verify_token(token: str) -> Optional[Dict]: |
| try: |
| return jwt.decode(token, _secret(), algorithms=[ALGORITHM]) |
| except jwt.ExpiredSignatureError: |
| return None |
| except jwt.InvalidTokenError: |
| return None |
|
|
|
|
| |
| |
| |
|
|
|
|
| def create_admin_session_token(user_id: int) -> str: |
| """Short-lived (1-2 h) JWT exclusively for admin dashboard sessions.""" |
| from .config import config |
|
|
| return create_access_token( |
| data={"user_id": user_id, "type": "admin_session"}, |
| expires_delta=datetime.timedelta(hours=config.ADMIN_SESSION_EXPIRE_HOURS), |
| ) |
|
|
|
|
| def verify_admin_session_token(token: str) -> Optional[int]: |
| """Return user_id if the token is a valid, unexpired admin session token.""" |
| payload = verify_token(token) |
| if not payload: |
| return None |
| if payload.get("type") != "admin_session": |
| return None |
| return payload.get("user_id") |
|
|
|
|
| def create_admin_reauth_token(user_id: int) -> str: |
| from .config import config |
|
|
| return create_access_token( |
| data={"user_id": user_id, "type": "admin_reauth"}, |
| expires_delta=datetime.timedelta(minutes=config.ADMIN_REAUTH_EXPIRE_MINUTES), |
| ) |
|
|
|
|
| def verify_admin_reauth_token(token: str) -> Optional[int]: |
| payload = verify_token(token) |
| if not payload: |
| return None |
| if payload.get("type") != "admin_reauth": |
| return None |
| return payload.get("user_id") |
|
|
|
|
| |
| |
| |
|
|
|
|
| def generate_totp_secret() -> str: |
| """Generate a base32-encoded TOTP secret.""" |
| return base64.b32encode(secrets.token_bytes(20)).decode("ascii").rstrip("=") |
|
|
|
|
| def verify_totp(secret: str, code: str) -> bool: |
| """Verify a 6-digit TOTP code against a secret using time-step logic.""" |
| try: |
| padding = "=" * ((8 - len(secret) % 8) % 8) |
| key = base64.b32decode((secret + padding).upper(), casefold=True) |
| except Exception: |
| return False |
|
|
| time_step = 30 |
| t = int(_time.time()) // time_step |
|
|
| for offset in range(-1, 2): |
| counter = struct.pack(">Q", t + offset) |
| h = hmac.new(key, counter, hashlib.sha1).digest() |
| offset_byte = h[-1] & 0x0F |
| truncated = ( |
| ((h[offset_byte] & 0x7F) << 24) |
| | ((h[offset_byte + 1] & 0xFF) << 16) |
| | ((h[offset_byte + 2] & 0xFF) << 8) |
| | (h[offset_byte + 3] & 0xFF) |
| ) |
| otp = truncated % (10**6) |
| if str(otp).zfill(6) == str(code).strip(): |
| return True |
| return False |
|
|
|
|
| |
| |
| |
|
|
|
|
| def verify_google_id_token(id_token: str) -> Optional[Dict]: |
| """Verify a Google ID token and return claims dict, or None on failure.""" |
| from .config import config |
|
|
| try: |
| if not config.GOOGLE_CLIENT_ID: |
| return None |
|
|
| from authlib.jose import JsonWebKey, jwt as authlib_jwt |
|
|
| jwk_url = "https://www.googleapis.com/oauth2/v3/certs" |
| import httpx |
|
|
| resp = httpx.get(jwk_url, timeout=10) |
| resp.raise_for_status() |
| jwks = resp.json() |
|
|
| key_set = JsonWebKey.import_key_set(jwks) |
| claims = authlib_jwt.decode(id_token, key_set) |
| claims.validate() |
|
|
| |
| aud = claims.get("aud") |
| if aud != config.GOOGLE_CLIENT_ID: |
| return None |
|
|
| |
| iss = claims.get("iss") |
| if iss not in ("https://accounts.google.com", "accounts.google.com"): |
| return None |
|
|
| return dict(claims) |
| except Exception: |
| return None |
|
|
|
|
| def verify_apple_identity_token(identity_token: str) -> Optional[Dict]: |
| """Verify an Apple identity token and return claims dict, or None on failure.""" |
| from .config import config |
|
|
| try: |
| if not config.APPLE_CLIENT_ID: |
| return None |
|
|
| from authlib.jose import JsonWebKey, jwt as authlib_jwt |
|
|
| jwk_url = "https://appleid.apple.com/auth/keys" |
| import httpx |
|
|
| resp = httpx.get(jwk_url, timeout=10) |
| resp.raise_for_status() |
| jwks = resp.json() |
|
|
| key_set = JsonWebKey.import_key_set(jwks) |
| claims = authlib_jwt.decode(identity_token, key_set) |
| claims.validate() |
|
|
| |
| aud = claims.get("aud") |
| if aud != config.APPLE_CLIENT_ID: |
| return None |
|
|
| |
| iss = claims.get("iss") |
| if iss != "https://appleid.apple.com": |
| return None |
|
|
| return dict(claims) |
| except Exception: |
| return None |
|
|
|
|
| GUEST_UPGRADE_TOKEN_EXPIRE_MINUTES = 60 |
|
|
|
|
| def create_guest_upgrade_token(guest_user_id: int) -> str: |
| """Create a short-lived token the frontend carries across the auth flow.""" |
| return create_access_token( |
| data={"guest_user_id": guest_user_id, "type": "guest_upgrade"}, |
| expires_delta=datetime.timedelta(minutes=GUEST_UPGRADE_TOKEN_EXPIRE_MINUTES), |
| ) |
|
|
|
|
| def verify_guest_upgrade_token(token: str) -> Optional[int]: |
| """Return guest_user_id if the token is valid and unexpired, else None.""" |
| payload = verify_token(token) |
| if not payload: |
| return None |
| if payload.get("type") != "guest_upgrade": |
| return None |
| return payload.get("guest_user_id") |
|
|
|
|
| def claim_guest_for_authenticated_user( |
| db, guest_user_id: int, openwebui_user_id: int, email: Optional[str] |
| ) -> int: |
| """ |
| Merge guest history into the authenticated OpenWebUI user. |
| |
| Two paths: |
| 1. **Same ID** (guest already has the email) → convert in-place. |
| 2. **Different ID** (OpenWebUI auto-provisioned a new row) → migrate |
| conversations/memories/profile from the guest, then delete the guest. |
| |
| Returns openwebui_user_id (unchanged) so the caller can continue. |
| """ |
| if guest_user_id == openwebui_user_id: |
| guest = db.get_user_by_id(guest_user_id) |
| username = guest.get("username") if guest else None |
| if not username: |
| username = email.split("@")[0] if email else f"user_{secrets.token_hex(6)}" |
| if db.get_user_by_username(username): |
| username = f"{username}_{secrets.token_hex(4)}" |
| db.convert_guest_to_oauth(guest_user_id, username, email) |
| else: |
| db.migrate_guest_to_user(guest_user_id, openwebui_user_id) |
| return openwebui_user_id |
|
|
|
|
| def oauth_login_or_register( |
| db, |
| provider: str, |
| provider_user_id: str, |
| email: Optional[str], |
| display_name: Optional[str], |
| is_guest: bool, |
| guest_user_id: Optional[int], |
| ) -> Tuple[int, bool]: |
| """ |
| Find or create a user for an OAuth login. |
| Returns (user_id, is_new_user). |
| |
| Flow: |
| 1. Check oauth_accounts table for existing link |
| 2. If not found, check users table by email for account linking |
| 3. If still not found: |
| - If current user is a guest, convert that guest |
| - Otherwise create a new user |
| 4. Link the OAuth account to the user |
| """ |
| |
| oauth = db.get_oauth_account(provider, provider_user_id) |
| if oauth: |
| target_user_id = oauth["user_id"] |
| if is_guest and guest_user_id and guest_user_id != target_user_id: |
| db.migrate_guest_to_user(guest_user_id, target_user_id) |
| return target_user_id, False |
|
|
| |
| target_user_id = None |
| if email: |
| existing = db.get_user_by_email(email) |
| if existing and not existing["is_guest"]: |
| target_user_id = existing["id"] |
| if is_guest and guest_user_id and guest_user_id != target_user_id: |
| db.migrate_guest_to_user(guest_user_id, target_user_id) |
| if not existing.get("email"): |
| db.update_email(target_user_id, email) |
|
|
| |
| if target_user_id is None: |
| if is_guest and guest_user_id: |
| username = display_name or email or f"user_{secrets.token_hex(6)}" |
| if db.get_user_by_username(username): |
| username = f"{username}_{secrets.token_hex(4)}" |
| success = db.convert_guest_to_oauth(guest_user_id, username, email) |
| if not success: |
| target_user_id = db.create_oauth_user(username, email) |
| is_new = True |
| else: |
| target_user_id = guest_user_id |
| is_new = False |
| else: |
| username = display_name or email or f"user_{secrets.token_hex(6)}" |
| if db.get_user_by_username(username): |
| username = f"{username}_{secrets.token_hex(4)}" |
| target_user_id = db.create_oauth_user(username, email) |
| is_new = True |
| else: |
| is_new = False |
|
|
| |
| db.create_oauth_account(target_user_id, provider, provider_user_id, email) |
|
|
| return target_user_id, is_new |
|
|
