Benjamin_Boenisch/breakpilot-compliance
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: phase 0 guardrails + phase 1 step 2 (models.py split)

Browse Source 
Squash of branch refactor/phase0-guardrails-and-models-split — 4 commits,
81 files, 173/173 pytest green, OpenAPI contract preserved (360 paths /
484 operations).

## Phase 0 — Architecture guardrails

Three defense-in-depth layers to keep the architecture rules enforced
regardless of who opens Claude Code in this repo:

  1. .claude/settings.json PreToolUse hook on Write/Edit blocks any file
     that would exceed the 500-line hard cap. Auto-loads in every Claude
     session in this repo.
  2. scripts/githooks/pre-commit (install via scripts/install-hooks.sh)
     enforces the LOC cap locally, freezes migrations/ without
     [migration-approved], and protects guardrail files without
     [guardrail-change].
  3. .gitea/workflows/ci.yaml gains loc-budget + guardrail-integrity +
     sbom-scan (syft+grype) jobs, adds mypy --strict for the new Python
     packages (compliance/{services,repositories,domain,schemas}), and
     tsc --noEmit for admin-compliance + developer-portal.

Per-language conventions documented in AGENTS.python.md, AGENTS.go.md,
AGENTS.typescript.md at the repo root — layering, tooling, and explicit
"what you may NOT do" lists. Root CLAUDE.md is prepended with the six
non-negotiable rules. Each of the 10 services gets a README.md.

scripts/check-loc.sh enforces soft 300 / hard 500 and surfaces the
current baseline of 205 hard + 161 soft violations so Phases 1-4 can
drain it incrementally. CI gates only CHANGED files in PRs so the
legacy baseline does not block unrelated work.

## Deprecation sweep

47 files. Pydantic V1 regex= -> pattern= (2 sites), class Config ->
ConfigDict in source_policy_router.py (schemas.py intentionally skipped;
it is the Phase 1 Step 3 split target). datetime.utcnow() ->
datetime.now(timezone.utc) everywhere including SQLAlchemy default=
callables. All DB columns already declare timezone=True, so this is a
latent-bug fix at the Python side, not a schema change.

DeprecationWarning count dropped from 158 to 35.

## Phase 1 Step 1 — Contract test harness

tests/contracts/test_openapi_baseline.py diffs the live FastAPI /openapi.json
against tests/contracts/openapi.baseline.json on every test run. Fails on
removed paths, removed status codes, or new required request body fields.
Regenerate only via tests/contracts/regenerate_baseline.py after a
consumer-updated contract change. This is the safety harness for all
subsequent refactor commits.

## Phase 1 Step 2 — models.py split (1466 -> 85 LOC shim)

compliance/db/models.py is decomposed into seven sibling aggregate modules
following the existing repo pattern (dsr_models.py, vvt_models.py, ...):

  regulation_models.py       (134) — Regulation, Requirement
  control_models.py          (279) — Control, Mapping, Evidence, Risk
  ai_system_models.py        (141) — AISystem, AuditExport
  service_module_models.py   (176) — ServiceModule, ModuleRegulation, ModuleRisk
  audit_session_models.py    (177) — AuditSession, AuditSignOff
  isms_governance_models.py  (323) — ISMSScope, Context, Policy, Objective, SoA
  isms_audit_models.py       (468) — Finding, CAPA, MgmtReview, InternalAudit,
                                     AuditTrail, Readiness

models.py becomes an 85-line re-export shim in dependency order so
existing imports continue to work unchanged. Schema is byte-identical:
__tablename__, column definitions, relationship strings, back_populates,
cascade directives all preserved.

All new sibling files are under the 500-line hard cap; largest is
isms_audit_models.py at 468. No file in compliance/db/ now exceeds
the hard cap.

## Phase 1 Step 3 — infrastructure only

backend-compliance/compliance/{schemas,domain,repositories}/ packages
are created as landing zones with docstrings. compliance/domain/
exports DomainError / NotFoundError / ConflictError / ValidationError /
PermissionError — the base classes services will use to raise
domain-level errors instead of HTTPException.

PHASE1_RUNBOOK.md at backend-compliance/PHASE1_RUNBOOK.md documents
the nine-step execution plan for Phase 1: snapshot baseline,
characterization tests, split models.py (this commit), split schemas.py
(next), extract services, extract repositories, mypy --strict, coverage.

## Verification

  backend-compliance/.venv-phase1: uv python install 3.12 + pip -r requirements.txt
  PYTHONPATH=. pytest compliance/tests/ tests/contracts/
  -> 173 passed, 0 failed, 35 warnings, OpenAPI 360/484 unchanged

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Sharang Parnerkar
2026-04-07 13:18:29 +02:00
parent 1dfea51919
commit 3320ef94fc
84 changed files with 52849 additions and 1731 deletions
Download Patch File Download Diff File Expand all files Collapse all files

