Shulou(Shulou.com)06/01 Report--

In this section, the author will implement an own Pocsuite framework according to the PoC framework structure and engineering practice. In order to get off to a good start, we first choose a good-sounding name. If we want to be powerful and domineering, we can take ancient artifacts, such as Xuanyuan, Xia, Yu, Chixiao, and so on. If we love the game, we can also have the Mountain King (BladeMaster) and the moon goddess (Priess Of the moon). Since the author is rather lazy, let's take a simple name: AirPoc, and call it "air cannon" in Chinese.

Here, please give full play to your imagination and fantasize about its function. You have to remember that there is no function that we can't achieve. If so, just kill the product manager.

The detected website owners are required to pay "air coins" as compensation.

Although it is only a temporary fantasy, but the little brother of the product is also a little excited to sort out the functions we need.

Easy to use, do not have too many commands, can be used across platforms

There are many people and great power, so that more people can participate.

It can be built into other products with simple operation.

Verification speed and verification accuracy are extremely high!

I don't know what's good, but you can get something out of it when you run!

Of course, the little brother of the product may be afraid of being beaten and did not add the concept of distributed, blockchain.

Specific details

Below by the author to specifically achieve by the author part-time product manager randomly think of (digging holes) things. We analyze the problems one by one and give the final solution.

Speaking of easy to use, we willfully choose to use Python, do not believe you look at the hair of the father of Python. After installing Python, you can use more than one piece of code, but in order to be simple and native enough, we decided to use Python's third-party packages as little as possible. At present, the latest version of Python is 3.7. let's take this as an example.

Sqlmap

Routersploit

Contributing a piece of code to it is probably the most important thing security researchers want to do.

So the author has an idea that AirPoc's PoC repository can be open source to GitHub, and can call the above PoC online, so that it will not be bothered by PoC updates.

Built into other products is also easier, if it is Python software, you can directly call AirPoc as a package, if other software, AirPoc can open a RPC interface to provide use, if you do not want the Python environment, you can also package through tools such as pyinstaller, our design principle is to try not to rely on other third-party libraries, so we will avoid a lot of strange problems.

In order to achieve high verification speed and accuracy, we need to do a good job of multithreading or co-programming concurrency model, which we will describe in detail later.

Finally, "I don't know what's good, but you can make something when you run!" If we have done all the above, this should come naturally.

The framework of AirPoc

Before completing this "grand plan", we also need to design the overall code framework. As a code cleanliness addict, a good code structure is the first step of the long March. We establish the following directory structure, env is a virtual environment, set up two directories lib, pocs, lib to store the relevant core files, pocs to store poc files, and a file main.py as the initial entry.

Just like building a building needs to lay a good foundation, and then complete the foundation frame, we can not write the specific function, but understand the meaning of the function as the "foundation". As follows, an initial framework is completed with the following code in the main.py file.

Import osimport timedef banner (): msg =''_ / | _\ | _\ / _ | / / | | _ | | | _ | / / | | | _ / | _ / | | / | |\ | _ | | _ |\ | _ | _ |\ _ | _ |\ _ /\ _ | {} '.format (version) print (msg) def init (config: dict): print ("[ *] target: {} ".format (config [" url "]) def end (): print (" [*] shutting down at {0} ".format (time.strftime ("% X ")) def start (): passdef main (): banner () config = {" url ":" https://www.seebug.org/"} init (config) start () end () if _ name__ ='_ _ main__': version = "v0.00000001" main ()

But, as you can see, the version number is about the same as the number of my bitcoin wallet, and we need to add some ingredients to it.

⚪ singleton mode

In the initialization project of our software, we need to get a lot of environment-related information. For example, what is the current execution path? Where is the poc directory? Which path we output the result file to, and so on.

One specific thing they have in common is that they only need to be loaded once and can be used directly in later use. This pattern has a separate term in software design patterns, "singleton pattern".

Fortunately, python's module is a natural singleton mode, because the module generates a .pyc file the first time it is imported, and loads the .pyc file directly on the second import without executing the module code again. Therefore, we only need to define the relevant functions and data in a module to get a singleton object.

We create a new data.py in the lib directory to store this information. At the same time, put the version information here.

Import osPATHS_ROOT = os.path.join (os.path.dirname (os.path.realpath (_ _ file__)), ".. /") PATHS_POCS = os.path.join (PATHS_ROOT, "pocs") PATHS_OUTPUT = os.path.join (PATHS_ROOT, "output") VERSION = "v0.0000001"

