T09 · Insecure Skill Coding Practices
Error
- Location
- scripts/task_queue.py:142
- Finding
- Path Traversal Through Unsanitized Task Identifiers## Vulnerability Details **File Location**: `scripts/task_queue.py:142-160` and `scripts/task_queue.py:164-194` **Vulnerability Type**: Path traversal and arbitrary JSON-file overwrite **Risk Level**: High The same vulnerable implementation is duplicated in `dist/scripts/task_queue.py:142-160` and `dist/scripts/task_queue.py:164-194`. The attacker-controlled identifier enters these methods through `scripts/task_manager.py:31-38` and `scripts/task_manager.py:59-69`. ### Vulnerable Code ```python def complete_task(self, task_id: str, result_data: Dict[str, Any] = None) -> bool: """Mark a task as completed and move it to completed directory.""" for i, task in enumerate(self.task_queue): if task['id'] == task_id and task['status'] == 'running': task['status'] = 'completed' task['completed_at'] = datetime.now().isoformat() if result_data: task['result'] = result_data # Save to completed directory completed_file = self.completed_dir / f"{task_id}.json" try: with open(completed_file, 'w', encoding='utf-8') as f: json.dump(task, f, indent=2, ensure_ascii=False) except IOError as e: print(f"Warning: Failed to save completed task {task_id}: {e}") # Remove from queue self.task_queue.pop(i) self._save_queue() return True return False ``` ```python def fail_task(self, task_id: str, error_message: str = None) -> bool: """Mark a task as failed and handle retries.""" for task in self.task_queue: if task['id'] == task_id: task['retry_count'] += 1 if task['retry_count'] < task['max_retries']: task['status'] = 'queued' task['started_at'] = None self._save ...[truncated 3198 chars]
- Remediation
- ## Remediation Suggestions 1. Restrict task identifiers to a conservative allowlist before storing or using them: ```python import re TASK_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,128}$") def validate_task_id(task_id: str) -> str: if not TASK_ID_PATTERN.fullmatch(task_id): raise ValueError("Invalid task identifier") return task_id ``` 2. Reject identifiers containing `/`, `\`, `..`, null bytes, drive prefixes, or absolute-path syntax. 3. Resolve every destination and verify containment before opening it: ```python base = self.completed_dir.resolve() destination = (base / f"{task_id}.json").resolve() if destination.parent != base: raise ValueError("Task path escapes the completed-task directory") ``` 4. Apply equivalent containment checks to `failed_dir`. 5. Use generated internal identifiers, such as UUIDs, for filenames instead of user-provided display identifiers. 6. Write atomically through a securely created temporary file in the same directory, then use `os.replace()`. 7. Add tests for absolute paths, `../` traversal, Windows drive paths, alternate separators, Unicode separator edge cases, and symlink-based escapes.