test_server_pbt.py

"""
Property-based tests for server.py — Invalid token rejection (Property 10).

**Validates: Requirements 2.2, 2.5**

Property 10: Invalid token always rejected — for any string that is not a valid,
unexpired JWT signed with the correct secret (including the empty string, arbitrary
text, malformed JWT segments, and tokens with a past `exp` claim), a request to any
protected endpoint using that string as the Bearer token SHALL receive HTTP 401.
"""

import hashlib
import string
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

import jwt
import pytest
from hypothesis import given, settings, assume
from hypothesis import strategies as st
from httpx import ASGITransport, AsyncClient

from server import app

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

TEST_PASSWORD = "property-test-secret"
WRONG_SECRET = "completely-wrong-secret-key"


def _jwt_secret_for(password: str) -> str:
    """Mirror server._jwt_secret() logic."""
    return hashlib.sha256(password.encode()).hexdigest()


# ---------------------------------------------------------------------------
# Strategies
# ---------------------------------------------------------------------------

# Strategy 1: Arbitrary ASCII text strings (includes empty string)
# HTTP headers only support ASCII, so we constrain to printable ASCII characters
# to avoid httpx encoding errors (which are transport-level, not auth-level).
_ascii_printable = string.ascii_letters + string.digits + string.punctuation + " "
arbitrary_text = st.text(alphabet=_ascii_printable, min_size=0, max_size=500)

# Strategy 2: Malformed JWT segments (wrong number of dots)
# Valid JWTs have exactly 2 dots (header.payload.signature)
# Generate strings with 0, 1, 3, or 4+ dots
_base64_chars = string.ascii_letters + string.digits + "-_"
_base64_segment = st.text(alphabet=_base64_chars, min_size=1, max_size=100)

malformed_jwt_no_dots = st.text(
    alphabet=_base64_chars, min_size=1, max_size=200
).filter(lambda s: "." not in s)

malformed_jwt_one_dot = st.builds(
    lambda a, b: f"{a}.{b}",
    _base64_segment,
    _base64_segment,
)

malformed_jwt_three_dots = st.builds(
    lambda a, b, c, d: f"{a}.{b}.{c}.{d}",
    _base64_segment,
    _base64_segment,
    _base64_segment,
    _base64_segment,
)

malformed_jwt_segments = st.one_of(
    malformed_jwt_no_dots,
    malformed_jwt_one_dot,
    malformed_jwt_three_dots,
)

# Strategy 3: Valid-format JWTs with past `exp` (expired tokens)
expired_jwt_strategy = st.builds(
    lambda hours_ago: jwt.encode(
        {
            "sub": "authenticated",
            "exp": datetime.now(tz=timezone.utc) - timedelta(hours=hours_ago),
        },
        _jwt_secret_for(TEST_PASSWORD),
        algorithm="HS256",
    ),
    st.integers(min_value=1, max_value=8760),  # 1 hour to 1 year ago
)

# Strategy 4: JWTs signed with wrong secret
wrong_secret_jwt_strategy = st.builds(
    lambda secret_suffix: jwt.encode(
        {
            "sub": "authenticated",
            "exp": datetime.now(tz=timezone.utc) + timedelta(hours=24),
        },
        hashlib.sha256(f"wrong-password-{secret_suffix}".encode()).hexdigest(),
        algorithm="HS256",
    ),
    st.text(min_size=1, max_size=50),
)

# Combined strategy: all invalid token types
invalid_token_strategy = st.one_of(
    arbitrary_text,
    malformed_jwt_segments,
    expired_jwt_strategy,
    wrong_secret_jwt_strategy,
)


# ---------------------------------------------------------------------------
# Property Tests
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
@settings(max_examples=100, deadline=5000)
@given(token=arbitrary_text)
async def test_arbitrary_text_rejected(token: str):
    """Any arbitrary text string used as Bearer token returns HTTP 401."""
    with patch("server._APP_PASSWORD", TEST_PASSWORD):
        async with AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post(
                "/api/query/stream",
                json={"query": "test", "history": []},
                headers={"Authorization": f"Bearer {token}"},
            )
    assert resp.status_code == 401, (
        f"Expected 401 for arbitrary text token {token!r}, got {resp.status_code}"
    )


@pytest.mark.asyncio
@settings(max_examples=100, deadline=5000)
@given(token=malformed_jwt_segments)
async def test_malformed_jwt_segments_rejected(token: str):
    """Malformed JWT segments (wrong number of dots) return HTTP 401."""
    with patch("server._APP_PASSWORD", TEST_PASSWORD):
        async with AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post(
                "/api/query/stream",
                json={"query": "test", "history": []},
                headers={"Authorization": f"Bearer {token}"},
            )
    assert resp.status_code == 401, (
        f"Expected 401 for malformed JWT {token!r}, got {resp.status_code}"
    )


@pytest.mark.asyncio
@settings(max_examples=100, deadline=5000)
@given(token=expired_jwt_strategy)
async def test_expired_jwt_rejected(token: str):
    """Valid-format JWTs with past `exp` return HTTP 401."""
    with patch("server._APP_PASSWORD", TEST_PASSWORD):
        async with AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post(
                "/api/query/stream",
                json={"query": "test", "history": []},
                headers={"Authorization": f"Bearer {token}"},
            )
    assert resp.status_code == 401, (
        f"Expected 401 for expired JWT, got {resp.status_code}"
    )


@pytest.mark.asyncio
@settings(max_examples=100, deadline=5000)
@given(token=wrong_secret_jwt_strategy)
async def test_wrong_secret_jwt_rejected(token: str):
    """JWTs signed with wrong secret return HTTP 401."""
    with patch("server._APP_PASSWORD", TEST_PASSWORD):
        async with AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post(
                "/api/query/stream",
                json={"query": "test", "history": []},
                headers={"Authorization": f"Bearer {token}"},
            )
    assert resp.status_code == 401, (
        f"Expected 401 for wrong-secret JWT, got {resp.status_code}"
    )


@pytest.mark.asyncio
@settings(max_examples=100, deadline=5000)
@given(token=invalid_token_strategy)
async def test_invalid_token_always_rejected(token: str):
    """Combined property: any invalid token returns HTTP 401 from protected endpoints."""
    with patch("server._APP_PASSWORD", TEST_PASSWORD):
        async with AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test"
        ) as client:
            resp = await client.post(
                "/api/query/stream",
                json={"query": "test", "history": []},
                headers={"Authorization": f"Bearer {token}"},
            )
    assert resp.status_code == 401, (
        f"Expected 401 for invalid token {token!r}, got {resp.status_code}"
    )