Soc2 Compliance

SOC 2 Type II Compliance

Implement and document SOC 2 Type II controls for Trust Services Criteria to achieve enterprise-ready compliance and customer trust.

When to Use This Skill

Use this skill when you need to:

• Prepare for SOC 2 Type II audit
• Implement Trust Services Criteria controls
• Document compliance evidence
• Build enterprise SaaS security
• Meet customer security requirements
• Establish security governance

Trigger phrases: "SOC 2 compliance", "SOC 2 audit", "Trust Services Criteria", "enterprise security", "security controls", "audit preparation"

What This Skill Does

Trust Services Criteria

Instructions

Step 1: Security Controls (CC1-CC9)

Access Control (CC6)

# Role-Based Access Control (RBAC)
from enum import Enum
from functools import wraps

class Role(Enum):
    ADMIN = "admin"
    USER = "user"
    READONLY = "readonly"

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"

ROLE_PERMISSIONS = {
    Role.ADMIN: {Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN},
    Role.USER: {Permission.READ, Permission.WRITE},
    Role.READONLY: {Permission.READ},
}

def require_permission(permission: Permission):
    """Decorator to enforce permission checks."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user = get_current_user()
            user_permissions = ROLE_PERMISSIONS.get(user.role, set())
            if permission not in user_permissions:
                audit_log.warning(
                    "access_denied",
                    user_id=user.id,
                    required_permission=permission.value,
                    resource=func.__name__
                )
                raise PermissionDeniedError(f"Missing permission: {permission.value}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

Authentication (CC6.1)

from datetime import datetime, timedelta
import secrets
import bcrypt

class AuthenticationService:
    """SOC 2 compliant authentication service."""

    # CC6.1: Logical and physical access controls
    MIN_PASSWORD_LENGTH = 12
    PASSWORD_HISTORY_COUNT = 12
    MAX_LOGIN_ATTEMPTS = 5
    LOCKOUT_DURATION = timedelta(minutes=30)
    SESSION_TIMEOUT = timedelta(hours=24)
    MFA_REQUIRED = True

    def authenticate(self, email: str, password: str, mfa_token: str = None) -> Session:
        """Authenticate user with password and optional MFA."""
        user = self.user_repository.find_by_email(email)

        # Check account lockout
        if self._is_locked_out(user):
            audit_log.warning("auth_attempt_locked", user_id=user.id)
            raise AccountLockedError()

        # Verify password
        if not bcrypt.checkpw(password.encode(), user.password_hash):
            self._record_failed_attempt(user)
            audit_log.warning("auth_failed", user_id=user.id, reason="invalid_password")
            raise InvalidCredentialsError()

        # Verify MFA if required
        if self.MFA_REQUIRED:
            if not mfa_token or not self._verify_mfa(user, mfa_token):
                audit_log.warning("auth_failed", user_id=user.id, reason="invalid_mfa")
                raise MFARequiredError()

        # Create session
        session = self._create_session(user)
        audit_log.info("auth_success", user_id=user.id, session_id=session.id)

        return session

Encryption (CC6.7)

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

class EncryptionService:
    """SOC 2 compliant encryption service."""

    # CC6.7: Encryption of data in transit and at rest
    ALGORITHM = "AES-256-GCM"
    KEY_ROTATION_DAYS = 90

    def __init__(self, key_vault: KeyVault):
        self.key_vault = key_vault

    def encrypt_data(self, data: bytes, classification: str) -> EncryptedData:
        """Encrypt sensitive data with appropriate key."""
        key = self.key_vault.get_current_key(classification)

        fernet = Fernet(key.value)
        encrypted = fernet.encrypt(data)

        audit_log.info(
            "data_encrypted",
            classification=classification,
            key_id=key.id,
            data_size=len(data)
        )

        return EncryptedData(
            ciphertext=encrypted,
            key_id=key.id,
            algorithm=self.ALGORITHM
        )

Step 2: Availability Controls (A1)

class AvailabilityMonitor:
    """SOC 2 Availability criteria implementation."""

    # A1.1: System availability monitoring
    SLA_TARGET = 99.9  # percent
    HEALTH_CHECK_INTERVAL = 30  # seconds

    def __init__(self):
        self.metrics = MetricsCollector()
        self.alerting = AlertingService()

    async def health_check(self) -> HealthStatus:
        """Comprehensive system health check."""
        checks = {
            "database": self._check_database(),
            "cache": self._check_cache(),
            "storage": self._check_storage(),
            "external_apis": self._check_external_apis(),
        }

        results = await asyncio.gather(*checks.values(), return_exceptions=True)
        status = HealthStatus(
            healthy=all(r.healthy for r in results if not isinstance(r, Exception)),
            checks={k: r for k, r in zip(checks.keys(), results)},
            timestamp=datetime.utcnow()
        )

        self.metrics.record("health_check", status)

        if not status.healthy:
            await self.alerting.send_alert(
                severity="critical",
                message="System health check failed",
                details=status.to_dict()
            )

        return status

    def calculate_uptime(self, period: timedelta) -> float:
        """Calculate system uptime for SLA reporting."""
        incidents = self.incident_repository.find_by_period(period)
        total_downtime = sum(i.duration for i in incidents)
        total_time = period.total_seconds()

        uptime = ((total_time - total_downtime) / total_time) * 100
        return round(uptime, 3)

Step 3: Confidentiality Controls (C1)

class DataClassification(Enum):
    """Data classification levels per SOC 2 C1."""
    PUBLIC = "public"
    INTERNAL = "internal"
    CONFIDENTIAL = "confidential"
    RESTRICTED = "restricted"

class DataProtectionService:
    """SOC 2 Confidentiality controls."""

    # C1.1: Confidential information identification
    CLASSIFICATION_RULES = {
        "ssn": DataClassification.RESTRICTED,
        "credit_card": DataClassification.RESTRICTED,
        "email": DataClassification.CONFIDENTIAL,
        "name": DataClassification.INTERNAL,
    }

    def classify_data(self, data: dict) -> dict:
        """Automatically classify data based on content."""
        classification = DataClassification.PUBLIC

        for field, value in data.items():
            field_class = self._detect_classification(field, value)
            if field_class.value > classification.value:
                classification = field_class

        audit_log.info(
            "data_classified",
            classification=classification.value,
            fields=list(data.keys())
        )

        return {
            "data": data,
            "classification": classification,
            "handling_requirements": self._get_handling_requirements(classification)
        }

    def _get_handling_requirements(self, classification: DataClassification) -> dict:
        """Get data handling requirements by classification."""
        requirements = {
            DataClassification.RESTRICTED: {
                "encryption": "required",
                "access_logging": "required",
                "retention_days": 365,
                "deletion": "secure_wipe",
            },
            DataClassification.CONFIDENTIAL: {
                "encryption": "required",
                "access_logging": "required",
                "retention_days": 730,
                "deletion": "standard",
            },
        }
        return requirements.get(classification, {})

Step 4: Processing Integrity Controls (PI1)

class ProcessingIntegrityService:
    """SOC 2 Processing Integrity controls."""

    # PI1.1: Processing accuracy and completeness
    def validate_input(self, data: dict, schema: Schema) -> ValidationResult:
        """Validate input data against schema."""
        errors = []
        warnings = []

        for field, rules in schema.fields.items():
            value = data.get(field)

            if rules.required and value is None:
                errors.append(f"Missing required field: {field}")
                continue

            if value is not None:
                if not self._validate_type(value, rules.type):
                    errors.append(f"Invalid type for {field}")
                if not self._validate_constraints(value, rules.constraints):
                    errors.append(f"Constraint violation for {field}")

        result = ValidationResult(
            valid=len(errors) == 0,
            errors=errors,
            warnings=warnings
        )

        audit_log.info(
            "data_validated",
            valid=result.valid,
            error_count=len(errors),
            schema=schema.name
        )

        return result

    def ensure_idempotency(self, operation_id: str) -> IdempotencyResult:
        """Ensure operation idempotency to prevent duplicate processing."""
        existing = self.idempotency_store.get(operation_id)

        if existing:
            audit_log.info("operation_duplicate", operation_id=operation_id)
            return IdempotencyResult(duplicate=True, result=existing)

        return IdempotencyResult(duplicate=False, result=None)

Step 5: Privacy Controls (P1-P8)

class PrivacyService:
    """SOC 2 Privacy criteria implementation."""

    # P1: Privacy notice
    # P2: Choice and consent
    # P3: Collection
    # P4: Use, retention, and disposal
    # P5: Access
    # P6: Disclosure
    # P7: Quality
    # P8: Monitoring and enforcement

    def record_consent(
        self,
        user_id: str,
        purpose: str,
        scope: list[str],
        expiry: datetime = None
    ) -> Consent:
        """Record user consent for data processing."""
        consent = Consent(
            user_id=user_id,
            purpose=purpose,
            scope=scope,
            granted_at=datetime.utcnow(),
            expires_at=expiry,
            version=self.get_privacy_policy_version()
        )

        self.consent_repository.save(consent)

        audit_log.info(
            "consent_recorded",
            user_id=user_id,
            purpose=purpose,
            scope=scope
        )

        return consent

    def handle_data_subject_request(
        self,
        user_id: str,
        request_type: DataSubjectRequestType
    ) -> DataSubjectResponse:
        """Handle data subject rights requests (GDPR, CCPA)."""
        audit_log.info(
            "dsr_received",
            user_id=user_id,
            request_type=request_type.value
        )

        if request_type == DataSubjectRequestType.ACCESS:
            return self._handle_access_request(user_id)
        elif request_type == DataSubjectRequestType.DELETE:
            return self._handle_deletion_request(user_id)
        elif request_type == DataSubjectRequestType.PORTABILITY:
            return self._handle_portability_request(user_id)

Step 6: Audit Logging

import structlog
from datetime import datetime

class AuditLogger:
    """Comprehensive audit logging for SOC 2 compliance."""

    def __init__(self):
        self.logger = structlog.get_logger("audit")

    def log(
        self,
        event: str,
        severity: str = "info",
        user_id: str = None,
        resource: str = None,
        action: str = None,
        outcome: str = None,
        **details
    ):
        """Log audit event with required SOC 2 fields."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "event": event,
            "severity": severity,
            "user_id": user_id or get_current_user_id(),
            "source_ip": get_client_ip(),
            "resource": resource,
            "action": action,
            "outcome": outcome,
            "request_id": get_request_id(),
            "session_id": get_session_id(),
            **details
        }

        # Log to immutable audit store
        self.logger.info(event, **entry)

        # Also store in database for reporting
        self.audit_repository.save(AuditEntry(**entry))

