T09 · Insecure Skill Coding Practices
Error
- Location
scripts/pipeline_orchestrator.py:184- Finding
Executable Python Code Injection in Generated Pipeline Artifacts
- Content
View full analysis
str: """Generate default_args dictionary.""" return f''' default_args = {{ 'owner': '{config.owner}', 'depends_on_past': False, 'email_on_failure': True, 'email_on_retry': False, 'retries': {config.default_retries}, 'retry_delay': timedelta(minutes=5), }} ''' ``` The same issue affects generated Bash tasks: ```python def _generate_bash_task(self, task: TaskConfig) -> str: """Generate BashOperator task.""" command = task.params.get('command', 'echo "Hello World"') return f''' {task.task_id} = BashOperator( task_id='{task.task_id}', bash_command='{command}', retries={task.retries}, retry_delay=timedelta(minutes={task.retry_delay_minutes}), execution_timeout=timedelta(minutes={task.timeout_minutes}), ) ''' ``` Task names and dependencies are also inserted as Python expressions: ```python def _generate_dependencies(self, config: PipelineConfig) -> str: """Generate task dependencies.""" deps_code = "\n # Task dependencies\n" for task in config.tasks: if task.dependencies: for dep in task.dependencies: deps_code += f" {dep} >> {task.task_id}\n" return deps_code ``` Configuration is loaded and passed to the generators without contextual escaping or strict schema validation. Generated output is written even when validation reports problems: ```python if args.config: with open(args.config) as f: if HAS_YAML: config_data = yaml.safe_load(f) else: config_data ...[truncated 3353 chars]- Remediation
View remediation