T05 · Unauthorized Access and Privilege Escalation
Error
- Location
sql-memory/sql_memory.py:286- Finding
Task Queue Operations Lack Ownership and State Authorization
- Content
View full analysis
bool: """Mark a task as processing.""" return self._db.execute(""" UPDATE memory.TaskQueue SET status='processing', started_at=%s WHERE id=%s """, (datetime.now(timezone.utc), int(task_id))) def complete_task(self, task_id: Any, result: str = '') -> bool: """Mark a task as completed.""" ok = self._db.execute(""" UPDATE memory.TaskQueue SET status='completed', completed_at=%s, error_log=%s WHERE id=%s """, (datetime.now(timezone.utc), result[:500], int(task_id))) _log.info(f"complete_task({task_id})") return ok def fail_task(self, task_id: Any, error: str, retry_count: int = 0, max_retries: int = 3) -> bool: """Fail or re-queue a task based on retry count.""" new_status = 'pending' if retry_count < max_retries else 'failed' ok = self._db.execute(""" UPDATE memory.TaskQueue SET status=%s, retry_count=retry_count+1, error_log=%s WHERE id=%s """, (new_status, error[:800], int(task_id))) _log.info(f"fail_task({task_id}) → {new_status}") return ok ``` ### Technical Analysis The task mutation methods authorize operations solely by task ID. They do not verify that the caller is the task's assigned agent, that the task belongs to an authorized tenant or session, or that its current state permits the requested transition. For example, `claim_task()` does not require `status='pending'`. Two agents can therefore claim the same task, and an agent can reclaim an already completed or processing task. Similarly, `complete_task()` and `fail_task()` can modify any task whose ID is known or guessed. The methods also return only the connector's Boolean result and do not ver ...[truncated 1413 chars]- Remediation
View remediation