performing-soc2-type2-audit-preparation

执行 SOC 2 Type II 审计准备工作

使用场景

• 在准备与 CPA 事务所开展 SOC 2 Type II 审计业务时
• 在针对 AICPA 信任服务标准开展差距评估时
• 在跨云基础设施和身份提供商自动化收集证据时
• 在验证控制措施在审计期间（3-12 个月）是否有效运行时
• 在构建持续合规监控以在审计之间维持 SOC 2 状态时
• 在整改准备评估期间发现的控制缺口时

前提条件

• 熟悉 AICPA 信任服务标准（CC1-CC9）
• 具有对云提供商 API（AWS、Azure 或 GCP）的只读权限访问
• 具有对身份提供商（Okta、Azure AD、Google Workspace）的访问权限
• 具有对版本控制系统的访问权限（GitHub、GitLab）
• 具有对工单系统的访问权限（Jira、Linear、ServiceNow）
• Python 3.8+，并具有 boto3、requests、pyyaml 依赖项
• 具有收集合规证据的适当授权

操作说明

1. 了解信任服务标准

SOC 2 建立在 AICPA 定义的五类信任服务类别之上。安全性（通用标准 CC1-CC9）是强制性的；其他类别根据业务相关性进行选择：

类别	标准	重点
安全性（强制性）	CC1-CC9	控制环境、风险、访问、运营、变更管理
可用性	A1	系统正常运行时间和灾难恢复
处理完整性	PI1	准确且完整的数据处理
保密性	C1	保护机密信息
隐私	P1-P8	个人信息生命周期

2. 通用标准细分（CC1-CC9）

CC1 - 控制环境： 董事会监督、管理层结构、诚信和道德价值观、人力资源政策、问责制。

CC2 - 沟通与信息： 安全策略的内部/外部沟通、系统边界、角色和职责。

CC3 - 风险评估： 风险识别、欺诈风险分析、变更影响评估、风险容忍度定义。

CC4 - 监控活动： 持续控制评估、缺陷识别、整改跟踪、内部审计。

CC5 - 控制活动： 策略到程序的映射、技术控制、在整个实体中部署控制。

CC6 - 逻辑和物理访问控制： 认证、授权、访问权限授予/撤销、物理安全、加密。

CC7 - 系统运营： 异常检测、事件响应、漏洞管理、变更检测、事件监控。

CC8 - 变更管理： 变更授权、测试、审批工作流、紧急变更、回滚程序。

CC9 - 风险缓解： 供应商风险管理、业务连续性、保险、剩余风险接受。

3. 开展差距评估

在审计期开始前，提前 8-12 周执行准备评估：

# Define control matrix against CC criteria
gap_assessment = {
    "CC1": {
        "CC1.1": {
            "criteria": "COSO Principle 1: Demonstrates commitment to integrity",
            "control": "Code of conduct signed annually by all employees",
            "evidence": "Signed acknowledgments in HR system",
            "status": "implemented",
            "gap": None,
        },
        "CC1.2": {
            "criteria": "COSO Principle 2: Board exercises oversight",
            "control": "Quarterly board security reviews",
            "evidence": "Board meeting minutes with security agenda items",
            "status": "partial",
            "gap": "No documented security committee charter",
        },
    },
}

4. 自动化证据收集

在整个审计期内从集成系统持续收集证据：

import boto3

# CC6 Evidence: AWS IAM access controls
iam = boto3.client("iam")

# Collect MFA status for all IAM users
users = iam.list_users()["Users"]
mfa_evidence = []
for user in users:
    mfa_devices = iam.list_mfa_devices(UserName=user["UserName"])
    mfa_evidence.append({
        "user": user["UserName"],
        "mfa_enabled": len(mfa_devices["MFADevices"]) > 0,
        "created": user["CreateDate"].isoformat(),
    })

# CC7 Evidence: AWS CloudTrail logging status
cloudtrail = boto3.client("cloudtrail")
trails = cloudtrail.describe_trails()["trailList"]
logging_evidence = []
for trail in trails:
    status = cloudtrail.get_trail_status(Name=trail["TrailARN"])
    logging_evidence.append({
        "trail": trail["Name"],
        "is_logging": status["IsLogging"],
        "multi_region": trail.get("IsMultiRegionTrail", False),
        "log_validation": trail.get("LogFileValidationEnabled", False),
    })

5. 验证控制有效性

对于 Type II 审计，证明控制措施在整个审计期内有效运行：

import requests