In order to better represent these constants, we use the specifications in the PEP8 standard to uniformly agree to use capitals and underscores to represent constants. To illustrate the difference, we symbolically subtract a 0 from VERSION to show that our bitcoin has increased tenfold.

⚪ dynamic loading

After solving our related environmental problems, we are looking at how to load the module dynamically. As we said in the details, we expect PoC to be loaded from local or remote sites such as GitHub.

Here the score is divided into two cases, if the dynamically loaded module is loaded through the file path, you can directly use _ _ import__ () to load, but if you want to load remotely, it may be a little more complicated. According to the relevant documents of python, we have to implement the "finder" and "loader" https://docs.python.org/zh-cn/3/reference/import.html.

Of course, you can also save it from remote to local and load it according to the local load mode. But Pocsuite already has the complete loader code, so we can use it directly.

Create a new lib/loader.py file

Import hashlibimport importlibfrom importlib.abc import Loaderdef get_md5 (value): if isinstance (value, str): value = value.encode (encoding='UTF-8') return hashlib.md5 (value). Hexdigest () def load_string_to_module (code_string) Fullname=None): try: module_name = 'pocs_ {0}' .format (get_md5 (code_string)) if fullname is None else fullname file_path = 'airpoc:// {0}' .format (module_name) poc_loader = PocLoader (module_name, file_path) poc_loader.set_data (code_string) spec = importlib.util.spec_from_file_location (module_name, file_path) Loader=poc_loader) mod = importlib.util.module_from_spec (spec) spec.loader.exec_module (mod) return mod except ImportError: error_msg = "load module'{0} 'failed!" .format (fullname) print (error_msg) raiseclass PocLoader (Loader): def _ init__ (self, fullname) Path): self.fullname = fullname self.path = path self.data = None def set_data (self, data): self.data = data def get_filename (self, fullname): return self.path def get_data (self, filename): if filename.startswith ('airpoc://') and self.data: data = self.data else: with open (filename Encoding='utf-8') as f: data = f.read () return data def exec_module (self, module): filename = self.get_filename (self.fullname) poc_code = self.get_data (filename) obj = compile (poc_code, filename, 'exec', dont_inherit=True, optimize=-1) exec (obj, module.__dict__)

We don't need to care about how it is implemented, we just need to know that we can use load_string_to_module to load the module from the source code. If you are interested in learning about the specific implementation, you can refer to the python official documentation above.

The formulation of rules

After loading the module from the file or remotely, you can prepare for running. We need to make a uniform convention of rules for PoC so that the program can call them better.

You can define the rules in detail, or you can simplify everything, mainly depending on the usage scenario. As mentioned earlier, in order to protect the security issues of the Security Alliance, we need PoC to write more easily and quickly.

At the same time, we also need to consider what to do if PoC requires multiple parameters. The author's rules are defined in this way.

Def verify (arg, * * kwargs): result = {} if requests.get (arg). Status_code = 200: result = {"name": "vulnerability name", "url": arg} return result

Define a verify function in the PoC file for verification, and arg is passed as a normal parameter, which is received from kwargs when more parameters need to be passed. After the PoC verification is successful, you only need to return a dictionary, and if the verification fails, return False or None. The content of the dictionary is formulated by the PoC writer, which gives the writer the maximum flexibility.

But be careful! The quality of PoC depends on the maintenance of the writer.

V0.01

Our ultimate goal is to set the goal, and the program automatically loads one or more PoC or all the PoC specified, and detects the targets one by one. The rest is how to concatenate these functions.

We have already implemented the basic framework of AirPoc, and now we only need to implement the function on the basis of it.

For testing purposes, let's first create two rudimentary PoC in the pocs directory according to the rules defined earlier.

Now, the code in main.py is as follows

