From c5ca250f37eb304611b928a17d1dbd2272ecad41 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Fri, 13 Oct 2023 20:02:19 -0400 Subject: [PATCH] Add id-slug lookup and restrict slugs endpoints to superadmins (#1279) Fixes #1278 - Adds `GET /orgs/slug-lookup` endpoint returning `{id: slug}` for all orgs - Restricts new endpoint and existing `GET /orgs/slugs` to superadmins --- backend/btrixcloud/orgs.py | 17 ++++++++++++++++- backend/test/test_org.py | 26 ++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/backend/btrixcloud/orgs.py b/backend/btrixcloud/orgs.py index c2a339bb..028029c3 100644 --- a/backend/btrixcloud/orgs.py +++ b/backend/btrixcloud/orgs.py @@ -402,6 +402,13 @@ class OrgOps: 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 @@ def init_orgs_api(app, mdb, user_manager, invites, user_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 index f9886ada..fa6ce9cb 100644 --- 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"