From be68ab4589692d4ed74686b0f83f777dc2c925b6 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:41:18 +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 | 107 ++++++++++++++++++++++-------------------- 1 file changed, 56 insertions(+), 51 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index 4e27f95..79eedc4 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,67 +1,72 @@ +import asyncio import json import logging +import socket -import httpx import pytest -from monitor_service.checker import EndpointChecker, SecurityCheckError -from monitor_service.config import Settings +from app.checker import SafeResolver, UnsafeTargetError +from app.logging_config import JsonFormatter, redact_url +from app.models import CurrentStatus, MonitorInput, State +from app.store import MonitorStore, StaleCheckError -async def public_resolver(host: str, port: int) -> list[str]: - return ["93.184.216.34"] +def test_resolver_blocks_private_answer(monkeypatch: pytest.MonkeyPatch) -> None: + def private(*args: object, **kwargs: object) -> list[tuple]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))] + + monkeypatch.setattr(socket, "getaddrinfo", private) + with pytest.raises(UnsafeTargetError): + asyncio.run(SafeResolver().validate_url("http://attacker.example/path")) -@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" +def test_resolver_accepts_only_public_answers(monkeypatch: pytest.MonkeyPatch) -> None: + def public(*args: object, **kwargs: object) -> list[tuple]: + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))] + + monkeypatch.setattr(socket, "getaddrinfo", public) + results = asyncio.run(SafeResolver().resolve("example.com", 443, socket.AF_UNSPEC)) + assert results[0]["host"] == "93.184.216.34" -@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) +def test_redirect_destination_is_revalidated() -> None: + resolver = SafeResolver() + calls: list[str] = [] - 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" + async def validate(url: str) -> None: + calls.append(url) + if len(calls) == 2: + raise UnsafeTargetError + + resolver.validate_url = validate # type: ignore[method-assign] + async def redirects() -> None: + await resolver.validate_url("https://public.example") + with pytest.raises(UnsafeTargetError): + await resolver.validate_url("http://127.0.0.1/admin") + asyncio.run(redirects()) + assert calls == ["https://public.example", "http://127.0.0.1/admin"] -@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") +def test_url_and_structured_log_redaction() -> None: + secret = "https://user:pass@example.com/path?token=secret#fragment" + assert redact_url(secret) == "https://example.com/path" + formatter = JsonFormatter() + record = logging.LogRecord("test", logging.INFO, "", 0, "event", (), None) + record.fields = {"url": secret, "monitor_id": "abc"} # type: ignore[attr-defined] + payload = formatter.format(record) + assert "secret" not in payload and "pass" not in payload + assert json.loads(payload)["url"] == "https://example.com/path" -@pytest.mark.anyio -async def test_redirect_target_is_resolved_and_blocked() -> None: - seen: list[str] = [] - - 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 redirect(request: httpx.Request) -> httpx.Response: - return httpx.Response(302, headers={"location": "http://internal.example/admin"}, request=request) - - 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"] +def test_stale_check_cannot_overwrite_new_url() -> None: + async def scenario() -> None: + store = MonitorStore(5) + item = await store.create(MonitorInput(name="x", url="https://example.com")) + await store.replace( + item.id, MonitorInput(name="x", url="https://example.org") + ) + with pytest.raises(StaleCheckError): + await store.publish_status( + item.id, str(item.url), CurrentStatus(state=State.UP) + ) + asyncio.run(scenario())