import sys
import types

import pytest
from mjr_am_backend.routes.core import security as sec

TEST_TOKEN = "unit-test-token-01"


class _DummyTransport:
    def __init__(self, peername):
        self._peername = peername

    def get_extra_info(self, key):
        if key == "peername":
            return self._peername
        return None


class _DummyRequest:
    def __init__(self, method="POST", headers=None, peername=("127.0.0.1", 8188)):
        self.method = method
        self.headers = headers or {}
        self.transport = _DummyTransport(peername)
        self.app = {}
        self.remote = None


def test_env_truthy_and_token_extractors(monkeypatch) -> None:
    monkeypatch.setenv("X_FLAG", "yes")
    assert sec._env_truthy("X_FLAG")
    monkeypatch.setenv("X_FLAG", "0")
    assert not sec._env_truthy("X_FLAG")

    assert sec._extract_bearer_token({"Authorization": "Bearer abc"}) == "abc"
    assert sec._extract_write_token_from_headers({"X-MJR-Token": "x"}) == "x"
    assert sec._extract_write_token_from_headers({"Cookie": "mjr_write_token=cookie-token"}) == "cookie-token"
    oversized_cookie = "mjr_write_token=" + ("x" * 5000)
    assert sec._extract_write_token_from_headers({"Cookie": oversized_cookie}) == ""


def test_hash_and_write_token_hash(monkeypatch) -> None:
    monkeypatch.setenv("MAJOOR_API_TOKEN", TEST_TOKEN)
    monkeypatch.delenv("MAJOOR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MAJOOR_API_TOKEN_PEPPER", raising=False)
    h = sec._get_write_token_hash()
    assert h
    assert h == sec._hash_token(TEST_TOKEN)

    monkeypatch.setenv("MAJOOR_API_TOKEN_HASH", "deadbeef")
    assert sec._get_write_token_hash() == "deadbeef"


def test_validate_token_format_rejects_short_and_accepts_valid(monkeypatch) -> None:
    sec._reset_security_state_for_tests()

    assert sec._validate_token_format("short-token", "test-short") is None
    assert sec._validate_token_format("valid-token-1234", "test-valid") == "valid-token-1234"


