From ccb748a367589d197d224213b6e2ab3335008c47 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:46:55 +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 | 132 ++++++++++++++++++++++++------------------ 1 file changed, 76 insertions(+), 56 deletions(-) diff --git a/tests/test_checker.py b/tests/test_checker.py index 79eedc4..d893bd5 100644 --- a/tests/test_checker.py +++ b/tests/test_checker.py @@ -1,72 +1,92 @@ -import asyncio import json import logging -import socket +import httpx import pytest +from fastapi.testclient import TestClient -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 +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 -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")) +async def public_resolver(host: str, port: int) -> list[str]: + return ["93.184.216.34"] -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" +async def private_resolver(host: str, port: int) -> list[str]: + return ["127.0.0.1"] -def test_redirect_destination_is_revalidated() -> None: - resolver = SafeResolver() - calls: list[str] = [] +@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 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"] + 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" -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.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") + + calls = 0 + + async def changing_resolver(host: str, port: int) -> list[str]: + return ["127.0.0.1"] if host == "internal.test" else ["93.184.216.34"] + + 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 -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()) +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 + + 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" + + +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("?")