From 5c4b56b3e690a12e7616bb53d52106625e8b0fa7 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:01: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 | 143 +++++++++++++++++++++++++++--------------- 1 file changed, 92 insertions(+), 51 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index 55b67e4..319c617 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,68 +1,109 @@ +import json import logging -import socket -from unittest.mock import AsyncMock, patch import httpx import pytest -from app.checker import check_url +from app.checker import EndpointChecker, UnsafeTarget 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))] +from app.logging import StructuredJsonFormatter, redact_url -@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" +async def public_resolver(host: str, port: int) -> list[str]: + return ["93.184.216.34"] -@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") +def checker(handler, resolver=public_resolver, **settings): + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + return EndpointChecker(client, Settings(**settings), resolver), client -@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_http_status_and_latency(): + service, client = checker(lambda request: httpx.Response(503, request=request)) + try: + result = await service.check("https://example.com/path") + finally: + await client.aclose() + assert result.state == "down" + assert result.status_code == 503 + assert result.latency_ms is not None and result.latency_ms >= 0 -@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) +async def test_timeout_is_sanitized_error_result(): + def timeout(request): + raise httpx.ReadTimeout("secret low-level detail", request=request) + service, client = checker(timeout) + try: + result = await service.check("https://example.com/?token=secret") + finally: + await client.aclose() assert result.state == "error" - assert result.error == "ConnectError" + assert result.error == "request timed out" + assert "secret" not in result.error -@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 +async def test_dns_blocks_private_answer_before_outbound_request(): + calls = 0 + def handler(request): + nonlocal calls + calls += 1 + return httpx.Response(200, request=request) + async def private_resolver(host: str, port: int) -> list[str]: + return ["10.0.0.8"] + service, client = checker(handler, private_resolver) + try: + with pytest.raises(UnsafeTarget, match="non-public"): + await service.check("http://internal.invalid/") + finally: + await client.aclose() + assert calls == 0 + + +async def test_mixed_dns_answers_are_blocked(): + async def mixed(host: str, port: int) -> list[str]: + return ["93.184.216.34", "127.0.0.1"] + service, client = checker(lambda request: httpx.Response(200, request=request), mixed) + try: + with pytest.raises(UnsafeTarget): + await service.check("https://example.com") + finally: + await client.aclose() + + +async def test_redirect_destination_is_resolved_and_blocked(): + calls = [] + def handler(request): + calls.append(str(request.url)) + return httpx.Response(302, headers={"Location": "http://127.0.0.1/admin"}, request=request) + async def resolver(host: str, port: int) -> list[str]: + return ["127.0.0.1"] if host == "127.0.0.1" else ["93.184.216.34"] + service, client = checker(handler, resolver) + try: + with pytest.raises(UnsafeTarget): + await service.check("https://example.com/start") + finally: + await client.aclose() + assert calls == ["https://example.com/start"] + + +async def test_redirect_limit_is_error(): + def handler(request): + return httpx.Response(302, headers={"Location": "/again"}, request=request) + service, client = checker(handler, max_redirects=1) + try: + result = await service.check("https://example.com/start") + finally: + await client.aclose() + assert result.state == "error" + assert result.error == "redirect limit exceeded" + + +def test_url_redaction_and_structured_log_are_inspectable(): + safe = redact_url("https://user:password@example.com/path?token=secret#private") + assert "user" not in safe and "password" not in safe and "secret" not in safe + record = logging.LogRecord("checker", logging.INFO, __file__, 1, + "done", (), None) + record.target = safe + payload = json.loads(StructuredJsonFormatter().format(record)) + assert payload["target"] == "https://example.com/path?#" + assert payload["message"] == "done"