web安全扫描器组件-HTTP请求参数处理py脚本

[{'param_name': 'dadsa2', 'param_type': 'String', 'param_lenght': 5}, {'param_name': 'a.aa', 'param_type': 'String', 'param_lenght': 8}, {'param_name': 'a.b', 'param_type': 'Int', 'param_lenght': 4}, {'param_name': 'b.b1.aa', 'param_type': 'String', 'param_lenght': 8}, {'param_name': 'b.b1.b', 'param_type': 'Int', 'param_lenght': 4}, {'param_name': 'c', 'param_type': 'Int', 'param_lenght': 1}, {'param_name': 'url', 'param_type': 'Encode', 'param_lenght': 35}]

# coding: utf-8
"""
@Time :    9/22/2022 17:32 
@Author:  fff
@File: Param_Process.py
@Software: PyCharm

; 参数处理 两个函数接口
; http_request_param_list  导入http请求,遍历请求内参数,并返回参数名list
; set_http_request_param  设置http请求指定参数名的值,后追加/前追加/替换 并且返回http请求
"""
import copy,enchant
from urllib import parse
from urllib.parse import unquote,quote,urlparse,ParseResult
class Param_Process:
    def __init__(self,loop_num=36):
        self.loop_num=loop_num

    def callback_len(self,data):
        '''
        ;返回长度极致30
        :param data:
        :return:
        '''
        data=len(data)
        if data>30:
            return 30
        else:return data

    def static_filter(self,url):
        '''
        # 伪静态与url路径泛化处理
        :param url:
        :return:
        '''
        urlparse_process=urlparse(url)
        urls = urlparse_process.path
        folder_name_list=[]
        file_type=['htm', 'html', 'xhtml', 'shtml', 'php', 'jsp', 'jspx', 'do', 'action', 'aspx', 'asp', 'py']
        d_enchant = enchant.Dict("en_US")
        for folder_name in urls.split('/'):
            if self.type_param_value(folder_name.strip())=='Int':
                folder_name_list.append('{%s:%s}'%(self.type_param_value(folder_name.strip()),str(self.callback_len(folder_name.strip()))))
            elif len(folder_name.strip())>2 and len(folder_name.strip())<20 and d_enchant.check(folder_name.strip())==False:
                name = folder_name.split('.')
                if len(name) > 1 and name[-1].lower() in file_type and d_enchant.check(name[0])==False:
                    folder_name_list.append('{%s:%s}.%s' % (self.type_param_value('.'.join(name[0:-1]).strip()), str(self.callback_len('.'.join(name[0:-1]).strip())),name[-1]))
                elif len(name) ==2 and d_enchant.check(name[0]) and name[-1].lower() in file_type:
                    if self.type_param_value(name[0])!='int' and '-' not in name[0]:
                        folder_name_list.append('%s.%s' % (name[0],name[1]))
                    else:folder_name_list.append('{%s:%s}.%s' % (self.type_param_value(name[0]),str(self.callback_len(name[0])),name[1] ))
                else:
                    folder_name_list.append('{%s:%s}'%(self.type_param_value(folder_name.strip()),str(self.callback_len(folder_name.strip()))))
            elif folder_name!=''and len(folder_name.strip())<20 and d_enchant.check(folder_name.strip()) and '-' not in folder_name.strip():
                folder_name_list.append(folder_name.strip())
            elif folder_name == '':folder_name_list.append(folder_name.strip())
            else:folder_name_list.append('{%s:%s}'%(self.type_param_value(folder_name.strip()),str(self.callback_len(folder_name.strip()))))
        url_path = "/".join(folder_name_list)
        return urlparse_process.scheme + '://' + urlparse_process.netloc +'/'+  url_path.strip('/')

    def fitler_request_param_list(self,url,param_list):
        '''
        ; 请求参数过滤
        :param url:
        :param param_list:
        :return:
        '''
        fitler_json={}
        fitler_json['url_path']=self.static_filter(url)
        fitler_json['list_param']=list(set([param['param_name'] for param in param_list]))
        print(fitler_json)
    def type_param_value(self,value):
        '''
        : 判断http请求传参类型 String Int Json List Encode List_Param
        :param http_request:
        :return:
        '''
        if type(value)==type(1):
            return 'Int'
        elif type(value)==type({}):
            return 'Json'
        elif type(value)==type([]):
            return 'List'
        elif type(value)==type('a'):
            if value.startswith('http://') or value.startswith('https://'):
                return 'Url'
            elif type(parse.parse_qs(value))==type({}) and parse.parse_qs(value)!={}:
                return 'List_Param'
            elif value.isdigit():
                return 'Int'
            elif value.count('%')>3 and len(value)>22:
                return 'Encode'
            return 'String'

    def param_qsl_process(self,body,param_name=''):
        '''
        ; aa=aaaaaaaa&b=1123&c=http://www.qq.com
        :param body:
        :param param_name:
        :return:
        '''
        self.json_process(dict(parse.parse_qsl(body)),param_name)

    def json_process(self,body,param_name=''):
        '''
        : json与字典方式处理
        :param body:
        :return:
        '''
        #print(self.loop_num,param_name)
        for param in body:
            type_result=self.type_param_value(body[param])
            if type_result=='Json':
                self.json_process(body[param],param_name=param_name+'.'+param)
            elif type_result=='List':
                list_num=0
                for line in body[param]:
                    if self.type_param_value(line)=='Json':
                        self.json_process(line,param_name=param_name+'.'+param+'.lisT_'+str(list_num))
                        #print(list_num)
                    list_num = list_num+1
            elif type_result=='List_Param':
                self.param_qsl_process(body[param],param_name=param_name+'.'+param)
            elif type_result=='Encode':
                uncode=unquote(body[param])
                if self.type_param_value(uncode) in ['Json','List']:
                    self.json_process(uncode,param_name=param_name+'.'+param)
                elif self.type_param_value(uncode)=='List_Param':
                    self.param_qsl_process(uncode,param_name=param_name+'.'+param)
                else:
                    if param_name == '':
                        self.param_name_list.append(
                            {"param_name": param, "param_type": type_result, "param_lenght": len(str(body[param]))})
                    else:
                        self.param_name_list.append(
                            {"param_name": param_name.strip('.') + '.' + param, "param_type": type_result,
                             "param_lenght": len(str(body[param]))})
            else:
                if param_name=='':
                    self.param_name_list.append({"param_name": param, "param_type": type_result, "param_lenght": len(str(body[param]))})
                else:
                    self.param_name_list.append({"param_name": param_name.strip('.')+'.'+param, "param_type": type_result, "param_lenght": len(str(body[param]))})
            #self.loop_num=0

    def http_request_param_list(self,http_request):
        '''
        : 导入http请求,遍历请求内参数,并返回参数名list
        :return:
        '''
        self.param_name_list = []
        body=http_request['body']
        query=urlparse(http_request['url']).query
        #print(query)
        self.param_qsl_process(query)
        if self.type_param_value(body)=='Json':
            self.json_process(body)
        elif self.type_param_value(body)=='List_Param':
            self.param_qsl_process(body)
        #self.fitler_request_param_list(http_request['url'],self.param_name_list[:self.loop_num])
        return self.param_name_list[:self.loop_num]

    def set_payload(self,param_value,payload,status=0):
        '''
        ; 设置payload 后追加/前追加/替换
        :param param_value:
        :param payload:
        :param status:
        :return:
        '''
        if status==0:
            return str(param_value)+payload
        elif status==1:
            return payload+str(param_value)
        elif status==2:
            return payload

    def set_query(self,parameter,param,payload,status):
        '''
        ;处理aaa=1a&bb=asa转json类型，设置payload
        :param parameter:
        :param param:
        :param payload:
        :param status:
        :return:
        '''
        try:
            param_dict = dict(parse.parse_qsl(parameter))
            param_dict[param] = self.set_payload(param_dict[param], payload, status)
            query = parse.urlencode(param_dict)
            return query
        except:return parameter


    def json_to_param(self,parameter,param_list):
        '''

        :param parameter:
        :param param:
        :param payload:
        :param status:
        :return:
        '''
        json_task_bak = parameter
        for line in param_list:
            if line == param_list[-1]:
                json_task_bak[line] = parse.urlencode(json_task_bak[line])
            else:
                json_task_bak = json_task_bak[line]
        return parameter

    def set_param(self,parameter,param,payload,status=0):
        '''
        ;设置指定参数名的值
        :param parameter:
        :param param:
        :param payload:
        :param status:
        :return:
        '''
        #print(parameter)
        type_parameter=self.type_param_value(parameter)
        if type_parameter=='List_Param':
            #print(parameter)
            return self.set_query(parameter,param,payload,status)
        elif type_parameter=='Json':
            param_list=param.split('.')
            param_list_bak=[]
            num_bak = 0
            #print(param_list)
            for line in param_list:
                if line.startswith('lisT_'):
                    param_list_bak.append(int(line.replace('lisT_', '')))
                else:param_list_bak.append(line)
            param_list=param_list_bak
            if len(param_list)>0:
                parameter_task=parameter
                bak_list=[]
                for line in param_list:
                    #print(parameter_task)
                    type_value=self.type_param_value(parameter_task[line])
                    if type_value == 'Json':
                        bak_list.append(line)
                        parameter_task=parameter_task[line]
                    elif type_value == 'List':
                        bak_list.append(line)
                        parameter_task = parameter_task[line]
                    elif type_value=='List_Param':
                        parameter_task[line]=dict(parse.parse_qsl(parameter_task[line]))
                        parameter_task = parameter_task[line]
                        bak_list.append(line)
                        num_bak=1
                    elif type_value == 'Encode':
                        line_encode=unquote(parameter_task[line])
                        encode_type=self.type_param_value(line_encode)
                        if encode_type=='List_Param':
                            parameter_task[line] = dict(parse.parse_qsl(parameter_task[line]))
                            parameter_task = parameter_task[line]
                            bak_list.append(line)
                            num_bak = 1
                        else:parameter_task[line]=quote(self.set_payload(line_encode,payload,status))
                    else:
                        parameter_task[line]=self.set_payload(parameter_task[line],payload,status)
                if bak_list!=[] and num_bak==1:
                    self.json_to_param(parameter,bak_list)
                    #print(parameter)
                return parameter
            else:
                parameter[param]=self.set_payload(parameter[param],payload,status)
                return parameter

    def set_http_request_param(self,http_request,param,payload,status=0):
        '''
        ;设置http请求指定参数名的值,后追加/前追加/替换 并且返回http请求
        :param http_request:
        :param param: 参数名
        :param payload:
        :param status:0 1 2
        :return:
        '''
        try:
            self.param_name_list = []
            http_request=copy.deepcopy(http_request)
            urlparse_process=urlparse(http_request['url'])
            query=urlparse_process.query
            if '=' in query :
                query=self.set_param(query, param, payload, status)
                res = ParseResult(scheme=urlparse_process.scheme, netloc=urlparse_process.hostname, path=urlparse_process.path, params=urlparse_process.params, query=query,
                                  fragment=urlparse_process.fragment)
                if http_request['url']!=res.geturl():
                    http_request['url']=res.geturl()
                    return http_request
            body_type=self.type_param_value(http_request['body'])
            if body_type in ['Json','List_Param']:
                http_request['body']=self.set_param(http_request['body'],param, payload, status)
            return http_request
        except Exception as erorr:
            print(erorr,erorr.__traceback__.tb_lineno)
            return 0



