💻 安全编码 — SAST/DAST扫描结果修复

📖 安全编码的重要性

安全漏洞的修复成本随着开发阶段推进呈指数级增长——设计阶段修复一个漏洞的成本是1x，测试阶段是10x，上线后是100x。安全编码（Secure Coding）从源头预防漏洞，是性价比最高的安全投资。

🔎 SAST静态扫描实战

使用Semgrep扫描Python代码

# ===== 安装Semgrep =====
pip3 install semgrep 2>/dev/null || true
which semgrep &>/dev/null && echo "Semgrep已安装" || \
  echo "Semgrep未安装，使用替代方案"

# ===== 创建含漏洞的示例代码 =====
mkdir -p /tmp/vuln-app
cat > /tmp/vuln-app/app.py <<'PYEOF'
from flask import Flask, request, render_template_string
import subprocess
import sqlite3
import pickle

app = Flask(__name__)

# 漏洞1: SQL注入
@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    conn = sqlite3.connect('users.db')
    query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
    result = conn.execute(query)
    return "OK" if result.fetchone() else "Failed"

# 漏洞2: 命令注入
@app.route('/ping')
def ping():
    host = request.args.get('host')
    output = subprocess.check_output(f"ping -c 1 {host}", shell=True)
    return output

# 漏洞3: SSTI (服务端模板注入)
@app.route('/greet')
def greet():
    name = request.args.get('name', 'World')
    template = f"<h1>Hello {name}!</h1>"
    return render_template_string(template)

# 漏洞4: 不安全的反序列化
@app.route('/deserialize')
def deserialize():
    data = request.args.get('data')
    obj = pickle.loads(data.encode())
    return str(obj)

# 漏洞5: 硬编码密码
DB_PASSWORD = "admin123!@#"
API_KEY = "sk-1234567890abcdef"

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0')
PYEOF

echo "[+] 含漏洞的示例代码已创建"

# ===== 手动漏洞识别 =====
echo -e "\n=== 漏洞清单 ==="
echo "1. SQL注入: f-string直接拼接用户输入"
echo "2. 命令注入: shell=True + 用户输入"
echo "3. SSTI: render_template_string + 用户输入"
echo "4. 不安全反序列化: pickle.loads + 用户输入"
echo "5. 硬编码密码: 源码中明文存储密钥"

🔧 漏洞修复实战

逐个修复SAST扫描发现的漏洞，展示安全编码的正确做法。

修复1: SQL注入 → 参数化查询

# ===== 不安全代码 =====
# query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'"
# conn.execute(query)

# ===== 安全代码 =====
cat > /tmp/vuln-app/fix_sql.py <<'PYEOF'
import sqlite3
from werkzeug.security import check_password_hash

def login_safe(username, password):
    """使用参数化查询防止SQL注入"""
    conn = sqlite3.connect('users.db')
    # ✅ 参数化查询: ?占位符，参数作为元组传入
    query = "SELECT username, password_hash FROM users WHERE username = ?"
    result = conn.execute(query, (username,))
    row = result.fetchone()
    if row and check_password_hash(row[1], password):
        return True
    return False
PYEOF

echo "[+] SQL注入修复: 参数化查询 + 密码哈希验证"

修复2: 命令注入 → 安全API

# ===== 不安全代码 =====
# output = subprocess.check_output(f"ping -c 1 {host}", shell=True)

# ===== 安全代码 =====
cat > /tmp/vuln-app/fix_cmd.py <<'PYEOF'
import subprocess
import re

def ping_safe(host):
    """安全执行ping命令"""
    # ✅ 输入验证: 仅允许合法IP/域名
    if not re.match(r'^[a-zA-Z0-9.\-]+$', host):
        raise ValueError("Invalid host format")
    
    # ✅ 使用列表参数（不经过shell解析）
    result = subprocess.run(
        ['ping', '-c', '1', '-W', '3', host],
        capture_output=True, text=True, timeout=10
    )
    return result.stdout
PYEOF

echo "[+] 命令注入修复: 输入验证 + 列表参数"

修复3: SSTI → 安全模板渲染

# ===== 不安全代码 =====
# template = f"<h1>Hello {name}!</h1>"
# return render_template_string(template)

# ===== 安全代码 =====
cat > /tmp/vuln-app/fix_ssti.py <<'PYEOF'
from flask import render_template_string, escape

# ✅ 方法1: 使用Jinja2自动转义
TEMPLATE = "<h1>Hello {{ name|e }}!</h1>"

def greet_safe(name):
    """安全模板渲染"""
    # Jinja2的|e过滤器自动转义HTML
    return render_template_string(TEMPLATE, name=name)

# ✅ 方法2: 直接转义用户输入
def greet_safe2(name):
    return f"<h1>Hello {escape(name)}!</h1>"
PYEOF

echo "[+] SSTI修复: Jinja2自动转义 + escape()"

修复4: 不安全反序列化 → 安全替代方案

# ===== 不安全代码 =====
# obj = pickle.loads(data)

# ===== 安全代码 =====
cat > /tmp/vuln-app/fix_deserialize.py <<'PYEOF'
import json

# ✅ 使用JSON替代pickle（只支持基本数据类型）
def deserialize_safe(data):
    """安全反序列化"""
    try:
        return json.loads(data)
    except json.JSONDecodeError:
        raise ValueError("Invalid JSON data")

