T09 · Insecure Skill Coding Practices
Warning
- Location
- scripts/queue.py:121
- Finding
- Non-Persistent Task Transition Allows Repeated Autonomous Execution and Queue Corruption<![CDATA[ ## Vulnerability Details **File Location**: `scripts/queue.py:72-74, 121-140, 195-199`; `scripts/heartbeat.py:66-76, 89-91, 137-147` **Vulnerability Type**: Non-atomic state transition and inconsistent queue parsing **Risk Level**: Medium ### Vulnerable Code ```python # scripts/queue.py:72-74 elif line.startswith('- [ ]') and current_section: task = line.replace('- [ ]', '').strip() result[current_section].append(task) ``` ```python # scripts/queue.py:121-140 def take_task(self, queue: Dict) -> Optional[str]: """ 取最高优先级的任务 Args: queue: 当前队列 Returns: str: 任务内容 或 None """ if not queue.get('ready'): return None # 取第一个任务 task = queue['ready'].pop(0) # 移到进行中 queue['in_progress'].append(task) return task ``` ```python # scripts/queue.py:195-199 def take_task() -> Optional[str]: """取任务""" q = TaskQueue() queue = q.read_queue() return q.take_task(queue) ``` ```python # scripts/heartbeat.py:66-76 # 3. 从队列取任务 queue = self.queue.read_queue() task = self.queue.take_task(queue) if task: print(f"[{datetime.now().strftime('%H:%M:%S')}] 开始新任务: {task}") # 4. 标记状态 self.state.start_task(task, estimated_minutes=30) return task ``` ```python # scripts/heartbeat.py:89-91 # 更新队列 if success: complete_task(task) ``` ```python # scripts/heartbeat.py:137-147 if __name__ == '__main__': # 测试心跳 print("="*50) print("Abby Autonomy Heartbeat 测试") print("="*50) result = run_heartbeat() if result: print(f"\n开始执行任务: {result}") start_working(result) ``` ### Technical Analysis `TaskQueue.take_task()` removes the selected task from the in-memory `ready` list and appends it to the in-memory `in_progress` list, but it never calls `write_queue()` to persist this transition. The module-level `take_task()` and heartbeat workflow likewise return the selected task without saving the modified queue. When the task completes, `comp ...[truncated 2844 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. Persist task claiming before returning the selected task: ```python def take_task(self, queue: Dict) -> Optional[str]: if not queue.get("ready"): return None task = queue["ready"].pop(0) queue.setdefault("in_progress", []).append(task) if not self.write_queue(queue): # Do not execute a task whose claim was not persisted. return None return task ``` 2. Make the claim operation atomic: - Acquire an exclusive lock before reading and modifying the queue. - Read, validate, modify, and write while holding the same lock. - Write to a temporary file in the same directory, flush and `fsync()` it, and replace `QUEUE.md` with `os.replace()`. - Release the lock only after the replacement succeeds. 3. Correctly parse checked entries by section. Ready and Blocked may use `- [ ]`, while In Progress and Done use `- [x]`. Do not discard entries solely because they are checked. 4. Assign each task a stable unique identifier instead of using the full task description as its identity. Require completion and pause operations to reference that identifier. 5. Validate state transitions: - Permit completion only when the task exists in In Progress. - Prevent duplicate IDs across sections. - If persistence fails, fail closed and do not begin execution. - On task failure, atomically move the task to Ready or Blocked rather than only clearing runtime state. 6. Protect against concurrent heartbeat execution with a process-level singleton lock or transactional queue backend. 7. Add automated tests covering: - Persistence immediately after claiming a task. - Two simultaneous heartbeat processes. - Restart between claim and completion. - Parsing and preservation of checked In Progress and Done entries. - Write failures and malformed queue files. - Repeated heartbeat runs confirming that a completed task is not selected again. ]]>