turnstone/tests/test_auth.py

"""Tests for turnstone.core.auth — bearer token authentication and cookies."""

import os
import queue
import threading
from unittest.mock import MagicMock, patch

import pytest

from turnstone.core.auth import (
    AUTH_COOKIE,
    AUTH_COOKIE_CONSOLE,
    AUTH_COOKIE_SERVER,
    WRITE_PATHS,
    _extract_bearer,
    _extract_cookie,
    check_request,
    create_jwt,
    is_public_path,
    load_jwt_secret,
    make_clear_cookie,
    make_set_cookie,
    required_scope,
)

# ---------------------------------------------------------------------------
# TestIsPublicPath
# ---------------------------------------------------------------------------


class TestIsPublicPath:
    def test_root(self):
        assert is_public_path("/") is True

    def test_health(self):
        assert is_public_path("/health") is True

    def test_metrics(self):
        assert is_public_path("/metrics") is True

    def test_static_css(self):
        assert is_public_path("/static/style.css") is True

    def test_static_js(self):
        assert is_public_path("/static/app.js") is True

    def test_static_subdir(self):
        assert is_public_path("/static/fonts/mono.woff2") is True

    def test_shared_css_public(self):
        assert is_public_path("/shared/base.css") is True

    def test_shared_js_public(self):
        assert is_public_path("/shared/utils.js") is True

    def test_api_workstreams_not_public(self):
        assert is_public_path("/api/workstreams") is False

    def test_api_workstreams_send_not_public(self):
        assert is_public_path("/api/workstreams/abc/send") is False

    def test_api_cluster_overview_not_public(self):
        assert is_public_path("/api/cluster/overview") is False

    def test_api_events_not_public(self):
        assert is_public_path("/api/events") is False

    def test_v1_api_login_public(self):
        assert is_public_path("/v1/api/auth/login") is True

    def test_v1_api_logout_public(self):
        assert is_public_path("/v1/api/auth/logout") is True

    def test_v1_api_workstreams_not_public(self):
        assert is_public_path("/v1/api/workstreams") is False

    def test_v1_api_workstreams_send_not_public(self):
        assert is_public_path("/v1/api/workstreams/abc/send") is False

    def test_openapi_json_public(self):
        assert is_public_path("/openapi.json") is True

    def test_docs_public(self):
        assert is_public_path("/docs") is True

    def test_shared_static_public(self):
        assert is_public_path("/shared/base.css") is True

    # Console proxy: a public proxied path must still be public, otherwise
    # the login modal can never re-authenticate from inside a ``/node/{id}/``
    # proxied page once the cookie expires.
    def test_proxy_v1_login_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/login") is True

    def test_proxy_no_v1_login_public(self):
        assert is_public_path("/node/node-a/api/auth/login") is True

    def test_proxy_v1_status_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/status") is True

    def test_proxy_v1_setup_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/setup") is True

    def test_proxy_v1_logout_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/logout") is True

    def test_proxy_v1_oidc_authorize_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/oidc/authorize") is True

    def test_proxy_v1_oidc_callback_public(self):
        assert is_public_path("/node/node-a/v1/api/auth/oidc/callback") is True

    def test_proxy_v1_workstreams_still_not_public(self):
        """Proxy prefix must not turn protected paths into public ones."""
        assert is_public_path("/node/node-a/v1/api/workstreams") is False

    def test_proxy_v1_refresh_still_requires_auth(self):
        """Refresh isn't in PUBLIC_PATHS — the caller must already have
        a valid cookie. Proxy-prefix shouldn't change that."""
        assert is_public_path("/node/node-a/v1/api/auth/refresh") is False


# ---------------------------------------------------------------------------
# TestRequiredRole
# ---------------------------------------------------------------------------


class TestRequiredScope:
    def test_get_api_needs_read(self):
        assert required_scope("GET", "/api/workstreams") == "read"

    def test_get_events_needs_read(self):
        assert required_scope("GET", "/api/events") == "read"

    def test_post_send_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc/send") == "write"

    def test_delete_send_needs_write(self):
        assert required_scope("DELETE", "/api/workstreams/abc/send") == "write"

    def test_post_approve_needs_approve(self):
        assert required_scope("POST", "/api/workstreams/abc/approve") == "approve"

    def test_post_cancel_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc/cancel") == "write"

    def test_post_close_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc/close") == "write"

    def test_post_rewind_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc/rewind") == "write"

    def test_post_retry_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc/retry") == "write"

    def test_get_events_per_ws_needs_read(self):
        assert required_scope("GET", "/api/workstreams/abc/events") == "read"

    def test_post_command_needs_write(self):
        assert required_scope("POST", "/api/command") == "write"

    def test_post_workstreams_new_needs_write(self):
        assert required_scope("POST", "/api/workstreams/new") == "write"

    def test_all_write_paths_need_write(self):
        for path in WRITE_PATHS:
            scope = required_scope("POST", path)
            assert scope in ("write", "approve"), f"{path} should need write or approve"

    def test_post_unknown_path_needs_read(self):
        assert required_scope("POST", "/api/unknown") == "read"

    def test_model_auth_cache_invalidation_needs_approve(self):
        assert required_scope("POST", "/api/_internal/model-auth-cache-invalidate") == "approve"
        assert (
            required_scope(
                "POST",
                "/v1/api/_internal/model-auth-cache-invalidate",
            )
            == "approve"
        )

    def test_v1_post_send_needs_write(self):
        assert required_scope("POST", "/v1/api/workstreams/abc/send") == "write"

    def test_v1_post_approve_needs_approve(self):
        assert required_scope("POST", "/v1/api/workstreams/abc/approve") == "approve"

    def test_v1_get_workstreams_needs_read(self):
        assert required_scope("GET", "/v1/api/workstreams") == "read"

    def test_v1_post_cluster_ws_new_needs_write(self):
        assert required_scope("POST", "/v1/api/cluster/workstreams/new") == "write"

    def test_proxy_v1_send_needs_write(self):
        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/send") == "write"

    def test_proxy_v1_approve_needs_approve(self):
        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/approve") == "approve"

    def test_proxy_v1_rewind_needs_write(self):
        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/rewind") == "write"

    def test_proxy_v1_retry_needs_write(self):
        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/retry") == "write"

    def test_proxy_v1_read_endpoint_needs_read(self):
        assert required_scope("GET", "/node/node-a/v1/api/workstreams") == "read"

    # Memory endpoints
    def test_get_memories_needs_read(self):
        assert required_scope("GET", "/api/memories") == "read"

    def test_post_memories_needs_write(self):
        assert required_scope("POST", "/api/memories") == "write"

    def test_post_memories_search_needs_read(self):
        """Search via POST is non-mutating — requires only read scope."""
        assert required_scope("POST", "/api/memories/search") == "read"

    def test_delete_memory_needs_write(self):
        assert required_scope("DELETE", "/api/memories/my_key") == "write"

    def test_v1_post_memories_needs_write(self):
        assert required_scope("POST", "/v1/api/memories") == "write"

    def test_v1_delete_memory_needs_write(self):
        assert required_scope("DELETE", "/v1/api/memories/test_key") == "write"

    def test_admin_memories_needs_approve(self):
        assert required_scope("GET", "/api/admin/memories") == "approve"

    def test_admin_memory_delete_needs_approve(self):
        assert required_scope("DELETE", "/api/admin/memories/some-id") == "approve"

    # Internal endpoints
    def test_internal_mcp_reload_needs_approve(self):
        assert required_scope("POST", "/api/_internal/mcp-reload") == "approve"

    def test_v1_internal_mcp_reload_needs_approve(self):
        assert required_scope("POST", "/v1/api/_internal/mcp-reload") == "approve"

    def test_internal_config_reload_needs_approve(self):
        assert required_scope("POST", "/api/_internal/config-reload") == "approve"

    def test_v1_internal_config_reload_needs_approve(self):
        assert required_scope("POST", "/v1/api/_internal/config-reload") == "approve"

    def test_proxy_internal_config_reload_needs_approve(self):
        assert required_scope("POST", "/node/n1/v1/api/_internal/config-reload") == "approve"

    def test_proxy_no_v1_internal_config_reload_needs_approve(self):
        assert required_scope("POST", "/node/n1/api/_internal/config-reload") == "approve"

    def test_proxy_internal_mcp_reload_needs_approve(self):
        assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-reload") == "approve"

    def test_proxy_no_v1_internal_mcp_reload_needs_approve(self):
        assert required_scope("POST", "/node/n1/api/_internal/mcp-reload") == "approve"

    def test_get_internal_mcp_reload_needs_read(self):
        """Only POST is elevated — GET falls through to read."""
        assert required_scope("GET", "/api/_internal/mcp-reload") == "read"

    def test_internal_mcp_refresh_one_needs_approve(self):
        assert required_scope("POST", "/api/_internal/mcp-refresh/srv") == "approve"

    def test_v1_internal_mcp_refresh_one_needs_approve(self):
        assert required_scope("POST", "/v1/api/_internal/mcp-refresh/srv") == "approve"

    def test_proxy_internal_mcp_refresh_one_needs_approve(self):
        assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-refresh/srv") == "approve"

    def test_proxy_no_v1_internal_mcp_refresh_one_needs_approve(self):
        assert required_scope("POST", "/node/n1/api/_internal/mcp-refresh/srv") == "approve"

    def test_internal_mcp_reconnect_one_needs_approve(self):
        assert required_scope("POST", "/api/_internal/mcp-reconnect/srv") == "approve"

    def test_v1_internal_mcp_reconnect_one_needs_approve(self):
        assert required_scope("POST", "/v1/api/_internal/mcp-reconnect/srv") == "approve"

    def test_proxy_internal_mcp_reconnect_one_needs_approve(self):
        assert required_scope("POST", "/node/n1/v1/api/_internal/mcp-reconnect/srv") == "approve"

    def test_proxy_no_v1_internal_mcp_reconnect_one_needs_approve(self):
        assert required_scope("POST", "/node/n1/api/_internal/mcp-reconnect/srv") == "approve"

    # Workstream sub-resource mutations (parametric paths)
    def test_ws_delete_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc123/delete") == "write"

    def test_ws_open_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc123/open") == "write"

    def test_ws_refresh_title_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc123/refresh-title") == "write"

    def test_ws_title_needs_write(self):
        assert required_scope("POST", "/api/workstreams/abc123/title") == "write"

    def test_v1_ws_delete_needs_write(self):
        assert required_scope("POST", "/v1/api/workstreams/abc123/delete") == "write"

    def test_proxy_ws_delete_needs_write(self):
        assert required_scope("POST", "/node/n1/v1/api/workstreams/abc123/delete") == "write"

    def test_proxy_ws_open_needs_write(self):
        assert required_scope("POST", "/node/n1/v1/api/workstreams/abc123/open") == "write"

    def test_proxy_ws_title_needs_write(self):
        assert required_scope("POST", "/node/n1/v1/api/workstreams/abc123/title") == "write"

    def test_ws_get_is_still_read(self):
        """GET on workstream sub-resource is not elevated."""
        assert required_scope("GET", "/api/workstreams/abc123/delete") == "read"


