Add id-slug lookup and restrict slugs endpoints to superadmins (webre…

…corder#1279) Fixes webrecorder#1278 - Adds `GET /orgs/slug-lookup` endpoint returning `{id: slug}` for all orgs - Restricts new endpoint and existing `GET /orgs/slugs` to superadmins
urbanlab · Oct 14, 2023 · c5ca250 · c5ca250
1 parent 8466caf
commit c5ca250
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 1 deletion.
diff --git a/backend/btrixcloud/orgs.py b/backend/btrixcloud/orgs.py
@@ -402,6 +402,13 @@ async def get_all_org_slugs(self):
         slugs = await self.orgs.distinct("slug", {})
         return {"slugs": slugs}
 
+    async def get_all_org_slugs_with_ids(self):
+        """Return dict with {id: slug} for all orgs."""
+        slug_id_map = {}
+        async for org in self.orgs.find({}):
+            slug_id_map[org["_id"]] = org["slug"]
+        return slug_id_map
+
 
 # ============================================================================
 # pylint: disable=too-many-statements
@@ -671,7 +678,15 @@ async def get_org_metrics(org: Organization = Depends(org_dep)):
         return await ops.get_org_metrics(org)
 
     @app.get("/orgs/slugs", tags=["organizations"])
-    async def get_all_org_slugs():
+    async def get_all_org_slugs(user: User = Depends(user_dep)):
+        if not user.is_superuser:
+            raise HTTPException(status_code=403, detail="Not Allowed")
         return await ops.get_all_org_slugs()
 
+    @app.get("/orgs/slug-lookup", tags=["organizations"])
+    async def get_all_org_slugs_with_ids(user: User = Depends(user_dep)):
+        if not user.is_superuser:
+            raise HTTPException(status_code=403, detail="Not Allowed")
+        return await ops.get_all_org_slugs_with_ids()
+
     return ops
diff --git a/backend/test/test_org.py b/backend/test/test_org.py
@@ -409,3 +409,29 @@ def test_get_org_slugs(admin_auth_headers):
     assert len(slugs) == org_count
     for slug in slugs:
         assert slug in org_slugs
+
+
+def test_get_org_slugs_non_superadmin(crawler_auth_headers):
+    r = requests.get(f"{API_PREFIX}/orgs/slugs", headers=crawler_auth_headers)
+    assert r.status_code == 403
+    assert r.json()["detail"] == "Not Allowed"
+
+
+def test_get_org_slug_lookup(admin_auth_headers):
+    # Build an expected return from /orgs list to compare against
+    expected_return = {}
+    r = requests.get(f"{API_PREFIX}/orgs", headers=admin_auth_headers)
+    assert r.status_code == 200
+    for org in r.json()["items"]:
+        expected_return[org["id"]] = org["slug"]
+
+    # Fetch data from /orgs/slug-lookup and verify data is correct
+    r = requests.get(f"{API_PREFIX}/orgs/slug-lookup", headers=admin_auth_headers)
+    assert r.status_code == 200
+    assert r.json() == expected_return
+
+
+def test_get_org_slug_lookup_non_superadmin(crawler_auth_headers):
+    r = requests.get(f"{API_PREFIX}/orgs/slug-lookup", headers=crawler_auth_headers)
+    assert r.status_code == 403
+    assert r.json()["detail"] == "Not Allowed"