# vulnerable_parser.py
import sys
import json

def parse_data(input_data):
    """
    一個(gè)存在安全問題的解析函數(shù)
    1. 可能引發(fā)異常的處理邏輯
    2. 可能存在代碼注入漏洞
    3. 可能存在正則表達(dá)式拒絕服務(wù)(ReDoS)
    """
    try:
        # 模擬不安全的數(shù)據(jù)處理
        if input_data.startswith("{"):
            # 嘗試解析為JSON - 可能引發(fā)異常
            data = json.loads(input_data)
            return f"Parsed JSON: {data}"
        elif "system" in input_data:
            # 模擬命令注入漏洞
            import os
            # 危險(xiǎn)操作：實(shí)際應(yīng)用中絕不應(yīng)該這樣做！
            result = os.popen(f"echo {input_data}").read()
            return f"Command result: {result}"
        elif "calc" in input_data:
            # 模擬不安全的eval使用
            return f"Calculation: {eval(input_data)}"
        else:
            # 模擬正則表達(dá)式ReDoS漏洞
            import re
            # 危險(xiǎn)的正則模式，可能造成ReDoS
            pattern = r"^(a+)+$"
            if re.match(pattern, input_data):
                return "Pattern matched"
            return f"Processed: {input_data}"
    except Exception as e:
        return f"Error: {str(e)}"

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python vulnerable_parser.py <input_file>")
        sys.exit(1)
    
    with open(sys.argv[1], 'r', encoding='utf-8', errors='ignore') as f:
        input_data = f.read().strip()
    
    result = parse_data(input_data)
    print(result)

# 安裝python-afl
pip install python-afl

# 使用AFL++運(yùn)行Python Fuzz測(cè)試
PYTHON_AFL_PERSISTENT=1 python -m afl.vulnerable_parser

# fuzz_parser.py
import atheris
import sys

with atheris.instrument_imports():
    import vulnerable_parser

def TestOneInput(data):
    try:
        # 將輸入數(shù)據(jù)轉(zhuǎn)換為字符串
        input_str = data.decode("utf-8")
        vulnerable_parser.parse_data(input_str)
    except (UnicodeDecodeError, ValueError):
        # 忽略編碼錯(cuò)誤，這是正常的Fuzz行為
        pass
    except Exception as e:
        # 這里可以記錄異常，但通常讓Fuzzer自動(dòng)處理
        pass

if __name__ == "__main__":
    atheris.Setup(sys.argv, TestOneInput)
    atheris.Fuzz()

pip install atheris
python fuzz_parser.py

# test_with_hypothesis.py
from hypothesis import given, strategies as st
import vulnerable_parser

@given(st.text())
def test_parse_data_with_random_input(text):
    """使用隨機(jī)生成的文本測(cè)試解析器"""
    try:
        result = vulnerable_parser.parse_data(text)
        # 這里可以添加一些斷言
        assert result is not None
    except Exception:
        # 對(duì)于Fuzz測(cè)試，異常是可以接受的
        pass

if __name__ == "__main__":
    test_parse_data_with_random_input()

pip install hypothesis
python test_with_hypothesis.py

# simple_fuzzer.py
import random
import string
import subprocess
import os

def generate_random_input(length=100):
    """生成隨機(jī)輸入"""
    # 生成基本隨機(jī)字符串
    chars = string.ascii_letters + string.digits + string.punctuation + " "
    result = ''.join(random.choice(chars) for _ in range(length))
    
    # 有時(shí)添加一些特殊模式
    if random.random() < 0.3:
        result = "{" + result + "}"  # 模擬JSON
    if random.random() < 0.1:
        result = "a" * length  # 可能觸發(fā)ReDoS
    if random.random() < 0.05:
        result = "system(" + result + ")"  # 可能觸發(fā)命令注入
        
    return result

def run_fuzzer():
    """運(yùn)行Fuzzer"""
    crash_count = 0
    hang_count = 0
    total_tests = 10000
    
    for i in range(total_tests):
        # 生成測(cè)試輸入
        test_input = generate_random_input(random.randint(1, 1000))
        
        # 寫入臨時(shí)文件
        with open("temp_input.txt", "w") as f:
            f.write(test_input)
        
        try:
            # 運(yùn)行目標(biāo)程序，設(shè)置超時(shí)
            result = subprocess.run(
                ["python", "vulnerable_parser.py", "temp_input.txt"],
                capture_output=True,
                text=True,
                timeout=5  # 5秒超時(shí)
            )
            
            # 檢查結(jié)果
            if result.returncode != 0:
                crash_count += 1
                print(f"Crash found with input: {test_input[:100]}...")
                print(f"Error: {result.stderr}")
                # 保存崩潰輸入以供分析
                with open(f"crash_{crash_count}.txt", "w") as f:
                    f.write(test_input)
                    
        except subprocess.TimeoutExpired:
            hang_count += 1
            print(f"Timeout with input: {test_input[:100]}...")
            # 保存導(dǎo)致超時(shí)的輸入
            with open(f"hang_{hang_count}.txt", "w") as f:
                f.write(test_input)
    
    # 清理
    if os.path.exists("temp_input.txt"):
        os.remove("temp_input.txt")
        
    print(f"Fuzzing completed. Total tests: {total_tests}")
    print(f"Crashes: {crash_count}, Hangs: {hang_count}")

if __name__ == "__main__":
    run_fuzzer()

# fixed_parser.py
import sys
import json
import re

def safe_parse_data(input_data):
    """
    修復(fù)后的安全解析函數(shù)
    """
    # 限制輸入大小
    if len(input_data) > 1000:
        return "Error: Input too long"
    
    try:
        if input_data.startswith("{"):
            # 安全地解析JSON
            data = json.loads(input_data)
            return f"Parsed JSON: {data}"
        else:
            # 使用更安全的正則表達(dá)式
            pattern = r"^a+$"  # 簡(jiǎn)化模式，避免ReDoS
            if re.match(pattern, input_data):
                return "Pattern matched"
            return f"Processed: {input_data}"
    except json.JSONDecodeError:
        return "Error: Invalid JSON"
    except Exception as e:
        return f"Error: {str(e)}"

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python fixed_parser.py <input_file>")
        sys.exit(1)
    
    try:
        with open(sys.argv[1], 'r', encoding='utf-8', errors='ignore') as f:
            input_data = f.read().strip()
        
        # 移除危險(xiǎn)功能（命令執(zhí)行、eval等）
        if any(cmd in input_data for cmd in ["system", "eval", "exec", "os.", "subprocess"]):
            print("Error: Suspicious input detected")
            sys.exit(1)
            
        result = safe_parse_data(input_data)
        print(result)
    except Exception as e:
        print(f"Error: {str(e)}")