# ---------------------------------------------------------------------------
# TestExtractBearer
# ---------------------------------------------------------------------------


class TestExtractBearer:
    def test_valid_bearer(self):
        assert _extract_bearer("Bearer tok_abc123") == "tok_abc123"

    def test_case_insensitive(self):
        assert _extract_bearer("bearer tok_abc123") == "tok_abc123"

    def test_mixed_case(self):
        assert _extract_bearer("BEARER tok_abc123") == "tok_abc123"

    def test_no_bearer_prefix(self):
        assert _extract_bearer("tok_abc123") is None

    def test_basic_auth_ignored(self):
        assert _extract_bearer("Basic dXNlcjpwYXNz") is None

    def test_none(self):
        assert _extract_bearer(None) is None

    def test_empty(self):
        assert _extract_bearer("") is None

    def test_bearer_only_no_token(self):
        assert _extract_bearer("Bearer") is None

    def test_token_with_spaces(self):
        # Only the first space separates scheme from token
        assert _extract_bearer("Bearer tok with spaces") == "tok with spaces"


# ---------------------------------------------------------------------------
# TestExtractCookie
# ---------------------------------------------------------------------------


class TestExtractCookie:
    def test_single_cookie(self):
        assert _extract_cookie("turnstone_auth=tok_abc", "turnstone_auth") == "tok_abc"

    def test_multiple_cookies(self):
        header = "theme=dark; turnstone_auth=tok_abc; other=val"
        assert _extract_cookie(header, "turnstone_auth") == "tok_abc"

    def test_missing_cookie(self):
        assert _extract_cookie("theme=dark; other=val", "turnstone_auth") is None

    def test_none_header(self):
        assert _extract_cookie(None, "turnstone_auth") is None

    def test_empty_header(self):
        assert _extract_cookie("", "turnstone_auth") is None

    def test_spaces_around_value(self):
        assert _extract_cookie("turnstone_auth = tok_abc ", "turnstone_auth") == "tok_abc"

    def test_no_equals(self):
        assert _extract_cookie("malformed", "turnstone_auth") is None


# ---------------------------------------------------------------------------
# TestMakeSetCookie / TestMakeClearCookie
# ---------------------------------------------------------------------------


class TestMakeSetCookie:
    def test_contains_token(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)
        assert "turnstone_auth_server=tok_abc" in val

    def test_httponly(self):
        assert "HttpOnly" in make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)

    def test_samesite_lax(self):
        assert "SameSite=Lax" in make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)

    def test_path(self):
        assert "Path=/" in make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)

    def test_max_age_default(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)
        assert "Max-Age=86400" in val  # 24 hours (matches JWT expiry)

    def test_max_age_custom(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER, max_age=3600)
        assert "Max-Age=3600" in val

    def test_secure_default(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER)
        assert "; Secure" in val

    def test_secure_false(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER, secure=False)
        assert "; Secure" not in val

    def test_secure_true(self):
        val = make_set_cookie("tok_abc", AUTH_COOKIE_SERVER, secure=True)
        assert "; Secure" in val

    def test_uses_provided_name(self):
        # Name is honored verbatim; one surface's name never leaks into the other.
        val = make_set_cookie("tok_abc", AUTH_COOKIE_CONSOLE)
        assert "turnstone_auth_console=tok_abc" in val
        assert "turnstone_auth_server" not in val


class TestMakeClearCookie:
    def test_max_age_zero(self):
        assert "Max-Age=0" in make_clear_cookie(AUTH_COOKIE_SERVER)

    def test_empty_value(self):
        assert "turnstone_auth_server=;" in make_clear_cookie(AUTH_COOKIE_SERVER)

    def test_httponly(self):
        assert "HttpOnly" in make_clear_cookie(AUTH_COOKIE_SERVER)

    def test_uses_provided_name(self):
        val = make_clear_cookie(AUTH_COOKIE_CONSOLE)
        assert "turnstone_auth_console=;" in val
        assert "Max-Age=0" in val


# ---------------------------------------------------------------------------
# TestCheckRequest
# ---------------------------------------------------------------------------