#! / usr/bin/env python3#-*-coding: utf-8-*-# @ Time: 3:13 on 2019-4-25 PM# @ Author: w7ay# @ File: main.pyimport osimport timefrom lib.data import VERSION, PATHS_POCS POCSfrom lib.loader import load_string_to_moduledef banner (): msg =''_ / | _\ | _\ / _ | / / | | _ | | | _ | | / / | _ / | _ / | | / | |\ | _ _ | _ |\ _ | _ |\ _ / _ _ | {}''.format (VERSION) print (msg) def init (config: dict) : print ("[*] target: {}" .format (config ["url"])) # load poc First traverse the path _ pocs = [] for root, dirs, files in os.walk (PATHS_POCS): files = filter (lambda x: not x.startswith ("_") and x.endswith (".py") and x not in config.get ("poc", []) Files) # filter out _ _ init__.py file and specify poc file _ pocs.extend (map (lambda x: os.path.join (root, x), files)) # load PoC for poc in _ pocs: with open (poc) according to the path 'r') as f: model = load_string_to_module (f.read ()) POCS.append (model) def end (): print ("[*] shutting down at {0}" .format (time.strftime ("% X")) def start (config: dict): url_list = config.get ("url", []) # Loop url_list and pocs Execute one by one. For i in url_list: for poc in POCS: try: ret = poc.verify (I) except Exception as e: ret = None print (e) if ret: print (ret) def main (): banner () config = {"url": ["https://www.seebug.org/"," "https://paper.seebug.org/"]," poc ": []} init (config) start (config) end () if _ _ name__ ='_ _ main__': main ()

Our version has also reached 0.01, and it is already a "mature" framework that can run PoC on its own.

Multithreaded model

In order to make our framework run faster, we use multithreading to process each PoC, because most of the tasks we deal with are Imax O-intensive tasks, so we don't have to worry too much about whether python is pseudo-threaded.

The simplest of the multithreaded models is the producer / consumer model, where multiple threads are started to consume a queue. Create a new lib/threads.py

