# 第二十九章：DevSecOps — 安全左移

> "安全不是开发完成后的检查站，而是贯穿整个流水线的守护者。"

```{mermaid}
mindmap
  root((DevSecOps))
    CI/CD 安全
      SAST
      SCA
      DAST
      镜像扫描
    工具链
      Semgrep
      Snyk
      Trivy
      OWASP ZAP
    IaC 安全
      tfsec
      Checkov
      kubesec
    文化
      Security Champion
      安全培训
      安全门禁
```

## 29.1 DevSecOps 理念

DevSecOps 将安全融入 DevOps 的每个阶段：

```{mermaid}
flowchart LR
    Code["🖊️ Code<br/>───<br/>Pre-commit Hook<br/>Secret Detection<br/>Linting"]
    SAST["🔍 SAST<br/>───<br/>Semgrep / CodeQL<br/>Bandit / gosec<br/>SpotBugs"]
    SCA["📦 SCA<br/>───<br/>Snyk / pip-audit<br/>govulncheck<br/>OWASP Dep-Check"]
    Build["🔨 Build<br/>───<br/>编译打包<br/>SBOM 生成<br/>签名验证"]
    ImageScan["🐳 镜像扫描<br/>───<br/>Trivy / Grype<br/>漏洞检查<br/>合规基线"]
    DAST["⚡ DAST<br/>───<br/>OWASP ZAP<br/>Nuclei<br/>API Fuzzing"]
    Deploy["🚀 Deploy<br/>───<br/>安全配置<br/>IaC 扫描<br/>准入控制"]
    Monitor["📊 Monitor<br/>───<br/>SIEM / WAF<br/>运行时防护<br/>告警响应"]

    Code --> SAST --> SCA --> Build --> ImageScan --> DAST --> Deploy --> Monitor

    style Code fill:#e1f5fe,stroke:#0288d1
    style SAST fill:#f3e5f5,stroke:#7b1fa2
    style SCA fill:#fff3e0,stroke:#ef6c00
    style Build fill:#e8f5e9,stroke:#388e3c
    style ImageScan fill:#fce4ec,stroke:#c62828
    style DAST fill:#fff8e1,stroke:#f9a825
    style Deploy fill:#e0f2f1,stroke:#00695c
    style Monitor fill:#f1f8e9,stroke:#558b2f
```

## 29.2 SAST — 静态应用安全测试

| 工具 | 语言支持 | 特点 | 开源 |
|------|---------|------|------|
| Semgrep | 多语言 | 自定义规则、快速 | ✅ |
| CodeQL | 多语言 | GitHub 集成、深度分析 | ✅ |
| SonarQube | 多语言 | 全面、企业级 | 部分 |
| Bandit | Python | Python 专用 | ✅ |

```bash
# Semgrep 扫描
semgrep scan --config auto .

# 自定义规则示例
# rules/no-hardcoded-secrets.yaml
rules:
  - id: hardcoded-password
    patterns:
      - pattern: password = "..."
    message: "Hardcoded password detected"
    severity: ERROR
    languages: [python]
```

### 29.2.1 SAST/DAST/SCA 工具集成点

```{mermaid}
flowchart TB
    subgraph IDE["🖥️ IDE / 本地开发"]
        PreCommit["Pre-commit Hook<br/>GitLeaks / Semgrep"]
    end

    subgraph CI["⚙️ CI 流水线"]
        direction TB
        SAST_CI["SAST 扫描<br/>Semgrep / CodeQL / SpotBugs / gosec / Bandit"]
        SCA_CI["SCA 扫描<br/>Snyk / pip-audit / govulncheck / OWASP Dep-Check"]
        Secret_CI["密钥扫描<br/>GitLeaks / TruffleHog"]
        IaC_CI["IaC 扫描<br/>Checkov / tfsec"]
        Image_CI["镜像扫描<br/>Trivy / Grype"]
        SBOM_CI["SBOM 生成<br/>Syft / CycloneDX"]
    end

    subgraph Staging["🧪 Staging 环境"]
        DAST_Stage["DAST 扫描<br/>OWASP ZAP / Nuclei"]
        API_Fuzz["API Fuzzing<br/>RESTler / Schemathesis"]
    end

    subgraph Prod["🏭 Production"]
        RASP["运行时防护<br/>RASP / WAF"]
        SIEM["安全监控<br/>SIEM / 告警"]
    end

    PreCommit -->|"git push"| SAST_CI
    SAST_CI --> SCA_CI --> Secret_CI --> IaC_CI --> Image_CI --> SBOM_CI
    SBOM_CI -->|"部署到 Staging"| DAST_Stage
    DAST_Stage --> API_Fuzz
    API_Fuzz -->|"安全门禁通过"| RASP
    RASP --> SIEM

    style IDE fill:#e3f2fd,stroke:#1565c0
    style CI fill:#f3e5f5,stroke:#6a1b9a
    style Staging fill:#fff3e0,stroke:#e65100
    style Prod fill:#e8f5e9,stroke:#2e7d32
```

### 29.2.2 SAST 工具对比

| 特性 | Semgrep | SonarQube | CodeQL | Bandit (Python) | gosec (Go) | SpotBugs (Java) |
|------|---------|-----------|--------|-----------------|------------|-----------------|
| **语言覆盖** | 30+ 语言 | 30+ 语言 | 12+ 语言 | Python | Go | Java/Kotlin |
| **规则自定义** | ⭐⭐⭐ YAML DSL | ⭐⭐ Java API | ⭐⭐⭐ QL 语言 | ⭐⭐ Python 插件 | ⭐⭐ Go 插件 | ⭐⭐ Java 插件 |
| **CI 集成** | ⭐⭐⭐ 原生支持 | ⭐⭐ 需要服务端 | ⭐⭐⭐ GitHub 原生 | ⭐⭐ CLI | ⭐⭐ CLI | ⭐⭐ Maven/Gradle |
| **扫描速度** | 🚀 极快 | 🐢 较慢 | 🐢 较慢 | 🚀 快 | 🚀 快 | ⚡ 中等 |
| **误报率** | 低 | 中 | 低 | 中 | 低 | 中 |
| **数据流分析** | 基础 | 深度 | 深度 | 无 | 基础 | 深度 |
| **开源** | ✅ 社区版 | 部分开源 | ✅ | ✅ | ✅ | ✅ |
| **企业功能** | Semgrep Cloud | 商业版 | GitHub Advanced Security | — | — | — |
| **适用场景** | 通用快速扫描 | 企业全面治理 | GitHub 深度集成 | Python 项目 | Go 项目 | Java/JVM 项目 |