class TestCheckRequest:
    """Tests for the main check_request() entry point."""

    _SECRET = "test-jwt-secret-minimum-32-chars!"

    @pytest.fixture()
    def read_jwt(self):
        return f"Bearer {create_jwt('u1', frozenset({'read'}), 'test', self._SECRET)}"

    @pytest.fixture()
    def full_jwt(self):
        return f"Bearer {create_jwt('u1', frozenset({'read', 'write', 'approve'}), 'test', self._SECRET)}"

    def test_public_path_no_token_ok(self):
        allowed, status, msg, _result = check_request(
            "GET", "/health", None, cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is True
        assert status == 200

    def test_public_root_no_token_ok(self):
        allowed, status, msg, _result = check_request(
            "GET", "/", None, cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is True

    def test_public_static_no_token_ok(self):
        allowed, status, msg, _result = check_request(
            "GET", "/static/style.css", None, cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is True

    def test_api_no_token_401(self):
        allowed, status, msg, _result = check_request(
            "GET", "/api/workstreams", None, cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is False
        assert status == 401
        assert "Unauthorized" in msg

    def test_api_invalid_token_401(self):
        allowed, status, msg, _result = check_request(
            "GET", "/api/workstreams", "Bearer wrong_token", cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is False
        assert status == 401

    def test_api_read_token_ok(self, read_jwt):
        allowed, status, msg, _result = check_request(
            "GET",
            "/api/workstreams",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True
        assert status == 200

    def test_api_full_token_ok(self, full_jwt):
        allowed, status, msg, _result = check_request(
            "GET",
            "/api/workstreams",
            full_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_write_read_token_403(self, read_jwt):
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/workstreams/abc/send",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403
        assert "Forbidden" in msg

    def test_write_full_token_ok(self, full_jwt):
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/workstreams/abc/send",
            full_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True
        assert status == 200

    def test_approve_read_token_403(self, read_jwt):
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/workstreams/abc/approve",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_proxy_write_read_token_403(self, read_jwt):
        """Read tokens cannot escalate to write ops via proxy routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/api/workstreams/abc/send",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_proxy_write_trailing_slash_read_token_403(self, read_jwt):
        """Trailing slash must not bypass write-role check on proxy routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/api/workstreams/abc/send/",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_direct_write_trailing_slash_read_token_403(self, read_jwt):
        """Trailing slash must not bypass write-role check on direct routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/workstreams/abc/send/",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_proxy_write_full_token_ok(self, full_jwt):
        """Full tokens pass through proxy write routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/api/workstreams/abc/send",
            full_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_proxy_v1_write_read_token_403(self, read_jwt):
        """Read tokens cannot escalate to write ops via v1 proxy routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/v1/api/workstreams/abc/send",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_proxy_v1_write_full_token_ok(self, full_jwt):
        """Full tokens pass through v1 proxy write routes."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/v1/api/workstreams/abc/send",
            full_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_proxy_v1_cluster_ws_new_read_403(self, read_jwt):
        """Read tokens cannot create workstreams via v1 proxy."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/node/node-a/v1/api/cluster/workstreams/new",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_proxy_read_endpoint_read_token_ok(self, read_jwt):
        """Read tokens can access proxy read endpoints."""
        allowed, status, msg, _result = check_request(
            "GET",
            "/node/node-a/api/workstreams",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_console_create_ws_read_token_403(self, read_jwt):
        """Read tokens cannot create workstreams."""
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/cluster/workstreams/new",
            read_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_approve_full_token_ok(self, full_jwt):
        allowed, status, msg, _result = check_request(
            "POST",
            "/api/workstreams/abc/approve",
            full_jwt,
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_no_auth_header_string(self):
        allowed, status, msg, _result = check_request(
            "GET", "/api/dashboard", "", cookie_name=AUTH_COOKIE_SERVER
        )
        assert allowed is False
        assert status == 401


# ---------------------------------------------------------------------------
# TestCheckRequestWithCookie
# ---------------------------------------------------------------------------


class TestCheckRequestWithCookie:
    """Tests for cookie-based auth fallback in check_request."""

    _SECRET = "test-jwt-secret-minimum-32-chars!"

    @pytest.fixture()
    def read_jwt(self):
        return create_jwt("u1", frozenset({"read"}), "test", self._SECRET)

    @pytest.fixture()
    def full_jwt(self):
        return create_jwt("u1", frozenset({"read", "write", "approve"}), "test", self._SECRET)

    def test_cookie_fallback_when_no_bearer(self, read_jwt):
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=f"{AUTH_COOKIE_SERVER}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True
        assert status == 200

    def test_bearer_takes_precedence_over_cookie(self, read_jwt, full_jwt):
        allowed, status, _, _r = check_request(
            "POST",
            "/api/workstreams/abc/send",
            f"Bearer {full_jwt}",
            cookie_header=f"{AUTH_COOKIE_SERVER}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_invalid_cookie_401(self):
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=f"{AUTH_COOKIE_SERVER}=wrong_token",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 401

    def test_cookie_read_on_write_403(self, read_jwt):
        allowed, status, _, _r = check_request(
            "POST",
            "/api/workstreams/abc/send",
            None,
            cookie_header=f"{AUTH_COOKIE_SERVER}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 403

    def test_cookie_full_on_write_ok(self, full_jwt):
        allowed, status, _, _r = check_request(
            "POST",
            "/api/workstreams/abc/send",
            None,
            cookie_header=f"{AUTH_COOKIE_SERVER}={full_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_no_cookie_no_bearer_401(self):
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=None,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 401

    def test_login_path_public(self):
        allowed, status, _, _r = check_request(
            "POST",
            "/api/auth/login",
            None,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    def test_logout_path_public(self):
        allowed, status, _, _r = check_request(
            "POST",
            "/api/auth/logout",
            None,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True

    # --- Cookie-name isolation (the server/console fix) -----------------------

    def test_legacy_cookie_name_rejected(self, read_jwt):
        """A pre-isolation ``turnstone_auth`` cookie no longer authenticates once
        the surface is configured for a scoped name: existing sessions must
        re-login after the rename (intentional hard cutover, no read-fallback)."""
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=f"{AUTH_COOKIE}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 401

    def test_console_cookie_name_not_read_by_server(self, read_jwt):
        """The console's cookie name is invisible to a server-configured surface,
        so a console session can't satisfy a server request (no cross-read)."""
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=f"{AUTH_COOKIE_CONSOLE}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is False
        assert status == 401

    def test_both_cookies_present_no_clobber(self, read_jwt):
        """With distinct names both surfaces' cookies coexist in one jar; the
        server reads only its own and authenticates even with a console cookie
        also present — the core property the rename buys."""
        allowed, status, _, _r = check_request(
            "GET",
            "/api/workstreams",
            None,
            cookie_header=f"{AUTH_COOKIE_CONSOLE}=other; {AUTH_COOKIE_SERVER}={read_jwt}",
            jwt_secret=self._SECRET,
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed is True
        assert status == 200


# ---------------------------------------------------------------------------
# Integration tests — actual HTTP server with auth enabled
# ---------------------------------------------------------------------------


class TestServerAuth:
    """Test turnstone-server with auth enabled using Starlette TestClient."""

    @classmethod
    def setup_class(cls):
        import queue
        import threading
        from unittest.mock import MagicMock

        from starlette.testclient import TestClient

        import turnstone.server as srv_mod
        from turnstone.core.metrics import MetricsCollector
        from turnstone.core.workstream import WorkstreamState

        srv_mod._metrics = MetricsCollector()
        srv_mod._metrics.model = "test-model"

        mock_session = MagicMock()
        mock_session.ws_id = "test-session-id"

        mock_ws = MagicMock()
        mock_ws.id = "test-ws"
        mock_ws.name = "test"
        mock_ws.state = WorkstreamState.IDLE
        mock_ws.session = mock_session
        # Set kind / parent_ws_id / user_id explicitly so list_workstreams
        # JSON-serializes them — a bare MagicMock attribute returns another
        # MagicMock that fails json.dumps and surfaces as 500.
        mock_ws.kind = "interactive"
        mock_ws.parent_ws_id = None
        mock_ws.user_id = "u1"
        mock_mgr = MagicMock()
        mock_mgr.list_all.return_value = [mock_ws]
        mock_mgr.max_active = 10

        from turnstone.core.auth import JWT_AUD_SERVER

        cls._jwt_secret = "test-jwt-secret-minimum-32-chars!"
        cls._read_hdr = {
            "Authorization": f"Bearer {create_jwt('u1', frozenset({'read'}), 'test', cls._jwt_secret, audience=JWT_AUD_SERVER)}"
        }
        cls._full_hdr = {
            "Authorization": f"Bearer {create_jwt('u1', frozenset({'read', 'write', 'approve'}), 'test', cls._jwt_secret, audience=JWT_AUD_SERVER)}"
        }
        app = srv_mod.create_app(
            workstreams=mock_mgr,
            global_queue=queue.Queue(),
            global_listeners=[],
            global_listeners_lock=threading.Lock(),
            skip_permissions=False,
            jwt_secret=cls._jwt_secret,
            cors_origins=["*"],
        )
        cls.client = TestClient(app, raise_server_exceptions=False)

    @classmethod
    def teardown_class(cls):
        cls.client.close()

    def test_health_no_token_200(self):
        resp = self.client.get("/health")
        assert resp.status_code == 200

    def test_metrics_no_token_passes_auth(self):
        resp = self.client.get("/metrics")
        assert resp.status_code not in (401, 403)

    def test_root_no_token_200(self):
        resp = self.client.get("/")
        assert resp.status_code == 200

    def test_static_css_no_token_200(self):
        resp = self.client.get("/static/style.css")
        assert resp.status_code == 200

    def test_api_workstreams_no_token_401(self):
        resp = self.client.get("/v1/api/workstreams")
        assert resp.status_code == 401
        assert "Unauthorized" in resp.json().get("error", "")

    def test_api_workstreams_read_token_200(self):
        resp = self.client.get("/v1/api/workstreams", headers=self._read_hdr)
        assert resp.status_code == 200

    def test_api_workstreams_full_token_200(self):
        resp = self.client.get("/v1/api/workstreams", headers=self._full_hdr)
        assert resp.status_code == 200

    def test_api_send_read_token_403(self):
        resp = self.client.post(
            "/v1/api/workstreams/x/send",
            headers=self._read_hdr,
            json={"message": "hello"},
        )
        assert resp.status_code == 403
        assert "Forbidden" in resp.json().get("error", "")

    def test_api_send_full_token_passes_auth(self):
        resp = self.client.post(
            "/v1/api/workstreams/nonexistent/send",
            headers=self._full_hdr,
            json={"message": "hello"},
        )
        assert resp.status_code not in (401, 403)

    def test_api_send_no_token_401(self):
        resp = self.client.post(
            "/v1/api/workstreams/x/send",
            json={"message": "hello"},
        )
        assert resp.status_code == 401

    def test_invalid_token_401(self):
        resp = self.client.get(
            "/v1/api/workstreams",
            headers={"Authorization": "Bearer wrong_token"},
        )
        assert resp.status_code == 401

    def test_options_no_auth_required(self):
        resp = self.client.options(
            "/v1/api/workstreams/x/send",
            headers={
                "Origin": "http://example.com",
                "Access-Control-Request-Method": "POST",
            },
        )
        assert resp.status_code == 200
        allowed = resp.headers.get("access-control-allow-headers", "")
        assert "authorization" in allowed.lower()

    def test_cors_includes_authorization(self):
        resp = self.client.options(
            "/v1/api/workstreams",
            headers={
                "Origin": "http://example.com",
                "Access-Control-Request-Method": "GET",
            },
        )
        allowed = resp.headers.get("access-control-allow-headers", "")
        assert "authorization" in allowed.lower()

    def test_shared_css_no_token_200(self):
        resp = self.client.get("/shared/base.css")
        assert resp.status_code == 200


class TestConsoleAuth:
    """Test console server with auth enabled using TestClient."""

    @classmethod
    def setup_class(cls):
        from unittest.mock import MagicMock

        from starlette.testclient import TestClient

        from turnstone.console.collector import ClusterCollector
        from turnstone.console.server import _load_static, create_app

        _load_static()

        mock_collector = MagicMock(spec=ClusterCollector)
        mock_collector.get_overview.return_value = {
            "nodes": 1,
            "workstreams": 2,
            "states": {"running": 1, "idle": 1},
            "aggregate": {"total_tokens": 100},
        }

        from turnstone.core.auth import JWT_AUD_CONSOLE

        cls._jwt_secret = "test-jwt-secret-minimum-32-chars!"
        cls._read_hdr = {
            "Authorization": f"Bearer {create_jwt('u1', frozenset({'read'}), 'test', cls._jwt_secret, audience=JWT_AUD_CONSOLE)}"
        }
        cls._full_hdr = {
            "Authorization": f"Bearer {create_jwt('u1', frozenset({'read', 'write', 'approve'}), 'test', cls._jwt_secret, audience=JWT_AUD_CONSOLE)}"
        }
        app = create_app(
            collector=mock_collector,
            jwt_secret=cls._jwt_secret,
        )
        cls.test_client = TestClient(app, raise_server_exceptions=False)

    @classmethod
    def teardown_class(cls):
        cls.test_client.close()

    def test_health_no_token_200(self):
        resp = self.test_client.get("/health")
        assert resp.status_code == 200

    def test_root_no_token_200(self):
        resp = self.test_client.get("/")
        assert resp.status_code == 200

    def test_api_overview_no_token_401(self):
        resp = self.test_client.get("/v1/api/cluster/overview")
        assert resp.status_code == 401

    def test_api_overview_read_token_200(self):
        resp = self.test_client.get("/v1/api/cluster/overview", headers=self._read_hdr)
        assert resp.status_code == 200

    def test_api_overview_full_token_200(self):
        resp = self.test_client.get("/v1/api/cluster/overview", headers=self._full_hdr)
        assert resp.status_code == 200

    def test_invalid_token_401(self):
        resp = self.test_client.get(
            "/v1/api/cluster/overview",
            headers={"Authorization": "Bearer wrong"},
        )
        assert resp.status_code == 401


# ---------------------------------------------------------------------------
# Login / Logout integration tests
# ---------------------------------------------------------------------------


class TestServerLogin:
    """Test login/logout cookie flow on turnstone-server."""

    @classmethod
    def setup_class(cls):
        import queue
        import threading
        from unittest.mock import MagicMock

        from starlette.testclient import TestClient

        import turnstone.server as srv_mod
        from turnstone.core.metrics import MetricsCollector
        from turnstone.core.workstream import WorkstreamState

        srv_mod._metrics = MetricsCollector()
        srv_mod._metrics.model = "test-model"

        mock_session = MagicMock()
        mock_session.ws_id = "test-session-id"

        mock_ws = MagicMock()
        mock_ws.id = "test-ws"
        mock_ws.name = "test"
        mock_ws.state = WorkstreamState.IDLE
        mock_ws.session = mock_session
        # Set kind / parent_ws_id / user_id explicitly so list_workstreams
        # JSON-serializes them — a bare MagicMock attribute returns another
        # MagicMock that fails json.dumps and surfaces as 500.
        mock_ws.kind = "interactive"
        mock_ws.parent_ws_id = None
        mock_ws.user_id = "u1"
        mock_mgr = MagicMock()
        mock_mgr.list_all.return_value = [mock_ws]
        mock_mgr.max_active = 10

        # Mock storage with a test user for password login
        from turnstone.core.auth import hash_password

        mock_storage = MagicMock()
        mock_storage.get_user_by_username.side_effect = lambda u: (
            {
                "user_id": "uid_test",
                "username": "testuser",
                "password_hash": hash_password("testpass"),
                "display_name": "Test",
            }
            if u == "testuser"
            else None
        )
        # whoami resolves the human username/display-name by user_id for the UI.
        mock_storage.get_user.side_effect = lambda uid: (
            {"user_id": "uid_test", "username": "testuser", "display_name": "Test"}
            if uid == "uid_test"
            else None
        )
        mock_storage.list_user_roles.return_value = [
            {"role_id": "builtin-admin", "scopes": "read,write,approve"}
        ]

        cls._jwt_secret = "test-jwt-secret-minimum-32-chars!"
        app = srv_mod.create_app(
            workstreams=mock_mgr,
            global_queue=queue.Queue(),
            global_listeners=[],
            global_listeners_lock=threading.Lock(),
            skip_permissions=False,
            jwt_secret=cls._jwt_secret,
            auth_storage=mock_storage,
        )
        cls.test_client = TestClient(app, raise_server_exceptions=False)

    @classmethod
    def teardown_class(cls):
        cls.test_client.close()

    def test_login_config_token_rejected(self):
        """Config token exchange is no longer allowed."""
        resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"token": "tok_full"},
        )
        assert resp.status_code == 401

    def test_login_invalid_credentials_401(self):
        resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "wrong"},
        )
        assert resp.status_code == 401

    def test_login_password_ok(self):
        resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "jwt" in data
        # Server surface sets ITS OWN scoped cookie — not the console's, not the legacy name.
        set_cookie = resp.headers.get("set-cookie", "")
        assert AUTH_COOKIE_SERVER in set_cookie
        assert AUTH_COOKIE_CONSOLE not in set_cookie

    def test_cookie_auth_on_api(self):
        # Login to get cookie (TestClient tracks cookies automatically)
        login_resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert login_resp.status_code == 200

        # Use cookie to access API — TestClient forwards cookies
        resp = self.test_client.get("/v1/api/workstreams")
        assert resp.status_code == 200

    def test_logout_clears_cookie(self):
        self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )

        # Logout
        logout_resp = self.test_client.post("/v1/api/auth/logout")
        assert logout_resp.status_code == 200
        cookie = logout_resp.headers.get("set-cookie", "")
        assert "Max-Age=0" in cookie
        assert AUTH_COOKIE_SERVER in cookie

        # API should now fail (cookie cleared)
        resp = self.test_client.get("/v1/api/workstreams")
        assert resp.status_code == 401

    def test_whoami_includes_exp(self):
        """whoami exposes the JWT exp so the frontend can schedule refresh."""
        import time

        self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        resp = self.test_client.get("/v1/api/auth/whoami")
        assert resp.status_code == 200
        data = resp.json()
        assert "exp" in data
        # whoami surfaces the human display name for the UI, not the opaque
        # user_id uuid (the rail footer renders this).
        assert data.get("username") == "Test"
        # Default JWT TTL is 24h; exp should be > now and < now + 25h.
        now = int(time.time())
        assert now < data["exp"] < now + 25 * 3600

    def test_refresh_returns_new_jwt_and_cookie(self):
        """POST /api/auth/refresh re-mints the cookie with a fresh exp."""
        # Storage needs get_user_permissions for the refresh re-resolve path.
        # Mock is shared across tests in the class — re-arm here in case a
        # prior test left it default.
        self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
            "read",
            "write",
            "approve",
        }

        login = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert login.status_code == 200

        refresh = self.test_client.post("/v1/api/auth/refresh")
        assert refresh.status_code == 200
        body = refresh.json()
        assert body["status"] == "ok"
        assert body["user_id"] == "uid_test"
        assert "jwt" in body
        # Set-Cookie header must be present so the browser updates.  Don't
        # assert the new JWT differs from the original — sub-second login
        # and refresh produce identical iat/exp claims and therefore an
        # identical token, which is fine: the cookie still gets re-set.
        cookie_hdr = refresh.headers.get("set-cookie", "")
        assert AUTH_COOKIE_SERVER in cookie_hdr
        assert "HttpOnly" in cookie_hdr

        # The refreshed cookie must keep working.
        resp = self.test_client.get("/v1/api/workstreams")
        assert resp.status_code == 200

    def test_refresh_response_includes_exp_and_permissions(self):
        """Refresh response shape must match whoami so the frontend can
        populate sessionStorage + reschedule the next refresh off the
        single round-trip without a follow-up /whoami call."""
        import time

        self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
            "read",
            "write",
            "approve",
        }
        login = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert login.status_code == 200

        refresh = self.test_client.post("/v1/api/auth/refresh")
        assert refresh.status_code == 200
        body = refresh.json()
        # exp present + within the expected default JWT TTL window
        assert "exp" in body, body
        now = int(time.time())
        assert now < body["exp"] < now + 25 * 3600, body
        # permissions present + non-empty (matches the seeded role set)
        assert body.get("permissions"), body
        assert "write" in body["permissions"].split(",")

    def test_refresh_unauthenticated_401(self):
        """Refresh requires a currently-valid cookie — no cookie → 401."""
        # Clear cookies on the test client
        self.test_client.cookies.clear()
        resp = self.test_client.post("/v1/api/auth/refresh")
        assert resp.status_code == 401

    def test_refresh_storage_failure_falls_back(self):
        """Transient storage error → fall back to in-token claims, not 403.

        The earlier implementation called _load_user_permissions() which
        swallows exceptions and returns set(); that path was
        indistinguishable from a deleted user (legitimate 403).  The
        handler now calls storage.get_user_permissions() directly so
        DB hiccups fall through to in-token perms.
        """
        # Re-arm the storage so login works first
        self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
            "read",
            "write",
            "approve",
        }
        login = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert login.status_code == 200

        # Now make storage raise on the refresh re-resolve
        self.test_client.app.state.auth_storage.get_user_permissions.side_effect = RuntimeError(
            "db down"
        )
        try:
            resp = self.test_client.post("/v1/api/auth/refresh")
            assert resp.status_code == 200, resp.text
            body = resp.json()
            # Permissions should still be present (fell back to in-token claims)
            assert body.get("permissions"), body
        finally:
            # Restore for any subsequent tests
            self.test_client.app.state.auth_storage.get_user_permissions.side_effect = None
            self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
                "read",
                "write",
                "approve",
            }

    def test_refresh_user_with_no_perms_403(self):
        """Storage returns empty (user deleted/role-stripped) → 403.

        Distinguished from the storage-failure case above because
        get_user_permissions returned a value (the empty set) without
        raising — that's an authoritative "no roles", not a hiccup.
        """
        self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
            "read",
            "write",
            "approve",
        }
        login = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert login.status_code == 200

        self.test_client.app.state.auth_storage.get_user_permissions.return_value = set()
        try:
            resp = self.test_client.post("/v1/api/auth/refresh")
            assert resp.status_code == 403
        finally:
            self.test_client.app.state.auth_storage.get_user_permissions.return_value = {
                "read",
                "write",
                "approve",
            }


