CVE-2023-39508: Remove Run task action from UI by bbovenzi · Pull Request #29706 · apache/airflow

def run(self, session=None):

“""Runs Task Instance.""”

dag_id = request.form.get(“dag_id”)

task_id = request.form.get(“task_id”)

dag_run_id = request.form.get(“dag_run_id”)

map_index = request.args.get("map_index", -1, type=int)

origin = get_safe_url(request.form.get(“origin”))

dag = get_airflow_app().dag_bag.get_dag(dag_id)

if not dag:

return redirect_or_json(origin, "DAG not found", "error", 404)

task = dag.get_task(task_id)

ignore_all_deps = request.form.get(“ignore_all_deps”) == “true”

ignore_task_deps = request.form.get(“ignore_task_deps”) == “true”

ignore_ti_state = request.form.get(“ignore_ti_state”) == “true”

executor = ExecutorLoader.get_default_executor()

if not executor.supports_ad_hoc_ti_run:

msg = f"{executor.__class__.__name__} does not support ad hoc task runs"

return redirect_or_json(origin, msg, "error", 400)

dag_run = dag.get_dagrun(run_id=dag_run_id, session=session)

if not dag_run:

return redirect_or_json(origin, "DAG run not found", "error", 404)

ti = dag_run.get_task_instance(task_id=task.task_id, map_index=map_index, session=session)

if not ti:

msg = “Could not queue task instance for execution, task instance is missing”

return redirect_or_json(origin, msg, "error", 400)

ti.refresh_from_task(task)

# Make sure the task instance can be run

dep_context = DepContext(

deps=RUNNING_DEPS,

ignore_all_deps=ignore_all_deps,

ignore_task_deps=ignore_task_deps,

ignore_ti_state=ignore_ti_state,

)

failed_deps = list(ti.get_failed_dep_statuses(dep_context=dep_context))

if failed_deps:

failed_deps_str = ", “.join(f"{dep.dep_name}: {dep.reason}” for dep in failed_deps)

msg = f"Could not queue task instance for execution, dependencies not met: {failed_deps_str}"

return redirect_or_json(origin, msg, "error", 400)

executor.job_id = None

executor.start()

executor.queue_task_instance(

ti,

ignore_all_deps=ignore_all_deps,

ignore_task_deps=ignore_task_deps,

ignore_ti_state=ignore_ti_state,

)

executor.heartbeat()

ti.queued_dttm = timezone.utcnow()

session.merge(ti)

msg = f"Sent {ti} to the message queue, it should start any moment now."

return redirect_or_json(origin, msg)