From f1bb9f162b64046d309f1c6aceba84009d580a86 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:55:04 +0000 Subject: [PATCH] decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project. --- tests/test_checker.py | 109 ++++++++++++++++++++---------------------- 1 file changed, 53 insertions(+), 56 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index 592f62d..55b67e4 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,71 +1,68 @@ import logging +import socket +from unittest.mock import AsyncMock, patch import httpx import pytest -from endpoint_monitor.checker import EndpointChecker, TargetBlockedError, validate_target -from endpoint_monitor.logging import JsonFormatter, redact_url +from app.checker import check_url +from app.config import Settings +from app.security import UnsafeDestination, validate_destination + +PUBLIC_DNS = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))] -async def public_resolver(host: str, port: int) -> list[str]: - return ["93.184.216.34"] +@pytest.mark.asyncio +async def test_success_and_redirect_targets_are_each_validated(): + transport = httpx.MockTransport(lambda request: ( + httpx.Response(302, headers={"location": "https://target.example/final"}) + if request.url.host == "start.example" else httpx.Response(200) + )) + validator = AsyncMock() + with patch("app.checker.validate_destination", validator): + result = await check_url("https://start.example/a?token=x", 1, Settings(), transport) + assert result.state == "up" + assert validator.await_count == 2 + assert validator.await_args_list[1].args[0] == "https://target.example/final" -@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1", "fc00::1"]) -async def test_dns_answers_block_non_public(address): - async def resolver(host: str, port: int) -> list[str]: - return [address] - - with pytest.raises(TargetBlockedError, match="non-public"): - await validate_target("http://attacker.example/path", resolver) +@pytest.mark.asyncio +async def test_private_dns_answer_is_blocked(): + loop = __import__("asyncio").get_running_loop() + with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=[ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80)) + ])): + with pytest.raises(UnsafeDestination, match="non-public"): + await validate_destination("http://internal.example") -async def test_any_unsafe_dns_answer_blocks_target(): - async def resolver(host: str, port: int) -> list[str]: - return ["93.184.216.34", "127.0.0.1"] - - with pytest.raises(TargetBlockedError): - await validate_target("https://mixed.example", resolver) +@pytest.mark.asyncio +async def test_redirect_to_private_literal_is_blocked(): + transport = httpx.MockTransport(lambda request: httpx.Response( + 302, headers={"location": "http://169.254.169.254/latest"} + )) + loop = __import__("asyncio").get_running_loop() + with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=PUBLIC_DNS)): + with pytest.raises(UnsafeDestination): + await check_url("https://example.com", 1, Settings(), transport) -async def test_redirect_hop_is_resolved_and_blocked(): - seen = [] - - 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"] - - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(302, headers={"location": "http://internal.example/admin"}) - - checker = EndpointChecker(1, 3, resolver, httpx.MockTransport(handler)) - with pytest.raises(TargetBlockedError): - await checker.check("https://public.example/start", "id") - assert seen == ["public.example", "internal.example"] +@pytest.mark.asyncio +async def test_transport_error_maps_to_error_state(): + def fail(request): + raise httpx.ConnectError("no", request=request) + transport = httpx.MockTransport(fail) + with patch("app.checker.validate_destination", new=AsyncMock()): + result = await check_url("https://example.com", 1, Settings(), transport) + assert result.state == "error" + assert result.error == "ConnectError" -async def test_success_latency_status_and_redaction(caplog): - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response(503) - - checker = EndpointChecker(1, 1, public_resolver, httpx.MockTransport(handler)) - with caplog.at_level(logging.INFO): - result = await checker.check("https://example.com/path?token=supersecret#fragment", "abc") - assert result.status.value == "down" - assert result.http_status == 503 - assert result.latency_ms >= 0 - assert result.final_url == "https://example.com/path" - assert "supersecret" not in caplog.text - assert "fragment" not in caplog.text - - -def test_redact_url_removes_credentials_query_fragment(): - assert redact_url("https://user:pass@example.com:8443/a?q=secret#x") == "https://example.com:8443/a" - - -def test_json_formatter_is_structured_and_does_not_add_message_secrets(): - record = logging.LogRecord("x", logging.INFO, __file__, 1, "event", (), None) - record.url = redact_url("https://example.com/?api_key=secret") - rendered = JsonFormatter().format(record) - assert '"event":"event"' in rendered - assert "secret" not in rendered +@pytest.mark.asyncio +async def test_logs_redact_query(caplog): + transport = httpx.MockTransport(lambda request: httpx.Response(200)) + with patch("app.checker.validate_destination", new=AsyncMock()), caplog.at_level(logging.INFO): + await check_url("https://example.com/a?api_key=supersecret", 1, Settings(), transport) + text = caplog.text + assert "supersecret" not in text + assert "REDACTED" in text