From f6041e79655d195ea44235b870bb1df0b03fbe8b Mon Sep 17 00:00:00 2001 From: Raul Bardaji Date: Sun, 30 Aug 2026 20:30:31 +0200 Subject: [PATCH] feat(pelican): add GET /pelican/read for inline object contents The integration could list objects and hand back a file to save, but there was no way to get an object's contents inline, so a caller had to write to disk and read back just to work with the data. /pelican/read returns the contents in the response body instead. Text comes back as text and non-UTF-8 payloads are base64-encoded rather than refused, with an encoding field saying which. The read is capped by PELICAN_MAX_READ_BYTES and the cap bounds the read itself, so an oversized object is never pulled into memory just to be rejected. Part of #262. --- CHANGELOG.md | 6 + api/routes/pelican_routes.py | 94 +++++++++ api/services/pelican_services/read_file.py | 98 +++++++++ docs/configuration.md | 9 + example.env | 7 + tests/test_pelican_read.py | 220 +++++++++++++++++++++ 6 files changed, 434 insertions(+) create mode 100644 api/services/pelican_services/read_file.py create mode 100644 tests/test_pelican_read.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c0211c5..de99dc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`GET /pelican/read` returns the contents of a Pelican object in the response body.** The Pelican integration could list objects (`/browse`, `/info`) and hand back a file to save (`/download`), but there was no way to get an object's contents inline — a caller wanting to work with the data had to write it to disk first and read it back. The new route returns the contents directly, so a subscriber can take the object referenced by an event and feed it straight into its own code. Text is returned as text; anything that is not valid UTF-8 is base64-encoded rather than refused, since Pelican namespaces hold binary payloads too, and the response says which of the two it is in an `encoding` field alongside `path`, `size` and `content`. The route is capped by `PELICAN_MAX_READ_BYTES` (10 MiB by default) because the contents travel in the response body; the cap is enforced while reading rather than after, so an oversized object is never pulled into the API's memory just to be rejected, and the caller is told to use `/download` instead. Failures are distinguishable from the status code: 404 when the object is not in the federation, 413 when it is past the limit, 502 when the federation cannot be reached. + +### Backwards compatibility +- Purely additive. `PELICAN_MAX_READ_BYTES` is optional and defaults to 10 MiB, and a missing, non-numeric or non-positive value falls back to that default, so existing deployments need no change. The new route sits behind the same authorization as the rest of the Pelican routes, and is only mounted when `PELICAN_ENABLED` is set. + ## [0.34.22] - 2026-08-30 ### Fixed diff --git a/api/routes/pelican_routes.py b/api/routes/pelican_routes.py index 58bee13..fbbed21 100644 --- a/api/routes/pelican_routes.py +++ b/api/routes/pelican_routes.py @@ -15,6 +15,7 @@ get_file_info, ) from api.services.pelican_services.download_file import download_file, stream_file +from api.services.pelican_services.read_file import read_object from api.services.pelican_services.import_metadata import import_file_as_resource from api.services.auth_services import ( get_user_for_read_operation, @@ -25,6 +26,11 @@ logger = logging.getLogger(__name__) +# Largest object /pelican/read will return inline. Anything bigger is a +# download, not a read: the contents go into the response body, so an +# unbounded read would put an arbitrary object into the endpoint's memory. +DEFAULT_MAX_READ_BYTES = 10 * 1024 * 1024 + # The read gate is declared on the router rather than on each route: these # endpoints shipped completely unauthenticated (issue #261), and a # router-level dependency means a route added later cannot silently miss it. @@ -245,6 +251,94 @@ async def download( raise HTTPException(status_code=500, detail=f"Error downloading file: {str(e)}") +def _max_read_bytes() -> int: + """ + Resolve the inline read limit from ``PELICAN_MAX_READ_BYTES``. + + Settings are declared with ``extra: "allow"``, so a malformed value + reaches this point instead of failing at startup; fall back to the + default rather than letting a typo disable the limit. + + Returns + ------- + int + Limit in bytes. + """ + raw = os.getenv("PELICAN_MAX_READ_BYTES", "") + if not raw: + return DEFAULT_MAX_READ_BYTES + try: + value = int(raw) + except ValueError: + logger.warning( + f"PELICAN_MAX_READ_BYTES is not an integer ({raw!r}); " + f"using {DEFAULT_MAX_READ_BYTES}." + ) + return DEFAULT_MAX_READ_BYTES + if value <= 0: + logger.warning( + f"PELICAN_MAX_READ_BYTES must be positive (got {value}); " + f"using {DEFAULT_MAX_READ_BYTES}." + ) + return DEFAULT_MAX_READ_BYTES + return value + + +@router.get("/read") +async def read_file_contents( + path: str = Query(..., description="Path of the object to read"), + federation: str = Query("osdf", description="Federation to query"), +): + """ + Read a Pelican object and return its contents in the response body. + + ``/download`` hands back a file to save; this returns the contents + inline so they can be piped straight into the caller's own code. + + Parameters + ---------- + path : str + Path of the object to read + federation : str + Federation name (default "osdf") + + Returns + ------- + dict + ``path``, ``size``, ``encoding`` ("utf-8" or "base64") and + ``content`` + + Raises + ------ + HTTPException + - 404: Object not found in the federation + - 413: Object larger than the inline read limit + - 502: The federation could not be reached + """ + try: + pelican_repo = get_pelican_repo(federation) + result = read_object(pelican_repo, path, _max_read_bytes()) + + if not result["success"]: + status_by_reason = { + "not_found": 404, + "too_large": 413, + "unavailable": 502, + } + raise HTTPException( + status_code=status_by_reason.get(result.get("reason"), 502), + detail=result["error"], + ) + + return result + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error reading Pelican object {path}: {e}") + raise HTTPException(status_code=500, detail=f"Error reading object: {str(e)}") + + @router.post("/import-metadata") async def import_metadata( request: ImportMetadataRequest, diff --git a/api/services/pelican_services/read_file.py b/api/services/pelican_services/read_file.py new file mode 100644 index 0000000..c07165e --- /dev/null +++ b/api/services/pelican_services/read_file.py @@ -0,0 +1,98 @@ +# api/services/pelican_services/read_file.py +""" +Service for reading Pelican objects into the response body. +""" + +import base64 +import logging +from typing import Any, Dict + +from api.repositories.pelican_repository import PelicanRepository + +logger = logging.getLogger(__name__) + + +def read_object( + pelican_repo: PelicanRepository, path: str, max_bytes: int +) -> Dict[str, Any]: + """ + Read the contents of a Pelican object and return them inline. + + Unlike :func:`api.services.pelican_services.download_file.download_file`, + which hands the caller a file to save, this returns the contents in the + response body so they can be fed straight into the caller's own code + without a temporary file. + + The size cap is enforced *while* reading rather than after: an object + larger than the cap must not be pulled into the endpoint's memory just + to be rejected, which is why this reads through an open handle instead + of calling ``read_file``. + + Parameters + ---------- + pelican_repo : PelicanRepository + Initialized Pelican repository + path : str + Path of the object to read + max_bytes : int + Largest object this endpoint will return inline + + Returns + ------- + dict + On success, ``path``, ``size``, ``encoding`` and ``content``. + On failure, ``error`` plus a ``reason`` of ``not_found``, + ``too_large`` or ``unavailable`` for the caller to map to a + status code. + """ + try: + with pelican_repo.open_file(path, mode="rb") as handle: + # One byte past the cap, so an oversized object is detected + # without being read in full. + payload = handle.read(max_bytes + 1) + except FileNotFoundError as exc: + logger.info(f"Pelican object not found: {path}") + return { + "success": False, + "path": path, + "error": f"Object not found in the federation: {str(exc) or path}", + "reason": "not_found", + } + except Exception as exc: + logger.error(f"Error reading Pelican object {path}: {exc}") + return { + "success": False, + "path": path, + "error": f"{type(exc).__name__}: {exc}", + "reason": "unavailable", + } + + if len(payload) > max_bytes: + return { + "success": False, + "path": path, + "error": ( + f"Object is larger than the {max_bytes} byte inline read " + "limit. Use /pelican/download to retrieve it as a file." + ), + "reason": "too_large", + } + + # Text is returned as text so a caller can use it directly; anything + # that is not valid UTF-8 is base64-encoded rather than rejected, and + # says so, because Pelican namespaces hold binary payloads too. + try: + content = payload.decode("utf-8") + encoding = "utf-8" + except UnicodeDecodeError: + content = base64.b64encode(payload).decode("ascii") + encoding = "base64" + + logger.info(f"Read {len(payload)} bytes from Pelican object {path}") + return { + "success": True, + "path": path, + "size": len(payload), + "encoding": encoding, + "content": content, + } diff --git a/docs/configuration.md b/docs/configuration.md index 08ed867..5246b36 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -340,6 +340,15 @@ Default Pelican federation, format `pelican://host` (e.g. `pelican://osg-htc.org Read straight from origin servers instead of caches. **Where:** keep `False` for better performance unless you have a reason. +#### `PELICAN_MAX_READ_BYTES` +*Optional · default: `10485760` (10 MiB).* +Largest object `GET /pelican/read` returns inline. That endpoint puts the +contents in the response body, so this caps what a single request can pull into +the API's memory; a larger object is refused with 413 and must be fetched with +`/pelican/download`. A non-numeric or non-positive value falls back to the +default. **Where:** raise it only if your callers genuinely read larger objects +inline. + --- ## Remote execution (Rexec) diff --git a/example.env b/example.env index 7dc0874..f4227dd 100644 --- a/example.env +++ b/example.env @@ -318,6 +318,13 @@ PELICAN_FEDERATION_URL= # Set to False to use caching infrastructure (recommended for better performance) PELICAN_DIRECT_READS=False +# Largest object /pelican/read will return inline, in bytes +# The read endpoint returns file contents in the response body, so this caps +# how much can be pulled into the API's memory by a single request +# Anything larger must be fetched with /pelican/download instead +# Default: 10485760 (10 MiB) +PELICAN_MAX_READ_BYTES=10485760 + # ============================================== # Rexec Deployment API Configuration # ============================================== diff --git a/tests/test_pelican_read.py b/tests/test_pelican_read.py new file mode 100644 index 0000000..b5ce328 --- /dev/null +++ b/tests/test_pelican_read.py @@ -0,0 +1,220 @@ +"""Tests for the Pelican inline read operation (issue #262).""" + +import base64 +from unittest.mock import MagicMock, patch + +import pytest + +from api.services.pelican_services.read_file import read_object + + +def _repo_returning(payload, error=None): + """Build a repository whose open_file yields ``payload`` or raises.""" + repo = MagicMock() + if error is not None: + repo.open_file.side_effect = error + return repo + handle = MagicMock() + handle.read.return_value = payload + repo.open_file.return_value.__enter__.return_value = handle + repo.open_file.return_value.__exit__.return_value = False + return repo + + +class TestReadObjectService: + """Tests for read_object.""" + + def test_text_is_returned_as_text(self): + repo = _repo_returning(b"time,value\n1,2\n") + + result = read_object(repo, "/public/data.csv", 1024) + + assert result["success"] is True + assert result["encoding"] == "utf-8" + assert result["content"] == "time,value\n1,2\n" + assert result["size"] == 15 + assert result["path"] == "/public/data.csv" + + def test_binary_is_base64_encoded_not_rejected(self): + """Pelican namespaces hold binary payloads; they must still read.""" + payload = b"\x89PNG\r\n\x1a\n\xff\xfe" + repo = _repo_returning(payload) + + result = read_object(repo, "/public/img.png", 1024) + + assert result["success"] is True + assert result["encoding"] == "base64" + assert base64.b64decode(result["content"]) == payload + + def test_oversized_object_is_refused(self): + """An object past the cap is refused rather than returned.""" + repo = _repo_returning(b"x" * 11) + + result = read_object(repo, "/public/big.bin", 10) + + assert result["success"] is False + assert result["reason"] == "too_large" + assert "/pelican/download" in result["error"] + + def test_cap_is_enforced_while_reading(self): + """ + The cap must bound the read itself, otherwise an arbitrarily large + object is pulled into memory just to be rejected. + """ + repo = _repo_returning(b"x" * 11) + + read_object(repo, "/public/big.bin", 10) + + handle = repo.open_file.return_value.__enter__.return_value + handle.read.assert_called_once_with(11) + + def test_missing_object_reports_not_found(self): + repo = _repo_returning(None, error=FileNotFoundError("no such object")) + + result = read_object(repo, "/public/gone.txt", 1024) + + assert result["success"] is False + assert result["reason"] == "not_found" + + def test_federation_failure_names_the_exception_type(self): + """The class name is carried so the cause is identifiable.""" + repo = _repo_returning(None, error=ConnectionError("origin unreachable")) + + result = read_object(repo, "/public/data.csv", 1024) + + assert result["success"] is False + assert result["reason"] == "unavailable" + assert "ConnectionError" in result["error"] + + def test_exactly_at_the_cap_is_allowed(self): + """The limit is inclusive; only past it is refused.""" + repo = _repo_returning(b"x" * 10) + + result = read_object(repo, "/public/edge.txt", 10) + + assert result["success"] is True + assert result["size"] == 10 + + +class TestMaxReadBytes: + """Tests for the PELICAN_MAX_READ_BYTES resolution.""" + + def test_default_when_unset(self): + from api.routes.pelican_routes import DEFAULT_MAX_READ_BYTES, _max_read_bytes + + with patch.dict("os.environ", {}, clear=False): + import os + + os.environ.pop("PELICAN_MAX_READ_BYTES", None) + assert _max_read_bytes() == DEFAULT_MAX_READ_BYTES + + def test_explicit_value_is_used(self): + from api.routes.pelican_routes import _max_read_bytes + + with patch.dict("os.environ", {"PELICAN_MAX_READ_BYTES": "2048"}): + assert _max_read_bytes() == 2048 + + def test_garbage_falls_back_to_default(self): + """Settings allow extra keys, so a typo has to be caught here.""" + from api.routes.pelican_routes import DEFAULT_MAX_READ_BYTES, _max_read_bytes + + with patch.dict("os.environ", {"PELICAN_MAX_READ_BYTES": "ten megabytes"}): + assert _max_read_bytes() == DEFAULT_MAX_READ_BYTES + + def test_non_positive_falls_back_to_default(self): + from api.routes.pelican_routes import DEFAULT_MAX_READ_BYTES, _max_read_bytes + + with patch.dict("os.environ", {"PELICAN_MAX_READ_BYTES": "0"}): + assert _max_read_bytes() == DEFAULT_MAX_READ_BYTES + + +class TestReadRoute: + """Tests for GET /pelican/read.""" + + @staticmethod + def _client(): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from api.routes.pelican_routes import router + + app = FastAPI() + app.include_router(router) + return app, TestClient(app) + + @staticmethod + def _as(roles): + return lambda: { + "roles": roles, + "groups": [], + "sub": "test_user", + "username": "Test User", + } + + def _get(self, app, client, roles=("ndp_viewer",)): + from api.services.auth_services import get_current_user + + app.dependency_overrides[get_current_user] = self._as(list(roles)) + try: + return client.get("/pelican/read", params={"path": "/public/data.csv"}) + finally: + app.dependency_overrides.clear() + + @patch("api.routes.pelican_routes.get_pelican_repo") + @patch("api.routes.pelican_routes.read_object") + def test_success(self, mock_read, mock_get_repo): + mock_get_repo.return_value = MagicMock() + mock_read.return_value = { + "success": True, + "path": "/public/data.csv", + "size": 4, + "encoding": "utf-8", + "content": "a,b\n", + } + app, client = self._client() + + response = self._get(app, client) + + assert response.status_code == 200 + assert response.json()["content"] == "a,b\n" + + @pytest.mark.parametrize( + "reason,expected_status", + [("not_found", 404), ("too_large", 413), ("unavailable", 502)], + ) + @patch("api.routes.pelican_routes.get_pelican_repo") + @patch("api.routes.pelican_routes.read_object") + def test_failure_reasons_map_to_status_codes( + self, mock_read, mock_get_repo, reason, expected_status + ): + mock_get_repo.return_value = MagicMock() + mock_read.return_value = { + "success": False, + "path": "/public/data.csv", + "error": "boom", + "reason": reason, + } + app, client = self._client() + + response = self._get(app, client) + + assert response.status_code == expected_status + assert response.json()["detail"] == "boom" + + @patch("api.routes.pelican_routes.read_object") + def test_requires_authentication(self, mock_read): + """The new route inherits the router gate added for issue #261.""" + _app, client = self._client() + + response = client.get("/pelican/read", params={"path": "/public/data.csv"}) + + assert response.status_code == 401 + mock_read.assert_not_called() + + @patch("api.routes.pelican_routes.read_object") + def test_rejects_user_without_role(self, mock_read): + app, client = self._client() + + response = self._get(app, client, roles=()) + + assert response.status_code == 403 + mock_read.assert_not_called()