Sprint 4 PR 4.5c — close /calendar/org/export auth gap

api/routers/calendar.py

@@ -10,6 +10,7 @@
     check_admin_permission,
     get_current_admin_user,
     get_current_user,
+    verify_org_member,
 )
 from api.models import Assignment, AuditAction, Event, Organization, Person, Resource, Solution
 from api.utils.audit_logger import log_audit_event
@@ -299,14 +300,16 @@ def admin_reset_calendar_token(
 @router.get("/org/export")
 def export_organization_events(
     org_id: str,
-    person_id: str,  # For auth - must be admin
     include_assignments: bool = True,
+    current_admin: Person = Depends(get_current_admin_user),
     db: Session = Depends(get_db),
 ):
     """
     Export all organization events as ICS file (admin only).
 
-    This endpoint is for administrators to export all events in the organization.
+    Caller must be authenticated and an admin in `org_id`. The legacy
+    `person_id` query param used as an auth proxy has been removed — the
+    caller is now identified solely by their JWT.
     """
     # Verify organization exists
     org = db.query(Organization).filter(Organization.id == org_id).first()
@@ -316,20 +319,7 @@ def export_organization_events(
             detail=f"Organization '{org_id}' not found",
         )
 
-    # Verify person exists and is admin
-    person = db.query(Person).filter(Person.id == person_id).first()
-    if not person or person.org_id != org_id:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Access denied. Admin privileges required.",
-        )
-
-    # Check if person is admin
-    if not person.roles or "admin" not in person.roles:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Access denied. Admin privileges required.",
-        )
+    verify_org_member(current_admin, org_id)
 
     # Get all events for this organization
     events = db.query(Event).filter(Event.org_id == org_id).all()

tests/api/test_calendar_auth.py

@@ -300,3 +300,80 @@ def test_admin_same_org_reaches_body(self, client, db):
         resp = client.get(f"/api/v1/calendar/export?person_id={other_id}", headers=admin_hdrs)
         assert resp.status_code == 404, resp.text
         assert "No assignments found" in resp.json()["detail"]
+
+
+@pytest.mark.no_mock_auth
+class TestOrgExportAuth:
+    """`/calendar/org/export` previously accepted a spoofable `person_id`
+    query param as the auth proxy: anyone who could guess an admin's id could
+    download the entire org's events. Auth is now JWT-only via
+    `Depends(get_current_admin_user)` + same-org check."""
+
+    def test_no_auth_rejected(self, client, db):
+        org_id = "cal-oe-noauth"
+        seed_org(client, org_id)
+
+        resp = client.get(f"/api/v1/calendar/org/export?org_id={org_id}")
+        assert resp.status_code in (401, 403)
+
+    def test_volunteer_in_same_org_rejected(self, client, db):
+        org_id = "cal-oe-vol"
+        seed_org(client, org_id)
+        _admin_for(client, org_id, "oe-vol-admin")
+        seed_user(
+            client,
+            org_id,
+            email="oev@o.org",
+            name="OEV",
+            password="OEVPass1!",
+            roles=["volunteer"],
+        )
+        v_hdrs = auth_headers(client, email="oev@o.org", password="OEVPass1!")
+
+        resp = client.get(f"/api/v1/calendar/org/export?org_id={org_id}", headers=v_hdrs)
+        assert resp.status_code == 403
+
+    def test_admin_cross_org_rejected(self, client, db):
+        seed_org(client, "cal-oe-cr-a")
+        seed_org(client, "cal-oe-cr-b")
+        a_hdrs = _admin_for(client, "cal-oe-cr-a", "oe-cr-a")
+        _admin_for(client, "cal-oe-cr-b", "oe-cr-b")
+
+        # Admin of org A asks to export org B's events.
+        resp = client.get("/api/v1/calendar/org/export?org_id=cal-oe-cr-b", headers=a_hdrs)
+        assert resp.status_code == 403
+
+    def test_admin_same_org_no_events_returns_404(self, client, db):
+        org_id = "cal-oe-empty"
+        seed_org(client, org_id)
+        a_hdrs = _admin_for(client, org_id, "oe-empty")
+
+        resp = client.get(f"/api/v1/calendar/org/export?org_id={org_id}", headers=a_hdrs)
+        # Same-org admin gets through auth; the empty-events 404 from the
+        # existing body logic is the next gate.
+        assert resp.status_code == 404
+        assert "No events" in resp.json()["detail"]
+
+    def test_spoofable_person_id_param_no_longer_grants_access(self, client, db):
+        """A volunteer cannot escalate by passing an admin's `person_id` in
+        the query string — the legacy spoof vector is gone."""
+        org_id = "cal-oe-spoof"
+        seed_org(client, org_id)
+        admin_hdrs = _admin_for(client, org_id, "oe-spoof-a")
+        admin_id = _person_id_for_email(client, admin_hdrs, "admin-oe-spoof-a@o.org")
+        seed_user(
+            client,
+            org_id,
+            email="oes@o.org",
+            name="OES",
+            password="OESPass1!",
+            roles=["volunteer"],
+        )
+        v_hdrs = auth_headers(client, email="oes@o.org", password="OESPass1!")
+
+        # Volunteer attempts to pass the admin's id as the legacy auth proxy.
+        resp = client.get(
+            f"/api/v1/calendar/org/export?org_id={org_id}&person_id={admin_id}",
+            headers=v_hdrs,
+        )
+        assert resp.status_code == 403

