T09 · Insecure Skill Coding Practices
- Location
scripts/task_queue.py:142- Finding
Path Traversal Through Unsanitized Task Identifiers
- Content
View full analysis
Vulnerability Details
File Location:
scripts/task_queue.py:142-160andscripts/task_queue.py:164-194
Vulnerability Type: Path traversal and arbitrary JSON-file overwrite
Risk Level: HighThe same vulnerable implementation is duplicated in
dist/scripts/task_queue.py:142-160anddist/scripts/task_queue.py:164-194. The attacker-controlled identifier enters these methods throughscripts/task_manager.py:31-38andscripts/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 Falsepython 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
View remediation
Remediation Suggestions
-
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 -
Reject identifiers containing
/,\,.., null bytes, drive prefixes, or absolute-path syntax. -
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") -
Apply equivalent containment checks to
failed_dir. -
Use generated internal identifiers, such as UUIDs, for filenames instead of user-provided display identifiers.
-
Write atomically through a securely created temporary file in the same directory, then use
os.replace(). -
Add tests for absolute paths,
../traversal, Windows drive paths, alternate separators, Unicode separator edge cases, and symlink-based escapes.
-