Prevent users from downloading each other's sandboxes

src/diracx/routers/job_manager/sandboxes.py

@@ -155,6 +155,7 @@ def pfn_to_key(pfn: str) -> str:
 async def get_sandbox_file(
     pfn: Annotated[str, Query(max_length=256, pattern=SANDBOX_PFN_REGEX)],
     settings: SandboxStoreSettings,
+    user_info: Annotated[AuthorizedUserInfo, Depends(verify_dirac_access_token)],
 ) -> SandboxDownloadResponse:
     """Get a presigned URL to download a sandbox file
 
@@ -164,7 +165,24 @@ async def get_sandbox_file(
     most storage backends return an error when they receive an authorization
     header for a presigned URL.
     """
-    # TODO: Prevent people from downloading other people's sandboxes?
+    required_prefix = (
+        "/"
+        + "/".join(
+            [
+                "S3",
+                settings.bucket_name,
+                user_info.vo,
+                user_info.dirac_group,
+                user_info.preferred_username,
+            ]
+        )
+        + "/"
+    )
+    if not pfn.startswith(required_prefix):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST,
+            detail=f"Invalid PFN. PFN must start with {required_prefix}",
+        )
     # TODO: Support by name and by job id?
     presigned_url = settings.s3_client.generate_presigned_url(
         ClientMethod="get_object",

tests/routers/jobs/test_sandboxes.py

@@ -2,13 +2,18 @@
 
 import hashlib
 import secrets
+from copy import deepcopy
 from io import BytesIO
 
 import requests
 from fastapi.testclient import TestClient
 
+from diracx.routers.auth import AuthSettings, create_token
 
-def test_upload_then_download(normal_user_client: TestClient):
+
+def test_upload_then_download(
+    normal_user_client: TestClient, test_auth_settings: AuthSettings
+):
     data = secrets.token_bytes(512)
     checksum = hashlib.sha256(data).hexdigest()
 
@@ -42,6 +47,20 @@ def test_upload_then_download(normal_user_client: TestClient):
     assert r.status_code == 200, r.text
     assert r.content == data
 
+    # Modify the authorization payload to be another user
+    other_user_payload = deepcopy(normal_user_client.dirac_token_payload)
+    other_user_payload["preferred_username"] = "other_user"
+    other_user_token = create_token(other_user_payload, test_auth_settings)
+
+    # Make sure another user can't download the sandbox
+    r = normal_user_client.get(
+        "/jobs/sandbox",
+        params={"pfn": sandbox_pfn},
+        headers={"Authorization": f"Bearer {other_user_token}"},
+    )
+    assert r.status_code == 400, r.text
+    assert "Invalid PFN. PFN must start with" in r.json()["detail"]
+
 
 def test_upload_oversized(normal_user_client: TestClient):
     data = secrets.token_bytes(512)