## 29.3 SCA — 软件组成分析

```bash
# Snyk — 依赖漏洞扫描
snyk test
snyk monitor  # 持续监控

# pip-audit — Python 依赖审计
pip-audit

# npm audit — Node.js 依赖审计
npm audit --audit-level=high
```

## 29.4 DAST — 动态应用安全测试

```bash
# OWASP ZAP — 动态扫描
docker run -t zaproxy/zap-stable zap-baseline.py \
    -t https://staging.example.com \
    -r report.html

# Nuclei — 快速漏洞扫描
nuclei -u https://staging.example.com -t cves/
```

## 29.5 IaC 安全

```bash
# tfsec — Terraform 安全扫描
tfsec .

# Checkov — 多 IaC 框架支持
checkov -d .
checkov --framework terraform,kubernetes

# kubesec — Kubernetes 清单安全评分
kubesec scan deployment.yaml
```

## 29.6 CI/CD 安全流水线

### 29.6.1 GitHub Actions 完整安全 Pipeline

```yaml
name: Security Pipeline
on: [push, pull_request]

jobs:
  sast:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Semgrep SAST
        uses: returntocorp/semgrep-action@v1
        with:
          config: auto

      - name: CodeQL Analysis
        uses: github/codeql-action/init@v3
        with:
          languages: python, java, go
      - name: CodeQL Autobuild
        uses: github/codeql-action/autobuild@v3
      - name: CodeQL Perform Analysis
        uses: github/codeql-action/analyze@v3

  sca:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Snyk SCA
        uses: snyk/actions/python@master
        env:
          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}

      - name: OWASP Dependency-Check
        uses: dependency-check/Dependency-Check_Action@main
        with:
          project: 'my-project'
          path: '.'
          format: 'HTML'
        env:
          JAVA_HOME: /opt/jdk

  container-scan:
    runs-on: ubuntu-latest
    needs: [sast, sca]
    steps:
      - uses: actions/checkout@v4
      - name: Build image
        run: docker build -t myapp:${{ github.sha }} .
      - name: Trivy scan
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: myapp:${{ github.sha }}
          severity: HIGH,CRITICAL
          exit-code: 1
          format: 'sarif'
          output: 'trivy-results.sarif'
      - name: Upload Trivy SARIF
        uses: github/codeql-action/upload-sarif@v3
        with:
          sarif_file: 'trivy-results.sarif'

      - name: Generate SBOM
        uses: anchore/sbom-action@v0
        with:
          image: myapp:${{ github.sha }}
          format: cyclonedx-json
          output-file: sbom.json

  secret-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
      - name: GitLeaks
        uses: gitleaks/gitleaks-action@v2
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

  iac-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Checkov
        uses: bridgecrewio/checkov-action@master
        with:
          directory: ./terraform

  dast:
    runs-on: ubuntu-latest
    needs: [container-scan]
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to staging
        run: |
          docker compose -f docker-compose.staging.yml up -d
          sleep 30  # 等待服务就绪
      - name: OWASP ZAP Baseline Scan
        uses: zaproxy/action-baseline@v0.12.0
        with:
          target: 'http://localhost:8080'
          rules_file_name: '.zap/rules.tsv'
          cmd_options: '-a'

  security-gate:
    runs-on: ubuntu-latest
    needs: [sast, sca, container-scan, secret-scan, iac-scan, dast]
    steps:
      - name: Evaluate Security Gate
        run: |
          echo "All security checks passed — deployment approved"
```

### 29.6.2 GitLab CI 等效安全流水线