# ✅ 如果必须使用pickle，限制白名单
import pickle
class SafeUnpickler(pickle.Unpickler):
    ALLOWED_CLASSES = {
        'builtins': {'dict', 'list', 'set', 'tuple', 'str', 'int', 'float', 'bool'}
    }
    
    def find_class(self, module, name):
        if module in self.ALLOWED_CLASSES:
            if name in self.ALLOWED_CLASSES[module]:
                return getattr(__import__(module), name)
        raise pickle.UnpicklingError(f"Forbidden: {module}.{name}")
PYEOF

echo "[+] 反序列化修复: JSON替代 + Pickle白名单"

修复5: 硬编码密码 → 环境变量/密钥管理

# ===== 不安全代码 =====
# DB_PASSWORD = "admin123!@#"
# API_KEY = "sk-1234567890abcdef"

# ===== 安全代码 =====
cat > /tmp/vuln-app/fix_secrets.py <<'PYEOF'
import os

# ✅ 方法1: 环境变量
DB_PASSWORD = os.environ.get('DB_PASSWORD')
API_KEY = os.environ.get('API_KEY')

# ✅ 方法2: .env文件（开发环境）
from dotenv import load_dotenv
load_dotenv()  # 从.env文件加载
DB_PASSWORD = os.environ.get('DB_PASSWORD')

# ✅ 方法3: 密钥管理服务（生产环境）
# AWS Secrets Manager / Azure Key Vault / HashiCorp Vault
# import boto3
# client = boto3.client('secretsmanager')
# response = client.get_secret_value(SecretId='prod/db/password')
# password = response['SecretString']

# ✅ 启动时检查
if not DB_PASSWORD:
    raise RuntimeError("DB_PASSWORD environment variable not set")
PYEOF

echo "[+] 硬编码密码修复: 环境变量 + 密钥管理服务"

🚀 DAST动态扫描实战

# ===== 使用OWASP ZAP进行DAST扫描 =====
# 安装ZAP（Docker版本）
# docker run -t owasp/zap2docker-stable zap-baseline.py -t http://target:8080

# ===== 使用Nuclei进行轻量DAST =====
# 安装nuclei
apt-get install -y nuclei 2>/dev/null || true

# ===== 创建安全测试靶场 =====
cat > /tmp/vuln-app/server.py <<'PYEOF'
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class TestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 安全响应头
        headers = {
            'X-Content-Type-Options': 'nosniff',
            'X-Frame-Options': 'DENY',
            'X-XSS-Protection': '1; mode=block',
            'Content-Security-Policy': "default-src 'self'",
            'Strict-Transport-Security': 'max-age=31536000'
        }
        for k, v in headers.items():
            self.send_header(k, v)
        
        self.send_response(200)
        self.send_header('Content-Type', 'text/html')
        self.end_headers()
        self.wfile.write(b'<h1>Secure App</h1>')
    
    def log_message(self, format, *args):
        print(f"[ZAP-TEST] {args[0]}")

if __name__ == '__main__':
    server = HTTPServer(('0.0.0.0', 9999), TestHandler)
    print("Test server running on :9999")
    server.handle_request()  # Handle one request then exit
PYEOF

# 启动测试服务器并检查安全头
python3 /tmp/vuln-app/server.py &
sleep 1

echo "=== DAST安全头检测 ==="
curl -sI http://127.0.0.1:9999 | grep -E "X-|Content-Security|Strict-Transport"

# 清理
kill %1 2>/dev/null

DAST扫描结果解读

# ===== 模拟DAST扫描报告 =====
cat > /tmp/dast_report.sh <<'SCRIPT'
#!/bin/bash
echo "========================================"
echo "  DAST扫描报告"
echo "========================================"

echo -e "\n[1] 安全头检测结果"
echo "  ✅ X-Content-Type-Options: nosniff"
echo "  ✅ X-Frame-Options: DENY"
echo "  ✅ X-XSS-Protection: 1; mode=block"
echo "  ✅ Content-Security-Policy: default-src 'self'"
echo "  ✅ Strict-Transport-Security: max-age=31536000"
echo "  ❌ Referrer-Policy: 缺失"
echo "  ❌ Permissions-Policy: 缺失"

echo -e "\n[2] Cookie安全检测"
echo "  ❌ Session Cookie: 缺少Secure标志"
echo "  ❌ Session Cookie: 缺少HttpOnly标志"
echo "  ❌ Session Cookie: 缺少SameSite属性"

echo -e "\n[3] TLS配置检测"
echo "  ✅ TLS 1.2+ 支持"
echo "  ❌ TLS 1.0/1.1 未禁用"
echo "  ❌ 弱密码套件存在"

echo -e "\n[4] 端点安全检测"
echo "  ❌ /admin: 未授权访问"
echo "  ❌ /api/users: 敏感数据暴露"
echo "  ✅ /api/health: 仅返回状态"

echo -e "\n[5] 修复优先级"
echo "  P0: Cookie安全标志 (立即)"
echo "  P1: 管理端点认证 (7天)"
echo "  P1: API数据过滤 (7天)"
echo "  P2: 安全头补全 (14天)"
echo "  P3: TLS配置加固 (30天)"

echo -e "\n========================================"
SCRIPT

chmod +x /tmp/dast_report.sh
bash /tmp/dast_report.sh

📊 安全编码速查表

📚 进阶资源

• Semgrep在线扫描
• SonarSource规则库
• OWASP WSTG
• OWASP Cheat Sheet系列
• Bandit Python安全扫描
• OWASP ZAP