import importlib
import sys
from types import SimpleNamespace

import bcrypt
import pytest
from fastapi import HTTPException
from starlette.status import HTTP_403_FORBIDDEN

from lightrag.api.passwords import BCRYPT_PASSWORD_PREFIX, hash_password
from lightrag.tools.hash_password import main as hash_password_main
from lightrag.utils import logger as lightrag_logger


def import_real_api_module(module_name: str):
    sys.modules.pop(module_name, None)

    package_name, _, child_name = module_name.rpartition(".")
    package = sys.modules.get(package_name)
    if package is not None and hasattr(package, child_name):
        delattr(package, child_name)

    return importlib.import_module(module_name)


@pytest.fixture
def auth_module(monkeypatch):
    config = import_real_api_module("lightrag.api.config")

    mock_global_args = SimpleNamespace(
        token_secret="test-jwt-secret",
        jwt_algorithm="HS256",
        token_expire_hours=48,
        guest_token_expire_hours=24,
        auth_accounts="admin:admin_pass",
    )

    monkeypatch.setattr(config, "global_args", mock_global_args)

    module = import_real_api_module("lightrag.api.auth")
    module = importlib.reload(module)
    yield module
    sys.modules.pop("lightrag.api.auth", None)


def build_bcrypt_value(password: str) -> str:
    hashed = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
    return f"{BCRYPT_PASSWORD_PREFIX}{hashed}"


def test_verify_plaintext_password(auth_module):
    handler = auth_module.AuthHandler()
    handler.accounts = {"admin": "admin_pass"}

    assert handler.verify_password("admin", "admin_pass")
    assert not handler.verify_password("admin", "wrong_pass")


def test_verify_prefixed_bcrypt_password(auth_module):
    handler = auth_module.AuthHandler()
    handler.accounts = {"user": build_bcrypt_value("user_pass")}

    assert handler.verify_password("user", "user_pass")
    assert not handler.verify_password("user", "wrong_pass")


def test_plaintext_password_with_bcrypt_prefix_stays_plaintext(auth_module):
    handler = auth_module.AuthHandler()
    handler.accounts = {"user": "$2b$not-a-real-hash"}

    assert handler.verify_password("user", "$2b$not-a-real-hash")
    assert not handler.verify_password("user", "anything-else")


def test_invalid_auth_accounts_raises(monkeypatch):
    config = import_real_api_module("lightrag.api.config")

    mock_global_args = SimpleNamespace(
        token_secret="test-jwt-secret",
        jwt_algorithm="HS256",
        token_expire_hours=48,
        guest_token_expire_hours=24,
        auth_accounts="admin",
    )

    monkeypatch.setattr(config, "global_args", mock_global_args)

    with pytest.raises(ValueError, match="AUTH_ACCOUNTS must use"):
        import_real_api_module("lightrag.api.auth")

    sys.modules.pop("lightrag.api.auth", None)


def test_initialize_config_rejects_default_token_secret_with_auth_accounts():
    config = import_real_api_module("lightrag.api.config")

    insecure_args = SimpleNamespace(
        auth_accounts="admin:admin_pass",
        token_secret=config.DEFAULT_TOKEN_SECRET,
    )

    with pytest.raises(ValueError, match="TOKEN_SECRET must be explicitly set"):
        config.initialize_config(insecure_args, force=True)


def test_initialize_config_allows_custom_token_secret_with_auth_accounts():
    config = import_real_api_module("lightrag.api.config")

    secure_args = SimpleNamespace(
        auth_accounts="admin:admin_pass",
        token_secret="custom-jwt-secret",
    )

    initialized = config.initialize_config(secure_args, force=True)

    assert initialized is secure_args


def test_guest_tokens_fall_back_to_default_secret_when_token_secret_missing(
    monkeypatch,
):
    config = import_real_api_module("lightrag.api.config")

    mock_global_args = SimpleNamespace(
        token_secret=None,
        jwt_algorithm="HS256",
        token_expire_hours=48,
        guest_token_expire_hours=24,
        auth_accounts="",
    )

    monkeypatch.setattr(config, "global_args", mock_global_args)
    warning_messages = []

    def capture_warning(message):
        warning_messages.append(message)

    monkeypatch.setattr(lightrag_logger, "warning", capture_warning)

    module = import_real_api_module("lightrag.api.auth")
    module = importlib.reload(module)
    handler = module.AuthHandler()

    token = handler.create_token("guest", role="guest")
    token_info = handler.validate_token(token)

    assert handler.secret == config.DEFAULT_TOKEN_SECRET
    assert token_info["username"] == "guest"
    assert token_info["role"] == "guest"
    assert any(
        "Falling back to the default guest-mode JWT secret" in msg
        for msg in warning_messages
    )

    sys.modules.pop("lightrag.api.auth", None)


