chwellofficial/nt360Slides

1import base642import hashlib3import hmac4import json5import os6import secrets7import time8from typing import Optional9 10from fastapi import Request11from starlette.responses import Response12 13from utils.get_env import get_user_config_path_env14 15SESSION_COOKIE_NAME = "presenton_session"16PBKDF2_ITERATIONS = 200_00017SESSION_TTL_SECONDS = 60 * 60 * 24 * 3018 19 20def _base64url_encode(data: bytes) -> str:21    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("utf-8")22 23 24def _base64url_decode(value: str) -> bytes:25    padded = value + "=" * (-len(value) % 4)26    return base64.urlsafe_b64decode(padded.encode("utf-8"))27 28 29def _load_user_config() -> dict:30    user_config_path = get_user_config_path_env()31    if not user_config_path or not os.path.exists(user_config_path):32        return {}33 34    try:35        with open(user_config_path, "r", encoding="utf-8") as config_file:36            data = json.load(config_file)37            return data if isinstance(data, dict) else {}38    except Exception:39        return {}40 41 42def _save_user_config(config: dict) -> None:43    user_config_path = get_user_config_path_env()44    if not user_config_path:45        raise ValueError("USER_CONFIG_PATH is not set")46 47    os.makedirs(os.path.dirname(user_config_path), exist_ok=True)48    with open(user_config_path, "w", encoding="utf-8") as config_file:49        json.dump(config, config_file)50 51 52def _hash_password(password: str, salt: bytes) -> bytes:53    return hashlib.pbkdf2_hmac(54        "sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS55    )56 57 58def _encode_password_hash(password: str) -> str:59    salt = secrets.token_bytes(16)60    digest = _hash_password(password, salt)61    salt_encoded = _base64url_encode(salt)62    digest_encoded = _base64url_encode(digest)63    return (64        f"pbkdf2_sha256${PBKDF2_ITERATIONS}${salt_encoded}${digest_encoded}"65    )66 67 68def _verify_password_hash(password: str, encoded_hash: str) -> bool:69    try:70        algorithm, iterations_str, salt_encoded, digest_encoded = encoded_hash.split("$")71        if algorithm != "pbkdf2_sha256":72            return False73 74        iterations = int(iterations_str)75        salt = _base64url_decode(salt_encoded)76        expected_digest = _base64url_decode(digest_encoded)77        actual_digest = hashlib.pbkdf2_hmac(78            "sha256", password.encode("utf-8"), salt, iterations79        )80        return hmac.compare_digest(actual_digest, expected_digest)81    except Exception:82        return False83 84 85def _get_or_create_auth_secret(config: dict) -> str:86    secret = config.get("AUTH_SECRET_KEY")87    if secret:88        return secret89 90    secret = _base64url_encode(secrets.token_bytes(32))91    config["AUTH_SECRET_KEY"] = secret92    _save_user_config(config)93    return secret94 95 96def is_auth_configured() -> bool:97    config = _load_user_config()98    return bool(config.get("AUTH_USERNAME") and config.get("AUTH_PASSWORD_HASH"))99 100 101def get_configured_auth_username() -> Optional[str]:102    config = _load_user_config()103    username = config.get("AUTH_USERNAME")104    if isinstance(username, str) and username.strip():105        return username.strip()106    return None107 108 109def setup_initial_credentials(username: str, password: str) -> None:110    cleaned_username = (username or "").strip()111    if len(cleaned_username) < 3:112        raise ValueError("Username must be at least 3 characters")113 114    if len(password or "") < 6:115        raise ValueError("Password must be at least 6 characters")116 117    config = _load_user_config()118    if config.get("AUTH_USERNAME") and config.get("AUTH_PASSWORD_HASH"):119        raise ValueError("Credentials already configured")120 121    config["AUTH_USERNAME"] = cleaned_username122    config["AUTH_PASSWORD_HASH"] = _encode_password_hash(password)123    _get_or_create_auth_secret(config)124    _save_user_config(config)125 126 127def force_set_credentials(username: str, password: str) -> None:128    """Overwrite stored credentials; used by env-based preseed/override."""129    cleaned_username = (username or "").strip()130    if len(cleaned_username) < 3:131        raise ValueError("Username must be at least 3 characters")132 133    if len(password or "") < 6:134        raise ValueError("Password must be at least 6 characters")135 136    config = _load_user_config()137    config["AUTH_USERNAME"] = cleaned_username138    config["AUTH_PASSWORD_HASH"] = _encode_password_hash(password)139    # Rotate the signing secret so any previously-issued tokens stop validating.140    config["AUTH_SECRET_KEY"] = _base64url_encode(secrets.token_bytes(32))141    _save_user_config(config)142 143 144def clear_stored_credentials() -> None:145    """Remove stored credentials; next boot will request setup again."""146    config = _load_user_config()147    removed = False148    for key in ("AUTH_USERNAME", "AUTH_PASSWORD_HASH", "AUTH_SECRET_KEY"):149        if key in config:150            config.pop(key, None)151            removed = True152    if removed:153        _save_user_config(config)154 155 156def verify_credentials(username: str, password: str) -> bool:157    config = _load_user_config()158    stored_username = config.get("AUTH_USERNAME")159    stored_hash = config.get("AUTH_PASSWORD_HASH")160 161    if not stored_username or not stored_hash:162        return False163 164    cleaned_username = (username or "").strip()165    if not hmac.compare_digest(cleaned_username, stored_username):166        return False167 168    return _verify_password_hash(password or "", stored_hash)169 170 171def _sign_payload(payload_encoded: str, secret: str) -> str:172    signature = hmac.new(173        secret.encode("utf-8"), payload_encoded.encode("utf-8"), hashlib.sha256174    ).digest()175    return _base64url_encode(signature)176 177 178def create_session_token(username: str) -> str:179    config = _load_user_config()180    secret = _get_or_create_auth_secret(config)181 182    issued_at = int(time.time())183    payload = {184        "v": 1,185        "u": username,186        "iat": issued_at,187        "exp": issued_at + SESSION_TTL_SECONDS,188    }189 190    payload_encoded = _base64url_encode(191        json.dumps(payload, separators=(",", ":")).encode("utf-8")192    )193    signature_encoded = _sign_payload(payload_encoded, secret)194    return f"{payload_encoded}.{signature_encoded}"195 196 197def validate_session_token(token: Optional[str]) -> Optional[str]:198    if not token:199        return None200 201    config = _load_user_config()202    stored_username = config.get("AUTH_USERNAME")203    if not stored_username:204        return None205 206    secret = config.get("AUTH_SECRET_KEY")207    if not secret:208        return None209 210    try:211        payload_encoded, signature_encoded = token.split(".", 1)212    except ValueError:213        return None214 215    expected_signature = _sign_payload(payload_encoded, secret)216    if not hmac.compare_digest(signature_encoded, expected_signature):217        return None218 219    try:220        payload_raw = _base64url_decode(payload_encoded)221        payload = json.loads(payload_raw)222    except Exception:223        return None224 225    username = payload.get("u")226    version = payload.get("v")227    expires_at = payload.get("exp")228    if not isinstance(username, str) or not isinstance(expires_at, int):229        return None230 231    if version != 1:232        return None233 234    if not hmac.compare_digest(username, stored_username):235        return None236 237    if expires_at < int(time.time()):238        return None239 240    return username241 242 243def get_session_token_from_request(request: Request) -> Optional[str]:244    cookie_token = request.cookies.get(SESSION_COOKIE_NAME)245    if cookie_token:246        return cookie_token247 248    auth_header = request.headers.get("Authorization", "")249    if auth_header.lower().startswith("bearer "):250        return auth_header[7:].strip() or None251 252    return None253 254 255def get_basic_auth_credentials_from_request(256    request: Request,257) -> Optional[tuple[str, str]]:258    auth_header = request.headers.get("Authorization", "")259    if not auth_header.lower().startswith("basic "):260        return None261 262    encoded_value = auth_header[6:].strip()263    if not encoded_value:264        return None265 266    try:267        decoded_value = base64.b64decode(encoded_value).decode("utf-8")268    except Exception:269        return None270 271    if ":" not in decoded_value:272        return None273 274    username, password = decoded_value.split(":", 1)275    return username, password276 277 278def get_auth_status(session_token: Optional[str] = None) -> dict:279    config = _load_user_config()280    configured = bool(config.get("AUTH_USERNAME") and config.get("AUTH_PASSWORD_HASH"))281 282    if not configured:283        return {284            "configured": False,285            "authenticated": False,286            "username": None,287        }288 289    username = validate_session_token(session_token)290    return {291        "configured": True,292        "authenticated": bool(username),293        "username": username,294    }295 296 297def _is_secure_request(request: Request) -> bool:298    forwarded_proto = request.headers.get("x-forwarded-proto", "")299    if forwarded_proto.lower() == "https":300        return True301    return request.url.scheme == "https"302 303 304def set_session_cookie(response: Response, token: str, request: Request) -> None:305    response.set_cookie(306        key=SESSION_COOKIE_NAME,307        value=token,308        max_age=SESSION_TTL_SECONDS,309        httponly=True,310        secure=_is_secure_request(request),311        samesite="lax",312        path="/",313    )314 315 316def clear_session_cookie(response: Response, request: Request) -> None:317    response.delete_cookie(318        key=SESSION_COOKIE_NAME,319        httponly=True,320        secure=_is_secure_request(request),321        samesite="lax",322        path="/",323    )324