diff --git a/tests/test_checker.py b/tests/test_checker.py index 8ad3631..4e27f95 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,108 +1,67 @@ -import asyncio import json import logging -import socket +import httpx import pytest -from app.checker import EndpointChecker, redact_url -from app.models import State -from app.settings import Settings +from monitor_service.checker import EndpointChecker, SecurityCheckError +from monitor_service.config import Settings -@pytest.fixture -def checker() -> EndpointChecker: - return EndpointChecker(Settings(request_timeout_seconds=1, connect_timeout_seconds=1, - max_redirects=2)) +async def public_resolver(host: str, port: int) -> list[str]: + return ["93.184.216.34"] -@pytest.mark.asyncio -async def test_public_address_is_pinned_for_request(checker: EndpointChecker, monkeypatch) -> None: - async def resolve(_host: str, _port: int) -> list[str]: - return ["93.184.216.34"] +@pytest.mark.anyio +async def test_success_latency_and_query_redaction(caplog: pytest.LogCaptureFixture) -> None: + transport = httpx.MockTransport(lambda request: httpx.Response(200, request=request)) + async with httpx.AsyncClient(transport=transport) as client: + checker = EndpointChecker(client, Settings(), public_resolver) + with caplog.at_level(logging.INFO, logger="monitor_service.checker"): + result = await checker.check("id", "https://example.com/path?token=very-secret") + assert result.status.state == "up" + assert result.status.http_status == 200 + assert result.status.latency_ms is not None + assert "very-secret" not in caplog.text + event = json.loads(caplog.records[-1].message) + assert event["url"] == "https://example.com/path" + + +@pytest.mark.anyio +async def test_transport_failure_maps_to_down() -> None: + def fail(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectTimeout("late", request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(fail)) as client: + result = await EndpointChecker(client, Settings(), public_resolver).check("id", "https://example.com") + assert result.status.state == "down" + assert result.status.error == "ConnectTimeout" + + +@pytest.mark.anyio +@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1"]) +async def test_dns_private_addresses_are_blocked(address: str) -> None: + async def resolver(host: str, port: int) -> list[str]: + return [address] + + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(200))) as client: + checker = EndpointChecker(client, Settings(), resolver) + with pytest.raises(SecurityCheckError, match="non-public"): + await checker.check("id", "http://attacker.example") + + +@pytest.mark.anyio +async def test_redirect_target_is_resolved_and_blocked() -> None: seen: list[str] = [] - async def request(_url: str, addresses: list[str]) -> tuple[int, None]: - seen.extend(addresses) - return 200, None - monkeypatch.setattr(checker, "_resolve", resolve) - monkeypatch.setattr(checker, "_request_once", request) - result = await checker.check("https://example.com/?token=secret") - assert result.state == State.UP - assert seen == ["93.184.216.34"] - assert "secret" not in (result.final_url or "") + async def resolver(host: str, port: int) -> list[str]: + seen.append(host) + return ["127.0.0.1"] if host == "internal.example" else ["93.184.216.34"] -@pytest.mark.asyncio -@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.1.1", "::1", "192.0.2.1"]) -async def test_dns_rejects_non_global_answers(checker: EndpointChecker, monkeypatch, address: str) -> None: - loop = asyncio.get_running_loop() - def answers(*_args, **_kwargs): - family = socket.AF_INET6 if ":" in address else socket.AF_INET - return [(family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (address, 443))] - monkeypatch.setattr(loop, "getaddrinfo", answers) - with pytest.raises(Exception, match="globally routable"): - await checker._resolve("attacker.invalid", 443) + def redirect(request: httpx.Request) -> httpx.Response: + return httpx.Response(302, headers={"location": "http://internal.example/admin"}, request=request) - -@pytest.mark.asyncio -async def test_mixed_dns_answer_is_rejected(checker: EndpointChecker, monkeypatch) -> None: - loop = asyncio.get_running_loop() - def answers(*_args, **_kwargs): - return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 80)) - for ip in ("93.184.216.34", "127.0.0.1")] - monkeypatch.setattr(loop, "getaddrinfo", answers) - with pytest.raises(Exception): - await checker._resolve("rebinding.invalid", 80) - - -@pytest.mark.asyncio -async def test_redirect_target_is_resolved_and_blocked(checker: EndpointChecker, monkeypatch) -> None: - calls = 0 - async def resolve(host: str, _port: int) -> list[str]: - if host == "localhost": - from app.checker import SecurityError - raise SecurityError("destination is not globally routable") - return ["93.184.216.34"] - async def request(_url: str, _addresses: list[str]) -> tuple[int, str]: - nonlocal calls - calls += 1 - return 302, "http://localhost/admin?key=secret" - monkeypatch.setattr(checker, "_resolve", resolve) - monkeypatch.setattr(checker, "_request_once", request) - result = await checker.check("https://example.com") - assert result.state == State.ERROR - assert result.error == "blocked_destination" - assert calls == 1 - - -@pytest.mark.asyncio -async def test_timeout_mapping(checker: EndpointChecker, monkeypatch) -> None: - async def resolve(_host: str, _port: int) -> list[str]: - return ["93.184.216.34"] - async def request(_url: str, _addresses: list[str]): - raise asyncio.TimeoutError - monkeypatch.setattr(checker, "_resolve", resolve) - monkeypatch.setattr(checker, "_request_once", request) - result = await checker.check("https://example.com") - assert result.state == State.ERROR and result.error == "timeout" - - -def test_redaction_removes_credentials_query_and_fragment() -> None: - value = redact_url("https://user:secret@example.com/path?token=abc&x=1#frag") - assert value == "https://example.com/path?REDACTED" - assert all(secret not in value for secret in ("user", "secret", "abc", "frag")) - - -@pytest.mark.asyncio -async def test_structured_logs_do_not_leak_query(checker: EndpointChecker, monkeypatch, caplog) -> None: - async def resolve(_host: str, _port: int) -> list[str]: - return ["93.184.216.34"] - async def request(_url: str, _addresses: list[str]) -> tuple[int, None]: - return 500, None - monkeypatch.setattr(checker, "_resolve", resolve) - monkeypatch.setattr(checker, "_request_once", request) - with caplog.at_level(logging.INFO, logger="endpoint_monitor.checker"): - result = await checker.check("https://example.com/a?api_key=supersecret") - assert result.state == State.DOWN - assert "supersecret" not in caplog.text - assert all("event" in json.loads(record.message) for record in caplog.records) + async with httpx.AsyncClient(transport=httpx.MockTransport(redirect)) as client: + with pytest.raises(SecurityCheckError): + await EndpointChecker(client, Settings(), resolver).check("id", "https://public.example") + assert seen == ["public.example", "internal.example"]