class TestConsoleLogin:
    """Test login/logout cookie flow on turnstone-console."""

    @classmethod
    def setup_class(cls):
        from unittest.mock import MagicMock

        from starlette.testclient import TestClient

        from turnstone.console.collector import ClusterCollector
        from turnstone.console.server import _load_static, create_app
        from turnstone.core.auth import hash_password

        _load_static()

        mock_collector = MagicMock(spec=ClusterCollector)
        mock_collector.get_overview.return_value = {
            "nodes": 1,
            "workstreams": 2,
            "states": {"running": 1, "idle": 1},
            "aggregate": {"total_tokens": 100},
        }

        mock_storage = MagicMock()
        mock_storage.get_user_by_username.side_effect = lambda u: (
            {
                "user_id": "uid_test",
                "username": "testuser",
                "password_hash": hash_password("testpass"),
                "display_name": "Test",
            }
            if u == "testuser"
            else None
        )
        # whoami resolves the human username/display-name by user_id for the UI.
        mock_storage.get_user.side_effect = lambda uid: (
            {"user_id": "uid_test", "username": "testuser", "display_name": "Test"}
            if uid == "uid_test"
            else None
        )
        mock_storage.list_user_roles.return_value = [
            {"role_id": "builtin-admin", "scopes": "read,write,approve"}
        ]

        cls._jwt_secret = "test-jwt-secret-minimum-32-chars!"
        app = create_app(
            collector=mock_collector,
            jwt_secret=cls._jwt_secret,
            auth_storage=mock_storage,
        )
        cls.test_client = TestClient(app, raise_server_exceptions=False)

    @classmethod
    def teardown_class(cls):
        cls.test_client.close()

    def test_login_config_token_rejected(self):
        resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"token": "tok_read"},
        )
        assert resp.status_code == 401

    def test_login_password_ok(self):
        resp = self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        assert resp.status_code == 200
        # Exact scoped name (not the legacy prefix) and no server-name bleed.
        set_cookie = resp.headers.get("set-cookie", "")
        assert AUTH_COOKIE_CONSOLE in set_cookie
        assert AUTH_COOKIE_SERVER not in set_cookie

    def test_cookie_auth_on_api(self):
        self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        resp = self.test_client.get("/v1/api/cluster/overview")
        assert resp.status_code == 200

    def test_logout_then_api_fails(self):
        self.test_client.post(
            "/v1/api/auth/login",
            json={"username": "testuser", "password": "testpass"},
        )
        self.test_client.post("/v1/api/auth/logout")
        resp = self.test_client.get("/v1/api/cluster/overview")
        assert resp.status_code == 401