# CC8 Evidence: Change management - verify all production changes
# had tickets, approvals, and testing before deployment
headers = {"Authorization": f"token {github_token}"}
prs = requests.get(
    "https://api.github.com/repos/org/repo/pulls",
    params={"state": "closed", "base": "main", "per_page": 100},
    headers=headers,
).json()

change_evidence = []
for pr in prs:
    if not pr.get("merged_at"):
        continue
    reviews = requests.get(pr["url"] + "/reviews", headers=headers).json()
    approved = any(r["state"] == "APPROVED" for r in reviews)
    change_evidence.append({
        "pr_number": pr["number"],
        "title": pr["title"],
        "merged_at": pr["merged_at"],
        "approved": approved,
    })

# Flag PRs merged without approval (control exception)
exceptions = [c for c in change_evidence if not c["approved"]]

6. 持续合规监控

设置每日运行的自动化检查以检测控制漂移：

# Daily compliance check - run via cron or Lambda
checks = [
    {"control": "CC6.1", "check": "All IAM users have MFA enabled"},
    {"control": "CC6.6", "check": "No public S3 buckets"},
    {"control": "CC7.1", "check": "CloudTrail logging enabled"},
    {"control": "CC7.2", "check": "GuardDuty findings under threshold"},
    {"control": "CC8.1", "check": "All PRs have required reviews"},
]

for check in checks:
    result = run_compliance_check(check["control"])
    if not result["passing"]:
        send_alert(
            channel="#compliance",
            message=f"Control drift: {check['control']} - {check['check']}",
            details=result["findings"],
        )

7. 为审计师准备证据包

将收集到的证据按标准整理为结构化证据包：

evidence_package = {
    "audit_period": {"start": "2025-04-01", "end": "2026-03-31"},
    "criteria_packages": {
        "CC1_Control_Environment": {
            "CC1.1": ["signed_acknowledgments.csv"],
            "CC1.2": ["board_minutes_q1.pdf", "board_minutes_q2.pdf"],
        },
        "CC6_Logical_Physical_Access": {
            "CC6.1": ["okta_mfa_policy.json", "iam_users_mfa_status.csv"],
            "CC6.2": ["access_review_q1.csv", "access_review_q2.csv"],
            "CC6.3": ["offboarding_tickets.csv", "terminated_user_audit.csv"],
        },
        "CC7_System_Operations": {
            "CC7.1": ["cloudtrail_config.json", "siem_dashboard.png"],
            "CC7.2": ["guardduty_findings_summary.csv"],
            "CC7.3": ["vulnerability_scan_reports/"],
        },
        "CC8_Change_Management": {
            "CC8.1": ["merged_prs_with_approvals.csv"],
        },
    },
}

示例

CC6.2 的自动化访问审查

import boto3
from datetime import datetime, timedelta

iam = boto3.client("iam")

# Find users with no activity in 90 days
inactive_threshold = datetime.utcnow() - timedelta(days=90)
report = iam.get_credential_report()["Content"].decode()

inactive_users = []
for line in report.strip().split("\n")[1:]:
    fields = line.split(",")
    username = fields[0]
    last_used = fields[4]
    if last_used not in ("N/A", "no_information"):
        last_date = datetime.strptime(last_used, "%Y-%m-%dT%H:%M:%S+00:00")
        if last_date < inactive_threshold:
            inactive_users.append({"user": username, "last_active": last_used})

CC7.2 的漏洞管理证据

import requests

headers = {"Authorization": f"Bearer {scanner_token}"}
scans = requests.get(
    "https://scanner.example.com/api/v1/scans",
    params={"status": "completed", "since": "2025-04-01"},
    headers=headers,
).json()

vuln_evidence = {"scan_count": len(scans), "critical_findings": 0, "high_findings": 0}
for scan in scans:
    findings = requests.get(
        f"https://scanner.example.com/api/v1/scans/{scan['id']}/findings",
        headers=headers,
    ).json()
    vuln_evidence["critical_findings"] += len([f for f in findings if f["severity"] == "critical"])
    vuln_evidence["high_findings"] += len([f for f in findings if f["severity"] == "high"])

CC7.3 的事件响应证据

incidents = requests.get(
    "https://pagerduty.com/api/v1/incidents",
    params={"since": "2025-04-01", "until": "2026-03-31"},
    headers={"Authorization": f"Token token={pd_token}"},
).json()

ir_evidence = {
    "total_incidents": len(incidents["incidents"]),
    "incidents": [
        {
            "id": inc["id"],
            "title": inc["title"],
            "severity": inc["urgency"],
            "created": inc["created_at"],
            "resolved": inc.get("last_status_change_at"),
        }
        for inc in incidents["incidents"]
    ],
}