安全开发生命周期：从需求分析到代码审计的 DevSecOps 实践

┌───────────────────────────────────────────────────────────┐
│                安全开发生命周期 (SDL)                       │
├───────────────────────────────────────────────────────────┤
│                                                           │
│  [需求] → 威胁建模 → STRIDE → 安全需求定义                 │
│     ↓                                                     │
│  [设计] → 架构评审 → 数据流分析 → 信任边界                 │
│     ↓                                                     │
│  [实现] → 安全编码规范 → Code Review → SAST               │
│     ↓                                                     │
│  [验证] → DAST → Fuzzing → 渗透测试                       │
│     ↓                                                     │
│  [发布] → 安全配置 → 监控告警 → 应急响应预案               │
│     ↓                                                     │
│  [运营] → 漏洞响应 → 补丁管理 → 持续改进                    │
│                                                           │
└───────────────────────────────────────────────────────────┘

系统: 用户 API (/api/v1/users/{id})

数据流图:
┌──────────┐     ┌──────────┐     ┌──────────┐
│ 客户端    │────▶│ API Gateway│───▶│ User DB  │
└──────────┘     └──────────┘     └──────────┘

STRIDE 分析:
S: 未实施 MFA → 实施 OAuth2 + MFA
T: 无 CSRF 防护 → 添加 SameSite Cookie
R: 无审计日志 → 实施 API 请求日志
I: IDOR 漏洞 → 验证资源归属
D: 无速率限制 → 添加 Rate Limiting
E: 越权访问 → 实施 RBAC + 服务端鉴权

# 1. 注入防护 - 参数化查询 (绝不拼接 SQL)
# ❌ 错误
query = f"SELECT * FROM users WHERE id = {user_id}"

# ✅ 正确
cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))

# 2. 认证失效 - 安全的 Session 管理
# ❌ 错误
session['user_id'] = user_id  # 未签名

# ✅ 正确
from itsdangerous import URLSafeTimedSerializer
serializer = URLSafeTimedSerializer(secret_key)
session['user_token'] = serializer.dumps(user_id)

# 3. 敏感信息泄露 - 环境变量
# ❌ 错误
DB_PASSWORD = "hardcoded_password_123"

# ✅ 正确
import os
DB_PASSWORD = os.environ['DB_PASSWORD']

# 4. XML 外部实体 (XXE)
# ❌ 错误
from lxml import etree
etree.fromstring(user_input)

# ✅ 正确
parser = etree.XMLParser(resolve_entities=False)
etree.fromstring(user_input, parser)

# 5. 访问控制 - 服务端验证
# ❌ 错误
if request.method == 'POST':
    delete_user(request.args['id'])

# ✅ 正确
if request.method == 'POST':
    user = get_current_user(request)
    if not user.is_admin:
        return abort(403)
    delete_user(request.args['id'])

# 6. CSRF - 验证 Token
from flask_wtf import CSRFProtect
csrf = CSRFProtect(app)

# 7. SSRF - 白名单域名
import ipaddress

def is_internal_ip(hostname):
    try:
        ip = ipaddress.ip_address(socket.gethostbyname(hostname))
        return ip.is_private or ip.is_loopback
    except:
        return False

# ❌ 错误 - 直接请求用户提供的 URL
response = requests.get(user_url)

# ✅ 正确 - 白名单验证
if not is_whitelisted(user_url):
    return abort(403)
response = requests.get(user_url, timeout=10)

# .github/workflows/security.yml
name: Security Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  sast:
    name: Static Application Security Testing
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      # Semgrep SAST
      - name: Semgrep Scan
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-
            p/security-audit
            p/owasp-top-ten
      
      # SonarQube 代码质量
      - name: SonarQube Scan
        uses: sonarsource/sonarcloud-github-action@v2
        env:
          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
      
      # Secret 检测
      - name: GitLeaks Scan
        uses: gitleaks/gitleaks-action@v2

  dependency-check:
    name: Dependency Security Audit
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      # npm audit
      - name: NPM Audit
        run: npm audit --production
      
      # Python safety check
      - name: Safety Check
        run: |
          pip install safety
          safety check --json > safety-report.json
      
      # Container scan
      - name: Trivy Image Scan
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: myapp:latest
          format: sarif
          output: trivy-results.sarif

  dast:
    name: Dynamic Application Security Testing
    runs-on: ubuntu-latest
    needs: [build]
    steps:
      # ZAP DAST Scan
      - name: OWASP ZAP Scan
        uses: zaproxy/action-baseline@v0.8.0
        with:
          target: 'https://staging.example.com'
          rules_file_name: 'zap-rules.tsv'