```yaml
# .gitlab-ci.yml
stages:
  - test
  - sast
  - sca
  - build
  - container-scan
  - dast
  - security-gate
  - deploy

variables:
  DOCKER_IMAGE: $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA

# ── SAST ──────────────────────────────────────────────
semgrep-sast:
  stage: sast
  image: returntocorp/semgrep
  script:
    - semgrep scan --config auto --sarif -o semgrep.sarif .
  artifacts:
    reports:
      sast: semgrep.sarif
    paths:
      - semgrep.sarif

bandit-sast:
  stage: sast
  image: python:3.12-slim
  script:
    - pip install bandit[toml]
    - bandit -r src/ -f json -o bandit-report.json --severity-level medium || true
  artifacts:
    paths:
      - bandit-report.json

gosec-sast:
  stage: sast
  image: securego/gosec:latest
  script:
    - gosec -fmt=json -out=gosec-report.json ./...
  artifacts:
    paths:
      - gosec-report.json
  rules:
    - exists:
        - go.mod

spotbugs-sast:
  stage: sast
  image: maven:3.9-eclipse-temurin-21
  script:
    - mvn compile com.github.spotbugs:spotbugs-maven-plugin:spotbugs
        -Dspotbugs.plugins.plugin.groupId=com.h3xstream.findsecbugs
        -Dspotbugs.plugins.plugin.artifactId=findsecbugs-plugin
  artifacts:
    paths:
      - target/spotbugsXml.xml
  rules:
    - exists:
        - pom.xml

# ── SCA ───────────────────────────────────────────────
snyk-sca:
  stage: sca
  image: snyk/snyk:python-3.12
  script:
    - snyk auth $SNYK_TOKEN
    - snyk test --severity-threshold=high --json-file-output=snyk-report.json
  artifacts:
    paths:
      - snyk-report.json

pip-audit-sca:
  stage: sca
  image: python:3.12-slim
  script:
    - pip install pip-audit
    - pip-audit -r requirements.txt --format json --output pip-audit-report.json || true
  artifacts:
    paths:
      - pip-audit-report.json

dependency-check-sca:
  stage: sca
  image: owasp/dependency-check:latest
  script:
    - /usr/share/dependency-check/bin/dependency-check.sh
        --project "my-project"
        --scan .
        --format JSON
        --out dependency-check-report
        --failOnCVSS 7
  artifacts:
    paths:
      - dependency-check-report/

# ── Build & Container Scan ────────────────────────────
build-image:
  stage: build
  image: docker:24
  services:
    - docker:24-dind
  script:
    - docker build -t $DOCKER_IMAGE .
    - docker push $DOCKER_IMAGE

trivy-container-scan:
  stage: container-scan
  image: aquasec/trivy:latest
  needs: [build-image]
  script:
    - trivy image --severity HIGH,CRITICAL --exit-code 1
        --format json --output trivy-report.json $DOCKER_IMAGE
  artifacts:
    paths:
      - trivy-report.json

# ── DAST ──────────────────────────────────────────────
zap-dast:
  stage: dast
  image: zaproxy/zap-stable
  needs: [trivy-container-scan]
  script:
    - zap-baseline.py -t $STAGING_URL -r zap-report.html -J zap-report.json
  artifacts:
    paths:
      - zap-report.html
      - zap-report.json

# ── Security Gate ─────────────────────────────────────
security-gate:
  stage: security-gate
  image: python:3.12-slim
  needs:
    - semgrep-sast
    - snyk-sca
    - trivy-container-scan
    - zap-dast
  script:
    - pip install json-query
    - |
      echo "=== Security Gate Evaluation ==="
      # 检查是否存在 Critical/High 漏洞
      python3 scripts/security_gate.py \
        --semgrep semgrep.sarif \
        --trivy trivy-report.json \
        --zap zap-report.json \
        --fail-on critical,high

# ── Deploy ────────────────────────────────────────────
deploy-production:
  stage: deploy
  needs: [security-gate]
  script:
    - echo "Deploying to production..."
  when: manual
  environment:
    name: production
```

## 29.7 安全门禁策略

| 严重程度 | CI 行为 | 说明 |
|---------|---------|------|
| Critical | ❌ 阻断 | 必须修复才能合并 |
| High | ❌ 阻断 | 必须修复或获得豁免 |
| Medium | ⚠️ 警告 | 记录并计划修复 |
| Low | ℹ️ 信息 | 记录，不阻断 |

### 29.7.1 安全门禁决策流程

```{mermaid}
flowchart TD
    Start(["🔍 安全扫描完成"]) --> Collect["汇总所有扫描报告<br/>SAST + SCA + 镜像扫描 + DAST"]
    Collect --> HasCritical{"存在 Critical<br/>漏洞？"}

    HasCritical -->|"是"| BlockCritical["❌ 阻断部署<br/>通知安全团队<br/>创建 P0 工单"]
    HasCritical -->|"否"| HasHigh{"存在 High<br/>漏洞？"}

    HasHigh -->|"是"| CheckExemption{"已获得安全<br/>团队豁免？"}
    CheckExemption -->|"否"| BlockHigh["❌ 阻断部署<br/>要求修复或申请豁免"]
    CheckExemption -->|"是"| LogExemption["📝 记录豁免<br/>设置修复截止日期"]
    LogExemption --> HasMedium

    HasHigh -->|"否"| HasMedium{"存在 Medium<br/>漏洞？"}
    HasMedium -->|"是"| WarnMedium["⚠️ 警告<br/>创建修复工单<br/>SLA: 30 天"]
    HasMedium -->|"否"| HasLow{"存在 Low<br/>漏洞？"}

    WarnMedium --> HasLow
    HasLow -->|"是"| InfoLow["ℹ️ 记录信息<br/>纳入技术债务"]
    HasLow -->|"否"| Clean["✅ 无漏洞"]

    InfoLow --> Approve
    Clean --> Approve
    Approve(["✅ 允许部署"])

    BlockCritical --> Review["安全团队评审"]
    BlockHigh --> Review
    Review --> Fixed{"漏洞已修复？"}
    Fixed -->|"是"| Rescan["重新扫描"]
    Fixed -->|"否"| Escalate["升级处理"]
    Rescan --> Start

    style BlockCritical fill:#ffcdd2,stroke:#c62828
    style BlockHigh fill:#ffcdd2,stroke:#c62828
    style WarnMedium fill:#fff9c4,stroke:#f9a825
    style InfoLow fill:#e3f2fd,stroke:#1565c0
    style Approve fill:#c8e6c9,stroke:#2e7d32
    style Clean fill:#c8e6c9,stroke:#2e7d32
```

### 29.7.2 安全门禁策略模板

```yaml
# security-gate-policy.yaml
# 安全门禁策略配置 — 定义各级别漏洞的处理方式
policy:
  name: "Default Security Gate Policy"
  version: "1.0"

  # ── 阻断规则 ──────────────────────────────────────
  blocking_rules:
    - severity: CRITICAL
      action: block
      message: "Critical 漏洞必须在合并前修复"
      notify:
        - security-team@example.com
        - oncall-slack-channel
      auto_create_ticket: true
      ticket_priority: P0

    - severity: HIGH
      action: block
      exemption_allowed: true
      exemption_approvers:
        - security-lead
        - engineering-director
      exemption_max_days: 14
      message: "High 漏洞需修复或获得豁免"

  # ── 警告规则 ──────────────────────────────────────
  warning_rules:
    - severity: MEDIUM
      action: warn
      sla_days: 30
      auto_create_ticket: true
      ticket_priority: P2

    - severity: LOW
      action: info
      sla_days: 90
      auto_create_ticket: false

  # ── 扫描器配置 ────────────────────────────────────
  scanners:
    sast:
      required: true
      tools: [semgrep, codeql]
      fail_on: [CRITICAL, HIGH]

    sca:
      required: true
      tools: [snyk, pip-audit]
      fail_on: [CRITICAL, HIGH]
      ignore_dev_dependencies: true

    container_scan:
      required: true
      tools: [trivy]
      fail_on: [CRITICAL, HIGH]
      base_image_policy: "only-approved"

    dast:
      required: false  # 仅在 staging 部署后执行
      tools: [zap]
      fail_on: [CRITICAL]

    secret_scan:
      required: true
      tools: [gitleaks]
      fail_on: [ANY]  # 任何密钥泄露都阻断

    iac_scan:
      required: true
      tools: [checkov]
      fail_on: [CRITICAL, HIGH]

  # ── 豁免管理 ──────────────────────────────────────
  exemptions:
    require_justification: true
    require_approval: true
    max_duration_days: 90
    review_cadence: "weekly"
    audit_log: true
```

