69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
import logging
|
|
import socket
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.checker import check_url
|
|
from app.config import Settings
|
|
from app.security import UnsafeDestination, validate_destination
|
|
|
|
PUBLIC_DNS = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_and_redirect_targets_are_each_validated():
|
|
transport = httpx.MockTransport(lambda request: (
|
|
httpx.Response(302, headers={"location": "https://target.example/final"})
|
|
if request.url.host == "start.example" else httpx.Response(200)
|
|
))
|
|
validator = AsyncMock()
|
|
with patch("app.checker.validate_destination", validator):
|
|
result = await check_url("https://start.example/a?token=x", 1, Settings(), transport)
|
|
assert result.state == "up"
|
|
assert validator.await_count == 2
|
|
assert validator.await_args_list[1].args[0] == "https://target.example/final"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_private_dns_answer_is_blocked():
|
|
loop = __import__("asyncio").get_running_loop()
|
|
with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=[
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 80))
|
|
])):
|
|
with pytest.raises(UnsafeDestination, match="non-public"):
|
|
await validate_destination("http://internal.example")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redirect_to_private_literal_is_blocked():
|
|
transport = httpx.MockTransport(lambda request: httpx.Response(
|
|
302, headers={"location": "http://169.254.169.254/latest"}
|
|
))
|
|
loop = __import__("asyncio").get_running_loop()
|
|
with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=PUBLIC_DNS)):
|
|
with pytest.raises(UnsafeDestination):
|
|
await check_url("https://example.com", 1, Settings(), transport)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transport_error_maps_to_error_state():
|
|
def fail(request):
|
|
raise httpx.ConnectError("no", request=request)
|
|
transport = httpx.MockTransport(fail)
|
|
with patch("app.checker.validate_destination", new=AsyncMock()):
|
|
result = await check_url("https://example.com", 1, Settings(), transport)
|
|
assert result.state == "error"
|
|
assert result.error == "ConnectError"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_redact_query(caplog):
|
|
transport = httpx.MockTransport(lambda request: httpx.Response(200))
|
|
with patch("app.checker.validate_destination", new=AsyncMock()), caplog.at_level(logging.INFO):
|
|
await check_url("https://example.com/a?api_key=supersecret", 1, Settings(), transport)
|
|
text = caplog.text
|
|
assert "supersecret" not in text
|
|
assert "REDACTED" in text
|