unsloth/studio/backend/tests/test_desktop_auth.py

import importlib.util
import asyncio
import hashlib
import json
import os
import platform
import secrets
import sqlite3
import subprocess
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace

import jwt
import pytest
from fastapi import APIRouter, FastAPI
from fastapi.security import HTTPAuthorizationCredentials
from fastapi.testclient import TestClient

from auth import storage


@pytest.fixture(autouse = True)
def isolated_auth_db(tmp_path, monkeypatch):
    monkeypatch.setattr(storage, "DB_PATH", tmp_path / "auth.db")
    monkeypatch.setattr(storage, "_BOOTSTRAP_PW_PATH", tmp_path / ".bootstrap_password")
    monkeypatch.setattr(storage, "_bootstrap_password", None)
    monkeypatch.setattr(storage, "_api_key_pbkdf2_salt_cache", None)
    yield


def seed_user(*, must_change_password = False):
    storage.create_initial_user(
        username = storage.DEFAULT_ADMIN_USERNAME,
        password = "human-password-123",
        jwt_secret = secrets.token_urlsafe(64),
        must_change_password = must_change_password,
    )


def auth_client():
    route_path = Path(__file__).resolve().parents[1] / "routes" / "auth.py"
    spec = importlib.util.spec_from_file_location("_desktop_auth_route", route_path)
    auth_route = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(auth_route)

    app = FastAPI()
    app.include_router(auth_route.router, prefix = "/api/auth")
    return TestClient(app)


def data_recipe_jobs_module():
    route_path = Path(__file__).resolve().parents[1] / "routes" / "data_recipe" / "jobs.py"
    spec = importlib.util.spec_from_file_location("_desktop_data_recipe_jobs", route_path)
    jobs_route = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(jobs_route)
    return jobs_route


def local_recipe():
    return {
        "model_providers": [{"name": "local", "is_local": True}],
        "model_configs": [{"alias": "local-model", "provider": "local"}],
        "columns": [{"column_type": "llm-text", "model_alias": "local-model"}],
    }


def local_recipe_request(token):
    return SimpleNamespace(
        headers = {"authorization": f"Bearer {token}"},
        app = SimpleNamespace(state = SimpleNamespace(server_port = 8888)),
        scope = {},
        base_url = "http://testserver/",
    )


@pytest.fixture
def loaded_local_model(monkeypatch):
    inference_module = SimpleNamespace(
        get_llama_cpp_backend = lambda: SimpleNamespace(is_loaded = True),
    )
    monkeypatch.setitem(sys.modules, "routes.inference", inference_module)


def test_desktop_secret_round_trip_uses_real_admin_subject():
    seed_user()
    raw = storage.create_desktop_secret()

    assert raw.startswith("desktop-")
    assert storage.validate_desktop_secret(raw) == storage.DEFAULT_ADMIN_USERNAME
    assert storage.validate_desktop_secret(raw + "x") is None


def test_create_desktop_secret_rotates_old_secret():
    seed_user()
    old = storage.create_desktop_secret()
    new = storage.create_desktop_secret()

    assert old != new
    assert storage.validate_desktop_secret(old) is None
    assert storage.validate_desktop_secret(new) == storage.DEFAULT_ADMIN_USERNAME


def test_clear_desktop_secret_invalidates_secret():
    seed_user()
    raw = storage.create_desktop_secret()

    storage.clear_desktop_secret()

    assert storage.validate_desktop_secret(raw) is None


def test_ensure_default_admin_does_not_recreate_bootstrap_for_existing_admin():
    seed_user()

    created = storage.ensure_default_admin()

    assert created is False
    assert not storage._BOOTSTRAP_PW_PATH.exists()


def test_ensure_default_admin_loads_existing_bootstrap_after_restart(monkeypatch):
    created = storage.ensure_default_admin()
    bootstrap_pw = storage._BOOTSTRAP_PW_PATH.read_text(encoding = "utf-8").strip()

    monkeypatch.setattr(storage, "_bootstrap_password", None)
    created_again = storage.ensure_default_admin()

    assert created is True
    assert storage._BOOTSTRAP_PW_PATH.exists()
    assert created_again is False
    assert storage.get_bootstrap_password() == bootstrap_pw