# ---------------------------------------------------------------------------
# Security hardening tests
# ---------------------------------------------------------------------------


class TestLoginRateLimiter:
    def test_allows_under_limit(self):
        from turnstone.core.auth import LoginRateLimiter

        limiter = LoginRateLimiter(max_attempts=3, window_seconds=60)
        for _ in range(3):
            ok, _ = limiter.check("ip:1.2.3.4")
            assert ok
            limiter.record("ip:1.2.3.4")
        # 4th should be blocked (3 recorded)
        ok, retry = limiter.check("ip:1.2.3.4")
        assert not ok
        assert retry > 0

    def test_different_keys_independent(self):
        from turnstone.core.auth import LoginRateLimiter

        limiter = LoginRateLimiter(max_attempts=2, window_seconds=60)
        limiter.record("ip:a")
        limiter.record("ip:a")
        ok_a, _ = limiter.check("ip:a")
        ok_b, _ = limiter.check("ip:b")
        assert not ok_a
        assert ok_b

    def test_cleanup(self):
        from turnstone.core.auth import LoginRateLimiter

        limiter = LoginRateLimiter(max_attempts=1, window_seconds=60)
        limiter.record("ip:old")
        removed = limiter.cleanup(max_age=0.0)
        assert removed == 1
        ok, _ = limiter.check("ip:old")
        assert ok

    def test_max_keys_protection(self):
        from turnstone.core.auth import LoginRateLimiter

        limiter = LoginRateLimiter(max_attempts=5, window_seconds=60)
        limiter.MAX_KEYS = 2
        limiter.record("a")
        limiter.record("b")
        limiter.record("c")  # should be silently dropped (at capacity)
        assert "c" not in limiter._attempts


