73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
import socket
|
|
|
|
import pytest
|
|
|
|
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
|
|
|
|
|
|
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"))
|
|
|
|
|
|
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"
|
|
|
|
|
|
def test_redirect_destination_is_revalidated() -> None:
|
|
resolver = SafeResolver()
|
|
calls: list[str] = []
|
|
|
|
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"]
|
|
|
|
|
|
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"
|
|
|
|
|
|
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())
|