Merge branch 'main' of https://git.mamahaha.work/sangge/BackDoorBuster into feature/pyc-detection
This commit is contained in:
commit
aeb4a33d98
2
MANIFEST.in
Normal file
2
MANIFEST.in
Normal file
@ -0,0 +1,2 @@
|
||||
include README.md
|
||||
include LICENSE
|
||||
64
README.md
64
README.md
@ -1,6 +1,7 @@
|
||||
# BackDoorBuster
|
||||
|
||||
|
||||
|
||||
## 项目背景
|
||||
|
||||
随着网络安全威胁的增加，恶意软件和后门的检测成为了保护个人和组织数据安全的重要任务。后门通常被隐藏在合法软件中，给黑客提供远程控制目标系统的能力。本项目旨在开发一个工具，能够有效识别和评估潜在的后门风险。
|
||||
@ -17,21 +18,66 @@
|
||||
- **报告生成**: 自动生成详细的检测报告，列出所有发现的敏感操作和对应的风险等级。
|
||||
- **持续更新与维护**: 随着新的后门技术和检测方法的出现，持续更新正则表达式库和评级标准。
|
||||
|
||||
## 打包
|
||||
|
||||
### pip
|
||||
|
||||
#### 打包命令
|
||||
|
||||
```bash
|
||||
pip install wheel
|
||||
python setup.py sdist bdist_wheel
|
||||
```
|
||||
|
||||
执行上述命令后，会在 dist 目录下生成 .tar.gz 和 .whl 文件。
|
||||
|
||||
#### 本地安装
|
||||
|
||||
- 安装 .whl 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0-py3-none-any.whl
|
||||
```
|
||||
|
||||
- 安装 .tar.gz 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0.tar.gz
|
||||
```
|
||||
|
||||
#### 上传到 PyPI
|
||||
|
||||
- 安装 twine：
|
||||
|
||||
``` bash
|
||||
pip install twine
|
||||
```
|
||||
|
||||
- 使用 twine 上传包到 PyPI：
|
||||
|
||||
``` bash
|
||||
twine upload dist/*
|
||||
```
|
||||
|
||||
需要提供 PyPI 的用户名和密码。如果没有 PyPI 账号，可以在 PyPI 注册。
|
||||
|
||||
#### 使用 PyPI 安装
|
||||
|
||||
包上传到 PyPI 后，可以通过以下命令安装：
|
||||
|
||||
``` bash
|
||||
pip install backdoor_buster
|
||||
```
|
||||
|
||||
## 使用说明
|
||||
|
||||
1. 安装依赖:
|
||||
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
2. 执行扫描:
|
||||
1. 执行扫描:
|
||||
|
||||
```bash
|
||||
python scan.py <project_directory>
|
||||
python -m detection <project_directory> -o <path> -m <mode>
|
||||
```
|
||||
|
||||
3. 查看报告:
|
||||
2. 查看报告:
|
||||
|
||||
报告将以文本形式输出在控制台，并可选择输出到指定文件。
|
||||
|
||||
|
||||
426
detection/__main__.py
Normal file
426
detection/__main__.py
Normal file
@ -0,0 +1,426 @@
|
||||
import os
|
||||
from typing import Dict, List, Tuple, Optional
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.lib.styles import getSampleStyleSheet
|
||||
from reportlab.platypus import Paragraph, Spacer, SimpleDocTemplate
|
||||
from .Regexdetection import find_dangerous_functions
|
||||
from .GPTdetection import detectGPT
|
||||
from .pyc_detection import disassemble_pyc
|
||||
from .utils import *
|
||||
import sys
|
||||
from colorama import init, Fore, Style
|
||||
|
||||
SUPPORTED_EXTENSIONS = {".py", ".js", ".cpp", ".pyc"}
|
||||
OUTPUT_FORMATS = ["html", "md", "txt", "pdf"]
|
||||
ORDERS = [
|
||||
"__import__",
|
||||
"system",
|
||||
"exec",
|
||||
"popen",
|
||||
"eval",
|
||||
"subprocess",
|
||||
"__getattribute__",
|
||||
"getattr",
|
||||
"child_process",
|
||||
]
|
||||
|
||||
# Initialize colorama
|
||||
init(autoreset=True)
|
||||
|
||||
ORANGE = "\033[38;5;214m"
|
||||
CYAN = Fore.CYAN
|
||||
|
||||
|
||||
def supports_color() -> bool:
|
||||
"""
|
||||
Checks if the running terminal supports color output.
|
||||
|
||||
Returns:
|
||||
bool: True if the terminal supports color, False otherwise.
|
||||
"""
|
||||
# Windows support
|
||||
if sys.platform == "win32":
|
||||
return True
|
||||
# Check if output is a TTY (terminal)
|
||||
if hasattr(sys.stdout, "isatty") and sys.stdout.isatty():
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def supports_emoji() -> bool:
|
||||
"""
|
||||
Checks if the running terminal supports emoji output.
|
||||
|
||||
Returns:
|
||||
bool: True if the terminal supports emoji, False otherwise.
|
||||
"""
|
||||
# This is a simple check. Modern terminals typically support emoji.
|
||||
return sys.platform != "win32" or os.getenv("WT_SESSION") is not None
|
||||
|
||||
|
||||
def highlight_orders(line: str, risk_level: str, use_color: bool) -> str:
|
||||
"""
|
||||
Highlights specific orders in the line based on risk level.
|
||||
|
||||
Args:
|
||||
line (str): The line to highlight.
|
||||
risk_level (str): The risk level of the line ("high", "medium", "low").
|
||||
use_color (bool): Whether to use color for highlighting.
|
||||
|
||||
Returns:
|
||||
str: The highlighted line.
|
||||
"""
|
||||
risk_colors = {
|
||||
"high": Fore.RED,
|
||||
"medium": Fore.YELLOW,
|
||||
"low": CYAN,
|
||||
}
|
||||
color = risk_colors.get(risk_level, Fore.WHITE) if use_color else ""
|
||||
reset = Style.RESET_ALL if use_color else ""
|
||||
|
||||
for order in ORDERS:
|
||||
line = line.replace(order, f"{color}{order}{reset}")
|
||||
return line
|
||||
|
||||
|
||||
def generate_text_content(results: Dict[str, List[Tuple[int, str]]]) -> str:
|
||||
"""
|
||||
Generates a formatted text report for security analysis results.
|
||||
|
||||
Args:
|
||||
results (Dict[str, List[Tuple[int, str]]]): The security analysis results categorized by risk levels.
|
||||
|
||||
Returns:
|
||||
str: The formatted text report as a string.
|
||||
"""
|
||||
use_color = supports_color()
|
||||
use_emoji = supports_emoji()
|
||||
|
||||
text_output = "Security Analysis Report\n"
|
||||
text_output += "=" * 30 + "\n\n"
|
||||
|
||||
for risk_level, entries in results.items():
|
||||
if entries and risk_level != "none":
|
||||
risk_color = (
|
||||
{
|
||||
"high": Fore.RED,
|
||||
"medium": Fore.YELLOW,
|
||||
"low": Fore.GREEN,
|
||||
}.get(risk_level, Fore.WHITE)
|
||||
if use_color
|
||||
else ""
|
||||
)
|
||||
|
||||
risk_title = (
|
||||
{
|
||||
"High": "👹",
|
||||
"Medium": "👾",
|
||||
"Low": "👻",
|
||||
}
|
||||
if use_emoji
|
||||
else {
|
||||
"High": "",
|
||||
"Medium": "",
|
||||
"Low": "",
|
||||
}
|
||||
)
|
||||
|
||||
text_output += f"{risk_color}{risk_level.capitalize()} Risk{risk_title[risk_level.capitalize()]}:{Style.RESET_ALL if use_color else ''}\n"
|
||||
text_output += "-" * (len(risk_level) + 6) + "\n"
|
||||
for line_num, line in entries:
|
||||
line = highlight_orders(line, risk_level, use_color)
|
||||
line_text = f"{Style.RESET_ALL if use_color else ''} {Fore.GREEN if use_color else ''}{line_num}{Style.RESET_ALL if use_color else ''}: {line}{Style.RESET_ALL if use_color else ''}\n"
|
||||
text_output += line_text
|
||||
text_output += "\n"
|
||||
|
||||
return text_output
|
||||
|
||||
|
||||
def output_results(
|
||||
results: Dict[str, List[Tuple[int, str]]],
|
||||
output_format: str,
|
||||
output_file: Optional[str] = None,
|
||||
) -> None:
|
||||
"""
|
||||
Outputs the security analysis results in the specified format.
|
||||
|
||||
Args:
|
||||
results (Dict[str, List[Tuple[int, str]]]): The security analysis results categorized by risk levels.
|
||||
output_format (str): The format to output the results in. Supported formats: "pdf", "html", "md", "txt".
|
||||
output_file (Optional[str]): The name of the file to save the output. If None, prints to the terminal.
|
||||
"""
|
||||
OUTPUT_FORMATS = {"pdf", "html", "md", "txt"}
|
||||
|
||||
if output_file:
|
||||
file_name, file_ext = os.path.splitext(output_file)
|
||||
if output_format not in OUTPUT_FORMATS:
|
||||
output_format = "txt"
|
||||
output_file = f"{file_name}.txt"
|
||||
results_dir = os.path.dirname(output_file)
|
||||
if not os.path.exists(results_dir) and results_dir != "":
|
||||
os.makedirs(results_dir)
|
||||
if output_format == "pdf":
|
||||
output_pdf(results, output_file)
|
||||
elif output_format == "html":
|
||||
output_html(results, output_file)
|
||||
elif output_format == "md":
|
||||
output_markdown(results, output_file)
|
||||
else: # Default to txt
|
||||
output_text(results, output_file)
|
||||
else:
|
||||
# If no output file is specified, default to text output to the terminal.
|
||||
txt_output = generate_text_content(results)
|
||||
print(txt_output)
|
||||
|
||||
|
||||
def output_pdf(results: Dict[str, List[Tuple[int, str]]], file_name):
|
||||
doc = SimpleDocTemplate(file_name, pagesize=letter)
|
||||
story = []
|
||||
styles = getSampleStyleSheet()
|
||||
|
||||
# Add the title centered
|
||||
title_style = styles["Title"]
|
||||
title_style.alignment = 1 # Center alignment
|
||||
title = Paragraph("Security Analysis Report", title_style)
|
||||
story.append(title)
|
||||
story.append(Spacer(1, 20)) # Space after title
|
||||
|
||||
# Add risk levels and entries
|
||||
normal_style = styles["BodyText"]
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
story.append(
|
||||
Paragraph(f"{risk_level.capitalize()} Risk:", styles["Heading2"])
|
||||
)
|
||||
for line_num, line in entries:
|
||||
entry = Paragraph(f"Line {line_num}: {line}", normal_style)
|
||||
story.append(entry)
|
||||
story.append(Spacer(1, 12)) # Space between sections
|
||||
|
||||
doc.build(story)
|
||||
|
||||
|
||||
def output_html(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
"""
|
||||
Generates an HTML report for security analysis results.
|
||||
|
||||
Args:
|
||||
results (Dict[str, List[Tuple[int, str]]]): The security analysis results categorized by risk levels.
|
||||
file_name (Optional[str]): The name of the file to save the HTML output. If None, returns the HTML string.
|
||||
|
||||
Returns:
|
||||
Optional[str]: The HTML string if file_name is None, otherwise None.
|
||||
"""
|
||||
html_output = """
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||
<link rel="icon" href="https://s2.loli.net/2024/05/30/WDc6MekjbuCU9Qo.png">
|
||||
<title>Security Analysis Report</title>
|
||||
<style>
|
||||
body {
|
||||
background-image: url('https://s2.loli.net/2024/05/30/85Mv7leB2IRWNp6.jpg');
|
||||
background-size: 100%, auto;
|
||||
background-attachment: fixed;
|
||||
font-family: Arial, sans-serif;
|
||||
}
|
||||
h1, h2 {
|
||||
color: white;
|
||||
}
|
||||
ul {
|
||||
list-style-type: none;
|
||||
padding: 0;
|
||||
}
|
||||
li {
|
||||
background: rgba(255, 255, 255, 0.8);
|
||||
margin: 5px 0;
|
||||
padding: 10px;
|
||||
border-radius: 5px;
|
||||
}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<h1>Security Analysis Report</h1>
|
||||
"""
|
||||
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
risk_title = {
|
||||
"High": f"<h2>{risk_level.capitalize()} Risk👹</h2><ul>",
|
||||
"Medium": f"<h2>{risk_level.capitalize()} Risk👾</h2><ul>",
|
||||
"Low": f"<h2>{risk_level.capitalize()} Risk👻</h2><ul>",
|
||||
}
|
||||
html_output += risk_title[risk_level.capitalize()]
|
||||
for line_num, line in entries:
|
||||
html_output += f"<li>{line_num}: {line}</li>"
|
||||
html_output += "</ul>"
|
||||
|
||||
html_output += "</body></html>"
|
||||
|
||||
if file_name:
|
||||
with open(file_name, "w", encoding="utf-8") as file:
|
||||
file.write(html_output)
|
||||
return None
|
||||
else:
|
||||
return html_output
|
||||
|
||||
|
||||
def output_markdown(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
"""
|
||||
Generates a Markdown report for security analysis results.
|
||||
|
||||
Args:
|
||||
results (Dict[str, List[Tuple[int, str]]]): The security analysis results categorized by risk levels.
|
||||
file_name (Optional[str]): The name of the file to save the Markdown output. If None, returns the Markdown string.
|
||||
|
||||
Returns:
|
||||
Optional[str]: The Markdown string if file_name is None, otherwise None.
|
||||
"""
|
||||
md_output = "# Security Analysis Report\n\n"
|
||||
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
md_output += f"## {risk_level.capitalize()} Risk\n\n"
|
||||
md_output += "| Line Number | Description |\n"
|
||||
md_output += "|-------------|-------------|\n"
|
||||
for line_num, line in entries:
|
||||
md_output += f"| {line_num} | {line} |\n"
|
||||
md_output += "\n"
|
||||
|
||||
if file_name:
|
||||
with open(file_name, "w") as file:
|
||||
file.write(md_output)
|
||||
return None
|
||||
else:
|
||||
return md_output
|
||||
|
||||
|
||||
def output_text(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
"""
|
||||
Generates a plain text report for security analysis results.
|
||||
|
||||
Args:
|
||||
results (Dict[str, List[Tuple[int, str]]]): The security analysis results categorized by risk levels.
|
||||
file_name (Optional[str]): The name of the file to save the text output. If None, returns the text string.
|
||||
|
||||
Returns:
|
||||
Optional[str]: The text string if file_name is None, otherwise None.
|
||||
"""
|
||||
text_output = "Security Analysis Report\n"
|
||||
text_output += "=" * len("Security Analysis Report") + "\n\n"
|
||||
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
text_output += f"{risk_level.capitalize()} Risk:\n"
|
||||
text_output += "-" * len(f"{risk_level.capitalize()} Risk:") + "\n"
|
||||
for line_num, line in entries:
|
||||
text_output += f" Line {line_num}: {line}\n"
|
||||
text_output += "\n"
|
||||
|
||||
if file_name:
|
||||
with open(file_name, "w") as file:
|
||||
file.write(text_output)
|
||||
return None
|
||||
else:
|
||||
return text_output
|
||||
|
||||
|
||||
def checkModeAndDetect(mode: str, filePath: str, fileExtension: str, pycdc_addr: str):
|
||||
# TODO:添加更多方式，这里提高代码的复用性和扩展性
|
||||
if fileExtension == ".pyc":
|
||||
# 反汇编pyc文件
|
||||
file_content = disassemble_pyc(filePath, pycdc_addr)
|
||||
if mode == "regex":
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
elif mode == "llm":
|
||||
return detectGPT(file_content)
|
||||
else:
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
else:
|
||||
file_content = read_file_content(filePath)
|
||||
if mode == "regex":
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
elif mode == "llm":
|
||||
return detectGPT(file_content)
|
||||
else:
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
|
||||
|
||||
def process_path(
|
||||
path: str, output_format: str, mode: str, pycdc_addr: str, output_file=None
|
||||
):
|
||||
results = {"high": [], "medium": [], "low": [], "none": []}
|
||||
if os.path.isdir(path):
|
||||
for root, dirs, files in os.walk(path):
|
||||
for file in files:
|
||||
file_extension = os.path.splitext(file)[1]
|
||||
if file_extension in SUPPORTED_EXTENSIONS:
|
||||
file_path = os.path.join(root, file)
|
||||
file_results = checkModeAndDetect(
|
||||
mode, file_path, file_extension, pycdc_addr
|
||||
)
|
||||
for key in file_results:
|
||||
if key != "none": # Exclude 'none' risk level
|
||||
results[key].extend(
|
||||
[
|
||||
(f"{file_path}: Line {line_num}", line)
|
||||
for line_num, line in file_results[key]
|
||||
]
|
||||
)
|
||||
elif os.path.isfile(path):
|
||||
file_extension = os.path.splitext(path)[1]
|
||||
if file_extension in SUPPORTED_EXTENSIONS:
|
||||
file_results = checkModeAndDetect(mode, path, file_extension, pycdc_addr)
|
||||
for key in file_results:
|
||||
if key != "none": # Exclude 'none' risk level
|
||||
results[key].extend(
|
||||
[
|
||||
(f"{path}: Line {line_num}", line)
|
||||
for line_num, line in file_results[key]
|
||||
]
|
||||
)
|
||||
else:
|
||||
print("Unsupported file type.")
|
||||
return
|
||||
else:
|
||||
print("Invalid path.")
|
||||
sys.exit(1)
|
||||
|
||||
output_results(results, output_format, output_file)
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Backdoor detection tool.", prog="detection"
|
||||
)
|
||||
parser.add_argument("path", help="Path to the code to analyze")
|
||||
parser.add_argument("-o", "--output", help="Output file path", default=None)
|
||||
parser.add_argument(
|
||||
"-m", "--mode", help="Mode of operation:[regex,llm]", default="regex"
|
||||
)
|
||||
parser.add_argument(
|
||||
"-p", "--pycdc", help="Path to pycdc.exe to decompile", default=None
|
||||
)
|
||||
args = parser.parse_args()
|
||||
output_format = "txt" # Default output format
|
||||
output_file = None
|
||||
if args.output:
|
||||
_, ext = os.path.splitext(args.output)
|
||||
ext = ext.lower()
|
||||
if ext in [".html", ".md", ".txt", ".pdf"]:
|
||||
output_format = ext.replace(".", "")
|
||||
output_file = args.output
|
||||
else:
|
||||
print(
|
||||
"Your input file format was incorrect, the output has been saved as a TXT file."
|
||||
)
|
||||
output_file = args.output.rsplit(".", 1)[0] + ".txt"
|
||||
# 如果未指定输出文件，则输出到 stdout；否则写入文件
|
||||
process_path(args.path, output_format, args.mode, args.pycdc, output_file)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -1,210 +0,0 @@
|
||||
import os
|
||||
from typing import Dict, List, Tuple
|
||||
from reportlab.lib.pagesizes import letter
|
||||
from reportlab.lib.styles import getSampleStyleSheet
|
||||
from reportlab.platypus import Paragraph, Spacer, SimpleDocTemplate
|
||||
from .Regexdetection import find_dangerous_functions
|
||||
from .GPTdetection import detectGPT
|
||||
from .utils import *
|
||||
from .pyc_detection import disassemble_pyc
|
||||
import sys
|
||||
|
||||
SUPPORTED_EXTENSIONS = {".py", ".js", ".cpp", ".pyc"}
|
||||
OUTPUT_FORMATS = ["html", "md", "txt", "pdf"]
|
||||
|
||||
|
||||
def generate_text_content(results):
|
||||
text_output = "Security Analysis Report\n"
|
||||
for risk_level, entries in results.items():
|
||||
if entries and risk_level != "none":
|
||||
text_output += f"{risk_level.capitalize()} Risk:\n"
|
||||
for line_num, line in entries:
|
||||
text_output += f" Line {line_num}: {line}\n"
|
||||
return text_output
|
||||
|
||||
|
||||
def output_results(results, output_format, output_file=None):
|
||||
if output_file:
|
||||
file_name = os.path.splitext(output_file)
|
||||
if output_format not in OUTPUT_FORMATS:
|
||||
output_format = "txt"
|
||||
output_file = f"{file_name}.txt"
|
||||
results_dir = os.path.dirname(output_file)
|
||||
if not os.path.exists(results_dir):
|
||||
os.makedirs(results_dir)
|
||||
if output_format == "pdf":
|
||||
output_pdf(results, output_file)
|
||||
elif output_format == "html":
|
||||
output_html(results, output_file)
|
||||
elif output_format == "md":
|
||||
output_markdown(results, output_file)
|
||||
else: # Default to txt
|
||||
output_text(results, output_file)
|
||||
else:
|
||||
# If no output file is specified, default to text output to the terminal.
|
||||
txt_output = generate_text_content(results)
|
||||
print(txt_output)
|
||||
|
||||
|
||||
def output_pdf(results: Dict[str, List[Tuple[int, str]]], file_name):
|
||||
doc = SimpleDocTemplate(file_name, pagesize=letter)
|
||||
story = []
|
||||
styles = getSampleStyleSheet()
|
||||
|
||||
# Add the title centered
|
||||
title_style = styles["Title"]
|
||||
title_style.alignment = 1 # Center alignment
|
||||
title = Paragraph("Security Analysis Report", title_style)
|
||||
story.append(title)
|
||||
story.append(Spacer(1, 20)) # Space after title
|
||||
|
||||
# Add risk levels and entries
|
||||
normal_style = styles["BodyText"]
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
story.append(
|
||||
Paragraph(f"{risk_level.capitalize()} Risk:", styles["Heading2"])
|
||||
)
|
||||
for line_num, line in entries:
|
||||
entry = Paragraph(f"Line {line_num}: {line}", normal_style)
|
||||
story.append(entry)
|
||||
story.append(Spacer(1, 12)) # Space between sections
|
||||
|
||||
doc.build(story)
|
||||
|
||||
|
||||
def output_html(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
html_output = "<html><head><title>Security Analysis Report</title></head><body>"
|
||||
html_output += "<h1>Security Analysis Report</h1>"
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
html_output += f"<h2>{risk_level.capitalize()} Risk</h2><ul>"
|
||||
for line_num, line in entries:
|
||||
html_output += f"<li>{line_num}: {line}</li>"
|
||||
html_output += "</ul>"
|
||||
html_output += "</body></html>"
|
||||
if file_name:
|
||||
with open(file_name, "w") as file:
|
||||
file.write(html_output)
|
||||
else:
|
||||
return html_output
|
||||
|
||||
|
||||
def output_markdown(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
md_output = "# Security Analysis Report\n"
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
md_output += f"## {risk_level.capitalize()} Risk\n"
|
||||
for line_num, line in entries:
|
||||
md_output += f"- {line_num}: {line}\n"
|
||||
if file_name:
|
||||
with open(file_name, "w") as file:
|
||||
file.write(md_output)
|
||||
else:
|
||||
return md_output
|
||||
|
||||
|
||||
def output_text(results: Dict[str, List[Tuple[int, str]]], file_name=None):
|
||||
text_output = "Security Analysis Report\n"
|
||||
for risk_level, entries in results.items():
|
||||
if risk_level != "none":
|
||||
text_output += f"{risk_level.capitalize()} Risk:\n"
|
||||
for line_num, line in entries:
|
||||
text_output += f" {line_num}: {line}\n"
|
||||
if file_name:
|
||||
with open(file_name, "w") as file:
|
||||
file.write(text_output)
|
||||
else:
|
||||
return text_output
|
||||
|
||||
|
||||
def checkModeAndDetect(mode: str, filePath: str, fileExtension: str):
|
||||
# TODO:添加更多方式，这里提高代码的复用性和扩展性
|
||||
if fileExtension == ".pyc":
|
||||
# 反汇编pyc文件
|
||||
file_content = disassemble_pyc(filePath)
|
||||
if mode == "regex":
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
elif mode == "llm":
|
||||
return detectGPT(file_content)
|
||||
else:
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
else:
|
||||
file_content = read_file_content(filePath)
|
||||
if mode == "regex":
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
elif mode == "llm":
|
||||
return detectGPT(file_content)
|
||||
else:
|
||||
return find_dangerous_functions(file_content, fileExtension)
|
||||
|
||||
|
||||
def process_path(path: str, output_format: str, mode: str, output_file=None):
|
||||
results = {"high": [], "medium": [], "low": [], "none": []}
|
||||
if os.path.isdir(path):
|
||||
for root, dirs, files in os.walk(path):
|
||||
for file in files:
|
||||
file_extension = os.path.splitext(file)[1]
|
||||
if file_extension in SUPPORTED_EXTENSIONS:
|
||||
file_path = os.path.join(root, file)
|
||||
|
||||
file_results = checkModeAndDetect(mode, file_path, file_extension)
|
||||
for key in file_results:
|
||||
if key != "none": # Exclude 'none' risk level
|
||||
results[key].extend(
|
||||
[
|
||||
(f"{file_path}: Line {line_num}", line)
|
||||
for line_num, line in file_results[key]
|
||||
]
|
||||
)
|
||||
elif os.path.isfile(path):
|
||||
file_extension = os.path.splitext(path)[1]
|
||||
if file_extension in SUPPORTED_EXTENSIONS:
|
||||
file_results = checkModeAndDetect(mode, path, file_extension)
|
||||
for key in file_results:
|
||||
if key != "none": # Exclude 'none' risk level
|
||||
results[key].extend(
|
||||
[
|
||||
(f"{path}: Line {line_num}", line)
|
||||
for line_num, line in file_results[key]
|
||||
]
|
||||
)
|
||||
else:
|
||||
print("Unsupported file type.")
|
||||
return
|
||||
else:
|
||||
print("Invalid path.")
|
||||
sys.exit(1)
|
||||
|
||||
output_results(results, output_format, output_file)
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
|
||||
parser = argparse.ArgumentParser(description="Backdoor detection tool.")
|
||||
parser.add_argument("path", help="Path to the code to analyze")
|
||||
parser.add_argument("-o", "--output", help="Output file path", default=None)
|
||||
parser.add_argument(
|
||||
"-m", "--mode", help="Mode of operation:[regex,llm]", default="regex"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
output_format = "txt" # Default output format
|
||||
output_file = None
|
||||
if args.output:
|
||||
_, ext = os.path.splitext(args.output)
|
||||
ext = ext.lower()
|
||||
if ext in [".html", ".md", ".txt", ".pdf"]:
|
||||
output_format = ext.replace(".", "")
|
||||
output_file = args.output
|
||||
else:
|
||||
print(
|
||||
"Your input file format was incorrect, the output has been saved as a TXT file."
|
||||
)
|
||||
output_file = args.output.rsplit(".", 1)[0] + ".txt"
|
||||
# 如果未指定输出文件，则输出到 stdout；否则写入文件
|
||||
process_path(args.path, output_format, args.mode, output_file)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -1,9 +1,36 @@
|
||||
from typing import List, Tuple
|
||||
import uncompyle6
|
||||
import io
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
|
||||
def disassemble_pyc(file_path: str) -> str:
|
||||
def run_pycdc(exe_path: str, pyc_file: str) -> str:
|
||||
"""
|
||||
Executes pycdc.exe with the given .pyc file using a command line string and captures the output.
|
||||
|
||||
Args:
|
||||
exe_path (str): Path to the pycdc.exe executable.
|
||||
pyc_file (str): Path to the .pyc file to decompile.
|
||||
|
||||
Returns:
|
||||
str: Output from pycdc.exe.
|
||||
"""
|
||||
if not os.path.isfile(exe_path):
|
||||
print(f"ERROR: The specified pycdc.exe path is not valid: {exe_path}")
|
||||
print("Please check your pycdc path.")
|
||||
exit(1)
|
||||
|
||||
command = f'"{exe_path}" "{pyc_file}"'
|
||||
result = subprocess.run(command, capture_output=True, text=True, shell=True)
|
||||
|
||||
if result.returncode != 0:
|
||||
raise Exception(f"Error running pycdc.exe: {result.stderr}")
|
||||
|
||||
return result.stdout
|
||||
|
||||
|
||||
def disassemble_pyc(file_path: str, pycdc_addr=None) -> str:
|
||||
"""
|
||||
Disassembles a .pyc file using uncompyle6.
|
||||
|
||||
@ -18,5 +45,11 @@ def disassemble_pyc(file_path: str) -> str:
|
||||
uncompyle6.main.decompile_file(file_path, output)
|
||||
return output.getvalue()
|
||||
except Exception as e:
|
||||
print(f"Error occurred while disassembling: {e}")
|
||||
return ""
|
||||
if pycdc_addr is None:
|
||||
print(
|
||||
"ERROR: For Python 3.11 and above, you need to install pycdc and compile it yourself to obtain pycdc.exe."
|
||||
)
|
||||
print("repo: https://github.com/zrax/pycdc.git")
|
||||
exit(1)
|
||||
else:
|
||||
return run_pycdc(pycdc_addr, file_path)
|
||||
|
||||
@ -1,9 +1,64 @@
|
||||
# 项目设计文档 - 后门检测系统
|
||||
|
||||
## 打包
|
||||
|
||||
### 简介
|
||||
|
||||
本项目需要将 Python 代码打包成`pip`包和`deb`包，以便于分发和安装。以下是如何实现和使用该打包功能的详细步骤。
|
||||
|
||||
### pip
|
||||
|
||||
#### 打包命令
|
||||
|
||||
```bash
|
||||
pip install wheel
|
||||
python setup.py sdist bdist_wheel
|
||||
```
|
||||
|
||||
执行上述命令后，会在 dist 目录下生成 .tar.gz 和 .whl 文件。
|
||||
|
||||
#### 本地安装
|
||||
|
||||
- 安装 .whl 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0-py3-none-any.whl
|
||||
```
|
||||
|
||||
- 安装 .tar.gz 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0.tar.gz
|
||||
```
|
||||
|
||||
#### 上传到 PyPI
|
||||
|
||||
- 安装 twine：
|
||||
|
||||
``` bash
|
||||
pip install twine
|
||||
```
|
||||
|
||||
- 使用 twine 上传包到 PyPI：
|
||||
|
||||
``` bash
|
||||
twine upload dist/*
|
||||
```
|
||||
|
||||
需要提供 PyPI 的用户名和密码。如果没有 PyPI 账号，可以在 PyPI 注册。
|
||||
|
||||
#### 使用 PyPI 安装
|
||||
|
||||
包上传到 PyPI 后，可以通过以下命令安装：
|
||||
|
||||
``` bash
|
||||
pip install backdoor_buster
|
||||
```
|
||||
|
||||
## 静态代码后门检测
|
||||
|
||||
**功能描述**:
|
||||
这个脚本用于扫描指定路径下的代码文件，检测潜在的危险函数调用，支持 `.py`, `.js`, `.cpp` 文件。
|
||||
这个脚本用于扫描指定路径下的代码文件，检测潜在的危险函数调用，支持 `.py`, `.js`, `.cpp`, `.pyc` 文件。
|
||||
|
||||
**主要组件**:
|
||||
|
||||
@ -67,7 +122,7 @@ python backdoor_detection.py ./src -o ./output/report.pdf
|
||||
**使用示例**:
|
||||
|
||||
```bash
|
||||
python requirements_detection.py ./requirements.txt -o ./output/report.md
|
||||
python -m detection.requirements_detection ./requirements.txt -o ./output/report.md
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
@ -46,7 +46,18 @@
|
||||
|
||||
- **主要应用**：通过爬虫收集漏洞依赖信息并进行汇总，用于判断依赖是否存在漏洞版本。
|
||||
|
||||
## 8. 代码和风险分析
|
||||
## 8. 打包
|
||||
|
||||
本项目支持打包作为`pip`包进行发布
|
||||
|
||||
- **主要应用**：
|
||||
- `pip`通过`wheel`并自行撰写`setup.py`以及`MANIFEST.in`，将项目打包发布
|
||||
|
||||
## 9. 反汇编
|
||||
|
||||
项目通过`uncompyle6`库提供的反汇编模块可以实现对python字节码进行反汇编之后扫描危险代码
|
||||
|
||||
## 10. 代码和风险分析
|
||||
|
||||
项目中实现了基本的静态代码分析功能，用于识别和报告潜在的安全风险函数调用，如 `system`、`exec` 等。
|
||||
|
||||
|
||||
@ -2,31 +2,68 @@
|
||||
|
||||
本文档提供了后门检测系统的使用方法，包括依赖版本漏洞检测和静态代码后门检测两部分。这将帮助用户正确执行安全检测，并理解输出结果。
|
||||
|
||||
## 安装需求
|
||||
|
||||
在开始使用本系统之前，请确保您的环境中安装了以下依赖：
|
||||
|
||||
- Python 3.6 或更高版本
|
||||
- `packaging` 库：用于版本控制和比较
|
||||
- `reportlab` 库：用于生成 PDF 报告
|
||||
|
||||
您可以通过以下命令安装必要的 Python 库：
|
||||
|
||||
```bash
|
||||
pip install packaging reportlab
|
||||
```
|
||||
|
||||
## 下载和配置
|
||||
|
||||
- 克隆或下载后门检测系统到您的本地环境。
|
||||
- 确保脚本文件 (`requirements_detection.py` 和 `backdoor_detection.py`) 在您的工作目录中。
|
||||
|
||||
## 打包
|
||||
|
||||
### pip
|
||||
|
||||
#### 打包命令
|
||||
|
||||
```bash
|
||||
pip install wheel
|
||||
python setup.py sdist bdist_wheel
|
||||
```
|
||||
|
||||
执行上述命令后，会在 dist 目录下生成 .tar.gz 和 .whl 文件。
|
||||
|
||||
#### 本地安装
|
||||
|
||||
- 安装 .whl 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0-py3-none-any.whl
|
||||
```
|
||||
|
||||
- 安装 .tar.gz 文件：
|
||||
|
||||
``` bash
|
||||
pip install dist/backdoor_buster-0.1.0.tar.gz
|
||||
```
|
||||
|
||||
#### 上传到 PyPI
|
||||
|
||||
- 安装 twine：
|
||||
|
||||
``` bash
|
||||
pip install twine
|
||||
```
|
||||
|
||||
- 使用 twine 上传包到 PyPI：
|
||||
|
||||
``` bash
|
||||
twine upload dist/*
|
||||
```
|
||||
|
||||
需要提供 PyPI 的用户名和密码。如果没有 PyPI 账号，可以在 PyPI 注册。
|
||||
|
||||
#### 使用 PyPI 安装
|
||||
|
||||
包上传到 PyPI 后，可以通过以下命令安装：
|
||||
|
||||
``` bash
|
||||
pip install backdoor_buster
|
||||
```
|
||||
|
||||
## 运行依赖版本漏洞检测脚本
|
||||
|
||||
**命令格式**：
|
||||
|
||||
```bash
|
||||
python requirements_detection.py <requirements_file> -o <output_file>
|
||||
python -m detection.requirements_detection <requirements_file> -o <output_file>
|
||||
```
|
||||
|
||||
**参数说明**：
|
||||
@ -37,7 +74,7 @@ python requirements_detection.py <requirements_file> -o <output_file>
|
||||
**示例**：
|
||||
|
||||
```bash
|
||||
python requirements_detection.py requirements.txt -o output/report.md
|
||||
python -m detection.requirements_detection requirements.txt -o output/report.md
|
||||
```
|
||||
|
||||
## 运行静态代码后门检测脚本
|
||||
@ -45,7 +82,7 @@ python requirements_detection.py requirements.txt -o output/report.md
|
||||
**命令格式**：
|
||||
|
||||
```bash
|
||||
python backdoor_detection.py <code_path> -o <output_file> -m <mode>
|
||||
python -m detection <code_path> -o <output_file> -m <mode>
|
||||
```
|
||||
|
||||
**参数说明**：
|
||||
@ -57,7 +94,7 @@ python backdoor_detection.py <code_path> -o <output_file> -m <mode>
|
||||
**示例**：
|
||||
|
||||
```bash
|
||||
python backdoor_detection.py ./src -o output/report.pdf -m regex
|
||||
python -m detection ./src -o output/report.pdf -m regex
|
||||
```
|
||||
|
||||
## 结果解读
|
||||
|
||||
43
setup.py
Normal file
43
setup.py
Normal file
@ -0,0 +1,43 @@
|
||||
# pip install wheel
|
||||
# python setup.py sdist bdist_wheel
|
||||
|
||||
from setuptools import setup, find_packages
|
||||
|
||||
|
||||
def read_file(filename: str) -> str:
|
||||
"""Read a file and return its content as a string.
|
||||
|
||||
Args:
|
||||
filename (str): The name of the file to read.
|
||||
|
||||
Returns:
|
||||
str: The content of the file.
|
||||
"""
|
||||
with open(filename, encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
setup(
|
||||
name="backdoor_buster",
|
||||
version="0.1.0",
|
||||
author="ciscn",
|
||||
description="A tool for integrated backdoor detection",
|
||||
long_description=read_file("README.md"),
|
||||
long_description_content_type="text/markdown",
|
||||
url="https://git.mamahaha.work/sangge/BackDoorBuster",
|
||||
packages=find_packages(),
|
||||
classifiers=[
|
||||
"Programming Language :: Python :: 3",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Operating System :: OS Independent",
|
||||
],
|
||||
python_requires=">=3.6",
|
||||
install_requires=[
|
||||
"reportlab",
|
||||
"requests",
|
||||
"packaging",
|
||||
"openai",
|
||||
"bs4",
|
||||
"uncompyle6",
|
||||
],
|
||||
)
|
||||
@ -1,7 +1,7 @@
|
||||
import unittest
|
||||
import warnings
|
||||
|
||||
from detection.backdoor_detection import find_dangerous_functions
|
||||
from detection.__main__ import find_dangerous_functions
|
||||
from detection.GPTdetection import detectGPT
|
||||
import os
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user