if __name__ == '__main__':
    task_json1={
        "url":"http://www.qq.com",
        "method":"POST",
        "body":{
        "j": ["a", "c", {"d": "http://www.163.com", "aaa2": {"c": 1,"c1":{"d1":"aaaa"}}},{"d": "http://www.163.com"}],
        "ccc":12222,
        "b":{"aa1":"cc245"}
        }
    }

    task_json={
        "url":"http://www.qq.com",
        "method": "POST",
        "body":{
            "a":"aaaaaaaa",
            "b":1,
            "c":[{"c1":"c1c1c1"},{"c2":"c2c2c2"},{"c3":2,"c4":"aaaa","a":1}],
            "d":{"d1":"d1d1d1","d2":3},
            "e":"eeeeeeeeeee",
            "url":"http://www.qq.com",
            "f":"dasdsa%E9%98%BF%E6%96%AF%E8%92%82%E8%8A%AC%E6%98%AF",
            "g":"aa=aaaaaaaa&b=1123&c=http://www.qq.com",
            "h":"145",
            "i":{"i1":"a","g":{"ia":"cccc12"}},
            "j":["a","c",{"d":"http://www.163.com","aaa2":{"c":1}}],
            "k":"http%3A%2F%2Fwww.163.com",
            "l":"asdasd%3Ddsadsa1%26gngf%3D1%26da12%3Dhttp%3A%2F%2Ftaw.target.com%2Faaa.html"
        }
    }
    task_param={
        "url":"http://www.qq.com?aaa=dasdsa&fsdfsd=1",
        "method": "POST",
        "body":"aa=aaaaaaaa&b=1123&c=http://www.qq.com&encode=asdasd%3Ddsadsa1%26gngf%3D1%26da12%3Dhttp%3A%2F%2Ftaw.target.com%2F1.html"
               }
    task_param_get={
        "url":"http://www.target.com/xxxx/1/asasdfdsadsadsasdasadas/ping.html?aaa=dasdsa&fsdfsd=1&bbb=%7B%22a%22%3A%22ccccccccc%22%7D",
        "method": "POST",
        "body":"Null"
               }
    task_param_post={
        "url":"http://www.target.com/xxxx?aaa=dasdsa&fsdfsd=1&bbb=%7B%22a%22%3A%22ccccccccc%22%7D",
        "method": "POST",
        "body":"bb=aassdfds&gfdgdf=2&sadsa=aaa%E9%AB%98%E6%9D%A0%E6%9D%86"
               }
    task_json2={
        "url":"http://www.target.com?dadsa2=aaa43",
        "method": "POST",
        "body":{"a":"aa=aaaaaaaa&b=1123","b": {"b1":"aa=aaaaaaaa&b=1123"},"c":"1","url":"http%3A%2F%2Ftest.target.com%2Faaaa"}
    }
    task_json3={
                    "url":"http://www.qq.com",
                    "method": "POST",
                    "body":
                    {
                        "aaaa":
                            {"b":"asdasd%3Ddsadsa1%26gngf%3D1%26da12%3Dhttp%3A%2F%2Ftst.qq.com%2Fflag.html"},
                        "c":1
                            }
    }

    task=Param_Process()
    #print(task.http_request_param_list(task_json2))
    #print(task.set_http_request_param(task_json2, 'url', '"xss', 0))
    #for line in task.http_request_param_list(task_param_post):
        #print(line['param_name'])
    #    print(task.set_http_request_param(task_param_post, line['param_name'], '"xss'))
    #print(task.set_http_request_param(task_json2,'url','"xss',0))、
    print(task.static_filter('http://www.target.com/dasdas/2131/log/dsadsa/aa/23/xxx/new'))