proyectos/claudia-docs-api
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: API token project ownership check

Browse Source 
- API tokens now verify project belongs to token owner before access
- Researcher tokens only access research/general docs in owner's projects
- Developer tokens only access development/general docs in owner's projects
- Viewer tokens have read-only access to all doc types in owner's projects
- Add test for cross-user project access prevention
This commit is contained in:
Motoko
2026-03-31 01:56:26 +00:00
parent 204badb964
commit ae2409ef46
2 changed files with 247 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
app/routers/documents.py
View File

@@ -230,21 +230,20 @@ async def list_documents(
):
agent, api_role = await get_current_agent_or_api_token(request, db)
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Project not found")
else:
# API tokens don't have project-level access control here
# Access is controlled at document level via agent_type
pass
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="Project not found")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
result = await db.execute(
select(Document).where(
@@ -283,17 +282,20 @@ async def create_document(
):
agent, api_role = await get_current_agent_or_api_token(request, db)
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Project not found")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="Project not found")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Determine agent_type for the document
doc_agent_type = payload.agent_type or "general"
@@ -353,21 +355,24 @@ async def get_document(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Document not found")
else:
# API tokens check role-based access
if not _can_access_document(api_role, doc.agent_type, require_write=False):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="Document not found")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=False):
raise HTTPException(status_code=403, detail="Forbidden")
return await document_to_response(db, doc)
@@ -391,21 +396,24 @@ async def update_document(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
if payload.title is not None:
doc.title = payload.title
@@ -449,21 +457,24 @@ async def delete_document(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
doc.is_deleted = True
doc.deleted_at = datetime.utcnow()
@@ -496,21 +507,24 @@ async def update_document_content(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
# Determine actual format based on content type (backward compatibility)
# If content is a string, treat as markdown regardless of format field
@@ -566,21 +580,24 @@ async def restore_document(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
doc.is_deleted = False
doc.deleted_at = None
@@ -608,21 +625,24 @@ async def assign_tags(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
for tag_id in payload.tag_ids:
tag_result = await db.execute(
@@ -668,21 +688,24 @@ async def remove_tag(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based write access
if not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based write access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=True):
raise HTTPException(status_code=403, detail="Forbidden")
await db.execute(
delete(DocumentTag).where(
@@ -717,21 +740,24 @@ async def _get_doc_with_access(
if not doc:
raise HTTPException(status_code=404, detail="Document not found")
# JWT tokens check project ownership
if api_role is None:
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.agent_id == agent.id,
Project.is_deleted == False,
)
# Check project ownership
proj_result = await db.execute(
select(Project).where(
Project.id == doc.project_id,
Project.is_deleted == False,
)
if not proj_result.scalar_one_or_none():
raise HTTPException(status_code=403, detail="Forbidden")
else:
# API tokens check role-based access
if not _can_access_document(api_role, doc.agent_type, require_write=require_write):
raise HTTPException(status_code=403, detail="Forbidden")
)
project = proj_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=403, detail="Forbidden")
# For API tokens, verify project belongs to the token owner
if api_role is not None and project.agent_id != agent.id:
raise HTTPException(status_code=403, detail="Forbidden")
# Check role-based access for API tokens
if api_role is not None and not _can_access_document(api_role, doc.agent_type, require_write=require_write):
raise HTTPException(status_code=403, detail="Forbidden")
return doc, api_role