Import threadingimport timedef exception_handled_function (thread_function, args= ()): try: thread_function (* args) except KeyboardInterrupt: raise except Exception as ex: print ("thread {0}: {1}" .format (threading.currentThread (). GetName (), str (ex)) def run_threads (num_threads, thread_function) Args: tuple = (): threads = [] # start multiple threads for num_threads in range (num_threads): thread = threading.Thread (target=exception_handled_function, name=str (num_threads), args= (thread_function) Args) thread.setDaemon (True) try: thread.start () except Exception as ex: err_msg = "error occurred while starting new thread ('{0}')" .format (str (ex)) print (err_msg) break threads.append (thread) # wait for all threads to finish alive = True while alive : alive = False for thread in threads: if thread.isAlive (): alive = True time.sleep

Then the main program is transformed into a multi-threaded mode, and the "consumer" in the original start () is extracted and used as a function to receive data with the queue. As follows

Def worker (): if not WORKER.empty (): arg, poc = WORKER.get () try: ret = poc.verify (arg) except Exception as e: ret = None print (e) if ret: print (ret) def start (config: dict): url_list = config.get ("url" []) # production for arg in url_list: for poc in POCS: WORKER.put ((arg, poc)) # consumption run_threads (10, worker)

In addition, the number of threads is configurable, and we change it to read from the configuration.

Run_threads (config.get ("thread_num", 10), worker)

If you run it again, you will find it much faster than before!

Unified network request

This is the last part of our entire framework, how to unify network requests. Sometimes we need to unify the settings of proxies, UA headers, and so on in the network requests issued by our PoC framework, which requires our framework to deal with them uniformly. Before achieving our goal, we also need to make an agreement in the framework that all our network requests need to be uniformly sent using requests. We said at the beginning that we would try not to use third-party modules, but the requests module was so easy to use that we excluded it.

The dynamic mechanism of the Python language, we can easily Hook it before using a function and redirect its original method to our custom method, which is a prerequisite that we can unify the network request.

Def hello (arg): return "hello" + argdef hook (arg): arg = arg.upper () return "hello" + arghello = hookprint (hello ("aa"))

Through hook a function to achieve our own purpose.

Tools like sqlmap are based on python's built-in urllib module, but there is a lot of code dealing with web requests, and even to deal with chunked sending packets, hook rewrites the lower-level httplib library.

In order to schedule network requests uniformly, pocsuite hook the related methods of requests module. We can refer to the code specifically.

The pocsuite3/lib/request/patch/__init__.py code clearly illustrates the function of hook.

From. Remove _ ssl_verify import remove_ssl_verifyfrom. Remove _ warnings import disable_warningsfrom. Hook _ request import patch_sessionfrom. Add _ httpraw import patch_addrawfrom. Hook _ request_redirect import patch_redirectdef patch_all (): disable_warnings () # disabled warning prompt remove_ssl_verify () # disable ssl verification patch_session () # hook seesion function patch_addraw () # add raw native package support Patch_redirect () # hook redirect function

If you have seen the source code of requests, you will know that the key point is to see how it functions hook seesion.

Pocsuite3/lib/request/patch/hook_request.py

From pocsuite3.lib.core.data import conffrom requests.models import Requestfrom requests.sessions import Sessionfrom requests.sessions import merge_setting, merge_cookiesfrom requests.cookies import RequestsCookieJarfrom requests.utils import get_encodings_from_contentdef session_request (self, method, url, params=None, data=None, headers=None, cookies=None, files=None, auth=None, timeout=conf.timeout if 'timeout' in conf else None, allow_redirects=True, proxies=None, hooks=None, stream=None, verify=False, cert=None Json=None): # Create the Request merged_cookies = merge_cookies (merge_cookies (RequestsCookieJar (), self.cookies), cookies or (conf.cookie if 'cookie' in conf else None)) req = Request (method=method.upper (), url=url, headers=merge_setting (headers, conf.http_headers if' http_headers' in conf else {}), files=files, data=data or {} Json=json, params=params or {}, auth=auth, cookies=merged_cookies, hooks=hooks,) prep = self.prepare_request (req) proxies = proxies or (conf.proxies if 'proxies' in conf else {}) settings = self.merge_environment_settings (prep.url, proxies, stream, verify, cert) # Send the request. Send_kwargs = {'timeout': timeout,' allow_redirects': allow_redirects,} send_kwargs.update (settings) resp = self.send (prep) * * send_kwargs) if resp.encoding = = 'ISO-8859-1): encodings = get_encodings_from_content (resp.text) if encodings: encoding = encodings [0] else: encoding = resp.apparent_encoding resp.encoding = encoding return respdef patch_session (): Session.request = session_request

It overrides the method of the session_request function so that we can customize the file first class information that we customize. The above code may require you to read requests to understand him, but it doesn't matter, let's just use it in the spirit of borrowlism.

In order to achieve this goal and better optimize the framework structure, we also need to make some minor adjustments.

Create a new lib/requests.py

From lib.data import CONFfrom requests.models import Requestfrom requests.sessions import Sessionfrom requests.sessions import merge_setting, merge_cookiesfrom requests.cookies import RequestsCookieJarfrom requests.utils import get_encodings_from_contentdef session_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=False, cert=None, json=None): # Create the Request. Conf = CONF.get ("requests", {}) if timeout is None and "timeout" in conf: timeout = conf ["timeout"] merged_cookies = merge_cookies (merge_cookies (RequestsCookieJar (), self.cookies), cookies or (conf.cookie if 'cookie' in conf else None) req = Request (method=method.upper (), url=url, headers=merge_setting (headers) Conf ["headers"] if 'headers' in conf else {}), files=files, data=data or {}, json=json, params=params or {}, auth=auth, cookies=merged_cookies, hooks=hooks,) prep = self.prepare_request (req) proxies = proxies or (conf ["proxies"] if' proxies' in conf else {}) settings = self.merge_environment_settings (prep.url, proxies, stream, verify Cert) # Send the request. Send_kwargs = {'timeout': timeout,' allow_redirects': allow_redirects,} send_kwargs.update (settings) resp = self.send (prep) * * send_kwargs) if resp.encoding = = 'ISO-8859-1): encodings = get_encodings_from_content (resp.text) if encodings: encoding = encodings [0] else: encoding = resp.apparent_encoding resp.encoding = encoding return respdef patch_session (): Session.request = session_request

At the same time, reserve the interface of requests in config.

And execute our hook when init.

Let's write a new PoC and test the final effect http://www.httpbin.org/get with this site.

Pocs/poc.py

Import requestsdef verify (arg, * * kwargs): r = requests.get (arg) if r.status_code = = 200: return {"url": arg, "text": r.text}

Also refer to the method of Pocsuite to disable warning information

From urllib3 import disable_warningsdisable_warnings ()

Finally, the version number was changed to 0.1 with a sense of ritual, and the framework part of AirPoc was largely completed.

Last

Many of the structural ideas of AirPoc come from Pocsuite. If you read Pocsuite directly, you may be able to gain a lot. At present, the AirPoc v0.1 infrastructure is almost complete, and one or more PoC can be loaded locally for batch testing. We'll try some more interesting things later, how to verify the situation without echo, how to generate shellcode, and how to operate the linked shell. Please look forward to the next "feature article" ~.

AirPoc download: https://images.seebug.org/archive/airpoc.zip

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.