saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
"""SSRF guard tests - see SECURITY.md."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.core.http import SsrfBlockedError, safe_fetch, validate_url
|
|
|
|
|
|
def test_validate_url_rejects_disallowed_scheme():
|
|
with pytest.raises(SsrfBlockedError):
|
|
validate_url("file:///etc/passwd")
|
|
|
|
|
|
def test_validate_url_rejects_url_with_no_hostname():
|
|
with pytest.raises(SsrfBlockedError):
|
|
validate_url("http://")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"hostname,ip",
|
|
[
|
|
("localhost-test", "127.0.0.1"),
|
|
("private-test", "10.0.0.5"),
|
|
("private-test-2", "192.168.1.1"),
|
|
("link-local-test", "169.254.1.1"),
|
|
("metadata-test", "169.254.169.254"),
|
|
],
|
|
)
|
|
def test_validate_url_blocks_private_and_metadata_addresses(hostname, ip):
|
|
with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", (ip, 0))]):
|
|
with pytest.raises(SsrfBlockedError):
|
|
validate_url(f"http://{hostname}/")
|
|
|
|
|
|
def test_validate_url_allows_public_address():
|
|
with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
|
|
validate_url("http://example.com/") # should not raise
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_safe_fetch_revalidates_each_redirect_hop():
|
|
"""A redirect to a private address must be blocked even if the initial
|
|
URL resolves to a public one."""
|
|
with patch("socket.getaddrinfo") as mock_resolve:
|
|
|
|
def resolve(hostname, *_args, **_kwargs):
|
|
if hostname == "public.example":
|
|
return [(2, 1, 6, "", ("93.184.216.34", 0))]
|
|
if hostname == "internal.example":
|
|
return [(2, 1, 6, "", ("10.0.0.5", 0))]
|
|
raise AssertionError(f"unexpected hostname {hostname}")
|
|
|
|
mock_resolve.side_effect = resolve
|
|
|
|
with respx.mock:
|
|
respx.get("http://public.example/").mock(
|
|
return_value=httpx.Response(302, headers={"Location": "http://internal.example/"})
|
|
)
|
|
with pytest.raises(SsrfBlockedError):
|
|
await safe_fetch("http://public.example/")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_safe_fetch_returns_final_response_body():
|
|
with patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]):
|
|
with respx.mock:
|
|
respx.get("http://public.example/").mock(
|
|
return_value=httpx.Response(200, text="hello world")
|
|
)
|
|
result = await safe_fetch("http://public.example/")
|
|
assert result.status_code == 200
|
|
assert result.text == "hello world"
|