"""Tests for the per-purpose ACL gate on ``POST /api/v1/uploads/presign-get``. Background: the presigned URL bypasses Helix auth once minted, so the endpoint must refuse any caller who isn't (a) the uploader or (b) a read-role holder for the purpose. Bare authentication is not enough — any logged-in user who observed an object_key (audit logs, shared screen, copy-paste) could otherwise mint a download URL. Two layers covered: * The shape parser (:func:`_parse_object_key`) extracts the uploader's person_id from the key path independent of any storage prefix. * The endpoint enforces the per-purpose policy: ``inventory_evidence`` allows uploader + asset-read roles; everything else returns 403. """ from __future__ import annotations import uuid from unittest.mock import AsyncMock, patch import pytest from httpx import AsyncClient from app.storage_router import _parse_object_key #: FIXED, not generated. This value is interpolated into ``@parametrize`` ids, #: which pytest computes at COLLECTION time — so a fresh uuid4 per process gives #: every xdist worker a different set of test ids, and xdist aborts the whole #: run with "Different tests were collected between gw0 and gw1". That takes the #: ENTIRE backend suite out of parallel execution, not just this file. Same #: defect as #3403 fixed in tests/test_cmdb; it is the parametrize argument that #: has to be stable, not the uuid. _UPLOADER = uuid.UUID("0f8f7e2a-6c1d-4a3b-9e55-2b7c1d0e4a91") @pytest.mark.parametrize( "key", ( f"inventory_evidence/2026/05/21/{_UPLOADER}/abc12345-photo.jpg", f"dev/inventory_evidence/2026/05/21/{_UPLOADER}/abc12345-photo.jpg", f"prod-v2/inventory_evidence/2026/05/21/{_UPLOADER}/x.jpg", ), ) def test_parse_object_key_accepts_registered_purpose_with_optional_prefix(key: str) -> None: assert _parse_object_key(key) == ("inventory_evidence", _UPLOADER) @pytest.mark.parametrize( "key", ( "some_other_purpose/2026/05/21/x/y.jpg", "dev/random/blob.txt", "", "inventory_evidence/2026/05", "inventory_evidence/2026/05/21", "inventory_evidence", "inventory_evidence/2026/05/21/not-a-uuid/x.jpg", f"inventory_evidence/2026/05/21/{_UPLOADER}/../secret.jpg", f"inventory_evidence/2026/05/21/{_UPLOADER}//x.jpg", f"inventory_evidence/2026/05/21/{_UPLOADER}/real.jpg/extra/segment", ), ) def test_parse_object_key_rejects_untrusted_shape(key: str) -> None: assert _parse_object_key(key) is None @pytest.mark.asyncio async def test_presign_get_refuses_other_employees_key( client: AsyncClient, employee_user, db_session, ): """Employee A cannot mint a presigned URL for an inventory_evidence key uploaded by Employee B — the IDOR class the ACL guard closes.""" _user, token = employee_user other_uploader = uuid.uuid4() # NOT the caller key = ( f"inventory_evidence/2026/05/21/{other_uploader}/" "abcd1234-photo.jpg" ) resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403, resp.text body = resp.json() # Standard error envelope — bare HTTPException(detail=...) is wrapped # by the StarletteHTTPException handler into ``errors[0].message`` # (see app/main.py), NOT a top-level ``detail`` field. err_msg = (body.get("errors") or [{}])[0].get("message", "").lower() assert "not authorised" in err_msg or "not authorized" in err_msg @pytest.mark.asyncio async def test_presign_get_allows_uploader_for_own_key( client: AsyncClient, employee_user, db_session, ): """The uploader may always presign-GET their own object.""" user, token = employee_user key = ( f"inventory_evidence/2026/05/21/{user.id}/" "abcd1234-photo.jpg" ) # The endpoint also resolves storage_config from the DB to mint a # URL — mock the boto3 layer so the test doesn't touch real S3. with patch( "app.storage_router.presign_get", new=AsyncMock(return_value="https://signed.example/url"), ): resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text assert resp.json()["data"]["url"] == "https://signed.example/url" @pytest.mark.asyncio async def test_presign_get_allows_asset_admin_for_any_inventory_key( client: AsyncClient, asset_admin_user, db_session, ): """Asset admins are the audience for the Resolve modal — they may view evidence photos uploaded by anyone.""" _admin, token = asset_admin_user other_uploader = uuid.uuid4() key = ( f"inventory_evidence/2026/05/21/{other_uploader}/" "abcd1234-photo.jpg" ) with patch( "app.storage_router.presign_get", new=AsyncMock(return_value="https://signed.example/url"), ): resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text @pytest.mark.asyncio async def test_presign_get_refuses_malformed_key( client: AsyncClient, employee_user, db_session, ): """A key that doesn't match the presign-put layout returns 403 rather than falling through to a presigned URL — closes the "guess a path to /etc/secret" style fishing attempt.""" _user, token = employee_user resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": "totally/random/path.txt"}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_presign_get_refuses_employee_for_disposal_evidence( client: AsyncClient, employee_user, db_session, ): """处置凭证 is an admin-desk artefact — a plain employee who observes a key (audit log, shared screen) must not be able to mint a URL.""" _user, token = employee_user key = ( f"asset_disposal_evidence/2026/09/02/{uuid.uuid4()}/" "abcd1234-recycle-note.jpg" ) resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403, resp.text @pytest.mark.asyncio async def test_presign_get_allows_asset_admin_for_disposal_evidence( client: AsyncClient, asset_admin_user, db_session, ): """Asset-read roles are the audience for the disposed asset's detail page — they may view evidence uploaded by any other admin.""" _admin, token = asset_admin_user key = ( f"dev/asset_disposal_evidence/2026/09/02/{uuid.uuid4()}/" "abcd1234-recycle-note.jpg" ) with patch( "app.storage_router.presign_get", new=AsyncMock(return_value="https://signed.example/url"), ): resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text @pytest.mark.asyncio async def test_every_allowed_purpose_has_an_explicit_read_policy( client: AsyncClient, employee_user, db_session, ): """A purpose registered for upload but with no ACL clause must 403, never fall through to "anyone authenticated". Pins the guard that forces the next contributor to make the policy decision explicitly — the alternative failure mode is silent and un-noticeable.""" from app.storage_router import ( _ALLOWED_PURPOSES, WORKFORCE_READABLE_PURPOSES, ) _user, token = employee_user unpoliced: list[str] = [] for purpose in sorted(_ALLOWED_PURPOSES): key = f"{purpose}/2026/09/02/{uuid.uuid4()}/abcd1234-x.jpg" with patch( "app.storage_router.presign_get", new=AsyncMock(return_value="https://signed.example/url"), ): resp = await client.post( "/api/v1/uploads/presign-get", json={"object_key": key}, headers={"Authorization": f"Bearer {token}"}, ) # 200 is legitimate only where the policy is deliberately "any # authenticated employee". That set is declared once, in # ``storage_router.WORKFORCE_READABLE_PURPOSES``, and read from # there rather than restated here — a duplicated literal is how # this guard would drift away from the ACL it guards. if ( resp.status_code == 200 and purpose not in WORKFORCE_READABLE_PURPOSES ): unpoliced.append(purpose) assert resp.status_code in (200, 403), (purpose, resp.text) assert unpoliced == [], ( f"purposes readable by any employee without an explicit policy: {unpoliced}" )