59974be696
Reject private/loopback/link-local hosts on POST /api/ingest/rss and
URL-shaped GDELT queries via camera_scraper.is_public_url (HTTP 400).
PATCH /api/sources/{id} only accepts name, url, config, enabled (422 else).
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""SSRF guard on ingest triggers + PATCH /api/sources allowlist."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from main import app
|
|
|
|
BASE = "http://test"
|
|
LINK_LOCAL_META = "http://169.254.169.254/latest/meta-data/"
|
|
LOOPBACK = "http://127.0.0.1/secret"
|
|
|
|
|
|
async def _req(method: str, path: str, **kw) -> httpx.Response:
|
|
transport = httpx.ASGITransport(app=app)
|
|
async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
|
|
return await client.request(method, path, **kw)
|
|
|
|
|
|
def test_rss_ingest_rejects_link_local_metadata_url(monkeypatch):
|
|
called = {"n": 0}
|
|
|
|
async def _boom(*_a, **_k):
|
|
called["n"] += 1
|
|
raise AssertionError("ingest_rss_feed must not run for a private URL")
|
|
|
|
monkeypatch.setattr("main.ingest_rss_feed", _boom)
|
|
resp = asyncio.run(_req("POST", "/api/ingest/rss", params={"feed_url": LINK_LOCAL_META}))
|
|
assert resp.status_code == 400
|
|
assert called["n"] == 0
|
|
|
|
|
|
def test_gdelt_ingest_rejects_private_query_url(monkeypatch):
|
|
called = {"n": 0}
|
|
|
|
async def _boom(*_a, **_k):
|
|
called["n"] += 1
|
|
raise AssertionError("ingest_gdelt must not run for a private URL query")
|
|
|
|
monkeypatch.setattr("main.ingest_gdelt", _boom)
|
|
resp = asyncio.run(_req("POST", "/api/ingest/gdelt", params={"query": LOOPBACK}))
|
|
assert resp.status_code == 400
|
|
assert called["n"] == 0
|
|
|
|
|
|
def test_update_source_rejects_unknown_fields():
|
|
sid = "00000000-0000-0000-0000-000000000001"
|
|
resp = asyncio.run(_req("PATCH", f"/api/sources/{sid}", json={"enabled": True, "source_type": "rss"}))
|
|
assert resp.status_code == 422
|
|
|
|
|
|
def test_feed_source_update_allowlist_only():
|
|
from schemas import FeedSourceUpdate
|
|
|
|
payload = FeedSourceUpdate(name="n", url="https://example.com/rss", config={"k": 1}, enabled=False)
|
|
assert payload.model_dump(exclude_unset=True) == {
|
|
"name": "n",
|
|
"url": "https://example.com/rss",
|
|
"config": {"k": 1},
|
|
"enabled": False,
|
|
}
|
|
with pytest.raises(ValidationError):
|
|
FeedSourceUpdate.model_validate({"enabled": True, "id": "00000000-0000-0000-0000-000000000001"})
|