## 29.8 Python DevSecOps 实例

### 29.8.1 Bandit SAST 自定义规则配置

```toml
# pyproject.toml — Bandit 配置
[tool.bandit]
# 排除测试目录
exclude_dirs = ["tests", "venv", ".tox"]
# 跳过特定检查
skips = ["B101"]  # 允许 assert（仅在测试中）
# 设置严重级别阈值
severity = "medium"
confidence = "medium"

[tool.bandit.assert_used]
skips = ["**/test_*.py", "**/conftest.py"]
```

```yaml
# .bandit.yaml — 高级自定义规则配置
# 自定义 Bandit profile
profile:
  name: custom_security_profile
  include:
    - any_other_function_with_shell_equals_true  # B604
    - exec_used                                   # B102
    - hardcoded_password_string                   # B105
    - hardcoded_password_funcarg                  # B106
    - hardcoded_sql_expressions                   # B608
    - jinja2_autoescape_false                     # B701
    - request_without_timeout                     # B113
    - snmp_insecure_version                       # B508
    - sql_statements                              # B608
    - ssl_with_bad_defaults                       # B503
    - try_except_pass                             # B110

# 自定义黑名单调用
blacklist_calls:
  bad_name_sets:
    - pickle:
        qualnames:
          - pickle.loads
          - pickle.load
          - pickle.Unpickler
          - cPickle.loads
          - cPickle.load
          - cPickle.Unpickler
        message: "Pickle 反序列化可导致远程代码执行，请使用 json 替代"
        level: HIGH
    - yaml_unsafe:
        qualnames:
          - yaml.load
          - yaml.unsafe_load
        message: "yaml.load 不安全，请使用 yaml.safe_load"
        level: HIGH

blacklist_imports:
  bad_import_sets:
    - telnet:
        imports:
          - telnetlib
        message: "Telnet 协议不加密，请使用 SSH"
        level: HIGH
```

```bash
# 运行 Bandit 扫描
bandit -r src/ -f json -o bandit-report.json --severity-level medium --confidence-level medium
bandit -r src/ -f html -o bandit-report.html  # HTML 报告
bandit -r src/ -f sarif -o bandit-report.sarif  # SARIF 格式（CI 集成）
```

### 29.8.2 Safety / pip-audit SCA 集成脚本