def test_bootstrap_password_file_ends_with_a_newline():
    # Otherwise `cat` welds the passphrase onto the shell prompt.
    storage.ensure_default_admin()

    # Bytes: read_text would decode CRLF back to "\n" and hide a CR.
    raw = storage._BOOTSTRAP_PW_PATH.read_bytes()

    assert raw == storage.get_bootstrap_password().encode("utf-8") + b"\n"


def test_bootstrap_password_round_trips_across_a_restart_with_the_newline():
    storage.ensure_default_admin()
    original = storage.get_bootstrap_password()

    storage._bootstrap_password = None

    assert storage.generate_bootstrap_password() == original


def test_upgrade_normalises_the_bootstrap_file():
    # Upgrade path: the admin row exists, so generate_bootstrap_password() never runs.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    storage.ensure_default_admin()

    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret\n"
    assert storage.get_bootstrap_password() == "legacy-bootstrap-secret"


@pytest.mark.parametrize(
    "other",
    [
        b"legacy-bootstrap-secret\r\n",  # only an unreleased build wrote this
        b"legacy-bootstrap-secret\r",
        b"legacy-bootstrap-secret   ",
    ],
)
def test_only_an_exactly_unterminated_bootstrap_file_is_touched(other):
    # Appending is safe only because it is restricted to the one released shape.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(other)

    storage.ensure_default_admin()

    assert storage.get_bootstrap_password() == "legacy-bootstrap-secret"
    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == other


def test_upgrade_normalises_when_the_admin_row_is_missing():
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    assert storage.generate_bootstrap_password() == "legacy-bootstrap-secret"
    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret\n"


def test_a_well_formed_bootstrap_file_is_not_rewritten():
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret\n")
    mtime = storage._BOOTSTRAP_PW_PATH.stat().st_mtime_ns

    storage.ensure_default_admin()

    assert storage._BOOTSTRAP_PW_PATH.stat().st_mtime_ns == mtime


def test_migration_failure_does_not_break_startup(monkeypatch):
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    real_open = storage.os.open

    def refuse(path, flags, *args, **kwargs):
        if str(path) == str(storage._BOOTSTRAP_PW_PATH):
            raise PermissionError("read-only auth dir")
        return real_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(storage.os, "open", refuse)

    storage.ensure_default_admin()

    assert storage.get_bootstrap_password() == "legacy-bootstrap-secret"
    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret"


def test_normalising_never_recreates_a_cleared_bootstrap_file(monkeypatch):
    # A rename would resurrect revoked plaintext if the password changed after the read.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    real_open = storage.os.open

    def clear_then_open(path, flags, *args, **kwargs):
        if str(path) == str(storage._BOOTSTRAP_PW_PATH):
            storage._BOOTSTRAP_PW_PATH.unlink(missing_ok = True)
        return real_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(storage.os, "open", clear_then_open)

    assert storage._read_persisted_bootstrap_password() == "legacy-bootstrap-secret"
    assert not storage._BOOTSTRAP_PW_PATH.exists()


def test_normalising_does_not_overwrite_a_rotated_bootstrap_file(monkeypatch):
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    real_open = storage.os.open

    def rotate_then_open(path, flags, *args, **kwargs):
        if str(path) == str(storage._BOOTSTRAP_PW_PATH):
            storage._BOOTSTRAP_PW_PATH.write_bytes(b"brand-new-secret\n")
        return real_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(storage.os, "open", rotate_then_open)

    storage._read_persisted_bootstrap_password()

    # The append may add a second newline; the rotated credential must survive.
    raw = storage._BOOTSTRAP_PW_PATH.read_bytes()
    assert raw.strip() == b"brand-new-secret"
    storage._bootstrap_password = None
    assert storage._load_bootstrap_password() == "brand-new-secret"


def test_leading_whitespace_bootstrap_file_is_left_alone(monkeypatch):
    # An in-place rewrite is not atomic, so only the exact unterminated shape is touched.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"  legacy-bootstrap-secret  ")

    storage.ensure_default_admin()

    assert storage.get_bootstrap_password() == "legacy-bootstrap-secret"
    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"  legacy-bootstrap-secret  "


def test_normalising_opens_the_file_in_binary_mode(monkeypatch):
    # Without O_BINARY, Windows text mode turns the written LF back into CRLF.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")
    monkeypatch.setattr(storage.os, "O_BINARY", 0x8000, raising = False)
    seen = []
    real_open = storage.os.open

    def spy(path, flags, *args, **kwargs):
        if str(path) == str(storage._BOOTSTRAP_PW_PATH):
            seen.append(flags)
        return real_open(path, flags & ~0x8000, *args, **kwargs)

    monkeypatch.setattr(storage.os, "open", spy)

    storage.ensure_default_admin()

    assert seen and all(f & 0x8000 for f in seen), seen


