BLEN/lib/core/common.py

# coding:utf-8
from __future__ import print_function
from logging import fatal
from typing import Text
from bs4 import BeautifulSoup
import sys
import string
import argparse
from urllib.parse import urlparse
    
import random
import configparser
import requests
import time
import os
import re
import base64

from lib.core.data import vulnoutput,unvulnoutput,unreachoutput,lock,AliveList,root_path
from lib.core.log import logverifyerror,logvuln,logunvuln

from requests.exceptions import ConnectTimeout
from requests.exceptions import ConnectionError
from requests.exceptions import HTTPError
from requests.exceptions import TooManyRedirects

def get_title(htmlcode):
    """
    获取网站title  

    use:
    get_title(html_source)  

    return:  
    title  
    """
    soup = BeautifulSoup(htmlcode, 'html.parser')

    end = str(soup.title).find("</title>")
    title = str(soup.title)[7:end] + "<title>"
    if "<title>" in title:
        title = title.replace("<title>","",10000)

    if "|" in title:
        title = title.replace("|"," ",100000)

    if "\n" in title:
        title = title.replace("\n","",100000)

    if "\t" in title:
        title = title.replace("\t","",100000)

    if "\r" in title:
        title = title.replace("\r","",100000)

    if "`" in title:
        title = title.replace("`","",100000)

    if "," in title:
        title = title.replace(",","",100000)

    return title



def url_handle(url):
    """
    url处理函数  

    return:
    dict urldict  
    """

    if url.startswith("http"):
        p = urlparse(url)
        # pass

    else:
        url = "http://" + url 
        p = urlparse(url)
        if ":" in p.netloc:
            if "443" in p.netloc.split(":")[1]:
                url = url.replace("http://","https://")
                p = urlparse(url)

            else:
                pass
        else:
            pass
        

    return p.scheme+"://"+p.netloc

def get_random_ua():
    with open("./data/user_agents.txt","r") as f:
        UAs = [i.strip() for i in f.readlines()]
    return random.choice(UAs)


def random_str(length=10, chars=string.ascii_letters + string.digits):
    return ''.join(random.sample(chars, length))

def get_local_version(path):
    cp = configparser.ConfigParser()
    cp.read(path)
    return cp["info"]["version"]

def get_ceye_dns():
    """
    根据你配置的dns，生成一个随机dns子域名  
    该函数会检查info.ini配置是否正确，并通过ture/false来反馈给调用者  
    如果配置读取失败会立即停止本次扫描并将错误信息打印到控制台提示用户  
    
    Returns:
        [type]: [description]
    """
    cp = configparser.ConfigParser()
    cp.read(root_path+"/info.ini")
    dns = cp["ceye"]["dns"]
    token = cp["ceye"]["token"]
    flag = random_str()
    
    if len(dns) == 0 or len(token) == 0:
        return False,"<title>ceye dns或token 配置为空，请重新确认</title>"
    else:
        return True,flag + "." + dns

def verify_ceye_dns(flag):
    """
    接通ceye的dns的api  
    内置sleep功能，必须在3秒以上，否则极其不稳定，考虑到批量扫描时对ceye的负载，我决定最少延时6秒   

    Args:
        flag ([type]): [description]

    Returns:
        [type]: [description]
    """
    cp = configparser.ConfigParser()
    cp.read(root_path+"/info.ini")
    token = cp["ceye"]["token"]
    api_url = "http://api.ceye.io/v1/records?token="+token+"&type=dns"
    
    time.sleep(random.randint(6,10))
    
    req = requests.get(api_url,timeout=10)
    if "User Not Exists" in req.text:
        return "<title>ceye 配置错误，无法获取数据</title>"
    elif flag.lower() in req.text.lower():
        return True
    else:
        return False

def get_latest_revision():
    lv = None
    cp = configparser.ConfigParser()
    try:
        req = requests.get("https://raw.githubusercontent.com/openx-org/BLEN/master/info.ini",timeout = 15)
        cp.read_string(req.text)

        lv = cp["info"]["version"]
    except Exception:
        pass
    return lv