def test_has_configured_write_token(monkeypatch) -> None:
    monkeypatch.delenv("MAJOOR_API_TOKEN", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN", raising=False)
    monkeypatch.delenv("MAJOOR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN_HASH", raising=False)
    assert sec._has_configured_write_token() is False

    monkeypatch.setenv("MAJOOR_API_TOKEN_HASH", "abc123")
    assert sec._has_configured_write_token() is True


def test_ip_helpers_and_resolve_client_ip(monkeypatch) -> None:
    assert sec._is_valid_ip("127.0.0.1")
    assert not sec._is_valid_ip("bad")
    assert sec._is_loopback_ip("::1")

    monkeypatch.setattr(sec, "_is_trusted_proxy", lambda ip: ip in {"127.0.0.1", "10.0.0.5"})
    headers = {"X-Forwarded-For": "8.8.8.8, 10.0.0.5"}
    assert sec._resolve_client_ip("127.0.0.1", headers) == "8.8.8.8"
    assert sec._resolve_client_ip("9.9.9.9", headers) == "9.9.9.9"

    # Spoofed leading XFF value should not win when a trusted proxy appends the real client.
    monkeypatch.setattr(sec, "_is_trusted_proxy", lambda ip: ip == "127.0.0.1")
    spoofed = {"X-Forwarded-For": "1.2.3.4, 203.0.113.10"}
    assert sec._resolve_client_ip("127.0.0.1", spoofed) == "203.0.113.10"


def test_require_operation_enabled_paths(monkeypatch) -> None:
    monkeypatch.delenv("MAJOOR_SAFE_MODE", raising=False)
    monkeypatch.delenv("MAJOOR_ALLOW_WRITE", raising=False)

    r = sec._require_operation_enabled("write", prefs={"safe_mode": True, "allow_write": True})
    assert r.ok

    r2 = sec._require_operation_enabled("delete", prefs={"allow_delete": False})
    assert not r2.ok
    assert r2.code == "FORBIDDEN"

    r3 = sec._require_operation_enabled("unknown-op", prefs={"safe_mode": False})
    assert r3.ok


def test_require_operation_enabled_string_bool_prefs() -> None:
    # "false" should not be interpreted as truthy for sensitive ops.
    denied = sec._require_operation_enabled("delete", prefs={"allow_delete": "false"})
    assert not denied.ok
    assert denied.code == "FORBIDDEN"

    allowed = sec._require_operation_enabled("delete", prefs={"allow_delete": "true"})
    assert allowed.ok


def test_auth_helpers_and_authenticated_user() -> None:
    class _UM:
        enabled = True

        @staticmethod
        def get_request_user_id(_request):
            return "u1"

    req = _DummyRequest()
    req.app["_mjr_user_manager"] = _UM()
    res = sec._require_authenticated_user(req)
    assert res.ok
    assert res.data == "u1"

    req2 = _DummyRequest()
    req2.app["_mjr_user_manager"] = types.SimpleNamespace(enabled=True, get_request_user_id=lambda _r: "")
    res2 = sec._require_authenticated_user(req2)
    assert not res2.ok
    assert res2.code == "AUTH_REQUIRED"


def test_server_module_user_manager_prefers_prompt_server_instance(monkeypatch) -> None:
    prompt_server = types.SimpleNamespace(instance=types.SimpleNamespace(user_manager="ctx-um"))
    monkeypatch.setitem(sys.modules, "server", types.SimpleNamespace(PromptServer=prompt_server))
    assert sec._server_module_user_manager() == "ctx-um"


def test_parse_trusted_proxies_and_is_trusted_proxy_cache(monkeypatch) -> None:
    monkeypatch.setenv("MAJOOR_TRUSTED_PROXIES", "127.0.0.1,10.0.0.0/8,0.0.0.0/0")
    monkeypatch.delenv("MAJOOR_ALLOW_INSECURE_TRUSTED_PROXIES", raising=False)

    nets = sec._parse_trusted_proxies()
    assert nets
    assert all(getattr(n, "prefixlen", 0) != 0 for n in nets)

    monkeypatch.setattr(sec, "_TRUSTED_PROXY_NETS", nets)
    sec._is_trusted_proxy.cache_clear()
    assert sec._is_trusted_proxy("127.0.0.1")
    assert not sec._is_trusted_proxy("8.8.8.8")


def test_client_identifier_and_rate_limit(monkeypatch) -> None:
    sec._reset_security_state_for_tests()
    monkeypatch.setattr(sec, "_is_trusted_proxy", lambda _ip: False)

    req = _DummyRequest(headers={})
    cid = sec._get_client_identifier(req)
    assert isinstance(cid, str)
    assert cid

    monkeypatch.setattr(sec.time, "time", lambda: 100.0)
    allowed, retry = sec._check_rate_limit(req, "ep", max_requests=2, window_seconds=60)
    assert allowed and retry is None
    allowed, retry = sec._check_rate_limit(req, "ep", max_requests=2, window_seconds=60)
    assert allowed and retry is None
    allowed, retry = sec._check_rate_limit(req, "ep", max_requests=2, window_seconds=60)
    assert not allowed
    assert isinstance(retry, int)


def test_check_write_access_uses_constant_time_compare(monkeypatch) -> None:
    import hmac

    token = TEST_TOKEN
    token_hash = sec._hash_token(token)
    monkeypatch.setenv("MAJOOR_API_TOKEN_HASH", token_hash)
    monkeypatch.delenv("MAJOOR_API_TOKEN", raising=False)

    calls = {"count": 0}
    real_compare = hmac.compare_digest

    def _compare(a, b):
        calls["count"] += 1
        return real_compare(a, b)

    monkeypatch.setattr(hmac, "compare_digest", _compare)

    out = sec._check_write_access(peer_ip="127.0.0.1", headers={"X-MJR-Token": token})
    assert out.ok
    assert calls["count"] >= 1


def test_check_write_access_blocks_insecure_remote_token_transport(monkeypatch) -> None:
    monkeypatch.setenv("MAJOOR_API_TOKEN", TEST_TOKEN)
    monkeypatch.delenv("MAJOOR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MAJOOR_ALLOW_INSECURE_TOKEN_TRANSPORT", raising=False)

    out = sec._check_write_access(
        peer_ip="203.0.113.10",
        headers={"X-MJR-Token": TEST_TOKEN},
        request_scheme="http",
    )
    assert not out.ok
    assert out.code == "FORBIDDEN"
    assert (out.meta or {}).get("auth") == "token_insecure_transport"


def test_check_write_access_accepts_https_via_trusted_proxy(monkeypatch) -> None:
    monkeypatch.setenv("MAJOOR_API_TOKEN", TEST_TOKEN)
    monkeypatch.delenv("MAJOOR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN_HASH", raising=False)
    monkeypatch.setattr(sec, "_is_trusted_proxy", lambda ip: ip == "127.0.0.1")

    out = sec._check_write_access(
        peer_ip="127.0.0.1",
        headers={
            "X-MJR-Token": TEST_TOKEN,
            "X-Forwarded-For": "203.0.113.10",
            "X-Forwarded-Proto": "https",
        },
        request_scheme="http",
    )
    assert out.ok
    assert (out.meta or {}).get("auth") == "token"


def test_csrf_and_origin_checks(monkeypatch) -> None:
    monkeypatch.setattr(sec, "_is_trusted_proxy", lambda _ip: False)

    req = _DummyRequest(method="POST", headers={"Host": "127.0.0.1:8188"})
    assert "Missing anti-CSRF" in str(sec._csrf_error(req))

    req2 = _DummyRequest(
        method="POST",
        headers={
            "Host": "127.0.0.1:8188",
            "X-Requested-With": "fetch",
            "Origin": "http://127.0.0.1:8188",
        },
    )
    assert sec._csrf_error(req2) is None

    req3 = _DummyRequest(
        method="POST",
        headers={
            "Host": "127.0.0.1:8188",
            "X-Requested-With": "fetch",
            "Origin": "http://evil.com",
        },
    )
    assert "Cross-site request blocked" in str(sec._csrf_error(req3))


def test_split_host_port_and_loopback_origin_match() -> None:
    assert sec._split_host_port("127.0.0.1:8188") == ("127.0.0.1", 8188)
    assert sec._split_host_port("localhost") == ("localhost", None)

    parsed = sec._parse_origin("http://127.0.0.1:8188")
    assert parsed is not None
    assert sec._is_loopback_origin_host_match(parsed, "localhost:8188")


@pytest.mark.asyncio
async def test_resolve_security_prefs_best_effort() -> None:
    class _Settings:
        async def get_security_prefs(self):
            return {"safe_mode": True}

    prefs = await sec._resolve_security_prefs({"settings": _Settings()})
    assert prefs == {"safe_mode": True}

    prefs2 = await sec._resolve_security_prefs(None)
    assert prefs2 is None


def test_check_write_access_allows_unknown_peer_on_truly_local(monkeypatch) -> None:
    """When peer_ip extraction returns 'unknown' and NO headers dict is passed (direct/test call),
    allowwrite access on local connections without a token. This handles edge cases where
    peer IP extraction fails (e.g., WebSocket, unix domain sockets)."""
    
    monkeypatch.delenv("MAJOOR_API_TOKEN", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN", raising=False)
    monkeypatch.delenv("MAJOOR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MJR_API_TOKEN_HASH", raising=False)
    monkeypatch.delenv("MAJOOR_REQUIRE_AUTH", raising=False)
    monkeypatch.delenv("MAJOOR_ALLOW_REMOTE_WRITE", raising=False)
    
    # Calling _check_write_access directly without headers (bypassing _require_write_access wrapper)
    out = sec._check_write_access(
        peer_ip="unknown",
        headers=None,
    )
    assert out.ok, f"Expected write access allowed for unknown peer when headers=None, got {out}"
    assert (out.meta or {}).get("auth") == "loopback_unknown_peer"


def test_check_write_access_allows_unknown_peer_with_token_not_provided(monkeypatch) -> None:
    """When a token IS configured but NOT provided in the request, and peer_ip is 'unknown'
    and headers is None, still allow write access (local edge case with token configured)."""
    
    monkeypatch.setenv("MAJOOR_API_TOKEN", "test-token-1234567890")
    monkeypatch.delenv("MAJOOR_REQUIRE_AUTH", raising=False)
    monkeypatch.delenv("MAJOOR_ALLOW_REMOTE_WRITE", raising=False)
    
    # Calling _check_write_access with token configured but not provided, unknown peer
    out = sec._check_write_access(
        peer_ip="unknown",
        headers=None,  # No headers dict = true direct call
    )
    assert out.ok, f"Expected write access allowed for unknown peer + token configured, got {out}"
    assert (out.meta or {}).get("auth") == "loopback_unknown_peer"


@pytest.fixture(autouse=True)
def monkeypatch():
    _mp = pytest.MonkeyPatch()
    yield _mp
    _mp.undo()