def test_clearing_by_truncation_mid_normalisation_is_not_undone(monkeypatch):
    # clear_bootstrap_password() truncates through its own descriptor when the unlink
    # fails (Windows, while ours is open); the append must not restore the plaintext.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")

    real_open = storage.os.open

    def truncate_then_open(path, flags, *args, **kwargs):
        fd = real_open(path, flags, *args, **kwargs)
        if str(path) == str(storage._BOOTSTRAP_PW_PATH):
            storage._BOOTSTRAP_PW_PATH.write_text("", encoding = "utf-8")
        return fd

    monkeypatch.setattr(storage.os, "open", truncate_then_open)

    storage._read_persisted_bootstrap_password()

    # A lone newline over a cleared file still reads back as no password.
    assert storage._BOOTSTRAP_PW_PATH.read_bytes().strip() == b""
    storage._bootstrap_password = None
    assert storage._load_bootstrap_password() is None


def test_normalising_works_without_fchmod(monkeypatch):
    # os.fchmod only reached Windows in 3.13; its absence must not raise.
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret")
    monkeypatch.delattr(storage.os, "fchmod", raising = False)

    storage.ensure_default_admin()

    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret\n"
    assert storage.get_bootstrap_password() == "legacy-bootstrap-secret"


def test_persisting_the_bootstrap_password_is_atomic(monkeypatch, tmp_path):
    # A partial write would destroy the only plaintext recovery credential.
    storage._persist_bootstrap_password("original-secret")

    def boom(src, dst):
        raise OSError("crash before replace")

    monkeypatch.setattr(storage.os, "replace", boom)
    with pytest.raises(OSError):
        storage._persist_bootstrap_password("new-secret")

    assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"original-secret\n"
    leftovers = [
        p.name
        for p in storage._BOOTSTRAP_PW_PATH.parent.iterdir()
        if "bootstrap_password." in p.name
    ]
    assert leftovers == []


def test_ensure_default_admin_does_not_generate_for_empty_existing_bootstrap():
    seed_user()
    storage._BOOTSTRAP_PW_PATH.write_text(" \n", encoding = "utf-8")

    created = storage.ensure_default_admin()

    assert created is False
    assert storage._BOOTSTRAP_PW_PATH.read_text(encoding = "utf-8") == " \n"
    assert storage.get_bootstrap_password() is None


def test_web_login_token_has_no_desktop_marker_and_keeps_password_gate():
    seed_user(must_change_password = True)
    client = auth_client()

    response = client.post(
        "/api/auth/login",
        json = {
            "username": storage.DEFAULT_ADMIN_USERNAME,
            "password": "human-password-123",
        },
    )

    assert response.status_code == 200
    body = response.json()
    assert body["must_change_password"] is True
    payload = jwt.decode(
        body["access_token"],
        storage.get_jwt_secret(storage.DEFAULT_ADMIN_USERNAME),
        algorithms = ["HS256"],
    )
    assert payload["sub"] == storage.DEFAULT_ADMIN_USERNAME
    assert "desktop" not in payload

    gated = client.post(
        "/api/auth/api-keys",
        headers = {"Authorization": f"Bearer {body['access_token']}"},
        json = {"name": "web"},
    )
    assert gated.status_code == 403


def test_desktop_login_mints_admin_token_without_clearing_web_password_change():
    seed_user(must_change_password = True)
    raw = storage.create_desktop_secret()
    client = auth_client()

    response = client.post("/api/auth/desktop-login", json = {"secret": raw})

    assert response.status_code == 200
    body = response.json()
    assert body["access_token"]
    assert body["refresh_token"]
    assert body["token_type"] == "bearer"
    assert body["must_change_password"] is False
    assert storage.requires_password_change(storage.DEFAULT_ADMIN_USERNAME) is True

    payload = jwt.decode(
        body["access_token"],
        storage.get_jwt_secret(storage.DEFAULT_ADMIN_USERNAME),
        algorithms = ["HS256"],
    )
    assert payload["sub"] == storage.DEFAULT_ADMIN_USERNAME
    assert payload["desktop"] is True