class TestJWTAudienceIssuer:
    SECRET = "test-secret-that-is-at-least-32-chars"

    def test_create_jwt_includes_iss(self):
        import jwt as pyjwt

        from turnstone.core.auth import JWT_ISSUER, create_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET)
        payload = pyjwt.decode(
            token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        assert payload["iss"] == JWT_ISSUER

    def test_create_jwt_with_audience(self):
        import jwt as pyjwt

        from turnstone.core.auth import JWT_AUD_SERVER, create_jwt

        token = create_jwt(
            "user1", frozenset({"read"}), "test", self.SECRET, audience=JWT_AUD_SERVER
        )
        payload = pyjwt.decode(token, self.SECRET, algorithms=["HS256"], audience=JWT_AUD_SERVER)
        assert payload["aud"] == JWT_AUD_SERVER

    def test_validate_jwt_wrong_audience_rejected(self):
        from turnstone.core.auth import JWT_AUD_CONSOLE, JWT_AUD_SERVER, create_jwt, validate_jwt

        token = create_jwt(
            "user1", frozenset({"read"}), "test", self.SECRET, audience=JWT_AUD_SERVER
        )
        result = validate_jwt(token, self.SECRET, audience=JWT_AUD_CONSOLE)
        assert result is None

    def test_validate_jwt_correct_audience_accepted(self):
        from turnstone.core.auth import JWT_AUD_SERVER, create_jwt, validate_jwt

        token = create_jwt(
            "user1", frozenset({"read"}), "test", self.SECRET, audience=JWT_AUD_SERVER
        )
        result = validate_jwt(token, self.SECRET, audience=JWT_AUD_SERVER)
        assert result is not None
        assert result.user_id == "user1"

    def test_validate_jwt_no_audience_backward_compat(self):
        from turnstone.core.auth import create_jwt, validate_jwt

        # Token without aud claim should be accepted when audience="" (backward compat)
        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET)
        result = validate_jwt(token, self.SECRET, audience="")
        assert result is not None

    def test_validate_jwt_accepts_within_leeway_after_expiry(self):
        """validate_jwt has 30s leeway for clock skew across hosts/processes."""
        import time

        import jwt as pyjwt

        from turnstone.core.auth import JWT_ISSUER, validate_jwt

        # Mint a token that "expired" 10 seconds ago — still within 30s leeway.
        now = int(time.time())
        token = pyjwt.encode(
            {
                "sub": "user1",
                "scopes": "read",
                "src": "test",
                "iss": JWT_ISSUER,
                "iat": now - 100,
                "exp": now - 10,
            },
            self.SECRET,
            algorithm="HS256",
        )
        result = validate_jwt(token, self.SECRET, audience="")
        assert result is not None
        assert result.user_id == "user1"

    def test_validate_jwt_rejects_past_leeway(self):
        """Tokens expired beyond the 30s leeway must still be rejected."""
        import time

        import jwt as pyjwt

        from turnstone.core.auth import JWT_ISSUER, validate_jwt

        now = int(time.time())
        token = pyjwt.encode(
            {
                "sub": "user1",
                "scopes": "read",
                "src": "test",
                "iss": JWT_ISSUER,
                "iat": now - 200,
                "exp": now - 60,
            },
            self.SECRET,
            algorithm="HS256",
        )
        result = validate_jwt(token, self.SECRET, audience="")
        assert result is None

    def test_create_jwt_expiry_seconds(self):
        import jwt as pyjwt

        from turnstone.core.auth import create_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET, expiry_seconds=300)
        payload = pyjwt.decode(
            token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        assert payload["exp"] - payload["iat"] == 300

    def test_create_jwt_expiry_seconds_overrides_hours(self):
        import jwt as pyjwt

        from turnstone.core.auth import create_jwt

        token = create_jwt(
            "user1",
            frozenset({"read"}),
            "test",
            self.SECRET,
            expiry_hours=24,
            expiry_seconds=60,
        )
        payload = pyjwt.decode(
            token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        # expiry_seconds takes precedence over expiry_hours
        assert payload["exp"] - payload["iat"] == 60

    def test_create_jwt_expiry_seconds_rejects_zero(self):
        import pytest

        from turnstone.core.auth import create_jwt

        with pytest.raises(ValueError, match="expiry_seconds must be positive"):
            create_jwt("user1", frozenset({"read"}), "test", self.SECRET, expiry_seconds=0)

    def test_create_jwt_expiry_seconds_rejects_negative(self):
        import pytest

        from turnstone.core.auth import create_jwt

        with pytest.raises(ValueError, match="expiry_seconds must be positive"):
            create_jwt("user1", frozenset({"read"}), "test", self.SECRET, expiry_seconds=-1)


class TestJWTVersionClaim:
    SECRET = "test-secret-that-is-at-least-32-chars"

    def test_create_jwt_with_version(self):
        import jwt as pyjwt

        from turnstone.core.auth import create_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET, version="1.2")
        payload = pyjwt.decode(
            token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        assert payload["ver"] == "1.2"

    def test_create_jwt_without_version(self):
        import jwt as pyjwt

        from turnstone.core.auth import create_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET)
        payload = pyjwt.decode(
            token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        assert "ver" not in payload

    def test_validate_jwt_carries_token_version(self):
        from turnstone.core.auth import create_jwt, validate_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET, version="1.2")
        result = validate_jwt(token, self.SECRET)
        assert result is not None
        assert result.user_id == "user1"
        assert result.token_version == "1.2"

    def test_validate_jwt_no_ver_returns_empty_token_version(self):
        from turnstone.core.auth import create_jwt, validate_jwt

        token = create_jwt("user1", frozenset({"read"}), "test", self.SECRET)
        result = validate_jwt(token, self.SECRET)
        assert result is not None
        assert result.token_version == ""

    def test_check_request_accepts_matching_version(self):
        from turnstone.core.auth import JWT_AUD_SERVER, check_request, create_jwt

        token = create_jwt(
            "user1",
            frozenset({"read"}),
            "test",
            self.SECRET,
            audience=JWT_AUD_SERVER,
            version="1.2",
        )
        allowed, _status, _msg, result = check_request(
            "GET",
            "/v1/api/workstreams",
            f"Bearer {token}",
            jwt_secret=self.SECRET,
            jwt_audience=JWT_AUD_SERVER,
            jwt_version="1.2",
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed
        assert result is not None

    def test_check_request_accepts_no_ver_backward_compat(self):
        from turnstone.core.auth import JWT_AUD_SERVER, check_request, create_jwt

        # Token without ver claim should be accepted (backward compat)
        token = create_jwt(
            "user1",
            frozenset({"read"}),
            "test",
            self.SECRET,
            audience=JWT_AUD_SERVER,
        )
        allowed, _status, _msg, _result = check_request(
            "GET",
            "/v1/api/workstreams",
            f"Bearer {token}",
            jwt_secret=self.SECRET,
            jwt_audience=JWT_AUD_SERVER,
            jwt_version="1.2",
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert allowed

    def test_check_request_rejects_old_version_jwt(self):
        from turnstone.core.auth import JWT_AUD_SERVER, check_request, create_jwt

        token = create_jwt(
            "user1",
            frozenset({"read"}),
            "test",
            self.SECRET,
            audience=JWT_AUD_SERVER,
            version="1.1",
        )
        allowed, status, msg, _result = check_request(
            "GET",
            "/v1/api/workstreams",
            f"Bearer {token}",
            jwt_secret=self.SECRET,
            jwt_audience=JWT_AUD_SERVER,
            jwt_version="1.2",
            cookie_name=AUTH_COOKIE_SERVER,
        )
        assert not allowed
        assert status == 401
        assert msg == "version_mismatch"


class TestVersionSlot:
    def test_returns_major_minor(self):
        from turnstone.core.auth import jwt_version_slot

        slot = jwt_version_slot()
        parts = slot.split(".")
        assert len(parts) == 2

    def test_strips_patch_and_prerelease(self):
        from unittest.mock import patch

        with patch("turnstone.__version__", "2.3.1a5"):
            from turnstone.core.auth import jwt_version_slot

            assert jwt_version_slot() == "2.3"


class TestServiceTokenManager:
    SECRET = "test-secret-that-is-at-least-32-chars"

    def test_auto_mints_on_first_access(self):
        from turnstone.core.auth import ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
        )
        token = mgr.token
        assert token  # non-empty
        assert isinstance(token, str)

    def test_bearer_header_format(self):
        from turnstone.core.auth import ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
        )
        header = mgr.bearer_header
        assert "Authorization" in header
        assert header["Authorization"].startswith("Bearer ")

    def test_token_stable_within_window(self):
        from turnstone.core.auth import ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
            expiry_hours=1,
        )
        t1 = mgr.token
        t2 = mgr.token
        assert t1 == t2

    def test_token_rotates_near_expiry(self):
        from turnstone.core.auth import ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
            expiry_hours=1,
        )
        _ = mgr.token  # initial mint
        # Simulate expiry by backdating _expires_at
        mgr._expires_at = 0.0
        t2 = mgr.token
        # Token was re-minted (even if payload matches within same second,
        # the internal state was refreshed)
        assert t2  # non-empty, valid token
        assert mgr._expires_at > 0.0  # was refreshed

    def test_audience_included(self):
        import jwt as pyjwt

        from turnstone.core.auth import JWT_AUD_SERVER, ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
            audience=JWT_AUD_SERVER,
        )
        payload = pyjwt.decode(
            mgr.token, self.SECRET, algorithms=["HS256"], audience=JWT_AUD_SERVER
        )
        assert payload["aud"] == JWT_AUD_SERVER

    def test_service_token_no_version_claim(self):
        import jwt as pyjwt

        from turnstone.core.auth import ServiceTokenManager

        mgr = ServiceTokenManager(
            user_id="svc",
            scopes=frozenset({"read"}),
            source="test",
            secret=self.SECRET,
        )
        payload = pyjwt.decode(
            mgr.token, self.SECRET, algorithms=["HS256"], options={"verify_aud": False}
        )
        assert "ver" not in payload


