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.
Some checks failed
ci / validate (push) Has been cancelled
Some checks failed
ci / validate (push) Has been cancelled
This commit is contained in:
@@ -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?<redacted>"
|
||||
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("?<redacted>")
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user