def test_desktop_refresh_preserves_desktop_marker():
    seed_user(must_change_password = True)
    raw = storage.create_desktop_secret()
    client = auth_client()
    login_body = client.post("/api/auth/desktop-login", json = {"secret": raw}).json()

    response = client.post(
        "/api/auth/refresh",
        json = {"refresh_token": login_body["refresh_token"]},
    )

    assert response.status_code == 200
    body = response.json()
    assert body["must_change_password"] is False
    payload = jwt.decode(
        body["access_token"],
        storage.get_jwt_secret(storage.DEFAULT_ADMIN_USERNAME),
        algorithms = ["HS256"],
    )
    assert payload["sub"] == storage.DEFAULT_ADMIN_USERNAME
    assert payload["desktop"] is True


def test_consume_refresh_token_second_call_returns_none():
    """Single-use rotation rejects the same token on a second consume."""
    seed_user()
    from datetime import datetime, timedelta, timezone

    raw = secrets.token_urlsafe(48)
    expires = (datetime.now(timezone.utc) + timedelta(days = 30)).isoformat()
    storage.save_refresh_token(raw, storage.DEFAULT_ADMIN_USERNAME, expires)

    first = storage.consume_refresh_token(raw)
    assert first == (storage.DEFAULT_ADMIN_USERNAME, False)
    second = storage.consume_refresh_token(raw)
    assert second is None


def test_consume_refresh_token_concurrent_only_one_succeeds(tmp_path, monkeypatch):
    """64-thread pile-up on one token; DELETE RETURNING permits one winner."""
    seed_user()
    from concurrent.futures import ThreadPoolExecutor
    from datetime import datetime, timedelta, timezone

    raw = secrets.token_urlsafe(48)
    expires = (datetime.now(timezone.utc) + timedelta(days = 30)).isoformat()
    storage.save_refresh_token(raw, storage.DEFAULT_ADMIN_USERNAME, expires)

    workers = 64

    def attempt(_idx: int):
        try:
            return storage.consume_refresh_token(raw)
        except sqlite3.OperationalError:
            # "database is locked" under contention; treat as losing the race.
            return None

    with ThreadPoolExecutor(max_workers = workers) as pool:
        results = list(pool.map(attempt, range(workers)))

    successes = [r for r in results if r is not None]
    assert len(successes) == 1, f"expected exactly one consumer to win, got {len(successes)}"
    assert successes[0] == (storage.DEFAULT_ADMIN_USERNAME, False)


def test_consume_refresh_token_expired_returns_none():
    seed_user()
    from datetime import datetime, timedelta, timezone

    raw = secrets.token_urlsafe(48)
    expires = (datetime.now(timezone.utc) - timedelta(hours = 1)).isoformat()
    storage.save_refresh_token(raw, storage.DEFAULT_ADMIN_USERNAME, expires)
    assert storage.consume_refresh_token(raw) is None


def test_desktop_session_uses_real_admin_identity_for_api_keys():
    seed_user(must_change_password = True)
    raw = storage.create_desktop_secret()
    client = auth_client()
    token = client.post("/api/auth/desktop-login", json = {"secret": raw}).json()["access_token"]

    response = client.post(
        "/api/auth/api-keys",
        headers = {"Authorization": f"Bearer {token}"},
        json = {"name": "desktop"},
    )

    assert response.status_code == 200
    rows = storage.list_api_keys(storage.DEFAULT_ADMIN_USERNAME)
    assert [row["name"] for row in rows] == ["desktop"]


def test_local_recipe_token_authenticates_as_admin_for_desktop_user(loaded_local_model):
    # _inject_local_providers mints an internal sk-unsloth-* API key (not a
    # forwarded JWT) that validates as admin whether the session was desktop or web.
    from auth.authentication import create_access_token, get_current_subject

    seed_user(must_change_password = True)
    jobs_route = data_recipe_jobs_module()
    incoming_token = create_access_token(
        subject = storage.DEFAULT_ADMIN_USERNAME,
        desktop = True,
    )
    recipe = local_recipe()

    jobs_route._inject_local_providers(recipe, local_recipe_request(incoming_token))

    local_token = recipe["model_providers"][0]["api_key"]
    assert local_token.startswith(storage.API_KEY_PREFIX)
    credentials = HTTPAuthorizationCredentials(
        scheme = "Bearer",
        credentials = local_token,
    )
    assert asyncio.run(get_current_subject(credentials)) == storage.DEFAULT_ADMIN_USERNAME


