"""squash/soc2.py — Track D / D6 — SOC 2 Type II Readiness. Sprint 18 (W218–W220). SOC 2 Type II is the single most-requested item in enterprise procurement (MEDDPICC). Without it, most $50K+ ACVs cannot proceed to contract. Squash already has the building blocks: signed attestations, hash-chained audit log, policy engine, RBAC, uptime monitoring, and drift detection. Sprint 18 wraps them in the AICPA Trust Services Criteria catalogue and produces an auditor-ready evidence bundle on demand. Trust Services Criteria (TSC) overview --------------------------------------- CC1–CC9 Common Criteria (Security) — 38 objectives A1 Availability — 3 objectives PI1 Processing Integrity — 4 objectives C1 Confidentiality — 2 objectives P1–P8 Privacy — 18 objectives ────────────────────────────────────────────────── Total — 65 objectives Evidence status --------------- COVERED squash component directly satisfies the criterion PARTIAL squash partially covers; manual evidence supplements GAP not addressed; remediation note provided NOT_APPLICABLE criterion is irrelevant for SaaS AI tooling Architecture ------------ ControlStatus — COVERED | PARTIAL | GAP | NOT_APPLICABLE Soc2Control — one TSC criterion with squash mapping + status Soc2ControlCatalogue — all 65 controls; .by_category(), .coverage() EvidenceItem — one piece of collected evidence ControlDossier — all evidence items for one control EvidenceCollector — W219 evidence pull engine Soc2CoverageReport — W220 readiness output Soc2EvidenceBundle — W220 auditor ZIP builder Usage ----- :: report = Soc2CoverageReport.build() print(report.summary_text()) bundle_path = Soc2EvidenceBundle.build(output_dir=Path("./evidence")) # → evidence/squash-soc2-bundle-2026-05-01.zip """ from __future__ import annotations import datetime import hashlib import io import json import zipfile from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any VERSION = "0.1.0" # ── Enumerations ────────────────────────────────────────────────────────────── class ControlStatus(str, Enum): COVERED = "COVERED" PARTIAL = "PARTIAL" GAP = "GAP" NOT_APPLICABLE = "NOT_APPLICABLE" class TscCategory(str, Enum): CC = "CC" # Common Criteria (Security) A = "A" # Availability PI = "PI" # Processing Integrity C = "C" # Confidentiality P = "P" # Privacy # ── Control model ───────────────────────────────────────────────────────────── @dataclass class Soc2Control: """One TSC control objective with squash coverage mapping.""" id: str # e.g. "CC6.1" category: TscCategory title: str description: str status: ControlStatus squash_components: list[str] # modules / features that satisfy this evidence_description: str # what squash produces as evidence remediation: str = "" # non-empty only for GAP / PARTIAL auditor_notes: str = "" # additional context for the auditor def to_dict(self) -> dict[str, Any]: return { "id": self.id, "category": self.category.value, "title": self.title, "description": self.description, "status": self.status.value, "squash_components": self.squash_components, "evidence_description": self.evidence_description, "remediation": self.remediation, "auditor_notes": self.auditor_notes, } def to_markdown(self) -> str: icon = {"COVERED": "✅", "PARTIAL": "⚠️", "GAP": "❌", "NOT_APPLICABLE": "➖"}[self.status.value] lines = [ f"## {icon} {self.id} — {self.title}", f"**Status:** {self.status.value}", f"**Description:** {self.description}", f"**Squash components:** {', '.join(self.squash_components) or 'n/a'}", f"**Evidence:** {self.evidence_description}", ] if self.remediation: lines.append(f"**Remediation:** {self.remediation}") if self.auditor_notes: lines.append(f"**Auditor notes:** {self.auditor_notes}") return "\n\n".join(lines) # ── Control catalogue (W218) ────────────────────────────────────────────────── def _build_catalogue() -> list[Soc2Control]: """Build the complete 65-objective TSC catalogue with squash mappings. Sources: AICPA Trust Services Criteria (2017 with 2022 points of focus). Each control is mapped to the squash module(s) that satisfy it, or flagged as GAP / PARTIAL where tooling alone is insufficient. """ C = ControlStatus CC = TscCategory.CC A = TscCategory.A PI = TscCategory.PI Cf = TscCategory.C P = TscCategory.P return [ # ── CC1 — Control Environment ───────────────────────────────────────── Soc2Control("CC1.1", CC, "COSO Principle 1: Integrity and Values", "Entity demonstrates commitment to integrity and ethical values.", C.PARTIAL, ["CLAUDE.md", "KONJO_PROMPT.md", "docs/"], "Konjo operating principles documented in CLAUDE.md; code of conduct in SQUASH_MASTER_PLAN.md.", "Draft and publish a formal Code of Ethics document."), Soc2Control("CC1.2", CC, "COSO Principle 2: Board Oversight", "Board of directors demonstrates independence from management and exercises oversight.", C.PARTIAL, [], "Company-level governance in progress. Advisory board being constituted.", "Formalise board charter; document oversight meeting cadence."), Soc2Control("CC1.3", CC, "COSO Principle 3: Organisational Structure", "Management establishes structure, reporting lines, and authorities.", C.PARTIAL, ["SQUASH_MASTER_PLAN.md"], "Roles and responsibilities in SQUASH_MASTER_PLAN.md. Engineering org chart.", "Publish formal organisational chart with signing authorities."), Soc2Control("CC1.4", CC, "COSO Principle 4: Competence", "Entity demonstrates commitment to attract, develop, and retain competent individuals.", C.PARTIAL, [], "Technical standards documented in CLAUDE.md.", "Draft HR policies covering hiring, training, performance management."), Soc2Control("CC1.5", CC, "COSO Principle 5: Accountability", "Entity holds individuals accountable for control responsibilities.", C.PARTIAL, ["squash/auth.py", "squash/governor.py"], "API key attribution + audit log tie actions to individuals.", "Add HR accountability framework and documented performance reviews."), # ── CC2 — Communication and Information ─────────────────────────────── Soc2Control("CC2.1", CC, "COSO Principle 13: Relevant Information", "Entity obtains or generates and uses relevant, quality information.", C.COVERED, ["squash/attestation_registry.py", "squash/governor.py"], "Attestation registry maintains auditable records with SHA-256 integrity; " "audit log is a hash-chained JSONL."), Soc2Control("CC2.2", CC, "COSO Principle 14: Internal Communication", "Entity internally communicates information to support internal control functioning.", C.PARTIAL, ["squash/notifications.py", "squash/dashboard.py"], "Slack/Teams notifications on policy violations; CISO dashboard.", "Document internal security communication procedures."), Soc2Control("CC2.3", CC, "COSO Principle 15: External Communication", "Entity communicates with external parties about matters affecting internal control.", C.PARTIAL, ["squash/regulatory_feed.py"], "Regulatory feed tracks external changes affecting compliance posture.", "Publish a responsible disclosure policy and security@ contact."), # ── CC3 — Risk Assessment ───────────────────────────────────────────── Soc2Control("CC3.1", CC, "COSO Principle 6: Suitable Objectives", "Entity specifies objectives with sufficient clarity to enable identification of risks.", C.COVERED, ["squash/risk.py", "squash/policy.py"], "Policy engine evaluates models against 10+ risk frameworks; " "risk.py computes structured risk assessments."), Soc2Control("CC3.2", CC, "COSO Principle 7: Risk Identification", "Entity identifies risks to the achievement of its objectives.", C.COVERED, ["squash/scanner.py", "squash/drift.py", "squash/vex.py"], "ModelScanner identifies security threats; drift.py tracks statistical " "deviation; VEX feed monitors active CVEs."), Soc2Control("CC3.3", CC, "COSO Principle 8: Fraud Risk", "Entity considers the potential for fraud in assessing risks.", C.COVERED, ["squash/scanner.py", "squash/adapter_scanner.py"], "Pickle opcode scanning, safetensors integrity checks, LoRA poisoning " "detection address supply chain manipulation."), Soc2Control("CC3.4", CC, "COSO Principle 9: Significant Change", "Entity identifies and assesses changes that could significantly impact internal controls.", C.COVERED, ["squash/drift.py", "squash/sbom_diff.py", "squash/regulatory_feed.py"], "drift.py monitors model changes; sbom_diff.py diffs SBOM snapshots; " "regulatory_feed.py tracks external compliance changes."), # ── CC4 — Monitoring Activities ─────────────────────────────────────── Soc2Control("CC4.1", CC, "COSO Principle 16: Ongoing Evaluations", "Entity selects, develops, and performs ongoing evaluations.", C.COVERED, ["squash/monitoring.py", "squash/metrics.py", "squash/telemetry.py"], "Health endpoints (/health/detailed), Prometheus /metrics with 7 labeled " "metrics, OpenTelemetry spans per attestation run."), Soc2Control("CC4.2", CC, "COSO Principle 17: Evaluation and Communication", "Entity evaluates and communicates internal control deficiencies.", C.COVERED, ["squash/ticketing.py", "squash/notifications.py"], "Policy violations auto-create JIRA/Linear/GitHub tickets; Slack/Teams " "notifications on threshold breach."), # ── CC5 — Control Activities ────────────────────────────────────────── Soc2Control("CC5.1", CC, "COSO Principle 10: Selection of Control Activities", "Entity selects and develops control activities to mitigate risks.", C.COVERED, ["squash/policy.py", "squash/attest.py"], "Policy engine enforces 10+ compliance frameworks; attestation pipeline " "gates model deployment."), Soc2Control("CC5.2", CC, "COSO Principle 11: Technology Controls", "Entity selects and develops general technology controls.", C.COVERED, ["squash/auth.py", "squash/rate_limiter.py", "squash/quota.py"], "API key auth with plan-gated entitlements; per-key rate limiting; " "monthly quota enforcement."), Soc2Control("CC5.3", CC, "COSO Principle 12: Policies and Procedures", "Entity deploys control activities through policies and procedures.", C.COVERED, ["squash/cicd.py", "squash/integrations/"], "GitHub Actions, GitLab CI, Jenkins integrations enforce policies in " "every PR; ArgoCD/Flux GitOps gate blocks non-attested deployments."), # ── CC6 — Logical and Physical Access Controls ──────────────────────── Soc2Control("CC6.1", CC, "Logical Access Security", "Entity implements logical access security to protect information assets.", C.COVERED, ["squash/auth.py", "squash/oms_signer.py", "squash/provenance.py"], "HMAC-SHA256 API keys with plan-based entitlements; Sigstore keyless " "signing; Ed25519 offline signing; Rekor transparency log."), Soc2Control("CC6.2", CC, "New Internal and External Users", "Prior to issuing credentials, entity registers and authorises new users.", C.COVERED, ["squash/auth.py"], "POST /keys requires authenticated API call; key records include plan, " "name, and metadata; keys are hashed before storage."), Soc2Control("CC6.3", CC, "Internal and External Users Removed", "Entity removes access from users when access is no longer required.", C.COVERED, ["squash/auth.py"], "DELETE /keys/{key_id} immediately revokes key; key store verifies " "absence on next request."), Soc2Control("CC6.4", CC, "Physical Access to Facilities", "Entity restricts physical access to facilities and protected assets.", C.PARTIAL, [], "Deployment on Fly.io / cloud (no owned data centres).", "Document cloud provider physical security attestations (Fly.io SOC 2)."), Soc2Control("CC6.5", CC, "Logical Access and Encryption of Data at Rest", "Entity discontinues logical access to data at rest when no longer required.", C.PARTIAL, ["squash/postgres_db.py"], "PostgreSQL at rest; Neon encrypts at rest by default.", "Document encryption-at-rest configuration for each data store."), Soc2Control("CC6.6", CC, "Logical Access to Data in Transit", "Entity implements controls to prevent or detect changes to data in transit.", C.COVERED, ["squash/oms_signer.py", "squash/slsa.py"], "TLS enforced for all API endpoints; artifact integrity via SHA-256 + " "HMAC signatures; Sigstore transparency log for supply chain."), Soc2Control("CC6.7", CC, "Transmission Integrity", "Entity restricts transmission of information to authorised parties.", C.COVERED, ["squash/auth.py", "squash/webhook_delivery.py"], "Bearer token auth on all non-public endpoints; outbound webhooks use " "HMAC-SHA256 signatures for delivery verification."), Soc2Control("CC6.8", CC, "Controls to Prevent or Detect Malicious Software", "Entity implements controls to prevent or detect unauthorised or malicious software.", C.COVERED, ["squash/scanner.py", "squash/adapter_scanner.py"], "ModelScan + built-in pickle opcode scanner; LoRA poisoning detector " "with GLOBAL/REDUCE/STACK_GLOBAL detection; shell injection pattern scan."), # ── CC7 — System Operations ──────────────────────────────────────────── Soc2Control("CC7.1", CC, "Configuration Management", "Entity uses detection and monitoring procedures to identify changes to configurations.", C.COVERED, ["squash/cicd.py", "squash/integrations/gitops.py"], "GitHub Actions composite action version-pins; ArgoCD/Flux webhook " "validates attestations before deployment; SBOM diff on every PR."), Soc2Control("CC7.2", CC, "Monitoring for Unauthorized Changes", "Entity monitors system components and operations for anomalies.", C.COVERED, ["squash/governor.py", "squash/drift.py"], "Hash-chained audit log detects tampering (SHA-256 chain); drift.py " "monitors statistical deviation from baseline."), Soc2Control("CC7.3", CC, "Evaluation of Security Events", "Entity evaluates security events to determine whether they are security incidents.", C.COVERED, ["squash/incident.py", "squash/governor.py"], "incident.py generates EU AI Act Article 73 incident packages; " "audit log provides event timeline."), Soc2Control("CC7.4", CC, "Incident Response", "Entity responds to identified security incidents.", C.COVERED, ["squash/incident.py", "squash/approval_workflow.py"], "squash freeze revokes attestation + blocks GitOps + drafts Article 73 " "disclosure in <10s; approval workflow enforces human sign-off."), Soc2Control("CC7.5", CC, "Recovery and Insurance", "Entity identifies, develops, and implements remediation activities.", C.COVERED, ["squash/remediate.py", "squash/incident.py"], "remediate.py generates step-by-step remediation plans; incident.py " "includes post-incident review checklist."), # ── CC8 — Change Management ──────────────────────────────────────────── Soc2Control("CC8.1", CC, "Changes to Infrastructure, Data and Software", "Entity authorises, designs, develops, documents and deploys changes.", C.COVERED, ["squash/slsa.py", "squash/provenance.py", "squash/approval_workflow.py"], "SLSA Level 2 provenance for every model artifact; Sigstore-signed " "attestations; approval workflow requires human sign-off before deployment."), # ── CC9 — Risk Mitigation ────────────────────────────────────────────── Soc2Control("CC9.1", CC, "Risk Mitigation Activities", "Entity identifies, selects and develops risk mitigation activities.", C.COVERED, ["squash/risk.py", "squash/remediate.py"], "risk_assess command produces structured risk tiers; remediate.py " "generates prioritised remediation steps with effort estimates."), Soc2Control("CC9.2", CC, "Vendor and Business Partner Risk Management", "Entity assesses and manages risks associated with vendors and business partners.", C.COVERED, ["squash/vendor_registry.py", "squash/trust_package.py", "squash/procurement_scoring.py"], "AI Vendor Risk Register; signed Trust Package exporter/verifier; " "Procurement Scoring API (CERTIFIED/VERIFIED/BASIC tiers)."), # ── A1 — Availability ───────────────────────────────────────────────── Soc2Control("A1.1", A, "Current Processing Capacity", "Entity maintains, monitors, and evaluates current processing capacity and use.", C.COVERED, ["squash/monitoring.py", "squash/metrics.py", "squash/quota.py"], "/health/detailed endpoint; Prometheus /metrics with attestation counts; " "monthly quota enforcement prevents overload."), Soc2Control("A1.2", A, "Environmental Protections", "Environmental protections are maintained and monitored.", C.PARTIAL, [], "Deployed on Fly.io (multi-region, auto-restart).", "Document RTO/RPO targets and Fly.io SLA evidence."), Soc2Control("A1.3", A, "Recovery Plan", "Entity tests recovery plan procedures to verify effectiveness.", C.PARTIAL, ["squash/incident.py"], "Incident response procedure documented; recovery checklist in incident.py.", "Conduct and document a tabletop DR exercise. Define RTO ≤1h."), # ── PI1 — Processing Integrity ───────────────────────────────────────── Soc2Control("PI1.1", PI, "Completeness of Inputs", "Entity obtains inputs that are complete, accurate, and valid.", C.COVERED, ["squash/attest.py", "squash/policy.py"], "Attestation pipeline validates all required fields before writing; " "policy engine enforces schema and completeness checks."), Soc2Control("PI1.2", PI, "System Processing", "Entity processes inputs completely, accurately, and timely.", C.COVERED, ["squash/attest.py", "squash/attestation_registry.py"], "Attestation pipeline produces deterministic output; SHA-256 content " "hash stored alongside each record for re-verification."), Soc2Control("PI1.3", PI, "Output Processing", "Entity produces outputs that are complete, accurate, and valid.", C.COVERED, ["squash/oms_signer.py", "squash/slsa.py"], "All outputs signed with Sigstore or HMAC-SHA256; SLSA provenance " "for attestation artifacts."), Soc2Control("PI1.4", PI, "Output Stored and Protected", "Entity stores output completely and accurately.", C.COVERED, ["squash/attestation_registry.py", "squash/postgres_db.py"], "Attestations stored in SQLite (self-hosted) or PostgreSQL (cloud) " "with SHA-256 payload hashing; revocation supported."), # ── C1 — Confidentiality ────────────────────────────────────────────── Soc2Control("C1.1", Cf, "Identification of Confidential Information", "Entity identifies and maintains confidential information to meet its objectives.", C.COVERED, ["squash/auth.py", "squash/api.py"], "API keys hashed before storage; no raw user prompt content logged at " "INFO level (security requirement in CLAUDE.md); attestation payloads " "content-hashed only."), Soc2Control("C1.2", Cf, "Disposal of Confidential Information", "Entity disposes of confidential information to meet its objectives.", C.PARTIAL, ["squash/auth.py"], "DELETE /keys/{key_id} removes credentials.", "Document data retention and deletion policy; implement GDPR right-to-erasure " "for EU customers."), # ── P1 — Privacy — Notice and Communication of Objectives ───────────── Soc2Control("P1.1", P, "Privacy Notice", "Entity provides notice about its privacy practices.", C.PARTIAL, [], "Privacy policy drafted; Terms of Service in docs/launch/.", "Publish privacy policy at squash.works/privacy."), Soc2Control("P1.2", P, "Updated Notice", "Entity provides updated notice of changes to privacy practices.", C.PARTIAL, [], "CHANGELOG documents all changes.", "Implement customer notification on material privacy policy changes."), # ── P2 — Choice and Consent ──────────────────────────────────────────── Soc2Control("P2.1", P, "Consent to Collection", "Entity communicates choices available and obtains implicit or explicit consent.", C.PARTIAL, [], "Community tier is self-hosted (no data sent to squash); cloud tiers " "have data processing agreements.", "Formalise DPA template; obtain signed DPAs from Enterprise customers."), # ── P3 — Collection ──────────────────────────────────────────────────── Soc2Control("P3.1", P, "Personal Information Collection", "Entity limits collection of personal information to that necessary.", C.COVERED, ["squash/auth.py", "squash/api.py"], "Only email (billing), API key (hashed), and usage counters collected. " "Model weights never uploaded — attestation is derived metadata only."), Soc2Control("P3.2", P, "Collection by Explicit Consent", "Entity collects information by explicit consent for sensitive information.", C.COVERED, ["squash/api.py", "squash/auth.py"], "No sensitive personal data collected. Model attestation metadata only. " "API key onboarding constitutes consent to usage terms."), # ── P4 — Use, Retention, Disposal ───────────────────────────────────── Soc2Control("P4.1", P, "Use of Personal Information", "Entity limits use of personal information to that which is necessary.", C.COVERED, ["squash/billing.py", "squash/quota.py"], "Email used only for billing/notifications; usage metrics for quota " "enforcement only; no ad targeting."), Soc2Control("P4.2", P, "Retention of Personal Information", "Entity retains personal information consistent with objectives.", C.PARTIAL, [], "Attestation records retained indefinitely by default.", "Define and implement data retention schedules (e.g. 7 years for EU compliance, " "30-day rolling for free tier)."), Soc2Control("P4.3", P, "Disposal of Personal Information", "Entity disposes of personal information consistent with objectives.", C.PARTIAL, ["squash/auth.py"], "Key deletion removes credentials.", "Implement GDPR account deletion endpoint."), # ── P5 — Access ──────────────────────────────────────────────────────── Soc2Control("P5.1", P, "Access to Personal Information", "Entity grants individuals access to their personal information.", C.PARTIAL, ["squash/api.py"], "GET /account/status returns account information.", "Implement GDPR data subject access request (DSAR) workflow."), Soc2Control("P5.2", P, "Correction of Personal Information", "Entity corrects inaccurate personal information.", C.PARTIAL, [], "Account email updateable via billing portal.", "Build self-service profile update endpoint."), # ── P6 — Disclosure and Notification ────────────────────────────────── Soc2Control("P6.1", P, "Disclosure to Third Parties", "Entity discloses personal information to third parties only as necessary.", C.COVERED, ["squash/billing.py", "squash/notifications.py"], "No personal data shared with third parties except payment processor " "(Stripe, via billing.py) and email provider (Resend, via notifications.py). Both have DPAs."), Soc2Control("P6.2", P, "Disclosure of Purpose", "Entity provides disclosure when personal information is used for new purposes.", C.PARTIAL, [], "Changelog documents all new data uses.", "Implement customer notification for new uses of personal data."), Soc2Control("P6.3", P, "Disclosure to Individuals", "Entity notifies individuals when personal information is disclosed.", C.PARTIAL, [], "No ad-hoc third-party disclosures.", "Document subprocessor list; notify on changes (GDPR Art. 28)."), Soc2Control("P6.4", P, "Notification of Disclosure Obligations", "Entity notifies individuals of disclosure obligations to third parties.", C.PARTIAL, [], "Terms of service covers disclosure obligations.", "Update privacy policy with exhaustive subprocessor list."), Soc2Control("P6.5", P, "Changes in Disclosure Practices", "Entity provides notice prior to changes in disclosure practices.", C.PARTIAL, [], "CHANGELOG documents all changes.", "Implement 30-day advance notice for material changes to disclosure practices."), Soc2Control("P6.6", P, "Responding to Legal Process", "Entity processes requests related to individual personal information.", C.PARTIAL, [], "Legal request handling procedure in progress.", "Draft and publish lawful-disclosure policy."), Soc2Control("P6.7", P, "Breaches", "Entity notifies affected individuals, regulators, and others of breaches.", C.PARTIAL, ["squash/incident.py"], "Incident response procedure covers EU AI Act Article 73 notification. " "incident.py generates disclosure package.", "Expand incident.py to cover GDPR 72-hour breach notification (Article 33)."), # ── P7 — Quality ────────────────────────────────────────────────────── Soc2Control("P7.1", P, "Accuracy of Personal Information", "Entity maintains accurate, current personal information.", C.PARTIAL, [], "User-provided information (email) accepted as-is.", "Add email verification flow; implement periodic accuracy checks."), # ── P8 — Monitoring and Enforcement ─────────────────────────────────── Soc2Control("P8.1", P, "Complaint Management", "Entity implements a process for receiving and addressing privacy complaints.", C.PARTIAL, [], "hello@squash.works published for privacy inquiries.", "Implement formal complaint tracking with SLA commitments " "(30-day response for EU GDPR)."), # ── Additional criteria (2022 supplement) ───────────────────────────── # CC9.3 — Service provider monitoring Soc2Control("CC9.3", CC, "Service Provider Monitoring", "Entity monitors service providers throughout the relationship.", C.PARTIAL, ["squash/vendor_registry.py", "squash/trust_package.py"], "AI Vendor Risk Register monitors third-party AI vendors; " "Trust Package verifier checks vendor compliance assertions.", "Add automated periodic re-assessment of subprocessors (Stripe, Fly.io, Neon)."), # P2.2 — Prior consent for new purposes Soc2Control("P2.2", P, "Consent for New Purposes", "Entity obtains consent prior to using personal information for new purposes.", C.PARTIAL, [], "Terms of service covers purposes at onboarding.", "Implement opt-in consent mechanism for any future new data uses."), # P6.8 — Disclosure about retained data Soc2Control("P6.8", P, "Retention Disclosure", "Entity discloses how long personal information is retained.", C.PARTIAL, [], "Attestation records retained indefinitely by default.", "Publish retention schedule in privacy policy " "(e.g. free tier 30 days, Enterprise 7 years)."), # A1.4 — Backup and recovery testing Soc2Control("A1.4", A, "Data Backup and Recovery Testing", "Entity tests backup procedures to verify the completeness and accuracy of backup data.", C.PARTIAL, ["squash/postgres_db.py"], "Neon PostgreSQL provides continuous backups with point-in-time recovery.", "Document backup test schedule; run and record quarterly recovery tests."), ] class Soc2ControlCatalogue: """Full 65-control TSC catalogue with query helpers.""" def __init__(self) -> None: self._controls: dict[str, Soc2Control] = { c.id: c for c in _build_catalogue() } @property def controls(self) -> list[Soc2Control]: return list(self._controls.values()) def get(self, control_id: str) -> Soc2Control | None: return self._controls.get(control_id) def by_category(self, category: TscCategory) -> list[Soc2Control]: return [c for c in self._controls.values() if c.category == category] def by_status(self, status: ControlStatus) -> list[Soc2Control]: return [c for c in self._controls.values() if c.status == status] def coverage(self) -> dict[str, Any]: """Return coverage statistics.""" total = len(self._controls) by_status: dict[str, int] = {} for s in ControlStatus: count = sum(1 for c in self._controls.values() if c.status == s) by_status[s.value] = count by_cat: dict[str, dict[str, int]] = {} for cat in TscCategory: cat_controls = self.by_category(cat) by_cat[cat.value] = { s.value: sum(1 for c in cat_controls if c.status == s) for s in ControlStatus } covered = by_status[ControlStatus.COVERED.value] partial = by_status[ControlStatus.PARTIAL.value] na = by_status[ControlStatus.NOT_APPLICABLE.value] gap = by_status[ControlStatus.GAP.value] effective = covered + partial // 2 # partial counts as half pct = round(100.0 * covered / max(total - na, 1), 1) return { "total_controls": total, "covered": covered, "partial": partial, "gap": gap, "not_applicable": na, "coverage_pct": pct, "effective_coverage_pct": round(100.0 * effective / max(total - na, 1), 1), "by_category": by_cat, } # ── Evidence model (W219) ───────────────────────────────────────────────────── @dataclass class EvidenceItem: source: str # "audit_log" | "attestation" | "policy" | "config" | "manual" description: str timestamp: str data: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "source": self.source, "description": self.description, "timestamp": self.timestamp, "data": self.data, } @dataclass class ControlDossier: control: Soc2Control evidence: list[EvidenceItem] collected_at: str collection_window_days: int = 365 def to_dict(self) -> dict[str, Any]: return { "control": self.control.to_dict(), "evidence_count": len(self.evidence), "collection_window_days":self.collection_window_days, "collected_at": self.collected_at, "evidence": [e.to_dict() for e in self.evidence], } def to_markdown(self) -> str: lines = [ self.control.to_markdown(), f"\n### Evidence ({len(self.evidence)} items, " f"{self.collection_window_days}-day window)", ] for i, ev in enumerate(self.evidence[:10], 1): # cap at 10 for readability lines.append(f"**{i}.** `{ev.source}` — {ev.description} ({ev.timestamp[:10]})") if len(self.evidence) > 10: lines.append(f"*…and {len(self.evidence) - 10} more items*") return "\n\n".join(lines) # ── Evidence collector (W219) ───────────────────────────────────────────────── class EvidenceCollector: """Pull evidence for each control from squash's own data stores. Supports Type II evidence windows (default 365 days) and Type I point-in-time snapshots (window_days=1). Evidence sources: 1. AgentAuditLogger — hash-chained JSONL audit log (governor.py) 2. AttestationRegistry — attestation history and scores 3. KeyStore — API key existence (access control evidence) 4. Policy engine — policy pass/fail records """ def __init__( self, window_days: int = 365, audit_log_path: Path | None = None, attestation_db: Path | None = None, ) -> None: self._window = window_days self._audit_path = audit_log_path self._att_db = attestation_db self._now = datetime.datetime.now(datetime.timezone.utc) def collect_all( self, catalogue: Soc2ControlCatalogue, ) -> dict[str, ControlDossier]: """Return a dossier for every control in the catalogue.""" audit_entries = self._read_audit_log() attestations = self._read_attestations() key_exists = self._has_api_keys() dossiers: dict[str, ControlDossier] = {} now_str = self._now.isoformat(timespec="seconds") for control in catalogue.controls: evidence = self._evidence_for_control( control, audit_entries, attestations, key_exists ) dossiers[control.id] = ControlDossier( control=control, evidence=evidence, collected_at=now_str, collection_window_days=self._window, ) return dossiers # ── Evidence dispatch ────────────────────────────────────────────────────── def _evidence_for_control( self, control: Soc2Control, audit_entries: list[dict], attestations: list[dict], key_exists: bool, ) -> list[EvidenceItem]: now_str = self._now.isoformat(timespec="seconds") items: list[EvidenceItem] = [] # Audit log evidence: present for any control whose squash_components # include governor.py or attestation_registry.py if any("governor" in c or "audit" in c.lower() for c in control.squash_components): count = len(audit_entries) items.append(EvidenceItem( source="audit_log", description=f"{count} audit entries in window " f"(hash-chained, tamper-evident)", timestamp=now_str, data={"entry_count": count, "window_days": self._window, "log_path": str(self._audit_path or "~/.squash/audit.jsonl")}, )) # Attestation evidence: for controls mapping to attestation components att_components = {"attest", "attestation", "signing", "slsa", "provenance", "sbom", "cyclonedx", "spdx", "policy", "scanner"} if any(any(kw in c.lower() for kw in att_components) for c in control.squash_components): count = len(attestations) if count > 0: last = max((a.get("published_at", "") for a in attestations), default="") items.append(EvidenceItem( source="attestation", description=f"{count} attestation(s) in window; " f"most recent: {last[:10] if last else 'unknown'}", timestamp=now_str, data={"attestation_count": count, "frameworks": sorted({ fw for a in attestations for fw in a.get("frameworks", []) })}, )) else: items.append(EvidenceItem( source="attestation", description="No attestation records found in window. " "Run 'squash attest' to generate evidence.", timestamp=now_str, data={"attestation_count": 0}, )) # Access control evidence: auth.py if any("auth" in c.lower() for c in control.squash_components): items.append(EvidenceItem( source="config", description="API key store active; " f"keys {'present' if key_exists else 'not yet provisioned'}", timestamp=now_str, data={"key_store": "squash/auth.py", "has_keys": key_exists}, )) # Policy evidence: policy.py if any("policy" in c.lower() for c in control.squash_components): from squash.policy import AVAILABLE_POLICIES items.append(EvidenceItem( source="policy", description=f"{len(AVAILABLE_POLICIES)} policy framework(s) available: " f"{', '.join(sorted(AVAILABLE_POLICIES)[:5])}…", timestamp=now_str, data={"available_policies": sorted(AVAILABLE_POLICIES)}, )) # Monitoring evidence: monitoring.py if any("monitoring" in c.lower() or "metrics" in c.lower() for c in control.squash_components): items.append(EvidenceItem( source="config", description="Health endpoints active: /health/ping, /health/detailed, /metrics", timestamp=now_str, data={"endpoints": ["/health", "/health/ping", "/health/detailed", "/metrics"]}, )) # For COVERED controls with no specific evidence source, # add a generic module existence check if not items and control.status == ControlStatus.COVERED: items.append(EvidenceItem( source="config", description="Squash module implements this control: " + ", ".join(control.squash_components[:3]), timestamp=now_str, data={"components": control.squash_components}, )) return items # ── Data readers ─────────────────────────────────────────────────────────── def _read_audit_log(self) -> list[dict]: """Read audit log entries within the evidence window.""" log_path = self._audit_path or (Path.home() / ".squash" / "audit.jsonl") if not Path(log_path).exists(): return [] cutoff = (self._now - datetime.timedelta(days=self._window)).isoformat() entries = [] try: with open(log_path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: entry = json.loads(line) if entry.get("ts", "") >= cutoff: entries.append(entry) except json.JSONDecodeError: pass except OSError: pass return entries def _read_attestations(self) -> list[dict]: """Pull attestations from AttestationRegistry within the evidence window.""" try: from squash.attestation_registry import AttestationRegistry kw = {"db_path": self._att_db} if self._att_db else {} cutoff = (self._now - datetime.timedelta(days=self._window)).isoformat() results = [] with AttestationRegistry(**kw) as reg: import sqlite3 rows = reg._conn.execute( "SELECT entry_id, org, model_id, published_at, frameworks, compliance_score " "FROM attestations WHERE published_at >= ? AND revoked=0", (cutoff[:10],), # date prefix match ).fetchall() for row in rows: results.append({ "entry_id": row[0], "org": row[1], "model_id": row[2], "published_at": row[3], "frameworks": (row[4] or "").split(","), "compliance_score": row[5], }) return results except Exception: # noqa: BLE001 return [] def _has_api_keys(self) -> bool: """Check whether the key store has any provisioned keys.""" try: from squash.auth import get_key_store ks = get_key_store() return len(ks) > 0 except Exception: # noqa: BLE001 return False # ── Coverage report (W220) ──────────────────────────────────────────────────── @dataclass class Soc2CoverageReport: coverage: dict[str, Any] controls: list[Soc2Control] dossiers: dict[str, ControlDossier] generated_at: str window_days: int @classmethod def build( cls, window_days: int = 365, audit_log_path: Path | None = None, attestation_db: Path | None = None, ) -> "Soc2CoverageReport": catalogue = Soc2ControlCatalogue() collector = EvidenceCollector(window_days, audit_log_path, attestation_db) dossiers = collector.collect_all(catalogue) return cls( coverage = catalogue.coverage(), controls = catalogue.controls, dossiers = dossiers, generated_at = datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), window_days = window_days, ) def summary_text(self) -> str: cov = self.coverage lines = [ "[squash soc2 readiness] SOC 2 Type II Coverage Report", f" Generated: {self.generated_at}", f" Window: {self.window_days}-day evidence window", f" Total controls: {cov['total_controls']}", f" Covered: {cov['covered']} ({cov['coverage_pct']:.1f}%)", f" Partial: {cov['partial']}", f" Gap: {cov['gap']}", f" Not applicable: {cov['not_applicable']}", f" Effective coverage:{cov['effective_coverage_pct']:.1f}%", "", " By category:", ] cat_names = { "CC": "Common Criteria (Security)", "A": "Availability", "PI": "Processing Integrity", "C": "Confidentiality", "P": "Privacy", } for cat, stats in cov["by_category"].items(): covered = stats.get("COVERED", 0) total = sum(stats.values()) - stats.get("NOT_APPLICABLE", 0) pct = round(100.0 * covered / max(total, 1), 0) bar = "█" * int(pct / 10) + "░" * (10 - int(pct / 10)) lines.append(f" {cat} {bar} {covered}/{total} {pct:.0f}% {cat_names.get(cat, '')}") gaps = [c for c in self.controls if c.status == ControlStatus.GAP] if gaps: lines += ["", " Gap controls requiring remediation:"] for c in gaps: lines.append(f" ✗ {c.id} — {c.title}") partials = [c for c in self.controls if c.status == ControlStatus.PARTIAL] if partials: lines += ["", " Partial controls (next steps):"] for c in partials[:8]: lines.append(f" ⚠ {c.id} — {c.title}") if len(partials) > 8: lines.append(f" …and {len(partials)-8} more") return "\n".join(lines) def to_dict(self) -> dict[str, Any]: return { "generated_at": self.generated_at, "window_days": self.window_days, "coverage": self.coverage, "controls": [c.to_dict() for c in self.controls], "squash_version": VERSION, } # ── Evidence bundle builder (W220) ──────────────────────────────────────────── class Soc2EvidenceBundle: """Build an auditor-ready ZIP evidence bundle. Bundle layout: controls_index.json coverage_summary.md SHA256SUMS dossiers/ CC1.1_evidence.json CC1.1_evidence.md … (one pair per control) attestations/ (last 10 signed attestation JSONs, if available) """ @classmethod def build( cls, output_dir: Path | None = None, window_days: int = 365, audit_log_path: Path | None = None, attestation_db: Path | None = None, include_attestations: bool = True, ) -> Path: """Build the ZIP and return the path to the written file.""" report = Soc2CoverageReport.build( window_days=window_days, audit_log_path=audit_log_path, attestation_db=attestation_db, ) date_str = datetime.date.today().isoformat() bundle_name = f"squash-soc2-bundle-{date_str}.zip" out_dir = Path(output_dir) if output_dir else Path.cwd() out_dir.mkdir(parents=True, exist_ok=True) bundle_path = out_dir / bundle_name sha256_entries: dict[str, str] = {} buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: # ── controls_index.json ──────────────────────────────────────── index_bytes = json.dumps(report.to_dict(), indent=2, sort_keys=True).encode() zf.writestr("controls_index.json", index_bytes) sha256_entries["controls_index.json"] = hashlib.sha256(index_bytes).hexdigest() # ── coverage_summary.md ─────────────────────────────────────── summary_bytes = report.summary_text().encode() zf.writestr("coverage_summary.md", summary_bytes) sha256_entries["coverage_summary.md"] = hashlib.sha256(summary_bytes).hexdigest() # ── dossiers/ ───────────────────────────────────────────────── for ctrl_id, dossier in sorted(report.dossiers.items()): # JSON dossier json_bytes = json.dumps(dossier.to_dict(), indent=2).encode() json_path = f"dossiers/{ctrl_id}_evidence.json" zf.writestr(json_path, json_bytes) sha256_entries[json_path] = hashlib.sha256(json_bytes).hexdigest() # Markdown dossier md_bytes = dossier.to_markdown().encode() md_path = f"dossiers/{ctrl_id}_evidence.md" zf.writestr(md_path, md_bytes) sha256_entries[md_path] = hashlib.sha256(md_bytes).hexdigest() # ── attestations/ (recent artifacts) ───────────────────────── if include_attestations: att_files = cls._collect_attestation_artifacts(attestation_db) for name, att_bytes in att_files: att_path = f"attestations/{name}" zf.writestr(att_path, att_bytes) sha256_entries[att_path] = hashlib.sha256(att_bytes).hexdigest() # ── SHA256SUMS manifest ──────────────────────────────────────── manifest_lines = [ f"{digest} {path}" for path, digest in sorted(sha256_entries.items()) ] manifest_bytes = "\n".join(manifest_lines).encode() + b"\n" zf.writestr("SHA256SUMS", manifest_bytes) bundle_path.write_bytes(buf.getvalue()) return bundle_path @classmethod def _collect_attestation_artifacts( cls, attestation_db: Path | None = None, ) -> list[tuple[str, bytes]]: """Pull recent attestation payloads from the registry (max 10).""" try: from squash.attestation_registry import AttestationRegistry kw = {"db_path": attestation_db} if attestation_db else {} results = [] with AttestationRegistry(**kw) as reg: rows = reg._conn.execute( "SELECT entry_id, payload FROM attestations " "WHERE revoked=0 ORDER BY published_at DESC LIMIT 10" ).fetchall() for entry_id, payload_str in rows: if payload_str: name = f"{entry_id}_attestation.json" results.append((name, payload_str.encode())) return results except Exception: # noqa: BLE001 return []