From 16dedeb6705cc5c42b0b6b6a2d659852c79308ef Mon Sep 17 00:00:00 2001 From: Jens Date: Sat, 22 Aug 2026 15:59:01 +0200 Subject: [PATCH] claim analysis jobs atomically and keep acquisition on its official host MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects of the same kind: work that is supposed to be bounded is not. The analysis worker selected queued jobs and then set them to running in a second statement. A restarted process overlapping the previous one, or a second replica, could both select the same row and both start tiled GPU inference on it — duplicate analysis runs and double the GPU load. The AOI worker beside it already claims with FOR UPDATE SKIP LOCKED; this uses a conditional update, which is the same guarantee in one statement. run_once now reports jobs it actually claimed rather than jobs it looked at. urlopen follows redirects, so although every acquisition URL is built from settings and cannot be steered by a request payload, a misconfigured or compromised upstream could send the runtime to the loopback interface, to another container on the compose network, or to a cloud metadata endpoint — and the bytes would then be persisted under an official provenance. That is exactly the substitution the product forbids. All eight fetch sites now open through a guard that refuses private, loopback and link-local destinations (resolving the host first, so a DNS name cannot hide one) and refuses a redirect that leaves the configured origin or downgrades from HTTPS. The guard is proven by calling the services' own fetch paths, not by grepping for the call: every existing acquisition test injects an opener, which bypasses it by design. Co-Authored-By: Claude Opus 5 --- backend/app/services/analysis_job_worker.py | 30 +++- .../bathymetry_profile_acquisition_service.py | 3 +- .../app/services/dhmv_acquisition_service.py | 3 +- .../flood_hazard_acquisition_service.py | 3 +- .../mdk_bathymetry_acquisition_service.py | 3 +- .../services/mdk_bathymetry_probe_service.py | 2 +- .../orthophoto_acquisition_service.py | 3 +- .../app/services/outbound_request_guard.py | 142 +++++++++++++++ .../services/source_catalog_probe_service.py | 3 +- .../thematic_raster_acquisition_service.py | 3 +- .../test_analysis_job_claim_is_atomic.py | 135 ++++++++++++++ backend/tests/test_analysis_job_queue.py | 10 ++ backend/tests/test_outbound_request_guard.py | 164 ++++++++++++++++++ docs/KNOWN_LIMITATIONS.md | 5 + 14 files changed, 493 insertions(+), 16 deletions(-) create mode 100644 backend/app/services/outbound_request_guard.py create mode 100644 backend/tests/test_analysis_job_claim_is_atomic.py create mode 100644 backend/tests/test_outbound_request_guard.py diff --git a/backend/app/services/analysis_job_worker.py b/backend/app/services/analysis_job_worker.py index 1b9501f3..322c361b 100644 --- a/backend/app/services/analysis_job_worker.py +++ b/backend/app/services/analysis_job_worker.py @@ -65,16 +65,26 @@ class AnalysisJobWorker: return SegmentationService.run_segmentation(**common) @staticmethod - def _claim(db, job: Job) -> None: - """Take the job out of the queue before doing any work on it. + def claim(db, job: Job) -> bool: + """Take the job out of the queue, atomically. Returns whether we won. - Without this the next poll would pick the same row up again while the - first execution is still running on the GPU. + Selecting and then updating in a second statement lets two workers — + a restarted process overlapping the previous one, or a second replica — + both start tiled GPU inference on the same row. The conditional update + makes exactly one caller see a row count of 1; the AOI worker beside + this one already claims with FOR UPDATE SKIP LOCKED for the same reason. """ - job.status = "running" - db.add(job) + claimed = ( + db.query(Job) + .filter(Job.id == job.id, Job.status == "queued") + .update({Job.status: "running"}, synchronize_session=False) + ) db.commit() + if not claimed: + return False + job.status = "running" + return True @staticmethod def _finalize(db, job: Job, result: Any) -> None: @@ -132,9 +142,13 @@ class AnalysisJobWorker: ) if job.job_type in AnalysisJobWorker.HANDLED_JOB_TYPES and job.status == "queued" ] + claimed_count = 0 for job in rows: + if not AnalysisJobWorker.claim(session, job): + # Another worker took it between the select and the claim. + continue + claimed_count += 1 try: - AnalysisJobWorker._claim(session, job) result = AnalysisJobWorker._dispatch(session, job) AnalysisJobWorker._finalize(session, job, result) except Exception as exc: @@ -142,7 +156,7 @@ class AnalysisJobWorker: message = getattr(exc, "message", None) or str(exc) or "Unexpected analysis job failure" AnalysisJobWorker._mark_failed(session, job, code=str(code), message=str(message)) logger.exception("Analysis job failed job_id=%s job_type=%s", job.id, job.job_type) - return len(rows) + return claimed_count finally: if owns_session: session.close() diff --git a/backend/app/services/bathymetry_profile_acquisition_service.py b/backend/app/services/bathymetry_profile_acquisition_service.py index bf3a19c8..97e4eb78 100644 --- a/backend/app/services/bathymetry_profile_acquisition_service.py +++ b/backend/app/services/bathymetry_profile_acquisition_service.py @@ -15,6 +15,7 @@ from shapely.geometry import Point, box, mapping from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Area, Dataset, DatasetVersion, Project from app.schemas.bathymetry import ( BathymetryPartitionFinalizeRequest, @@ -224,7 +225,7 @@ class BathymetryProfileAcquisitionService: }, ) try: - with (opener or urlopen)(request, timeout=settings.bathymetry_profiles_timeout_seconds) as response: + with (opener or guarded_opener(url))(request, timeout=settings.bathymetry_profiles_timeout_seconds) as response: limit = settings.bathymetry_profiles_max_response_mb * 1024 * 1024 content = response.read(limit + 1) except HTTPError as exc: diff --git a/backend/app/services/dhmv_acquisition_service.py b/backend/app/services/dhmv_acquisition_service.py index a3f2db8f..d19a00fc 100644 --- a/backend/app/services/dhmv_acquisition_service.py +++ b/backend/app/services/dhmv_acquisition_service.py @@ -22,6 +22,7 @@ from shapely.ops import transform as shapely_transform from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Area, Dataset, Project from app.schemas.dhmv import DhmvAcquireRequest, DhmvAcquisitionResult, DhmvProductRead from app.services.dataset_service import DatasetService @@ -294,7 +295,7 @@ class DhmvAcquisitionService: ) max_bytes = settings.dhmv_max_response_mb * 1024 * 1024 try: - with (opener or urlopen)(request, timeout=settings.dhmv_timeout_seconds) as response: + with (opener or guarded_opener(request_url))(request, timeout=settings.dhmv_timeout_seconds) as response: content_type = str(response.headers.get("Content-Type", "")) content_length = response.headers.get("Content-Length") if content_length and int(content_length) > max_bytes: diff --git a/backend/app/services/flood_hazard_acquisition_service.py b/backend/app/services/flood_hazard_acquisition_service.py index 1da8b1df..29834d00 100644 --- a/backend/app/services/flood_hazard_acquisition_service.py +++ b/backend/app/services/flood_hazard_acquisition_service.py @@ -23,6 +23,7 @@ from shapely.ops import transform as shapely_transform from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Area, Dataset, Project from app.schemas.flood_hazard import FloodHazardAcquireRequest, FloodHazardAcquisitionResult, FloodHazardProductRead from app.services.dataset_service import DatasetService @@ -256,7 +257,7 @@ class FloodHazardAcquisitionService: request = Request(request_url, headers={"Accept": "*/*", "User-Agent": "GeoIntel/0.1 bounded-vmm-flood-hazard-acquisition"}) max_bytes = settings.flood_hazard_max_response_mb * 1024 * 1024 try: - with (opener or urlopen)(request, timeout=settings.flood_hazard_timeout_seconds) as response: + with (opener or guarded_opener(request_url))(request, timeout=settings.flood_hazard_timeout_seconds) as response: content_type = str(response.headers.get("Content-Type", "")) content_length = response.headers.get("Content-Length") if content_length and int(content_length) > max_bytes: diff --git a/backend/app/services/mdk_bathymetry_acquisition_service.py b/backend/app/services/mdk_bathymetry_acquisition_service.py index 2510c0d9..945e5113 100644 --- a/backend/app/services/mdk_bathymetry_acquisition_service.py +++ b/backend/app/services/mdk_bathymetry_acquisition_service.py @@ -10,6 +10,7 @@ from uuid import UUID from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Dataset from app.schemas.bathymetry import MdkBathymetryAcquireRequest, MdkBathymetryAcquisitionResult from app.services.dataset_service import DatasetService @@ -236,7 +237,7 @@ class MdkBathymetryAcquisitionService: ) max_bytes = settings.mdk_bathymetry_acquisition_max_response_mb * 1024 * 1024 try: - with (opener or urlopen)(request, timeout=settings.mdk_bathymetry_acquisition_timeout_seconds) as response: + with (opener or guarded_opener(request_url))(request, timeout=settings.mdk_bathymetry_acquisition_timeout_seconds) as response: content_type = str(response.headers.get("Content-Type", "")) if hasattr(response, "headers") else "" content = response.read(max_bytes + 1) except HTTPError as exc: diff --git a/backend/app/services/mdk_bathymetry_probe_service.py b/backend/app/services/mdk_bathymetry_probe_service.py index 6fd21167..6de5c4cb 100644 --- a/backend/app/services/mdk_bathymetry_probe_service.py +++ b/backend/app/services/mdk_bathymetry_probe_service.py @@ -43,7 +43,7 @@ class MdkBathymetryProbeService: "User-Agent": "GeoIntel/1.0 MDK-bathymetry-readiness-probe", }, ) - with (opener or urlopen)(request, timeout=settings.mdk_bathymetry_probe_timeout_seconds) as response: + with (opener or guarded_opener(capabilities_url))(request, timeout=settings.mdk_bathymetry_probe_timeout_seconds) as response: limit = settings.mdk_bathymetry_probe_max_response_mb * 1024 * 1024 content = response.read(limit + 1) if len(content) > limit: diff --git a/backend/app/services/orthophoto_acquisition_service.py b/backend/app/services/orthophoto_acquisition_service.py index 52c51816..c8ed9450 100644 --- a/backend/app/services/orthophoto_acquisition_service.py +++ b/backend/app/services/orthophoto_acquisition_service.py @@ -21,6 +21,7 @@ from shapely.ops import transform as shapely_transform from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Area, Dataset, Project from app.schemas.orthophoto import OrthophotoAcquireRequest, OrthophotoAcquisitionResult, OrthophotoProductRead from app.services.dataset_service import DatasetService @@ -469,7 +470,7 @@ class OrthophotoAcquisitionService: @staticmethod def _fetch(request_url: str, settings: Settings, opener: Callable[..., Any] | None = None) -> tuple[bytes, str]: request = Request(request_url, headers={"User-Agent": "GeoIntel/0.1 bounded-orthophoto-acquisition"}) - open_request = opener or urlopen + open_request = opener or guarded_opener(request_url) try: with open_request(request, timeout=settings.orthophoto_timeout_seconds) as response: content_type = str(response.headers.get("Content-Type", "")) diff --git a/backend/app/services/outbound_request_guard.py b/backend/app/services/outbound_request_guard.py new file mode 100644 index 00000000..0603bbeb --- /dev/null +++ b/backend/app/services/outbound_request_guard.py @@ -0,0 +1,142 @@ +"""Keep bounded acquisition bounded to the official host it was aimed at. + +Every acquisition service builds its URL from configured settings, so a request +payload cannot point the runtime somewhere else. The redirect chain can: +``urlopen`` follows redirects by default, so a misconfigured or compromised +upstream can send the runtime to the loopback interface, to another container +on the compose network, or to a cloud metadata endpoint — and whatever comes +back is then persisted as official source data. + +That is the substitution the product explicitly forbids, so a redirect that +leaves the configured origin fails closed instead. +""" + +from __future__ import annotations + +import ipaddress +import socket +from collections.abc import Callable +from typing import Any +from urllib.parse import urlparse +from urllib.request import urlopen + +from app.core.errors import AppError + +ALLOWED_SCHEMES = {"http", "https"} + + +def _reject(code: str, message: str, **details: Any) -> AppError: + return AppError(code=code, message=message, details=details or None, status_code=502) + + +def _resolved_addresses(host: str) -> list[str]: + """Every address the host resolves to, so a DNS name cannot hide a private one.""" + + try: + infos = socket.getaddrinfo(host, None) + except OSError: + # Resolution failure is not the guard's problem: the request itself will + # fail with a clear provider error a moment later. + return [] + return [str(info[4][0]) for info in infos] + + +def _is_public_address(value: str) -> bool: + try: + address = ipaddress.ip_address(value) + except ValueError: + return False + return not ( + address.is_private + or address.is_loopback + or address.is_link_local + or address.is_reserved + or address.is_multicast + or address.is_unspecified + ) + + +def assert_public_http_url(url: str) -> None: + """Refuse anything that is not an ordinary outbound HTTP(S) destination.""" + + parsed = urlparse(url) + if parsed.scheme not in ALLOWED_SCHEMES: + raise _reject( + "OUTBOUND_URL_NOT_ALLOWED", + "Bounded acquisition only performs HTTP(S) requests.", + scheme=parsed.scheme, + ) + host = parsed.hostname + if not host: + raise _reject("OUTBOUND_URL_NOT_ALLOWED", "Outbound request has no host.", url=url) + + literal = host.strip("[]") + candidates = [literal] if _looks_like_ip(literal) else _resolved_addresses(host) + if candidates and not all(_is_public_address(candidate) for candidate in candidates): + raise _reject( + "OUTBOUND_URL_NOT_ALLOWED", + "Bounded acquisition refuses a private, loopback or link-local destination.", + host=host, + ) + + +def _looks_like_ip(value: str) -> bool: + try: + ipaddress.ip_address(value) + except ValueError: + return False + return True + + +def assert_same_origin_redirect(original_url: str, final_url: str) -> None: + """Allow a redirect only within the origin the request was aimed at. + + A path change is normal — providers version their endpoints. A host change + means the bytes no longer come from the source the provenance will claim, + and a scheme downgrade means they are no longer protected in transit. + """ + + if not final_url or final_url == original_url: + return + + original = urlparse(original_url) + final = urlparse(final_url) + if (final.hostname or "").casefold() != (original.hostname or "").casefold(): + raise _reject( + "OUTBOUND_REDIRECT_NOT_ALLOWED", + "The official endpoint redirected to a different host; acquisition fails closed.", + expected_host=original.hostname, + redirect_host=final.hostname, + ) + if original.scheme == "https" and final.scheme != "https": + raise _reject( + "OUTBOUND_REDIRECT_NOT_ALLOWED", + "The official endpoint redirected from HTTPS to an unprotected scheme.", + redirect_scheme=final.scheme, + ) + assert_public_http_url(final_url) + + +def guarded_opener(expected_url: str) -> Callable[..., Any]: + """An ``urlopen`` replacement that verifies where the response came from. + + ``urlopen`` has already followed the redirect chain by the time it returns, + so the check is on ``response.url``: the body is still unread, and raising + here means nothing off-origin is ever parsed or persisted. + """ + + assert_public_http_url(expected_url) + + def _open(request: Any, *args: Any, _transport: Callable[..., Any] | None = None, **kwargs: Any) -> Any: + response = (_transport or urlopen)(request, *args, **kwargs) + final_url = str(getattr(response, "url", "") or "") + try: + assert_same_origin_redirect(expected_url, final_url) + except AppError: + close = getattr(response, "close", None) + if callable(close): + close() + raise + return response + + return _open diff --git a/backend/app/services/source_catalog_probe_service.py b/backend/app/services/source_catalog_probe_service.py index d54dfa21..cfe13f4a 100644 --- a/backend/app/services/source_catalog_probe_service.py +++ b/backend/app/services/source_catalog_probe_service.py @@ -18,6 +18,7 @@ from sqlalchemy.orm import Session from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Dataset, Project from app.schemas.source_catalog import ( SourceCatalogProbeItem, @@ -179,7 +180,7 @@ def _bounded_fetch( ) max_bytes = (max_response_mb or settings.source_catalog_probe_max_response_mb) * 1024 * 1024 try: - with (opener or urlopen)(request, timeout=settings.source_catalog_probe_timeout_seconds) as response: + with (opener or guarded_opener(url))(request, timeout=settings.source_catalog_probe_timeout_seconds) as response: content_length = _header(response.headers, "Content-Length") if content_length: try: diff --git a/backend/app/services/thematic_raster_acquisition_service.py b/backend/app/services/thematic_raster_acquisition_service.py index b1b444ba..366c5bfc 100644 --- a/backend/app/services/thematic_raster_acquisition_service.py +++ b/backend/app/services/thematic_raster_acquisition_service.py @@ -21,6 +21,7 @@ from shapely.ops import transform as shapely_transform from app.core.config import Settings, get_settings from app.core.errors import AppError +from app.services.outbound_request_guard import guarded_opener from app.models import Area, Dataset, Project from app.schemas.thematic_raster import ( ThematicRasterAcquireRequest, @@ -359,7 +360,7 @@ class ThematicRasterAcquisitionService: max_bytes = settings.thematic_raster_max_response_mb * 1024 * 1024 for attempt in range(1, ThematicRasterAcquisitionService.WCS_FETCH_ATTEMPTS + 1): try: - with (opener or urlopen)(request, timeout=settings.thematic_raster_timeout_seconds) as response: + with (opener or guarded_opener(request_url))(request, timeout=settings.thematic_raster_timeout_seconds) as response: content_type = str(response.headers.get("Content-Type", "")) content_length = response.headers.get("Content-Length") if content_length and int(content_length) > max_bytes: diff --git a/backend/tests/test_analysis_job_claim_is_atomic.py b/backend/tests/test_analysis_job_claim_is_atomic.py new file mode 100644 index 00000000..06429c47 --- /dev/null +++ b/backend/tests/test_analysis_job_claim_is_atomic.py @@ -0,0 +1,135 @@ +"""A queued run must be claimed once, even if two workers look at it. + +The worker selected queued jobs and then set them to running in a second +statement. Two workers — an API restart overlapping the previous process, or a +second replica — could both select the same row and both start tiled GPU +inference on it, producing duplicate analysis runs and doubling the GPU load. + +The AOI worker beside it already claims with ``FOR UPDATE SKIP LOCKED``. This +uses a conditional update, which is the same guarantee expressed in one +statement: exactly one caller sees a row count of 1. +""" + +from __future__ import annotations + +from uuid import uuid4 + +from app.models import Job +from app.services.analysis_job_worker import AnalysisJobWorker + + +class _Update: + """Mimics a conditional UPDATE: the first caller wins, the rest see zero.""" + + def __init__(self, store: dict, job_id): + self.store = store + self.job_id = job_id + + def update(self, values, **_kwargs) -> int: + if self.store.get(self.job_id) != "queued": + return 0 + self.store[self.job_id] = "running" + return 1 + + +class _Query: + def __init__(self, session, model): + self.session = session + self.model = model + self.job_id = None + + def filter(self, *criteria): + for criterion in criteria: + right = getattr(criterion, "right", None) + value = getattr(right, "value", None) + if isinstance(value, type(uuid4())): + self.job_id = value + return self + + def update(self, values, **kwargs) -> int: + return _Update(self.session.statuses, self.job_id).update(values, **kwargs) + + def order_by(self, *_args): + return self + + def limit(self, _count): + return self + + def all(self): + return list(self.session.rows) + + +class _Session: + def __init__(self, rows: list[Job]): + self.rows = rows + self.statuses = {row.id: row.status for row in rows} + self.committed = 0 + + def query(self, model): + return _Query(self, model) + + def get(self, _model, item_id): + return next((row for row in self.rows if row.id == item_id), None) + + def add(self, _item): + return None + + def commit(self): + self.committed += 1 + + def rollback(self): + return None + + def close(self): + return None + + +def _job() -> Job: + return Job( + id=uuid4(), + job_type="detection.run", + status="queued", + project_id=uuid4(), + parameters_json={}, + ) + + +def test_the_first_claim_wins() -> None: + job = _job() + session = _Session([job]) + + assert AnalysisJobWorker.claim(session, job) is True + assert job.status == "running" + + +def test_a_second_claim_on_the_same_job_is_refused() -> None: + job = _job() + session = _Session([job]) + + assert AnalysisJobWorker.claim(session, job) is True + assert AnalysisJobWorker.claim(session, job) is False + + +def test_a_job_that_is_no_longer_queued_cannot_be_claimed() -> None: + job = _job() + session = _Session([job]) + session.statuses[job.id] = "success" + + assert AnalysisJobWorker.claim(session, job) is False + + +def test_an_unclaimable_job_is_skipped_rather_than_run(monkeypatch) -> None: + job = _job() + session = _Session([job]) + session.statuses[job.id] = "running" + dispatched: list[Job] = [] + monkeypatch.setattr( + AnalysisJobWorker, + "_dispatch", + staticmethod(lambda _db, item: dispatched.append(item)), + ) + + processed = AnalysisJobWorker.run_once(db=session) + + assert processed == 0 + assert dispatched == [] diff --git a/backend/tests/test_analysis_job_queue.py b/backend/tests/test_analysis_job_queue.py index 2421d68b..543f279e 100644 --- a/backend/tests/test_analysis_job_queue.py +++ b/backend/tests/test_analysis_job_queue.py @@ -26,6 +26,16 @@ class FakeQuery: def filter(self, *criteria): return self + def update(self, values, **_kwargs) -> int: + """Stand in for the conditional claim: succeeds while still queued.""" + + claimed = 0 + for row in self.rows: + if getattr(row, "status", None) == "queued": + row.status = "running" + claimed += 1 + return claimed + def order_by(self, *_args): return self diff --git a/backend/tests/test_outbound_request_guard.py b/backend/tests/test_outbound_request_guard.py new file mode 100644 index 00000000..d398b4b6 --- /dev/null +++ b/backend/tests/test_outbound_request_guard.py @@ -0,0 +1,164 @@ +"""Bounded acquisition must stay bounded to the official host. + +Every acquisition service builds its URL from configured settings, so the +request payload cannot point the runtime anywhere. The redirect chain can: +``urlopen`` follows redirects by default, so a misconfigured or compromised +upstream can send the runtime to ``127.0.0.1``, to the container network, or to +a cloud metadata endpoint — and the response is then persisted as if it were +official source data. + +The product's stated rule is that acquisition fails closed and never +substitutes fabricated data for official data. A redirect off the configured +host is exactly that substitution. +""" + +from __future__ import annotations + +import pytest + +from app.core.errors import AppError +from app.services.outbound_request_guard import ( + assert_public_http_url, + assert_same_origin_redirect, +) + + +class TestUrlShape: + def test_an_official_https_endpoint_is_accepted(self) -> None: + assert_public_http_url("https://geo.api.vlaanderen.be/dhmv/wcs?SERVICE=WCS") + + def test_a_non_http_scheme_is_refused(self) -> None: + with pytest.raises(AppError) as exc_info: + assert_public_http_url("file:///etc/passwd") + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" + + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1:8000/internal", + "http://localhost/internal", + "http://10.1.2.3/internal", + "http://192.168.10.150/internal", + "http://172.16.0.9/internal", + "http://169.254.169.254/latest/meta-data/", + "http://[::1]/internal", + ], + ) + def test_private_and_loopback_destinations_are_refused(self, url: str) -> None: + with pytest.raises(AppError) as exc_info: + assert_public_http_url(url) + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" + + def test_a_url_without_a_host_is_refused(self) -> None: + with pytest.raises(AppError): + assert_public_http_url("https:///no-host") + + +class TestRedirects: + def test_a_redirect_within_the_same_origin_is_allowed(self) -> None: + assert_same_origin_redirect( + "https://geo.api.vlaanderen.be/dhmv/wcs", + "https://geo.api.vlaanderen.be/dhmv/wcs/v2?x=1", + ) + + def test_a_redirect_to_another_host_is_refused(self) -> None: + with pytest.raises(AppError) as exc_info: + assert_same_origin_redirect( + "https://geo.api.vlaanderen.be/dhmv/wcs", + "https://cdn.example.net/payload.tif", + ) + + assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED" + assert "cdn.example.net" in str(exc_info.value.details) + + def test_a_downgrade_to_plain_http_is_refused(self) -> None: + with pytest.raises(AppError) as exc_info: + assert_same_origin_redirect( + "https://geo.api.vlaanderen.be/wcs", + "http://geo.api.vlaanderen.be/wcs", + ) + + assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED" + + def test_a_redirect_to_the_loopback_is_refused_even_on_the_same_scheme(self) -> None: + with pytest.raises(AppError): + assert_same_origin_redirect("https://geo.api.vlaanderen.be/wcs", "https://127.0.0.1/wcs") + + def test_an_upgrade_to_https_stays_allowed(self) -> None: + assert_same_origin_redirect("http://geo.example.be/wcs", "https://geo.example.be/wcs") + + +def test_the_guard_opener_refuses_a_cross_host_redirect() -> None: + """The opener is what the acquisition services actually call.""" + + from app.services.outbound_request_guard import guarded_opener + + opener = guarded_opener("https://geo.api.vlaanderen.be/wcs") + + class _Redirecting: + def __init__(self, location: str) -> None: + self.url = location + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + with pytest.raises(AppError) as exc_info: + with opener( + type("Req", (), {"full_url": "https://geo.api.vlaanderen.be/wcs"})(), + timeout=1, + _transport=lambda *_a, **_k: _Redirecting("https://evil.example.net/x"), + ): + pass + + assert exc_info.value.code == "OUTBOUND_REDIRECT_NOT_ALLOWED" + + +class TestTheGuardIsWiredIntoAcquisition: + """Behavioural, not a grep: each service is called on its real fetch path. + + Every existing acquisition test injects an ``opener``, which bypasses the + guard by design — that is how those tests stub the network. These call the + production default instead. + """ + + def _settings(self): + from app.core.config import Settings + + return Settings(_env_file=None) + + def test_dhmv_refuses_a_loopback_endpoint(self) -> None: + from app.services.dhmv_acquisition_service import DhmvAcquisitionService + + with pytest.raises(AppError) as exc_info: + DhmvAcquisitionService._fetch("http://127.0.0.1:9/wcs", self._settings()) + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" + + def test_flood_hazard_refuses_a_link_local_endpoint(self) -> None: + from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService + + with pytest.raises(AppError) as exc_info: + FloodHazardAcquisitionService._fetch("http://169.254.169.254/latest/", self._settings()) + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" + + def test_thematic_raster_refuses_a_private_endpoint(self) -> None: + from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService + + with pytest.raises(AppError) as exc_info: + ThematicRasterAcquisitionService._fetch("http://10.0.0.5/product.tif", self._settings()) + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" + + def test_orthophoto_refuses_a_private_endpoint(self) -> None: + from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService + + with pytest.raises(AppError) as exc_info: + OrthophotoAcquisitionService._fetch("http://192.168.10.150/wms", self._settings()) + + assert exc_info.value.code == "OUTBOUND_URL_NOT_ALLOWED" diff --git a/docs/KNOWN_LIMITATIONS.md b/docs/KNOWN_LIMITATIONS.md index 0100fe46..3c6f735d 100644 --- a/docs/KNOWN_LIMITATIONS.md +++ b/docs/KNOWN_LIMITATIONS.md @@ -62,6 +62,11 @@ runtime source of truth. - An export states its own completeness in `geointel_provenance`. A capped export is still a valid, usable file — it simply no longer implies it holds everything the selection contains. +- Bounded acquisition refuses a redirect that leaves the configured origin, and + refuses any private, loopback or link-local destination. An official endpoint + that legitimately moves to a new host therefore fails closed until the + operator updates the configured URL, which is the intended trade: bytes from + an unexpected host must never be persisted under an official provenance. ## Historical analysis