安全运营之API接口暴露探测

def timeCal(f):  # 装饰器，用于显示版本信息以及计算检测耗费时间

    print("#********************")

    print("# version 0.3")

    print("# edit by xxxxx")

    print("#********************")

    def func():

        start = time.time()

        f()

        end = time.time()

        duration = end - start

        print("Crack completed,Time cost %s seconds " % duration)

    return func

#这里使用threading模块完成多线程编写，对URL进行检测，查看返回的状态码是否为200，如果为200说明该URL对应的API接口可以被访问到，然后将其记录入列表。

class MyThread(threading.Thread):

    def __init__(self, queue,):

        threading.Thread.__init__(self)

        self._queue = queue

    def run(self):

        while not self._queue.empty():

            url = self._queue.get()

            urllib3.disable_warnings()

            try:

                print("Testing URL: %s" % url)

                res = requests.get(url, verify=False, allow_redirects=False, timeout=10)

                status_code = res.status_code

                if status_code == 200:

                    print("======> %s status is 200" % url)

                    results.append(url)

            except Exception as e:

                pass

#这部分是解决脚本入参的，读取用户的输入，指定IP地址列表清单、API接口清单以及多线程数量，这里默认线程数量设置为100

@timeCal

def main():

    parser = optparse.OptionParser("Usage:%prog -i <ipAddressFile> -w <wordFile> -t <thsCount>")

    parser.add_option("-i", dest="tips", type="string", help="specify ip address")

    parser.add_option("-w", dest="twordfile", type="string", help="specify words file")

    parser.add_option("-t", dest="thsCount", type="string", help="specify threads count", default=100)

    (options, args) = parser.parse_args()

    ips = options.tips

    wordfile = options.twordfile

    thsCount = options.thsCount

    if not ips or not wordfile or not thsCount:

        print(parser.usage)

        exit(0)


    threads = []

    threads_count = int(thsCount)

    q = queue.Queue()

#读取IP地址列表，对每个IP地址与API接口进行拼接为完整的URL

    with open(ips) as ips:

        for ip in ips:

            ip = ip.strip()

            with open(wordfile) as fd:

                for uri in fd:

                    uri = uri.strip()

                    url = "https://" + ip + "/" + uri

                    q.put(url)

    for i in range(threads_count):

        threads.append(MyThread(q))

    for i in threads:

        i.start()

    for i in threads:

        i.join()

def sendmail(ti, content):

    print("扫描完成，开始发送邮件……")

    mail_host = "smtp.163.com"

    mail_user = "xxxxxxxxx"

    mail_pass = "xxxxxxxxxxx"
    

    newContent = '''

    本次API接口扫描情况如下：

    1、本次扫描时间： %s

    2、本次扫描结果：\n %s

    ''' % (ti, content)


    # 发件人

    sender = " xxxxxxxxx @163.com"

    # 收件人

    receivers = ['67xxxxxxx31@qq.com']


    # 邮件正文

    message = MIMEText(newContent, "plain", "utf-8")

    # 邮件头部，多个收件人，注意收件人地址拼接

    message['From'] = sender

    message['To'] = Header(';'.join(receivers), 'utf-8')


    # 邮件主题

    subject = "XXXX公司 API接口泄露监测"

    message['Subject'] = Header(subject, 'utf-8')


    try:

        smtpObj = smtplib.SMTP(mail_host, 25)

        smtpObj.login(mail_user, mail_pass)

        smtpObj.sendmail(sender, receivers, message.as_string())

        print("邮件发送成功")

    except smtplib.SMTPException as Exception:

        print(Exception)

        print("Error: 发送邮件失败")

if __name__ == '__main__':

    main()

    ti = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

    content = '\n'

    print("=======================================")

    print("扫描时间： %s " % ti)

    print("Test finished! The results are bellow:")

    if len(results) != 0:

        for result in results:

            print(result)

        content = content.join(results)

        sendmail(ti, content)

    else:

        noResults = '本次扫描未发现高危API接口暴露信息'

        print(noResults)

        sendmail(ti, noResults)