```python
#!/usr/bin/env python3
"""
sca_scanner.py — Python SCA 集成扫描脚本
同时运行 pip-audit 和 safety，聚合结果并判断是否阻断。
"""

import json
import subprocess
import sys
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path


class Severity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    UNKNOWN = "unknown"

    @property
    def weight(self) -> int:
        return {"critical": 4, "high": 3, "medium": 2, "low": 1, "unknown": 0}[
            self.value
        ]


@dataclass
class Vulnerability:
    package: str
    installed_version: str
    fixed_version: str | None
    vuln_id: str
    severity: Severity
    description: str
    source: str  # "pip-audit" or "safety"


@dataclass
class SCAReport:
    vulnerabilities: list[Vulnerability] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    @property
    def critical_count(self) -> int:
        return sum(1 for v in self.vulnerabilities if v.severity == Severity.CRITICAL)

    @property
    def high_count(self) -> int:
        return sum(1 for v in self.vulnerabilities if v.severity == Severity.HIGH)

    def should_block(self, block_on: set[Severity] | None = None) -> bool:
        if block_on is None:
            block_on = {Severity.CRITICAL, Severity.HIGH}
        return any(v.severity in block_on for v in self.vulnerabilities)


def run_pip_audit(requirements: str = "requirements.txt") -> list[Vulnerability]:
    """运行 pip-audit 并解析结果。"""
    vulns = []
    try:
        result = subprocess.run(
            ["pip-audit", "-r", requirements, "--format", "json", "--output", "-"],
            capture_output=True,
            text=True,
            timeout=300,
        )
        data = json.loads(result.stdout) if result.stdout else {"dependencies": []}
        for dep in data.get("dependencies", []):
            for vuln in dep.get("vulns", []):
                vulns.append(
                    Vulnerability(
                        package=dep["name"],
                        installed_version=dep["version"],
                        fixed_version=vuln.get("fix_versions", [None])[0],
                        vuln_id=vuln["id"],
                        severity=_parse_severity(vuln.get("severity", "unknown")),
                        description=vuln.get("description", ""),
                        source="pip-audit",
                    )
                )
    except (subprocess.TimeoutExpired, FileNotFoundError) as e:
        print(f"[WARN] pip-audit 执行失败: {e}", file=sys.stderr)
    return vulns


def run_safety(requirements: str = "requirements.txt") -> list[Vulnerability]:
    """运行 safety check 并解析结果。"""
    vulns = []
    try:
        result = subprocess.run(
            ["safety", "check", "-r", requirements, "--json"],
            capture_output=True,
            text=True,
            timeout=300,
        )
        data = json.loads(result.stdout) if result.stdout else []
        for item in data:
            # safety JSON 格式: [package, affected, installed, description, vuln_id]
            if isinstance(item, list) and len(item) >= 5:
                vulns.append(
                    Vulnerability(
                        package=item[0],
                        installed_version=item[2],
                        fixed_version=None,
                        vuln_id=str(item[4]),
                        severity=Severity.UNKNOWN,
                        description=item[3],
                        source="safety",
                    )
                )
    except (subprocess.TimeoutExpired, FileNotFoundError) as e:
        print(f"[WARN] safety 执行失败: {e}", file=sys.stderr)
    return vulns


def _parse_severity(raw: str) -> Severity:
    try:
        return Severity(raw.lower())
    except ValueError:
        return Severity.UNKNOWN


def aggregate_report(requirements: str = "requirements.txt") -> SCAReport:
    """聚合多个 SCA 工具的扫描结果。"""
    report = SCAReport()
    report.vulnerabilities.extend(run_pip_audit(requirements))
    report.vulnerabilities.extend(run_safety(requirements))

    # 按 vuln_id 去重（保留严重级别更高的）
    seen: dict[str, Vulnerability] = {}
    for v in report.vulnerabilities:
        key = f"{v.package}:{v.vuln_id}"
        if key not in seen or v.severity.weight > seen[key].severity.weight:
            seen[key] = v
    report.vulnerabilities = list(seen.values())
    report.vulnerabilities.sort(key=lambda v: v.severity.weight, reverse=True)
    return report


def main() -> None:
    report = aggregate_report()

    print(f"\n{'='*60}")
    print(f"  SCA 扫描报告 — 共发现 {len(report.vulnerabilities)} 个漏洞")
    print(f"  Critical: {report.critical_count}  |  High: {report.high_count}")
    print(f"{'='*60}\n")

    for v in report.vulnerabilities:
        icon = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🔵"}.get(
            v.severity.value, "⚪"
        )
        print(f"  {icon} [{v.severity.value.upper()}] {v.package}=={v.installed_version}")
        print(f"     ID: {v.vuln_id}  |  来源: {v.source}")
        if v.fixed_version:
            print(f"     修复版本: {v.fixed_version}")
        print(f"     {v.description[:120]}")
        print()

    # 输出 JSON 报告
    report_path = Path("sca-report.json")
    report_path.write_text(
        json.dumps(
            {
                "total": len(report.vulnerabilities),
                "critical": report.critical_count,
                "high": report.high_count,
                "blocked": report.should_block(),
                "vulnerabilities": [
                    {
                        "package": v.package,
                        "version": v.installed_version,
                        "vuln_id": v.vuln_id,
                        "severity": v.severity.value,
                        "source": v.source,
                    }
                    for v in report.vulnerabilities
                ],
            },
            indent=2,
        )
    )
    print(f"  📄 报告已保存: {report_path}")

    if report.should_block():
        print("\n  ❌ 安全门禁: 存在 Critical/High 漏洞，阻断部署\n")
        sys.exit(1)
    else:
        print("\n  ✅ 安全门禁: 通过\n")


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

### 29.8.3 安全测试报告聚合器

```python
#!/usr/bin/env python3
"""
security_report_aggregator.py — 聚合多种安全扫描报告，生成统一仪表盘。
支持: Bandit (SAST), pip-audit (SCA), Trivy (Container), ZAP (DAST)
"""

import json
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path


@dataclass
class Finding:
    tool: str
    category: str  # SAST, SCA, Container, DAST
    severity: str
    title: str
    description: str
    location: str
    vuln_id: str = ""


@dataclass
class AggregatedReport:
    findings: list[Finding] = field(default_factory=list)
    generated_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

    def summary(self) -> dict[str, dict[str, int]]:
        result: dict[str, dict[str, int]] = {}
        for f in self.findings:
            cat = result.setdefault(f.category, {})
            cat[f.severity] = cat.get(f.severity, 0) + 1
        return result


def parse_bandit(path: Path) -> list[Finding]:
    data = json.loads(path.read_text())
    return [
        Finding(
            tool="Bandit",
            category="SAST",
            severity=r["issue_severity"].lower(),
            title=r["issue_text"],
            description=f"CWE: {r.get('issue_cwe', {}).get('id', 'N/A')}",
            location=f"{r['filename']}:{r['line_number']}",
            vuln_id=r.get("test_id", ""),
        )
        for r in data.get("results", [])
    ]


def parse_trivy(path: Path) -> list[Finding]:
    data = json.loads(path.read_text())
    findings = []
    for result in data.get("Results", []):
        for vuln in result.get("Vulnerabilities", []):
            findings.append(
                Finding(
                    tool="Trivy",
                    category="Container",
                    severity=vuln.get("Severity", "UNKNOWN").lower(),
                    title=f"{vuln['PkgName']}:{vuln.get('InstalledVersion', '?')}",
                    description=vuln.get("Title", ""),
                    location=result.get("Target", ""),
                    vuln_id=vuln.get("VulnerabilityID", ""),
                )
            )
    return findings


def parse_zap(path: Path) -> list[Finding]:
    data = json.loads(path.read_text())
    findings = []
    risk_map = {"0": "info", "1": "low", "2": "medium", "3": "high"}
    for site in data.get("site", []):
        for alert in site.get("alerts", []):
            findings.append(
                Finding(
                    tool="ZAP",
                    category="DAST",
                    severity=risk_map.get(alert.get("riskcode", "0"), "unknown"),
                    title=alert.get("name", ""),
                    description=alert.get("desc", "")[:200],
                    location=alert.get("uri", ""),
                    vuln_id=f"CWE-{alert.get('cweid', 'N/A')}",
                )
            )
    return findings


def aggregate(report_dir: Path) -> AggregatedReport:
    report = AggregatedReport()
    parsers = {
        "bandit-report.json": parse_bandit,
        "trivy-report.json": parse_trivy,
        "zap-report.json": parse_zap,
    }
    for filename, parser in parsers.items():
        path = report_dir / filename
        if path.exists():
            report.findings.extend(parser(path))
            print(f"  ✅ 已解析: {filename}")
        else:
            print(f"  ⏭️  跳过（未找到）: {filename}")
    return report