def test_local_recipe_token_authenticates_as_admin_for_web_user(loaded_local_model):
    # Mirror of the desktop variant: API-key issuance is identical for web/desktop tokens.
    from auth.authentication import create_access_token, get_current_subject

    seed_user(must_change_password = False)
    jobs_route = data_recipe_jobs_module()
    incoming_token = create_access_token(subject = storage.DEFAULT_ADMIN_USERNAME)
    recipe = local_recipe()

    jobs_route._inject_local_providers(recipe, local_recipe_request(incoming_token))

    local_token = recipe["model_providers"][0]["api_key"]
    assert local_token.startswith(storage.API_KEY_PREFIX)
    credentials = HTTPAuthorizationCredentials(
        scheme = "Bearer",
        credentials = local_token,
    )
    assert asyncio.run(get_current_subject(credentials)) == storage.DEFAULT_ADMIN_USERNAME


def test_desktop_login_rejects_invalid_secret():
    seed_user(must_change_password = False)
    client = auth_client()

    response = client.post(
        "/api/auth/desktop-login",
        json = {"secret": "desktop-invalid"},
    )

    assert response.status_code == 401


def test_write_desktop_secret_file_is_0600_on_unix(tmp_path):
    from unsloth_cli.commands import studio as studio_cli

    path = tmp_path / ".desktop_secret"
    if platform.system() != "Windows":
        path.write_text("old-secret")
        os.chmod(path, 0o644)

    studio_cli._write_auth_secret(path, "desktop-secret")

    assert path.read_bytes() == b"desktop-secret\n"
    if platform.system() != "Windows":
        assert oct(path.stat().st_mode & 0o777) == "0o600"


def test_reset_password_removes_desktop_secret_files(tmp_path, monkeypatch):
    from typer.testing import CliRunner
    from unsloth_cli.commands import studio as studio_cli

    auth_dir = tmp_path / "auth"
    auth_dir.mkdir()
    (auth_dir / "auth.db").write_text("db")
    (auth_dir / ".bootstrap_password").write_text("boot")
    (auth_dir / ".desktop_secret").write_text("new")
    monkeypatch.setattr(studio_cli, "STUDIO_HOME", tmp_path)

    result = CliRunner().invoke(studio_cli.studio_app, ["reset-password"])

    assert result.exit_code == 0
    assert not (auth_dir / "auth.db").exists()
    assert not (auth_dir / ".bootstrap_password").exists()
    assert not (auth_dir / ".desktop_secret").exists()


def test_reset_password_removes_desktop_secret_files_without_db(tmp_path, monkeypatch):
    from typer.testing import CliRunner
    from unsloth_cli.commands import studio as studio_cli

    auth_dir = tmp_path / "auth"
    auth_dir.mkdir()
    (auth_dir / ".desktop_secret").write_text("new")
    monkeypatch.setattr(studio_cli, "STUDIO_HOME", tmp_path)

    result = CliRunner().invoke(studio_cli.studio_app, ["reset-password"])

    assert result.exit_code == 0
    assert not (auth_dir / ".desktop_secret").exists()


def test_desktop_capabilities_json_reports_rollout_safe_flags():
    from typer.testing import CliRunner
    import unsloth_cli.commands.studio as studio_cli

    result = CliRunner().invoke(
        studio_cli.studio_app,
        ["desktop-capabilities", "--json"],
    )

    assert result.exit_code == 0
    body = json.loads(result.output)
    assert body["desktop_protocol_version"] == 1
    assert body["supports_provision_desktop_auth"] is True
    assert body["supports_api_only"] is True
    assert isinstance(body["version"], str)