class TestIsSecureRequest:
    def test_https_scheme(self):
        from turnstone.core.auth import is_secure_request

        assert is_secure_request({}, scheme="https") is True

    def test_http_scheme(self):
        from turnstone.core.auth import is_secure_request

        assert is_secure_request({}, scheme="http") is False

    def test_x_forwarded_proto_https(self):
        from turnstone.core.auth import is_secure_request

        assert is_secure_request({"x-forwarded-proto": "https"}, scheme="http") is True

    def test_x_forwarded_proto_http(self):
        from turnstone.core.auth import is_secure_request

        assert is_secure_request({"x-forwarded-proto": "http"}, scheme="http") is False


class TestSecretStrength:
    def test_short_secret_exits(self):
        old = os.environ.get("TURNSTONE_JWT_SECRET", "")
        os.environ["TURNSTONE_JWT_SECRET"] = "short"
        try:
            with pytest.raises(SystemExit):
                load_jwt_secret()
        finally:
            if old:
                os.environ["TURNSTONE_JWT_SECRET"] = old
            else:
                os.environ.pop("TURNSTONE_JWT_SECRET", None)

    def test_missing_secret_exits(self):
        with (
            patch("turnstone.core.config.load_config", return_value={}),
            patch.dict(os.environ, {}, clear=True),
            pytest.raises(SystemExit),
        ):
            load_jwt_secret()


class TestCorsConfigurable:
    """Verify CORS middleware is only added when origins are configured."""

    def test_no_cors_origins_no_cors_headers(self):
        """Without cors_origins, no Access-Control headers."""
        from starlette.testclient import TestClient

        import turnstone.server as srv_mod

        mgr = MagicMock()
        mgr.list_all.return_value = []
        mgr.max_active = 10
        app = srv_mod.create_app(
            workstreams=mgr,
            global_queue=queue.Queue(),
            global_listeners=[],
            global_listeners_lock=threading.Lock(),
            skip_permissions=False,
        )
        client = TestClient(app)
        resp = client.get("/health", headers={"Origin": "http://evil.com"})
        assert "Access-Control-Allow-Origin" not in resp.headers
        client.close()

    def test_cors_origins_set(self):
        """With cors_origins, CORS headers are present."""
        from starlette.testclient import TestClient

        import turnstone.server as srv_mod

        mgr = MagicMock()
        mgr.list_all.return_value = []
        mgr.max_active = 10
        app = srv_mod.create_app(
            workstreams=mgr,
            global_queue=queue.Queue(),
            global_listeners=[],
            global_listeners_lock=threading.Lock(),
            skip_permissions=False,
            cors_origins=["http://example.com"],
        )
        client = TestClient(app)
        resp = client.get(
            "/health",
            headers={"Origin": "http://example.com"},
        )
        assert resp.headers.get("Access-Control-Allow-Origin") == "http://example.com"
        client.close()


# ---------------------------------------------------------------------------
# TestVerifyPassword — OIDC sentinel handling
# ---------------------------------------------------------------------------


class TestVerifyPassword:
    def test_valid_bcrypt_hash(self):
        from turnstone.core.auth import hash_password, verify_password

        hashed = hash_password("mypassword")
        assert verify_password("mypassword", hashed) is True
        assert verify_password("wrongpassword", hashed) is False

    def test_oidc_sentinel_rejected(self):
        from turnstone.core.auth import verify_password

        # OIDC sentinel must return False, not crash with ValueError
        assert verify_password("anypassword", "!oidc") is False

    def test_non_bcrypt_hash_rejected(self):
        from turnstone.core.auth import verify_password

        assert verify_password("password", "not_a_hash") is False
        assert verify_password("password", "") is False

    def test_empty_password_against_oidc_sentinel(self):
        from turnstone.core.auth import verify_password

        assert verify_password("", "!oidc") is False


