From bcc64ea7ca496a3b3f2145a3d09bfa6a727e624a Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:52:32 +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 | 115 +++++++++++++++++------------------------- 1 file changed, 47 insertions(+), 68 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index d893bd5..592f62d 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,92 +1,71 @@ -import json import logging import httpx import pytest -from fastapi.testclient import TestClient -from app.checker import EndpointChecker -from app.config import Settings -from app.logging import JsonFormatter -from app.main import app -from app.models import State -from app.security import DestinationRejected, redacted_url +from endpoint_monitor.checker import EndpointChecker, TargetBlockedError, validate_target +from endpoint_monitor.logging import JsonFormatter, redact_url async def public_resolver(host: str, port: int) -> list[str]: return ["93.184.216.34"] -async def private_resolver(host: str, port: int) -> list[str]: - return ["127.0.0.1"] +@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_success_down_and_transport_error() -> None: - async def handler(request: httpx.Request) -> httpx.Response: - if request.url.path == "/ok": - return httpx.Response(204) - if request.url.path == "/down": - return httpx.Response(503) - raise httpx.ConnectError("secret host detail", request=request) +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"] - worker = EndpointChecker(Settings(), public_resolver, httpx.MockTransport(handler)) - assert (await worker.check("1", "https://example.com/ok?token=x")).state is State.UP - assert (await worker.check("1", "https://example.com/down")).state is State.DOWN - failed = await worker.check("1", "https://example.com/fail") - assert failed.state is State.ERROR - assert failed.error == "outbound request failed: ConnectError" + with pytest.raises(TargetBlockedError): + await validate_target("https://mixed.example", resolver) -@pytest.mark.asyncio -async def test_private_dns_and_private_redirect_are_blocked() -> None: - worker = EndpointChecker(Settings(), private_resolver, httpx.MockTransport(lambda r: httpx.Response(200))) - with pytest.raises(DestinationRejected): - await worker.check("1", "http://example.com") +async def test_redirect_hop_is_resolved_and_blocked(): + seen = [] - calls = 0 + 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"] - async def changing_resolver(host: str, port: int) -> list[str]: - return ["127.0.0.1"] if host == "internal.test" else ["93.184.216.34"] + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(302, headers={"location": "http://internal.example/admin"}) - async def redirect(request: httpx.Request) -> httpx.Response: - nonlocal calls - calls += 1 - return httpx.Response(302, headers={"location": "http://internal.test/admin"}) - - worker = EndpointChecker(Settings(), changing_resolver, httpx.MockTransport(redirect)) - with pytest.raises(DestinationRejected): - await worker.check("1", "https://example.com/start") - assert calls == 1 + 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"] -def test_api_check_updates_status_and_rejection(client: TestClient) -> None: - created = client.post( - "/v1/monitors", json={"name": "site", "url": "https://example.com/ok"} - ).json() - app.state.checker = EndpointChecker( - Settings(), public_resolver, httpx.MockTransport(lambda r: httpx.Response(200)) - ) - response = client.post(f"/v1/monitors/{created['id']}/check") - assert response.status_code == 200 - assert response.json()["state"] == "up" - assert client.get(f"/v1/monitors/{created['id']}/status").json()["status_code"] == 200 +async def test_success_latency_status_and_redaction(caplog): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(503) - app.state.checker = EndpointChecker( - Settings(), private_resolver, httpx.MockTransport(lambda r: httpx.Response(200)) - ) - response = client.post(f"/v1/monitors/{created['id']}/check") - assert response.status_code == 400 - assert client.get(f"/v1/monitors/{created['id']}/status").json()["state"] == "error" + 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_url_and_json_log_redaction() -> None: - safe = redacted_url("https://user:pass@example.com/a?token=secret#fragment") - assert safe == "https://example.com/a?" - record = logging.LogRecord("x", logging.INFO, "", 0, "done", (), None) - record.url = safe - payload = json.loads(JsonFormatter().format(record)) - serialized = json.dumps(payload) - assert "secret" not in serialized - assert "pass" not in serialized - assert payload["url"].endswith("?") +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