[verified] fix files list owner scoping
api-ci-deploy / test-build-deploy (push) Has been cancelled

This commit is contained in:
2026-06-08 02:08:38 +01:00
parent a6753d092f
commit e98fed661f
3 changed files with 145 additions and 0 deletions
+21
View File
@@ -36,6 +36,24 @@ DOCLING_NOOCR_TIMEOUT = int(os.getenv('DOCLING_NOOCR_TIMEOUT', '3600')) # 1 hou
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
def _user_id_from_payload(payload: Dict[str, Any]) -> str:
user_id = payload.get('sub') or payload.get('user_id')
if not user_id:
raise HTTPException(status_code=401, detail="Invalid token payload")
return user_id
def _cabinet_visible_to_user(client: SupabaseServiceRoleClient, cabinet_id: str, user_id: str) -> bool:
"""Require cabinet ownership before service-role reads file metadata."""
owned = (
client.supabase.table('file_cabinets')
.select('id')
.eq('id', cabinet_id)
.eq('user_id', user_id)
.limit(1)
.execute()
)
return bool(owned.data)
def _safe_filename(name: str) -> str:
base = os.path.basename(name or 'file')
return re.sub(r"[^A-Za-z0-9._-]+", "_", base)
@@ -117,7 +135,10 @@ async def upload_file(
@router.get("/files")
def list_files(cabinet_id: str, payload: Dict[str, Any] = Depends(auth)):
user_id = _user_id_from_payload(payload)
client = SupabaseServiceRoleClient()
if not _cabinet_visible_to_user(client, cabinet_id, user_id):
return []
res = client.supabase.table('files').select('*').eq('cabinet_id', cabinet_id).execute()
return res.data
@@ -26,6 +26,24 @@ auth = SupabaseBearer()
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
def _user_id_from_payload(payload: Dict[str, Any]) -> str:
user_id = payload.get('sub') or payload.get('user_id')
if not user_id:
raise HTTPException(status_code=401, detail="Invalid token payload")
return user_id
def _cabinet_visible_to_user(client: SupabaseServiceRoleClient, cabinet_id: str, user_id: str) -> bool:
"""Require cabinet ownership before service-role reads file metadata."""
owned = (
client.supabase.table('file_cabinets')
.select('id')
.eq('id', cabinet_id)
.eq('user_id', user_id)
.limit(1)
.execute()
)
return bool(owned.data)
def _choose_bucket(scope: str, user_id: str, school_id: Optional[str]) -> str:
"""Choose appropriate bucket based on scope - matches old system logic."""
scope = (scope or 'teacher').lower()
@@ -134,7 +152,10 @@ async def upload_file(
@router.get("/files")
def list_files(cabinet_id: str, payload: Dict[str, Any] = Depends(auth)):
"""List files in a cabinet."""
user_id = _user_id_from_payload(payload)
client = SupabaseServiceRoleClient()
if not _cabinet_visible_to_user(client, cabinet_id, user_id):
return []
res = client.supabase.table('files').select('*').eq('cabinet_id', cabinet_id).execute()
return res.data