From f963e1bad70232ced5ab514a60ed81e39646fe96 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:49:46 +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_security.py | 110 ++++++++++++++++++++++++++++------------- 1 file changed, 76 insertions(+), 34 deletions(-) diff --git a/tests/test_security.py b/tests/test_security.py index 54329b7..b55efb4 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -1,43 +1,85 @@ -import io +import asyncio import json import logging +import socket -import httpx +import pytest -from app.checker import EndpointChecker +from app.checker import EndpointChecker, HopResponse, PinnedResolver +from app.config import Settings from app.logging import JsonFormatter -from app.models import Monitor, MonitorCreate -from app.security import redacted_url +from app.models import HealthState +from app.security import TargetPolicyError, ValidatedTarget, redact_url, validate_target -async def test_dns_resolution_blocks_private_address_before_transport() -> None: - calls = 0 - - async def private_resolver(_: str, __: int) -> list[str]: - return ["10.0.0.7"] - - async def transport_handler(_: httpx.Request) -> httpx.Response: - nonlocal calls - calls += 1 - return httpx.Response(200) - - monitor = Monitor(**MonitorCreate(name="private", url="https://internal.example").model_dump()) - checker = EndpointChecker( - 1, 2, resolver=private_resolver, transport=httpx.MockTransport(transport_handler) - ) - result = await checker.check(monitor) - assert result.state == "blocked" - assert calls == 0 +async def resolver_for(*addresses: str): + async def resolve(host: str, port: int) -> list[tuple[object, ...]]: + return [(socket.AF_INET6 if ":" in address else socket.AF_INET, 1, 6, "", (address, port)) for address in addresses] + return resolve -def test_url_and_json_logs_redact_query_and_credentials() -> None: - assert redacted_url("https://user:pass@example.com/a?token=secret#x") == "https://example.com/a" - stream = io.StringIO() - handler = logging.StreamHandler(stream) - handler.setFormatter(JsonFormatter()) - record = logging.LogRecord("test", logging.INFO, "", 0, "check", (), None) - record.event_data = {"url": redacted_url("https://example.com/a?token=secret")} - handler.emit(record) - payload = json.loads(stream.getvalue()) - assert payload["url"] == "https://example.com/a" - assert "secret" not in stream.getvalue() +@pytest.mark.asyncio +@pytest.mark.parametrize("address", ["127.0.0.1", "10.1.2.3", "169.254.169.254", "::1", "fc00::1", "0.0.0.0", "224.0.0.1"]) +async def test_ip_classification_blocks_non_public_answers(address: str) -> None: + with pytest.raises(TargetPolicyError): + await validate_target("http://example.test/", await resolver_for(address)) + + +@pytest.mark.asyncio +async def test_mixed_dns_answer_is_blocked_and_public_is_pinned() -> None: + with pytest.raises(TargetPolicyError): + await validate_target("https://example.test", await resolver_for("93.184.216.34", "127.0.0.1")) + target = await validate_target("https://example.test", await resolver_for("93.184.216.34")) + assert target.addresses == ("93.184.216.34",) + pinned = PinnedResolver(target.hostname, target.addresses) + assert (await pinned.resolve("example.test", 443))[0]["host"] == "93.184.216.34" + with pytest.raises(OSError): + await pinned.resolve("attacker.test", 443) + + +class RedirectFetcher: + def __init__(self) -> None: + self.calls: list[ValidatedTarget] = [] + + async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse: + self.calls.append(target) + return HopResponse(302, "http://internal.test/admin") if len(self.calls) == 1 else HopResponse(200, None) + + +@pytest.mark.asyncio +async def test_redirect_hop_is_resolved_and_blocked_before_second_fetch() -> None: + seen: list[str] = [] + async def resolver(host: str, port: int) -> list[tuple[object, ...]]: + seen.append(host) + address = "93.184.216.34" if host == "public.test" else "127.0.0.1" + return [(socket.AF_INET, 1, 6, "", (address, port))] + fetcher = RedirectFetcher() + result = await EndpointChecker(Settings(), fetcher=fetcher, resolver=resolver).check("http://public.test/start") + assert result.error_code == "blocked_target" + assert seen == ["public.test", "internal.test"] + assert len(fetcher.calls) == 1 + + +class SlowFetcher: + async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse: + await asyncio.sleep(0.1) + return HopResponse(200, None) + + +@pytest.mark.asyncio +async def test_total_timeout_is_mapped_without_leaking_exception() -> None: + checker = EndpointChecker(Settings(total_timeout_seconds=0.01, connect_timeout_seconds=0.01, read_timeout_seconds=0.01), fetcher=SlowFetcher(), resolver=await resolver_for("93.184.216.34")) + result = await checker.check("https://example.test/?token=secret") + assert result.state == HealthState.down + assert result.error_code == "timeout" + assert result.error_message == "endpoint check timed out" + assert result.latency_ms >= 0 + + +def test_query_redaction_and_json_log_output() -> None: + assert redact_url("https://user:pass@example.com:8443/p?q=secret#frag") == "https://example.com:8443/p" + record = logging.LogRecord("x", logging.INFO, "", 1, "done", (), None) + record.url = redact_url("https://example.com/path?api_key=hunter2") + payload = JsonFormatter().format(record) + assert json.loads(payload)["url"] == "https://example.com/path" + assert "hunter2" not in payload