def print_dashboard(report: AggregatedReport) -> None:
    summary = report.summary()
    total = len(report.findings)
    print(f"\n{'='*60}")
    print(f"  🛡️  安全扫描聚合报告  |  {report.generated_at}")
    print(f"  总发现: {total}")
    print(f"{'='*60}")
    for category, counts in summary.items():
        print(f"\n  📂 {category}:")
        for sev in ["critical", "high", "medium", "low", "info"]:
            if sev in counts:
                icon = {"critical": "🔴", "high": "🟠", "medium": "🟡",
                        "low": "🔵", "info": "⚪"}[sev]
                print(f"     {icon} {sev.upper()}: {counts[sev]}")
    print(f"\n{'='*60}\n")


def main() -> None:
    report_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
    report = aggregate(report_dir)
    print_dashboard(report)

    output = report_dir / "aggregated-security-report.json"
    output.write_text(
        json.dumps(
            {
                "generated_at": report.generated_at,
                "total_findings": len(report.findings),
                "summary": report.summary(),
                "findings": [
                    {
                        "tool": f.tool,
                        "category": f.category,
                        "severity": f.severity,
                        "title": f.title,
                        "location": f.location,
                        "vuln_id": f.vuln_id,
                    }
                    for f in report.findings
                ],
            },
            indent=2,
        )
    )
    print(f"  📄 聚合报告已保存: {output}")


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

## 29.9 Java DevSecOps 实例

### 29.9.1 SpotBugs + Find Security Bugs 配置（Maven）

```xml
<!-- pom.xml — SpotBugs + Find Security Bugs 插件配置 -->
<project>
    <properties>
        <spotbugs.version>4.8.4</spotbugs.version>
        <findsecbugs.version>1.13.0</findsecbugs.version>
        <dependency-check.version>9.0.9</dependency-check.version>
    </properties>

    <build>
        <plugins>
            <!-- SpotBugs + Find Security Bugs -->
            <plugin>
                <groupId>com.github.spotbugs</groupId>
                <artifactId>spotbugs-maven-plugin</artifactId>
                <version>${spotbugs.version}</version>
                <configuration>
                    <!-- 仅报告 Medium 及以上 -->
                    <effort>Max</effort>
                    <threshold>Medium</threshold>
                    <failOnError>true</failOnError>
                    <xmlOutput>true</xmlOutput>
                    <xmlOutputDirectory>${project.build.directory}/spotbugs</xmlOutputDirectory>
                    <!-- 排除文件 -->
                    <excludeFilterFile>spotbugs-exclude.xml</excludeFilterFile>
                    <plugins>
                        <plugin>
                            <groupId>com.h3xstream.findsecbugs</groupId>
                            <artifactId>findsecbugs-plugin</artifactId>
                            <version>${findsecbugs.version}</version>
                        </plugin>
                    </plugins>
                </configuration>
                <executions>
                    <execution>
                        <id>security-check</id>
                        <phase>verify</phase>
                        <goals>
                            <goal>check</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
```

```xml
<!-- spotbugs-exclude.xml — 排除规则 -->
<FindBugsFilter>
    <!-- 排除测试代码 -->
    <Match>
        <Source name="~.*Test\.java"/>
    </Match>
    <!-- 排除生成代码 -->
    <Match>
        <Package name="~.*\.generated\..*"/>
    </Match>
</FindBugsFilter>
```

### 29.9.2 OWASP Dependency-Check Maven 插件配置

```xml
<!-- pom.xml — OWASP Dependency-Check 插件 -->
<plugin>
    <groupId>org.owasp</groupId>
    <artifactId>dependency-check-maven</artifactId>
    <version>${dependency-check.version}</version>
    <configuration>
        <!-- CVSS 7.0 及以上阻断构建 -->
        <failBuildOnCVSS>7</failBuildOnCVSS>
        <formats>
            <format>HTML</format>
            <format>JSON</format>
            <format>SARIF</format>
        </formats>
        <outputDirectory>${project.build.directory}/dependency-check</outputDirectory>
        <!-- NVD API Key（加速数据库更新） -->
        <nvdApiKey>${env.NVD_API_KEY}</nvdApiKey>
        <!-- 排除测试依赖 -->
        <skipTestScope>true</skipTestScope>
        <!-- 排除特定 CVE（已评估为误报） -->
        <suppressionFiles>
            <suppressionFile>dependency-check-suppression.xml</suppressionFile>
        </suppressionFiles>
    </configuration>
    <executions>
        <execution>
            <id>dependency-check</id>
            <phase>verify</phase>
            <goals>
                <goal>check</goal>
            </goals>
        </execution>
    </executions>
</plugin>
```

```xml
<!-- dependency-check-suppression.xml — 误报抑制 -->
<suppressions xmlns="https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.3.xsd">
    <suppress>
        <notes>误报: 此 CVE 不影响我们的使用方式</notes>
        <cve>CVE-2023-XXXXX</cve>
    </suppress>
    <suppress>
        <notes>已通过其他方式缓解</notes>
        <packageUrl regex="true">^pkg:maven/com\.example/.*$</packageUrl>
        <cvssBelow>4</cvssBelow>
    </suppress>
</suppressions>
```

### 29.9.3 Gradle 安全扫描任务