def test_health_response_reports_desktop_capability_fields(monkeypatch):
    routes_module = ModuleType("routes")
    routes_module.__path__ = []
    settings_module = ModuleType("routes.settings")
    settings_module.router = APIRouter()
    llama_module = ModuleType("routes.llama")
    llama_module.router = APIRouter()
    prompts_module = ModuleType("routes.prompts")
    prompts_module.router = APIRouter()

    for name, router in {
        "auth_router": APIRouter(),
        "chat_history_router": APIRouter(),
        "data_recipe_router": APIRouter(),
        "datasets_router": APIRouter(),
        "export_router": APIRouter(),
        "inference_router": APIRouter(),
        "inference_studio_router": APIRouter(),
        "mcp_servers_router": APIRouter(),
        "models_router": APIRouter(),
        "providers_router": APIRouter(),
        "rag_router": APIRouter(),
        "research_runs_router": APIRouter(),
        "settings_router": settings_module.router,
        "training_history_router": APIRouter(),
        "training_router": APIRouter(),
    }.items():
        setattr(routes_module, name, router)
    routes_module.settings = settings_module
    routes_module.llama = llama_module

    monkeypatch.setitem(sys.modules, "routes", routes_module)
    monkeypatch.setitem(sys.modules, "routes.settings", settings_module)
    monkeypatch.setitem(sys.modules, "routes.llama", llama_module)
    monkeypatch.setitem(sys.modules, "routes.prompts", prompts_module)

    import studio.backend.main as backend_main

    monkeypatch.setattr(backend_main._hw_module, "CHAT_ONLY", True)
    monkeypatch.setattr(backend_main._hw_module, "CHAT_ONLY_REASON", "mlx_unavailable")

    seed_user()
    from auth.authentication import create_access_token

    token = create_access_token(storage.DEFAULT_ADMIN_USERNAME)

    app = FastAPI()
    app.add_api_route("/api/health", backend_main.health_check, methods = ["GET"])
    client = TestClient(app)

    unauthenticated = client.get("/api/health")
    assert unauthenticated.status_code == 200
    unauthenticated_body = unauthenticated.json()
    assert unauthenticated_body["chat_only"] is True
    assert "chat_only_reason" not in unauthenticated_body

    response = client.get(
        "/api/health",
        headers = {"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200
    body = response.json()

    assert body["desktop_protocol_version"] == 1
    assert body["supports_desktop_auth"] is True
    assert body["chat_only_reason"] == "mlx_unavailable"


def test_provision_desktop_auth_writes_secret_and_creates_db_without_backend_deps(
    tmp_path, monkeypatch
):
    auth_dir = tmp_path / "auth"
    auth_dir.mkdir()

    code = """
import builtins
import sys
from pathlib import Path
from typer.testing import CliRunner

studio_home = Path(sys.argv[1])
real_import = builtins.__import__

def guarded_import(name, globals = None, locals = None, fromlist = (), level = 0):
    # Only gate absolute imports; relative `from .utils import x` inside
    # third-party packages (e.g. typer._click.decorators) hits level > 0
    # with name="utils" and must pass through.
    blocked = ("auth", "fastapi", "structlog", "utils")
    if level == 0 and (name in blocked or name.startswith(("auth.", "utils."))):
        raise ModuleNotFoundError(name)
    return real_import(name, globals, locals, fromlist, level)

builtins.__import__ = guarded_import
from unsloth_cli.commands import studio as studio_cli

studio_cli.STUDIO_HOME = studio_home
result = CliRunner().invoke(studio_cli.studio_app, ["provision-desktop-auth"])
if result.exit_code != 0:
    print(result.output)
    if result.exception is not None:
        raise result.exception
    raise SystemExit(result.exit_code)
"""
    result = subprocess.run(
        [sys.executable, "-c", code, str(tmp_path)],
        cwd = Path(__file__).resolve().parents[3],
        env = {**os.environ, "PYTHONPATH": "."},
        text = True,
        capture_output = True,
    )
    assert result.returncode == 0, result.stderr + result.stdout
    # Strip like the src-tauri readers do.
    secret = (auth_dir / ".desktop_secret").read_text().strip()
    assert secret.startswith("desktop-")

    conn = sqlite3.connect(auth_dir / "auth.db")
    conn.row_factory = sqlite3.Row
    try:
        user = conn.execute(
            """
            SELECT username, password_salt, password_hash, must_change_password
            FROM auth_user
            """
        ).fetchone()
        app_secrets = {
            row["key"]: row["value"] for row in conn.execute("SELECT key, value FROM app_secrets")
        }
        refresh_columns = {row["name"] for row in conn.execute("PRAGMA table_info(refresh_tokens)")}
    finally:
        conn.close()

    bootstrap_password = (auth_dir / ".bootstrap_password").read_text().strip()
    bootstrap_hash = hashlib.pbkdf2_hmac(
        "sha256",
        bootstrap_password.encode("utf-8"),
        user["password_salt"].encode("utf-8"),
        100_000,
    ).hex()

    assert bootstrap_password
    assert user["username"] == "unsloth"
    assert user["must_change_password"] == 1
    assert bootstrap_hash == user["password_hash"]
    assert len(app_secrets["api_key_pbkdf2_salt"]) == 64
    assert len(app_secrets["desktop_secret_hash"]) == 64
    assert app_secrets["desktop_secret_created_at"]
    assert "is_desktop" in refresh_columns

    monkeypatch.setattr(storage, "DB_PATH", auth_dir / "auth.db")
    monkeypatch.setattr(storage, "_api_key_pbkdf2_salt_cache", None)
    assert storage.validate_desktop_secret(secret) == storage.DEFAULT_ADMIN_USERNAME
    assert storage.requires_password_change(storage.DEFAULT_ADMIN_USERNAME) is True


def test_provision_desktop_auth_keeps_existing_admin_password(tmp_path, monkeypatch):
    from typer.testing import CliRunner
    from unsloth_cli.commands import studio as studio_cli

    auth_dir = tmp_path / "auth"
    auth_dir.mkdir()
    monkeypatch.setattr(studio_cli, "STUDIO_HOME", tmp_path)

    conn = sqlite3.connect(auth_dir / "auth.db")
    try:
        conn.execute(
            """
            CREATE TABLE auth_user (
                id INTEGER PRIMARY KEY,
                username TEXT UNIQUE NOT NULL,
                password_salt TEXT NOT NULL,
                password_hash TEXT NOT NULL,
                jwt_secret TEXT NOT NULL,
                must_change_password INTEGER NOT NULL DEFAULT 0
            )
            """
        )
        conn.execute(
            """
            INSERT INTO auth_user (
                username, password_salt, password_hash, jwt_secret, must_change_password
            )
            VALUES (?, ?, ?, ?, ?)
            """,
            ("unsloth", "existing-salt", "existing-hash", "existing-jwt", 0),
        )
        conn.commit()
    finally:
        conn.close()

    result = CliRunner().invoke(studio_cli.studio_app, ["provision-desktop-auth"])

    assert result.exit_code == 0
    assert not (auth_dir / ".bootstrap_password").exists()
    conn = sqlite3.connect(auth_dir / "auth.db")
    conn.row_factory = sqlite3.Row
    try:
        user = conn.execute(
            """
            SELECT password_salt, password_hash, jwt_secret, must_change_password
            FROM auth_user WHERE username = ?
            """,
            ("unsloth",),
        ).fetchone()
    finally:
        conn.close()

    assert dict(user) == {
        "password_salt": "existing-salt",
        "password_hash": "existing-hash",
        "jwt_secret": "existing-jwt",
        "must_change_password": 0,
    }


def test_update_password_clears_desktop_secret():
    seed_user()
    raw = storage.create_desktop_secret()
    assert storage.validate_desktop_secret(raw) == storage.DEFAULT_ADMIN_USERNAME

    changed = storage.update_password(storage.DEFAULT_ADMIN_USERNAME, "new-admin-password")
    assert changed is True
    assert storage.validate_desktop_secret(raw) is None


def test_update_password_on_unknown_user_leaves_desktop_secret_intact():
    seed_user()
    raw = storage.create_desktop_secret()

    changed = storage.update_password("not-a-user", "irrelevant")
    assert changed is False
    assert storage.validate_desktop_secret(raw) == storage.DEFAULT_ADMIN_USERNAME


def test_desktop_auth_provision_has_bounded_timeout():
    rs_path = (
        Path(__file__).resolve().parents[3] / "studio" / "src-tauri" / "src" / "desktop_auth.rs"
    )
    src = rs_path.read_text(encoding = "utf-8")
    start = src.index("async fn provision_desktop_auth(")
    depth = 0
    body_start = src.index("{", start)
    body_end = None
    for i in range(body_start, len(src)):
        c = src[i]
        if c == "{":
            depth += 1
        elif c == "}":
            depth -= 1
            if depth == 0:
                body_end = i + 1
                break
    assert body_end is not None
    body = src[start:body_end]
    assert "tokio::time::timeout" in body
    import re

    m = re.search(r"Duration::from_secs\(\s*(\d+)\s*\)", body)
    assert m is not None
    seconds = int(m.group(1))
    assert 5 <= seconds <= 120