# Global audit logger
audit_log = AuditLogger()

Evidence Collection Checklist

Security Evidence

• Access control policies documented
• RBAC implementation screenshots
• Encryption configuration evidence
• Penetration test results
• Vulnerability scan reports

Availability Evidence

• Uptime reports for audit period
• Incident response documentation
• Disaster recovery test results
• SLA compliance reports

Confidentiality Evidence

• Data classification matrix
• Encryption key management procedures
• Access request/approval logs
• Data handling procedures

Processing Integrity Evidence

• Input validation documentation
• Error handling procedures
• Change management records
• QA test results

Privacy Evidence

• Privacy policy current version
• Consent management records
• Data subject request logs
• Retention policy documentation

Quality Checklist

• All 5 Trust Services Criteria addressed
• Control descriptions documented
• Evidence mapped to controls
• Testing procedures defined
• Exceptions documented
• Remediation plans in place
• Audit trail complete
• Policies reviewed and approved

Related Skills

• iso27001-compliance - ISO 27001 implementation
• security-review - Security vulnerability review
• gdpr-compliance - GDPR compliance

Version: 1.0.0 Last Updated: December 2025 Based on: AI Templates compliance_governance/compliance_frameworks/

Iterative Refinement Strategy

This skill is optimized for an iterative approach:

1. Execute: Perform the core steps defined above.
2. Review: Critically analyze the output (coverage, quality, completeness).
3. Refine: If targets aren't met, repeat the specific implementation steps with improved context.
4. Loop: Continue until the definition of done is satisfied.