def _load_api_key_only_modules(monkeypatch):
    """Import auth + utils_api in the API-key-only profile.

    API-key-only = an API key is set but AUTH_ACCOUNTS is empty, so
    ``auth_configured`` is False and AuthHandler falls back to the public
    DEFAULT_TOKEN_SECRET. Returns (config, auth, utils_api).
    """
    config = import_real_api_module("lightrag.api.config")

    mock_global_args = SimpleNamespace(
        token_secret=None,  # -> AuthHandler falls back to DEFAULT_TOKEN_SECRET
        jwt_algorithm="HS256",
        token_expire_hours=48,
        guest_token_expire_hours=24,
        auth_accounts="",  # no password accounts -> auth_configured is False
        whitelist_paths="/health",  # consumed by utils_api at import time
        token_auto_renew=False,
    )
    monkeypatch.setattr(config, "global_args", mock_global_args)

    auth = import_real_api_module("lightrag.api.auth")
    auth = importlib.reload(auth)
    utils_api = import_real_api_module("lightrag.api.utils_api")
    return config, auth, utils_api


async def test_combined_auth_rejects_guest_token_in_api_key_mode(monkeypatch):
    """A forged/obtained guest token must not bypass the X-API-Key check.

    Regression for GHSA-f4vv-55c2-5789 / GHSA-xr5c-v5r6-c9f9: in the
    API-key-only profile an anonymous caller can mint a guest JWT with the
    public default secret (or fetch one from /auth-status) and previously
    short-circuited authorization before the API key was ever checked.
    """
    config, auth, utils_api = _load_api_key_only_modules(monkeypatch)
    try:
        assert utils_api.auth_configured is False
        assert auth.auth_handler.secret == config.DEFAULT_TOKEN_SECRET

        forged_guest = auth.auth_handler.create_token(username="guest", role="guest")

        api_key = "operator-secret-key"
        dependency = utils_api.get_combined_auth_dependency(api_key=api_key)

        request = SimpleNamespace(url=SimpleNamespace(path="/documents"), scope={})
        response = SimpleNamespace(headers={})

        # Forged guest token, no API key -> rejected (was HTTP 200 before the fix).
        with pytest.raises(HTTPException) as rejected:
            await dependency(
                request=request,
                response=response,
                token=forged_guest,
                api_key_header_value=None,
            )
        assert rejected.value.status_code == HTTP_403_FORBIDDEN

        # Guest token together with the real API key -> allowed (the WebUI sends
        # both Authorization and X-API-Key headers).
        result = await dependency(
            request=request,
            response=response,
            token=forged_guest,
            api_key_header_value=api_key,
        )
        assert result is None
    finally:
        sys.modules.pop("lightrag.api.utils_api", None)
        sys.modules.pop("lightrag.api.auth", None)


async def test_combined_auth_allows_guest_token_when_fully_open(monkeypatch):
    """Fully-open mode (no API key, no AUTH_ACCOUNTS) keeps accepting guest tokens.

    Guards against over-fixing: the bypass fix must not break zero-config guest
    access, which is the intended convenience mode.
    """
    _config, auth, utils_api = _load_api_key_only_modules(monkeypatch)
    try:
        guest_token = auth.auth_handler.create_token(username="guest", role="guest")

        dependency = utils_api.get_combined_auth_dependency(api_key=None)
        request = SimpleNamespace(url=SimpleNamespace(path="/documents"), scope={})
        response = SimpleNamespace(headers={})

        result = await dependency(
            request=request,
            response=response,
            token=guest_token,
            api_key_header_value=None,
        )
        assert result is None
    finally:
        sys.modules.pop("lightrag.api.utils_api", None)
        sys.modules.pop("lightrag.api.auth", None)


def test_hash_password_returns_prefixed_value(auth_module):
    hashed = hash_password("new_password")

    assert hashed.startswith(BCRYPT_PASSWORD_PREFIX)
    raw_hash = hashed[len(BCRYPT_PASSWORD_PREFIX) :]
    assert bcrypt.checkpw("new_password".encode("utf-8"), raw_hash.encode("utf-8"))


def test_hash_password_cli_outputs_auth_accounts_entry(capsys):
    exit_code = hash_password_main(["--username", "admin", "secret"])

    assert exit_code == 0
    output = capsys.readouterr().out.strip()
    username, hashed = output.split(":", 1)
    assert username == "admin"
    assert hashed.startswith(BCRYPT_PASSWORD_PREFIX)
    raw_hash = hashed[len(BCRYPT_PASSWORD_PREFIX) :]
    assert bcrypt.checkpw("secret".encode("utf-8"), raw_hash.encode("utf-8"))