# ---------------------------------------------------------------------------
# TestOIDCPublicPaths — OIDC endpoints are public
# ---------------------------------------------------------------------------


class TestOIDCPublicPaths:
    def test_oidc_authorize_is_public(self):
        assert is_public_path("/api/auth/oidc/authorize") is True
        assert is_public_path("/v1/api/auth/oidc/authorize") is True

    def test_oidc_callback_is_public(self):
        assert is_public_path("/api/auth/oidc/callback") is True
        assert is_public_path("/v1/api/auth/oidc/callback") is True


# ---------------------------------------------------------------------------
# TestRequirePermissionServiceScope — service scope bypasses permission checks
# ---------------------------------------------------------------------------


class TestRequirePermissionServiceScope:
    """Verify require_permission() behaviour with the service scope."""

    def _make_request(self, auth_result):
        """Build a mock Starlette request with the given AuthResult on state."""
        request = MagicMock()
        request.state.auth_result = auth_result
        return request

    def test_service_scope_bypasses_permission(self):
        """Service-scoped tokens bypass all permission checks (returns None)."""
        from turnstone.core.auth import AuthResult, require_permission

        auth = AuthResult(
            user_id="svc-agent",
            scopes=frozenset({"service"}),
            token_source="jwt",
        )
        request = self._make_request(auth)
        result = require_permission(request, "admin.users")
        assert result is None  # bypass — no 403

    def test_without_service_scope_and_without_permission_returns_403(self):
        """Non-service tokens without the required permission get 403."""
        from turnstone.core.auth import AuthResult, require_permission

        auth = AuthResult(
            user_id="regular-user",
            scopes=frozenset({"read", "write"}),
            token_source="jwt",
        )
        request = self._make_request(auth)
        result = require_permission(request, "admin.users")
        assert result is not None
        assert result.status_code == 403

    def test_without_service_scope_with_permission_returns_none(self):
        """Non-service tokens with the required permission pass."""
        from turnstone.core.auth import AuthResult, require_permission

        auth = AuthResult(
            user_id="admin-user",
            scopes=frozenset({"read", "write", "approve"}),
            token_source="jwt",
            permissions=frozenset({"admin.users"}),
        )
        request = self._make_request(auth)
        result = require_permission(request, "admin.users")
        assert result is None  # granted — no 403

    def test_no_auth_result_returns_401(self):
        """Missing auth_result on request state returns 401."""
        from turnstone.core.auth import require_permission

        request = MagicMock()
        del request.state.auth_result  # ensure attribute is absent
        result = require_permission(request, "admin.users")
        assert result is not None
        assert result.status_code == 401


# ---------------------------------------------------------------------------
# TestUserHasPermission — in-process permission check for tool exec paths
# ---------------------------------------------------------------------------


class TestUserHasPermission:
    """In-process permission helper for model-facing tool exec paths.

    Distinct from ``require_permission`` (HTTP-only, returns JSONResponse);
    this helper returns a plain bool so tool callers can shape the denial
    themselves.  Loads permissions through storage on every call — there's
    no per-session cache, by design: a permission revocation should take
    effect on the next tool call, not require a session restart.
    """

    def test_returns_true_when_user_holds_permission(self):
        from turnstone.core.auth import user_has_permission

        storage = MagicMock()
        storage.get_user_permissions.return_value = {"model.skills.write", "read"}
        assert user_has_permission("alice", "model.skills.write", storage=storage) is True

    def test_returns_false_when_user_lacks_permission(self):
        from turnstone.core.auth import user_has_permission

        storage = MagicMock()
        storage.get_user_permissions.return_value = {"read", "write"}
        assert user_has_permission("alice", "model.skills.write", storage=storage) is False

    def test_empty_user_id_returns_false_without_storage_lookup(self):
        """Empty user_id short-circuits — no anonymous permission holder."""
        from turnstone.core.auth import user_has_permission

        storage = MagicMock()
        assert user_has_permission("", "model.skills.write", storage=storage) is False
        storage.get_user_permissions.assert_not_called()

    def test_storage_failure_returns_false_fail_closed(self):
        """Roles backend hiccups must deny, not allow (fail-closed)."""
        from turnstone.core.auth import user_has_permission

        storage = MagicMock()
        storage.get_user_permissions.side_effect = RuntimeError("DB down")
        assert user_has_permission("alice", "model.skills.write", storage=storage) is False

    def test_unregistered_storage_returns_false(self, monkeypatch):
        """Storage registry returning None (pre-init) denies without raising.

        Only the model-tool path can land here — HTTP handlers run after
        the auth middleware which already requires storage.
        """
        from turnstone.core import auth as _auth_mod

        monkeypatch.setattr(
            "turnstone.core.storage._registry.get_storage", lambda: None, raising=True
        )
        assert _auth_mod.user_has_permission("alice", "model.skills.write") is False

    def test_each_call_hits_storage_no_implicit_cache(self):
        """Pin the load-bearing 'no caching' contract from the class docstring.

        Future refactor that adds an ``lru_cache`` decorator, a per-session
        cache, or any process-wide memoization would silently break
        revocation latency (an admin revoking ``model.skills.write`` from a
        role would see the model still able to write skills until cache
        expiry / session restart).  If a cache is added intentionally, this
        test should be rewritten to assert the invalidation contract — not
        deleted.
        """
        from turnstone.core.auth import user_has_permission

        storage = MagicMock()
        storage.get_user_permissions.return_value = {"model.skills.write"}
        user_has_permission("alice", "model.skills.write", storage=storage)
        user_has_permission("alice", "model.skills.write", storage=storage)
        assert storage.get_user_permissions.call_count == 2


# ---------------------------------------------------------------------------
# TestBuiltinAdminDefaultPermissions — lock the "ungranted by default" invariant
# ---------------------------------------------------------------------------


class TestBuiltinAdminDefaultPermissions:
    """Regression guards on what builtin-admin gets out of the box.

    The migration chain (008 seed + 017 catch-up + later additive
    migrations) is the source of truth for builtin-admin's permission
    set.  Permissions intentionally ungranted by default — currently
    ``model.skills.write`` — must stay absent from that chain, or
    operators upgrading from older versions silently inherit a
    capability they never consented to.  Mirrors the
    ``tests/test_migration_049.py`` pattern: drive Alembic forward
    against an isolated SQLite DB and inspect the resulting row.
    """

    def _alembic_cfg(self, db_path):
        from pathlib import Path

        from alembic.config import Config

        migrations_dir = str(
            Path(__file__).resolve().parent.parent / "turnstone" / "core" / "storage" / "migrations"
        )
        cfg = Config()
        cfg.set_main_option("script_location", migrations_dir)
        cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
        return cfg

    def test_model_skills_write_not_in_builtin_admin_after_full_migration(self, tmp_path):
        """After every shipped migration, ``builtin-admin.permissions`` must
        not contain ``model.skills.write``.  A migration that grants it
        breaks the explicit-opt-in security contract documented in the
        ``_VALID_PERMISSIONS`` block in ``console/server.py``.
        """
        import sqlalchemy as sa
        from alembic import command

        db_path = tmp_path / "perm.db"
        cfg = self._alembic_cfg(db_path)
        command.upgrade(cfg, "head")

        engine = sa.create_engine(f"sqlite:///{db_path}")
        try:
            with engine.connect() as conn:
                row = conn.execute(
                    sa.text("SELECT permissions FROM roles WHERE role_id = 'builtin-admin'")
                ).fetchone()
        finally:
            engine.dispose()

        assert row is not None, "builtin-admin role not seeded by migration chain"
        perms = {p.strip() for p in (row[0] or "").split(",") if p.strip()}
        assert "model.skills.write" not in perms, (
            "builtin-admin must NOT hold model.skills.write by default — "
            f"got perms={sorted(perms)}.  If a migration intentionally "
            "added this grant, update the security contract in "
            "``console/server.py`` _VALID_PERMISSIONS docstring first."
        )