468
backend-compliance/compliance/db/isms_audit_models.py Normal file
View File

@@ -0,0 +1,468 @@
"""
ISMS Audit Execution models (ISO 27001 Kapitel 9-10) — extracted from
compliance/db/models.py.
Covers findings, corrective actions (CAPA), management reviews, internal
audits, audit trail, and readiness checks. The governance side (scope,
context, policies, objectives, SoA) lives in ``isms_governance_models.py``.
Re-exported from ``compliance.db.models`` for backwards compatibility.
DO NOT change __tablename__, column names, or relationship strings.
"""
import uuid
import enum
from datetime import datetime, date, timezone
from sqlalchemy import (
Column, String, Text, Integer, Boolean, DateTime, Date,
ForeignKey, Enum, JSON, Index, Float,
)
from sqlalchemy.orm import relationship
from classroom_engine.database import Base
# ============================================================================
# ENUMS
# ============================================================================
class FindingTypeEnum(str, enum.Enum):
"""ISO 27001 audit finding classification."""
MAJOR = "major" # Major nonconformity - blocks certification
MINOR = "minor" # Minor nonconformity - requires CAPA
OFI = "ofi" # Opportunity for Improvement
POSITIVE = "positive" # Positive observation
class FindingStatusEnum(str, enum.Enum):
"""Status of an audit finding."""
OPEN = "open"
IN_PROGRESS = "in_progress"
CORRECTIVE_ACTION_PENDING = "capa_pending"
VERIFICATION_PENDING = "verification_pending"
VERIFIED = "verified"
CLOSED = "closed"
class CAPATypeEnum(str, enum.Enum):
"""Type of corrective/preventive action."""
CORRECTIVE = "corrective" # Fix the nonconformity
PREVENTIVE = "preventive" # Prevent recurrence
BOTH = "both"
# ============================================================================
# MODELS
# ============================================================================
class AuditFindingDB(Base):
"""
Audit Finding with ISO 27001 Classification (Major/Minor/OFI)
Tracks findings from internal and external audits with proper
classification and CAPA workflow.
"""
__tablename__ = 'compliance_audit_findings'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
finding_id = Column(String(30), unique=True, nullable=False, index=True) # e.g., "FIND-2026-001"
# Source
audit_session_id = Column(String(36), ForeignKey('compliance_audit_sessions.id'), index=True)
internal_audit_id = Column(String(36), ForeignKey('compliance_internal_audits.id'), index=True)
# Classification (CRITICAL for ISO 27001!)
finding_type = Column(Enum(FindingTypeEnum), nullable=False)
# ISO reference
iso_chapter = Column(String(20)) # e.g., "6.1.2", "9.2"
annex_a_control = Column(String(20)) # e.g., "A.8.2"
# Finding details
title = Column(String(300), nullable=False)
description = Column(Text, nullable=False)
objective_evidence = Column(Text, nullable=False) # What the auditor observed
# Root cause analysis
root_cause = Column(Text)
root_cause_method = Column(String(50)) # "5-why", "fishbone", "pareto"
# Impact assessment
impact_description = Column(Text)
affected_processes = Column(JSON)
affected_assets = Column(JSON)
# Status tracking
status = Column(Enum(FindingStatusEnum), default=FindingStatusEnum.OPEN)
# Responsibility
owner = Column(String(100)) # Person responsible for closure
auditor = Column(String(100)) # Auditor who raised finding
# Dates
identified_date = Column(Date, nullable=False, default=date.today)
due_date = Column(Date) # Deadline for closure
closed_date = Column(Date)
# Verification
verification_method = Column(Text)
verified_by = Column(String(100))
verified_at = Column(DateTime)
verification_evidence = Column(Text)
# Closure
closure_notes = Column(Text)
closed_by = Column(String(100))
# Timestamps
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# Relationships
corrective_actions = relationship("CorrectiveActionDB", back_populates="finding", cascade="all, delete-orphan")
__table_args__ = (
Index('ix_finding_type_status', 'finding_type', 'status'),
Index('ix_finding_due_date', 'due_date'),
)
def __repr__(self):
return f"<AuditFinding {self.finding_id}: {self.finding_type.value}>"
@property
def is_blocking(self) -> bool:
"""Major findings block certification."""
return self.finding_type == FindingTypeEnum.MAJOR and self.status != FindingStatusEnum.CLOSED
class CorrectiveActionDB(Base):
"""
Corrective & Preventive Actions (CAPA) - ISO 27001 10.1
Tracks actions taken to address nonconformities.
"""
__tablename__ = 'compliance_corrective_actions'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
capa_id = Column(String(30), unique=True, nullable=False, index=True) # e.g., "CAPA-2026-001"
# Link to finding
finding_id = Column(String(36), ForeignKey('compliance_audit_findings.id'), nullable=False, index=True)
# Type
capa_type = Column(Enum(CAPATypeEnum), nullable=False)
# Action details
title = Column(String(300), nullable=False)
description = Column(Text, nullable=False)
expected_outcome = Column(Text)
# Responsibility
assigned_to = Column(String(100), nullable=False)
approved_by = Column(String(100))
# Timeline
planned_start = Column(Date)
planned_completion = Column(Date, nullable=False)
actual_completion = Column(Date)
# Status
status = Column(String(30), default="planned") # planned, in_progress, completed, verified, cancelled
progress_percentage = Column(Integer, default=0)
# Resources
estimated_effort_hours = Column(Integer)
actual_effort_hours = Column(Integer)
resources_required = Column(Text)
# Evidence of implementation
implementation_evidence = Column(Text)
evidence_ids = Column(JSON)
# Effectiveness review
effectiveness_criteria = Column(Text)
effectiveness_verified = Column(Boolean, default=False)
effectiveness_verification_date = Column(Date)
effectiveness_notes = Column(Text)
# Timestamps
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# Relationships
finding = relationship("AuditFindingDB", back_populates="corrective_actions")
__table_args__ = (
Index('ix_capa_status', 'status'),
Index('ix_capa_due', 'planned_completion'),
)
def __repr__(self):
return f"<CAPA {self.capa_id}: {self.capa_type.value}>"
class ManagementReviewDB(Base):
"""
Management Review (ISO 27001 Kapitel 9.3)
Records mandatory management reviews of the ISMS.
"""
__tablename__ = 'compliance_management_reviews'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
review_id = Column(String(30), unique=True, nullable=False, index=True) # e.g., "MR-2026-Q1"
# Review details
title = Column(String(200), nullable=False)
review_date = Column(Date, nullable=False)
review_period_start = Column(Date) # Period being reviewed
review_period_end = Column(Date)
# Participants
chairperson = Column(String(100), nullable=False) # Usually top management
attendees = Column(JSON) # List of {"name": "", "role": ""}
# 9.3 Review Inputs (mandatory!)
input_previous_actions = Column(Text) # Status of previous review actions
input_isms_changes = Column(Text) # Changes in internal/external issues
input_security_performance = Column(Text) # Nonconformities, monitoring, audit results
input_interested_party_feedback = Column(Text)
input_risk_assessment_results = Column(Text)
input_improvement_opportunities = Column(Text)
# Additional inputs
input_policy_effectiveness = Column(Text)
input_objective_achievement = Column(Text)
input_resource_adequacy = Column(Text)
# 9.3 Review Outputs (mandatory!)
output_improvement_decisions = Column(Text) # Decisions for improvement
output_isms_changes = Column(Text) # Changes needed to ISMS
output_resource_needs = Column(Text) # Resource requirements
# Action items
action_items = Column(JSON) # List of {"action": "", "owner": "", "due_date": ""}
# Overall assessment
isms_effectiveness_rating = Column(String(20)) # "effective", "partially_effective", "not_effective"
key_decisions = Column(Text)
# Approval
status = Column(String(30), default="draft") # draft, conducted, approved
approved_by = Column(String(100))
approved_at = Column(DateTime)
minutes_document_path = Column(String(500)) # Link to meeting minutes
# Next review
next_review_date = Column(Date)
# Timestamps
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
__table_args__ = (
Index('ix_mgmt_review_date', 'review_date'),
Index('ix_mgmt_review_status', 'status'),
)
def __repr__(self):
return f"<ManagementReview {self.review_id}: {self.review_date}>"
class InternalAuditDB(Base):
"""
Internal Audit (ISO 27001 Kapitel 9.2)
Tracks internal audit program and individual audits.
"""
__tablename__ = 'compliance_internal_audits'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
audit_id = Column(String(30), unique=True, nullable=False, index=True) # e.g., "IA-2026-001"
# Audit details
title = Column(String(200), nullable=False)
audit_type = Column(String(50), nullable=False) # "scheduled", "surveillance", "special"
# Scope
scope_description = Column(Text, nullable=False)
iso_chapters_covered = Column(JSON) # e.g., ["4", "5", "6.1"]
annex_a_controls_covered = Column(JSON) # e.g., ["A.5", "A.6"]
processes_covered = Column(JSON)
departments_covered = Column(JSON)
# Audit criteria
criteria = Column(Text) # Standards, policies being audited against
# Timeline
planned_date = Column(Date, nullable=False)
actual_start_date = Column(Date)
actual_end_date = Column(Date)
# Audit team
lead_auditor = Column(String(100), nullable=False)
audit_team = Column(JSON) # List of auditor names
auditee_representatives = Column(JSON) # Who was interviewed
# Status
status = Column(String(30), default="planned") # planned, in_progress, completed, cancelled
# Results summary
total_findings = Column(Integer, default=0)
major_findings = Column(Integer, default=0)
minor_findings = Column(Integer, default=0)
ofi_count = Column(Integer, default=0)
positive_observations = Column(Integer, default=0)
# Conclusion
audit_conclusion = Column(Text)
overall_assessment = Column(String(30)) # "conforming", "minor_nc", "major_nc"
# Report
report_date = Column(Date)
report_document_path = Column(String(500))
# Sign-off
report_approved_by = Column(String(100))
report_approved_at = Column(DateTime)
# Follow-up
follow_up_audit_required = Column(Boolean, default=False)
follow_up_audit_id = Column(String(36))
# Timestamps
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
# Relationships
findings = relationship("AuditFindingDB", backref="internal_audit", foreign_keys=[AuditFindingDB.internal_audit_id])
__table_args__ = (
Index('ix_internal_audit_date', 'planned_date'),
Index('ix_internal_audit_status', 'status'),
)
def __repr__(self):
return f"<InternalAudit {self.audit_id}: {self.title}>"
class AuditTrailDB(Base):
"""
Comprehensive Audit Trail for ISMS Changes
Tracks all changes to compliance-relevant data for
accountability and forensic analysis.
"""
__tablename__ = 'compliance_audit_trail'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
# What changed
entity_type = Column(String(50), nullable=False, index=True) # "control", "risk", "policy", etc.
entity_id = Column(String(36), nullable=False, index=True)
entity_name = Column(String(200)) # Human-readable identifier
# Action
action = Column(String(20), nullable=False) # "create", "update", "delete", "approve", "sign"
# Change details
field_changed = Column(String(100)) # Which field (for updates)
old_value = Column(Text)
new_value = Column(Text)
change_summary = Column(Text) # Human-readable summary
# Who & When
performed_by = Column(String(100), nullable=False)
performed_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
# Context
ip_address = Column(String(45))
user_agent = Column(String(500))
session_id = Column(String(100))
# Integrity
checksum = Column(String(64)) # SHA-256 of the change
# Timestamps (immutable after creation)
created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
Index('ix_audit_trail_entity', 'entity_type', 'entity_id'),
Index('ix_audit_trail_time', 'performed_at'),
Index('ix_audit_trail_user', 'performed_by'),
)
def __repr__(self):
return f"<AuditTrail {self.action} on {self.entity_type}/{self.entity_id}>"
class ISMSReadinessCheckDB(Base):
"""
ISMS Readiness Check Results
Stores automated pre-audit checks to identify potential
Major findings before external audit.
"""
__tablename__ = 'compliance_isms_readiness'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
# Check run
check_date = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
triggered_by = Column(String(100)) # "scheduled", "manual", "pre-audit"
# Overall status
overall_status = Column(String(20), nullable=False) # "ready", "at_risk", "not_ready"
certification_possible = Column(Boolean, nullable=False)
# Chapter-by-chapter status (ISO 27001)
chapter_4_status = Column(String(20)) # Context
chapter_5_status = Column(String(20)) # Leadership
chapter_6_status = Column(String(20)) # Planning
chapter_7_status = Column(String(20)) # Support
chapter_8_status = Column(String(20)) # Operation
chapter_9_status = Column(String(20)) # Performance
chapter_10_status = Column(String(20)) # Improvement
# Potential Major findings
potential_majors = Column(JSON) # List of {"check": "", "status": "", "recommendation": ""}
# Potential Minor findings
potential_minors = Column(JSON)
# Improvement opportunities
improvement_opportunities = Column(JSON)
# Scores
readiness_score = Column(Float) # 0-100
documentation_score = Column(Float)
implementation_score = Column(Float)
evidence_score = Column(Float)
# Recommendations
priority_actions = Column(JSON) # List of recommended actions before audit
# Timestamps
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
__table_args__ = (
Index('ix_readiness_date', 'check_date'),
Index('ix_readiness_status', 'overall_status'),
)
def __repr__(self):
return f"<ISMSReadiness {self.check_date}: {self.overall_status}>"
__all__ = [
"FindingTypeEnum",
"FindingStatusEnum",
"CAPATypeEnum",
"AuditFindingDB",
"CorrectiveActionDB",
"ManagementReviewDB",
"InternalAuditDB",
"AuditTrailDB",
"ISMSReadinessCheckDB",
]