"""Unit tests for SSRF allowlist guard (_assert_allowed_url) — issue #756. Approach: unit-tests on the helper directly (no DB/scraper fixtures needed). The guard is a pure synchronous function; TestClient integration is not required because the guard raises before any async scraper is called. """ from __future__ import annotations import os import sys from unittest.mock import MagicMock os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") _wp_mock = MagicMock() sys.modules.setdefault("weasyprint", _wp_mock) sys.modules.setdefault("weasyprint.CSS", _wp_mock) sys.modules.setdefault("weasyprint.HTML", _wp_mock) import pytest # noqa: E402 from fastapi import HTTPException # noqa: E402 from app.api.v1.admin import _assert_allowed_url # noqa: E402 from app.core.config import settings # noqa: E402 # --------------------------------------------------------------------------- # Absolute URL — host NOT in allowlist → 400 # --------------------------------------------------------------------------- def test_absolute_ssrf_metadata_endpoint_blocked() -> None: """http://169.254.169.254/... (AWS metadata) должен быть отклонён с 400.""" with pytest.raises(HTTPException) as exc_info: _assert_allowed_url("http://169.254.169.254/latest/meta-data/") assert exc_info.value.status_code == 400 assert "host not allowed" in exc_info.value.detail def test_absolute_localhost_blocked() -> None: with pytest.raises(HTTPException) as exc_info: _assert_allowed_url("http://localhost/internal/admin") assert exc_info.value.status_code == 400 def test_absolute_arbitrary_host_blocked() -> None: with pytest.raises(HTTPException) as exc_info: _assert_allowed_url("https://evil.example.com/scrape") assert exc_info.value.status_code == 400 # --------------------------------------------------------------------------- # Absolute URL — host IN allowlist → passes # --------------------------------------------------------------------------- def test_absolute_avito_allowed() -> None: _assert_allowed_url("https://www.avito.ru/catalog/houses/some-slug/12345") def test_absolute_cian_allowed() -> None: _assert_allowed_url("https://ekb.cian.ru/sale/flat/327830237/") def test_absolute_yandex_realty_allowed() -> None: _assert_allowed_url("https://realty.yandex.ru/ekaterinburg/kupit/kvartira/123/") def test_absolute_n1_allowed() -> None: _assert_allowed_url("https://ekaterinburg.n1.ru/kupit/kvartiry/vtorichka/") # --------------------------------------------------------------------------- # Relative path — no netloc → always passes # --------------------------------------------------------------------------- def test_relative_path_passes_the_guard() -> None: """Путь без схемы/хоста (авито-стиль /ekaterinburg/...) разрешается.""" _assert_allowed_url("/ekaterinburg/kvartiry/prodam-123") def test_relative_path_house_catalog_passes() -> None: _assert_allowed_url("/catalog/houses/ekb-slug/99999") def test_empty_string_passes() -> None: """Пустая строка — нет netloc → guard не блокирует (upstream validation задача scraper'а).""" _assert_allowed_url("") # --------------------------------------------------------------------------- # Allowlist integrity: verify expected hosts present # --------------------------------------------------------------------------- def test_allowlist_contains_required_scraper_hosts() -> None: required = { "www.avito.ru", "ekb.cian.ru", "www.cian.ru", "realty.yandex.ru", "ekaterinburg.n1.ru", } missing = required - settings.scrape_allowed_hosts assert not missing, f"Missing hosts in scrape_allowed_hosts: {missing}"