T05 · Unauthorized Access and Privilege Escalation
Error
- Location
taskqueue/app.py:34- Finding
Unauthenticated Access to Task Data and State-Changing Operations
- Content
View full analysis
Vulnerability Details
File Location:
taskqueue/app.py:34-90
Vulnerability Type: Missing authentication and authorization
Risk Level: HighVulnerable Code
python @app.post("/v1/tasks", response_model=TaskResponse) async def create(request: CreateTaskRequest) -> TaskResponse: task = create_task(request.title, request.payload, request.priority, request.tags) return _task_to_response(task) @app.get("/v1/tasks", response_model=TaskListResponse) async def list_all( status: str | None = Query(default=None), tag: str | None = Query(default=None), ) -> TaskListResponse: tasks = list_tasks(status, tag) return TaskListResponse(tasks=[_task_to_response(t) for t in tasks], total=len(tasks)) @app.get("/v1/tasks/{task_id}", response_model=TaskResponse) async def get_by_id(task_id: str) -> TaskResponse: task = get_task(task_id) if task is None: raise HTTPException(status_code=404, detail=f"Task not found: {task_id}") return _task_to_response(task) @app.post("/v1/claim", response_model=TaskResponse) async def claim() -> TaskResponse: task = claim_next() if task is None: raise HTTPException(status_code=404, detail="No pending tasks to claim") return _task_to_response(task) @app.post("/v1/tasks/{task_id}/complete", response_model=TaskResponse) async def complete(task_id: str, request: CompleteRequest) -> TaskResponse: task = complete_task(task_id, request.result) if task is None: raise HTTPException(status_code=404, detail=f"Task not found: {task_id}") return _task_to_response(task) @app.post("/v1/tasks/{task_id}/fail", response_model=TaskResponse) async def fail(task_id: str, request: FailRequest) -> TaskResponse: task = fail_task(task_id, request.error) if task is None: raise HTTPException(status_code=404, detail=f"Task not found: {task_id}") return _task_to_resp ...[truncated 1773 chars]- Remediation
View remediation
Remediation Suggestions
- Require authentication on every API endpoint, using a suitably protected API token, mTLS, OAuth2, or another mechanism appropriate to the deployment.
- Apply authorization checks based on queue, tenant, task ownership, and operation type.
- Do not treat knowledge of a task identifier as authorization.
- Return only the fields each caller is permitted to access.
- Bind the service to
127.0.0.1by default when it is intended only for local agent workflows. - Place remotely reachable deployments behind TLS and an authenticated reverse proxy.
- Add audit logging for task creation, reads, claims, completion, and failure operations.
- Document the service's trust boundary and secure deployment requirements.