tests/conftest.py

@@ -80,6 +80,7 @@ def override_verify_org_member(person: Person, org_id: str) -> None:
     app.dependency_overrides[get_current_user] = override_get_user
 
     import api.dependencies
+    import api.routers.calendar
     import api.routers.events
     import api.routers.people
     import api.routers.teams
@@ -91,6 +92,8 @@ def override_verify_org_member(person: Person, org_id: str) -> None:
         api.routers.events.verify_org_member = override_verify_org_member
     if hasattr(api.routers.teams, "verify_org_member"):
         api.routers.teams.verify_org_member = override_verify_org_member
+    if hasattr(api.routers.calendar, "verify_org_member"):
+        api.routers.calendar.verify_org_member = override_verify_org_member
 
     yield
 
@@ -102,6 +105,8 @@ def override_verify_org_member(person: Person, org_id: str) -> None:
         api.routers.events.verify_org_member = original_verify
     if hasattr(api.routers.teams, "verify_org_member"):
         api.routers.teams.verify_org_member = original_verify
+    if hasattr(api.routers.calendar, "verify_org_member"):
+        api.routers.calendar.verify_org_member = original_verify
 
 
 @pytest.fixture(scope="session", autouse=True)

tests/contract/openapi.snapshot.json

@@ -7856,7 +7856,7 @@
     },
     "/api/v1/calendar/org/export": {
       "get": {
-        "description": "Export all organization events as ICS file (admin only).\n\nThis endpoint is for administrators to export all events in the organization.",
+        "description": "Export all organization events as ICS file (admin only).\n\nCaller must be authenticated and an admin in `org_id`. The legacy\n`person_id` query param used as an auth proxy has been removed \u2014 the\ncaller is now identified solely by their JWT.",
         "operationId": "exportOrganizationEvents",
         "parameters": [
           {
@@ -7868,15 +7868,6 @@
               "type": "string"
             }
           },
-          {
-            "in": "query",
-            "name": "person_id",
-            "required": true,
-            "schema": {
-              "title": "Person Id",
-              "type": "string"
-            }
-          },
           {
             "in": "query",
             "name": "include_assignments",
@@ -7908,6 +7899,11 @@
             "description": "Validation Error"
           }
         },
+        "security": [
+          {
+            "HTTPBearer": []
+          }
+        ],
         "summary": "Export Organization Events",
         "tags": [
           "calendar"

tests/unit/test_calendar.py

@@ -302,10 +302,14 @@ def setup_test_data(self, client):
         )
 
     def test_org_export_as_admin(self, client):
-        """Test organization export as admin."""
-        response = client.get(
-            f"{API_BASE}/calendar/org/export?org_id=org_export_test&person_id=admin_person_1"
-        )
+        """Test organization export as admin.
+
+        Auth is the mocked admin (org = `test_org`). The seeded events
+        under `org_export_test` are reachable only because the unit-tier
+        `verify_org_member` override is a no-op. The same-org check is
+        exercised by `tests/api/test_calendar_auth.py::TestOrgExportAuth`.
+        """
+        response = client.get(f"{API_BASE}/calendar/org/export?org_id=org_export_test")
 
         assert response.status_code == 200
         assert "text/calendar" in response.headers["content-type"]
@@ -314,20 +318,9 @@ def test_org_export_as_admin(self, client):
         assert "BEGIN:VCALENDAR" in content
         assert "Team Meeting" in content
 
-    def test_org_export_as_volunteer_denied(self, client):
-        """Test organization export as volunteer is denied."""
-        response = client.get(
-            f"{API_BASE}/calendar/org/export?org_id=org_export_test&person_id=volunteer_person_1"
-        )
-
-        assert response.status_code == 403
-        assert "Admin privileges required" in response.json()["detail"]
-
     def test_org_export_nonexistent_org(self, client):
         """Test organization export for non-existent org."""
-        response = client.get(
-            f"{API_BASE}/calendar/org/export?org_id=nonexistent_org&person_id=admin_person_1"
-        )
+        response = client.get(f"{API_BASE}/calendar/org/export?org_id=nonexistent_org")
 
         assert response.status_code == 404
         assert "not found" in response.json()["detail"]
@@ -337,19 +330,7 @@ def test_org_export_no_events(self, client):
         # Create new org with no events
         client.post(f"{API_BASE}/organizations/", json={"id": "empty_org", "name": "Empty Org"})
 
-        client.post(
-            f"{API_BASE}/people/",
-            json={
-                "id": "empty_org_admin",
-                "org_id": "empty_org",
-                "name": "Empty Admin",
-                "roles": ["admin"],
-            },
-        )
-
-        response = client.get(
-            f"{API_BASE}/calendar/org/export?org_id=empty_org&person_id=empty_org_admin"
-        )
+        response = client.get(f"{API_BASE}/calendar/org/export?org_id=empty_org")
 
         assert response.status_code == 404
         assert "No events found" in response.json()["detail"]