```groovy
// build.gradle — Gradle 安全扫描配置

plugins {
    id 'java'
    id 'com.github.spotbugs' version '6.0.9'
    id 'org.owasp.dependencycheck' version '9.0.9'
}

// ── SpotBugs + Find Security Bugs ────────────────────
spotbugs {
    effort = 'max'
    reportLevel = 'medium'
    excludeFilter = file('spotbugs-exclude.xml')
}

dependencies {
    spotbugsPlugins 'com.h3xstream.findsecbugs:findsecbugs-plugin:1.13.0'
}

tasks.withType(com.github.spotbugs.snom.SpotBugsTask) {
    reports {
        xml.required = true
        html.required = true
        sarif.required = true
    }
}

// ── OWASP Dependency-Check ───────────────────────────
dependencyCheck {
    failBuildOnCVSS = 7.0f
    formats = ['HTML', 'JSON', 'SARIF']
    suppressionFile = 'dependency-check-suppression.xml'
    nvd {
        apiKey = System.getenv('NVD_API_KEY') ?: ''
    }
    analyzers {
        assemblyEnabled = false  // 非 .NET 项目禁用
    }
}

// ── 聚合安全扫描任务 ─────────────────────────────────
tasks.register('securityScan') {
    group = 'verification'
    description = 'Run all security scans (SAST + SCA)'
    dependsOn 'spotbugsMain', 'dependencyCheckAnalyze'

    doLast {
        println '=== Security Scan Complete ==='
        println "SpotBugs report: ${project.buildDir}/reports/spotbugs/main.html"
        println "Dependency-Check report: ${project.buildDir}/reports/dependency-check-report.html"
    }
}

// 将安全扫描集成到 check 生命周期
check.dependsOn 'securityScan'
```

```bash
# 运行 Gradle 安全扫描
./gradlew securityScan

# 单独运行 SpotBugs
./gradlew spotbugsMain

# 单独运行 Dependency-Check
./gradlew dependencyCheckAnalyze
```

## 29.10 Go DevSecOps 实例

### 29.10.1 gosec 配置和自定义规则

```yaml
# .gosec.yaml — gosec 配置文件
global:
  # 排除测试文件
  exclude-dir:
    - vendor
    - testdata
    - mocks

  # 排除特定规则
  exclude:
    - G104  # 未检查的错误（某些场景下可接受）

  # 设置严重级别
  severity: medium
  confidence: medium

# 输出配置
output:
  format: json
  file: gosec-report.json
```

```bash
#!/bin/bash
# scripts/gosec-scan.sh — gosec 扫描脚本

set -euo pipefail

REPORT_DIR="security-reports"
mkdir -p "$REPORT_DIR"

echo "=== gosec SAST 扫描 ==="

# 运行 gosec，生成多种格式报告
gosec -fmt=json -out="${REPORT_DIR}/gosec-report.json" \
      -severity=medium -confidence=medium \
      -exclude-dir=vendor -exclude-dir=testdata \
      ./... || GOSEC_EXIT=$?

# 同时生成 SARIF 格式（用于 GitHub/GitLab 集成）
gosec -fmt=sarif -out="${REPORT_DIR}/gosec-report.sarif" \
      -severity=medium -confidence=medium \
      -exclude-dir=vendor -exclude-dir=testdata \
      ./... 2>/dev/null || true

# 解析结果
if [ -f "${REPORT_DIR}/gosec-report.json" ]; then
    TOTAL=$(jq '.Issues | length' "${REPORT_DIR}/gosec-report.json")
    HIGH=$(jq '[.Issues[] | select(.severity == "HIGH")] | length' "${REPORT_DIR}/gosec-report.json")
    MEDIUM=$(jq '[.Issues[] | select(.severity == "MEDIUM")] | length' "${REPORT_DIR}/gosec-report.json")

    echo "  总发现: ${TOTAL}"
    echo "  HIGH: ${HIGH}  |  MEDIUM: ${MEDIUM}"

    if [ "${HIGH}" -gt 0 ]; then
        echo "  ❌ 存在 HIGH 级别问题，阻断构建"
        exit 1
    fi
fi

echo "  ✅ gosec 扫描通过"
```

### 29.10.2 govulncheck 集成脚本

```bash
#!/bin/bash
# scripts/govulncheck-scan.sh — Go 依赖漏洞扫描

set -euo pipefail

REPORT_DIR="security-reports"
mkdir -p "$REPORT_DIR"

echo "=== govulncheck SCA 扫描 ==="

# 安装 govulncheck（如果未安装）
if ! command -v govulncheck &> /dev/null; then
    go install golang.org/x/vuln/cmd/govulncheck@latest
fi

# 运行 govulncheck
govulncheck -json ./... > "${REPORT_DIR}/govulncheck-report.json" 2>&1 || VULN_EXIT=$?

# 解析 JSON 结果
if [ -f "${REPORT_DIR}/govulncheck-report.json" ]; then
    # 提取被调用的漏洞（实际影响的）
    CALLED=$(jq '[.[] | select(.finding != null) | select(.finding.trace[0].function != null)] | length' \
        "${REPORT_DIR}/govulncheck-report.json" 2>/dev/null || echo "0")

    echo "  实际被调用的漏洞: ${CALLED}"

    if [ "${CALLED}" -gt 0 ]; then
        echo ""
        echo "  受影响的漏洞详情:"
        jq -r '.[] | select(.finding != null) | select(.finding.trace[0].function != null) |
            "  🔴 \(.finding.osv) — \(.finding.trace[0].module)@\(.finding.trace[0].version)"' \
            "${REPORT_DIR}/govulncheck-report.json" 2>/dev/null || true
        echo ""
        echo "  ❌ 存在被调用的已知漏洞，阻断构建"
        exit 1
    fi
fi

echo "  ✅ govulncheck 扫描通过"
```

```go
// tools/vulncheck/main.go — 编程方式调用 govulncheck
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"os"

	"golang.org/x/vuln/scan"
)

// VulnReport 漏洞报告
type VulnReport struct {
	TotalVulns  int         `json:"total_vulns"`
	CalledVulns int         `json:"called_vulns"`
	Blocked     bool        `json:"blocked"`
	Details     []VulnEntry `json:"details"`
}

// VulnEntry 单条漏洞
type VulnEntry struct {
	ID       string `json:"id"`
	Module   string `json:"module"`
	Package  string `json:"package"`
	Function string `json:"function,omitempty"`
	Called   bool   `json:"called"`
}

func main() {
	ctx := context.Background()

	// 使用 govulncheck 的 scan 命令
	cmd := scan.Command(ctx, "./...")
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr

	if err := cmd.Start(); err != nil {
		fmt.Fprintf(os.Stderr, "govulncheck 启动失败: %v\n", err)
		os.Exit(1)
	}

	if err := cmd.Wait(); err != nil {
		fmt.Fprintf(os.Stderr, "发现漏洞，请检查报告\n")
		os.Exit(1)
	}

	fmt.Println("✅ 未发现已知漏洞")
}

func writeReport(report VulnReport, path string) error {
	data, err := json.MarshalIndent(report, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(path, data, 0644)
}
```