def run(POC_Class,target,proxy=False,output=True,PocRemain="",Alive_mode = False):

    global vulnoutput,unvulnoutput,unreachoutput,AliveList
    while not target.empty():

        try:
            target_url = target.get()
            rVerify = POC_Class(target_url,proxy)
            poc_name = rVerify._info["name"]
            # print(poc_name)
            vuln = rVerify._verify()
            
            if Alive_mode == True:
                if vuln[0] == True:
                    AliveList.add(target_url)
                    continue
                    # print("yes")
                    
                else:
                    continue
            if vuln[0] == True:
                try:
                    vulntitle=get_title(vuln[1])
                except:
                    vulntitle = ""
                lock.acquire()
                logvuln("POC 剩余 : %s ╭☞ Target 剩余 : %d Vuln %s | WebSite Title：%s "%(PocRemain,target.qsize(),target_url,vulntitle))
                # logvuln("%s ╭☞ %d Vuln %s | WebSite Title：%s | Server Response : %s"%(PocRemain,target.qsize(),target_url,vulntitle,vuln[1]))
                if poc_name in vulnoutput:
                    vulnoutput[poc_name].append(target_url+" || "+vulntitle)
                else:
                    vulnoutput.update({poc_name:list()})
                    vulnoutput[poc_name].append(target_url+" || "+vulntitle)
                lock.release()
            else:
                lock.acquire()
                logunvuln("POC 剩余 : %s ╭☞ Target 剩余 : %d UnVuln %s "%(PocRemain,target.qsize(),target_url))
                unvulnoutput.append(target_url)
                lock.release()
        
        except NotImplementedError as e :
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d The POC does not support virtualized depiction scan mode  Error details：%s "%(PocRemain,target.qsize(),str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

        except TimeoutError as e:
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d Connection timed out %s Error details：%s "%(PocRemain,target.qsize(),target,str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

        except HTTPError as e:
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d HTTPError occurred %s Error details：%s "%(PocRemain,target.qsize(),target,str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

        except ConnectionError as e:
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d Connection error %s Error details：%s "%(PocRemain,target.qsize(),target,str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

        except TooManyRedirects as e:
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d The number of resets exceeds the limit, and the goal is discarded %s Error details：%s "%(PocRemain,target.qsize(),target,str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

        except BaseException as e:
            lock.acquire()
            logverifyerror("POC 剩余 : %s ╭☞ %d unknown mistake %s Error details：%s "%(PocRemain,target.qsize(),target,str(e)))
            unreachoutput.append(target_url+" || Error details"+str(e))
            lock.release()
            pass

def GetCommand():
    parser = argparse.ArgumentParser(description="Blen framewark of POC test",
    usage="python3 Blen.py -f [target_path] / -u [url]   -s [poc_path]  --thread 50\n\
       python3 Blen.py --fofa-search")

    searchengine = parser.add_argument_group("SearchEngine")
    searchengine.add_argument("--fofa-search",action="store_true",help="Fofa Search模式，该参数不需要跟值")

    # target = parser.add_argument_group("TARGET")
    target = parser.add_mutually_exclusive_group()
    target.add_argument("-u","--url",type=str,help="指定单个url，该模式不支持多POC或全量POC (e.g. www.baidu.com)")
    target.add_argument("-f","--file",type=str,help="指定存有url列表的文件路径，该模式支持多POC或全量POC (e.g. /root/urllist.txt)")

    script = parser.add_argument_group("Script")
    script.add_argument("-s","--script",type=str,help="指定POC相对路径，格式见readme.md (e.g. -s poc/jellyfin/jellyfin_fileread_scan/poc.py OR -s all)")
    
    system = parser.add_argument_group("System")
    system.add_argument("--thread",default=10,type=int,help="指定线程数，默认为10，仅扫描时指定线程数有效")
    system.add_argument("--proxy",default=False,help="指定Http Proxy，仅扫描时指定线程数有效，Example：127.0.0.1:8080 OR http://127.0.0.1:8080")
    system.add_argument("--output",default=True,help="不用管，扫完了看output目录即可")
    system.add_argument("--sound",action="store_true",help="扫完了会有铃声提醒，虚拟机里不一定能用")
    system.add_argument("--version",action="store_true",help="显示本地oFx版本，并根据网络状态给出最新版本号")

    developer = parser.add_argument_group("Developer(POC开发者工具箱)")
    developer.add_argument("--add-poc",action="store_true",help="生成POC标准目录结构，该参数不需要跟值")
    developer.add_argument("--show-error",action = "store_true",help="single mode下展示详细报错信息")
    if len(sys.argv) == 1:
        sys.argv.append("-h")
    args=parser.parse_args()
    return args 

def Str2Base64(string):
    """
    str => base64
    """
    return base64.b64encode(str.encode(string)).decode()