#!/usr/bin/env python3
"""
安全代码扫描器
检测常见不安全编码模式
"""
import re
import ast
from pathlib import Path
from dataclasses import dataclass


@dataclass
class Finding:
    file: str
    line: int
    pattern: str
    severity: str  # high, medium, low
    message: str


class SecureCodeScanner:
    """安全代码扫描器"""
    
    PATTERNS = {
        # SQL 注入
        'sql_injection': {
            'regex': r'(execute|cursor\.execute)\s*\(\s*f["\']',
            'severity': 'high',
            'message': 'Possible SQL injection via f-string',
        },
        # 硬编码凭证
        'hardcoded_secret': {
            'regex': r'(password|secret|api_key|token)\s*=\s*["\'][^"\']{8,}["\']',
            'severity': 'high',
            'message': 'Hardcoded credential detected',
        },
        # eval 使用
        'eval_usage': {
            'regex': r'\beval\s*\(',
            'severity': 'high',
            'message': 'eval() can execute arbitrary code',
        },
        # subprocess shell=True
        'shell_injection': {
            'regex': r'subprocess\.\w+\(.*shell\s*=\s*True',
            'severity': 'high',
            'message': 'Shell injection via subprocess shell=True',
        },
        # TLS 验证关闭
        'tls_disabled': {
            'regex': r'verify\s*=\s*False',
            'severity': 'medium',
            'message': 'TLS certificate verification disabled',
        },
        # 弱随机数
        'weak_random': {
            'regex': r'random\.(randint|random|choice)',
            'severity': 'medium',
            'message': 'Use secrets module for cryptographic randomness',
        },
        # pickle 反序列化
        'pickle_usage': {
            'regex': r'pickle\.(loads|load)',
            'severity': 'high',
            'message': 'Pickle deserialization can execute arbitrary code',
        },
    }
    
    def scan_file(self, filepath: Path) -> list:
        """扫描单个文件"""
        findings = []
        
        try:
            content = filepath.read_text(encoding='utf-8')
        except (UnicodeDecodeError, PermissionError):
            return findings
        
        for pattern_name, config in self.PATTERNS.items():
            for i, line in enumerate(content.split('\n'), 1):
                if re.search(config['regex'], line):
                    findings.append(Finding(
                        file=str(filepath),
                        line=i,
                        pattern=pattern_name,
                        severity=config['severity'],
                        message=config['message'],
                    ))
        
        return findings
    
    def scan_directory(self, directory: Path) -> list:
        """递归扫描"""
        all_findings = []
        
        extensions = {'.py', '.js', '.ts', '.go', '.java', '.rb'}
        
        for filepath in directory.rglob('*'):
            if filepath.suffix in extensions and '.git' not in str(filepath):
                all_findings.extend(self.scan_file(filepath))
        
        return all_findings
    
    def report(self, findings: list) -> str:
        """生成报告"""
        output = []
        
        severity_counts = {'high': 0, 'medium': 0, 'low': 0}
        
        for finding in findings:
            severity_counts[finding.severity] += 1
            output.append(
                f"[{finding.severity.upper()}] {finding.file}:{finding.line} - {finding.message}"
            )
        
        header = f"Found {len(findings)} issues: "
        header += f"{severity_counts['high']} high, "
        header += f"{severity_counts['medium']} medium, "
        header += f"{severity_counts['low']} low"
        
        return header + '\n' + '\n'.join(output)


if __name__ == '__main__':
    import sys
    
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <directory_or_file>")
        sys.exit(1)
    
    scanner = SecureCodeScanner()
    target = Path(sys.argv[1])
    
    if target.is_file():
        findings = scanner.scan_file(target)
    else:
        findings = scanner.scan_directory(target)
    
    print(scanner.report(findings))

# 代码安全审查清单

## 输入验证
- [ ] 所有用户输入是否经过服务端验证？
- [ ] 输入长度是否有上限？
- [ ] 特殊字符是否转义？

## 认证与授权
- [ ] 敏感操作是否重新认证？
- [ ] API 端点是否有授权检查？
- [ ] 权限验证在服务端执行？

## 加密
- [ ] 使用安全的随机数生成器？
- [ ] 密钥长度 >= 256 bit？
- [ ] 禁用已知不安全的算法 (MD5, SHA1)？

## 错误处理
- [ ] 错误消息不泄露敏感信息？
- [ ] 异常是否被正确捕获？
- [ ] 生产环境禁用详细错误输出？

## 日志审计
- [ ] 登录/登出是否记录？
- [ ] 敏感操作是否审计？
- [ ] 日志不包含密码/token？