### 29.10.3 golangci-lint 安全 Linter 配置

```yaml
# .golangci.yml — golangci-lint 安全相关配置
run:
  timeout: 5m
  skip-dirs:
    - vendor
    - testdata
    - mocks

linters:
  enable:
    # ── 安全相关 linter ──────────────────────────────
    - gosec          # Go 安全检查（等同于独立 gosec）
    - bodyclose      # HTTP response body 未关闭
    - noctx          # HTTP 请求未传递 context
    - rowserrcheck   # sql.Rows.Err() 未检查
    - sqlclosecheck  # sql.Rows 未关闭
    - makezero       # 切片初始化后 append 可能导致 bug
    - exportloopref  # 循环变量引用导出

    # ── 代码质量 linter ──────────────────────────────
    - errcheck       # 未检查的错误
    - govet          # Go vet 检查
    - staticcheck    # 静态分析
    - ineffassign    # 无效赋值
    - unused         # 未使用的代码

linters-settings:
  gosec:
    # 包含所有规则
    includes:
      - G101  # 硬编码凭据
      - G102  # 绑定到所有接口
      - G103  # unsafe 包使用
      - G104  # 未检查的错误
      - G106  # ssh.InsecureIgnoreHostKey
      - G107  # URL 作为 HTTP 请求参数
      - G108  # /debug/pprof 自动暴露
      - G109  # 整数溢出
      - G110  # 解压炸弹
      - G111  # 目录遍历
      - G112  # 慢速 HTTP 攻击
      - G114  # net/http 默认 ServeMux
      - G201  # SQL 查询拼接
      - G202  # SQL 字符串格式化
      - G203  # HTML 模板未转义
      - G204  # 命令注入
      - G301  # 目录权限过大
      - G302  # 文件权限过大
      - G303  # 可预测的临时文件
      - G304  # 文件路径注入
      - G305  # Zip Slip
      - G306  # 写入文件权限过大
      - G401  # 弱加密算法 (MD5/SHA1)
      - G402  # TLS 不安全配置
      - G403  # RSA 密钥过短
      - G404  # 不安全的随机数
      - G501  # 导入黑名单 crypto/md5
      - G502  # 导入黑名单 crypto/des
      - G503  # 导入黑名单 crypto/rc4
      - G504  # 导入黑名单 net/http/cgi
      - G505  # 导入黑名单 crypto/sha1
      - G601  # 隐式内存别名
    # 排除特定规则（如果需要）
    excludes: []
    # 设置置信度
    confidence: medium
    severity: medium
    config:
      # G101: 硬编码凭据检测的正则
      G101:
        pattern: "(?i)(password|secret|token|api_key|apikey|access_key)\\s*=\\s*['\"][^'\"]{8,}"

  bodyclose:
    # 无额外配置

  noctx:
    # 无额外配置

issues:
  # 不排除任何安全相关的 issue
  exclude-rules:
    - path: _test\.go
      linters:
        - gosec  # 测试文件中放宽安全检查

severity:
  default-severity: warning
  rules:
    - linters:
        - gosec
      severity: error
```

```bash
# 运行安全相关 lint
golangci-lint run --enable gosec,bodyclose,noctx,rowserrcheck,sqlclosecheck

# 仅运行 gosec
golangci-lint run --disable-all --enable gosec

# 生成 JSON 报告
golangci-lint run --out-format json > golangci-lint-report.json
```

## 29.11 漏洞管理 SLA

| 严重程度 | CVSS 范围 | 修复 SLA | 升级时间 | 示例 |
|---------|----------|---------|---------|------|
| **Critical** | 9.0 – 10.0 | **24 小时** | 4 小时未响应 → VP | RCE、SQL 注入（无认证） |
| **High** | 7.0 – 8.9 | **7 天** | 3 天未响应 → Director | 认证绕过、SSRF |
| **Medium** | 4.0 – 6.9 | **30 天** | 14 天未响应 → Manager | XSS（存储型）、信息泄露 |
| **Low** | 0.1 – 3.9 | **90 天** | 60 天未响应 → Tech Lead | 低风险信息泄露、版本暴露 |
| **Info** | 0.0 | **下个季度** | 不升级 | 最佳实践建议 |

**SLA 补充说明：**

- **修复 SLA** 从漏洞确认（非误报）之日起计算
- **豁免申请** 必须在 SLA 到期前提交，需安全团队审批
- **SLA 违规** 纳入团队安全 KPI，季度安全评审中回顾
- **零日漏洞** 不受常规 SLA 约束，启动应急响应流程

## 29.12 Security Champion 计划

每个团队指定 1–2 名 Security Champion：

- 参加安全培训（每季度）
- 审查团队的安全实践
- 推动安全工具的采用
- 作为安全团队和开发团队的桥梁
- 参与威胁建模和安全设计评审

## 29.13 小结

- **DevSecOps** 将安全融入 CI/CD 的每个阶段
- **SAST**（Semgrep）在编码阶段发现漏洞
- **SCA**（Snyk）检测依赖中的已知漏洞
- **DAST**（ZAP）在运行时发现安全问题
- **IaC 扫描**（Checkov）确保基础设施配置安全
- **安全门禁** 根据严重程度决定阻断还是警告
- **Security Champion** 是安全文化落地的关键
- **多语言支持**：Python（Bandit + pip-audit）、Java（SpotBugs + Dependency-Check）、Go（gosec + govulncheck）各有成熟工具链
- **漏洞管理 SLA** 确保漏洞在规定时间内得到修复
- **安全门禁策略** 应可配置、可审计、支持豁免流程