fix: enforce per-table access control on catalog profile endpoints

Add can_access_table check to GET /api/catalog/profile/{table_name} and
POST /api/catalog/profile/{table_name}/refresh, returning 403 for
unauthorized tables. Update test_api_complete to cover new 403 behaviour
and fix the existing 404 test to use admin token.
This commit is contained in:
ZdenekSrotyr 2026-04-09 16:30:24 +02:00
parent ad6b3a96e4
commit 449053bf8a
2 changed files with 26 additions and 1 deletions

View file

@ -22,6 +22,9 @@ async def get_table_profile(
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
):
"""Get profiler data for a specific table."""
# Check table-level access
if not can_access_table(user, table_name, conn):
raise HTTPException(status_code=403, detail=f"Access denied to table '{table_name}'")
repo = ProfileRepository(conn)
profile = repo.get(table_name)
if not profile:
@ -100,6 +103,9 @@ async def refresh_profile(
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
):
"""Re-generate profile for a table on demand."""
# Check table-level access
if not can_access_table(user, table_name, conn):
raise HTTPException(status_code=403, detail=f"Access denied to table '{table_name}'")
from src.profiler import profile_table, TableInfo
data_dir = _get_data_dir()

View file

@ -53,9 +53,28 @@ class TestCatalog:
assert resp.status_code == 200
def test_catalog_profile_not_found(self, client):
resp = client["client"].get("/api/catalog/profile/nonexistent", headers=_h(client["analyst"]))
# Admin can see 404 for truly missing tables (bypasses access control)
resp = client["client"].get("/api/catalog/profile/nonexistent", headers=_h(client["admin"]))
assert resp.status_code == 404
def test_catalog_profile_access_denied_for_analyst(self, client):
# Non-registered (non-public) table returns 403 for analyst
resp = client["client"].get("/api/catalog/profile/private_table", headers=_h(client["analyst"]))
assert resp.status_code == 403
def test_catalog_profile_refresh_access_denied_for_analyst(self, client):
# Refresh endpoint also enforces access control
resp = client["client"].post("/api/catalog/profile/private_table/refresh", headers=_h(client["analyst"]))
assert resp.status_code == 403
def test_catalog_profile_public_table_accessible_to_analyst(self, client):
# Register a public table — analyst can access its profile (404 since no profile data)
client["client"].post("/api/admin/register-table",
json={"name": "public_table", "source_type": "keboola"},
headers=_h(client["admin"]))
resp = client["client"].get("/api/catalog/profile/public_table", headers=_h(client["analyst"]))
assert resp.status_code == 404 # access granted, but no profile data yet
# ---- Telegram ----