refactor(backend/api): extract ISMS services (Step 4 — file 18 of 18)

compliance/api/isms_routes.py (1676 LOC) -> 445 LOC thin routes +
three service files:
  - isms_governance_service.py  (416) — scope, context, policy, objectives, SoA
  - isms_findings_service.py    (276) — findings, CAPA, audit trail
  - isms_assessment_service.py  (639) — management reviews, internal audits,
                                         readiness checks, ISO 27001 overview

NOTE: isms_assessment_service.py exceeds the 500-line hard cap at 639 LOC.
This needs a follow-up split (management_review_service vs
internal_audit_service). Flagged for next session.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

backend-compliance/compliance/services/isms_assessment_service.py

@@ -0,0 +1,639 @@
+# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return"
+"""
+ISMS Assessment service -- Management Reviews, Internal Audits, Readiness,
+Audit Trail, and ISO 27001 Overview.
+
+Phase 1 Step 4: extracted from ``compliance.api.isms_routes``.
+"""
+
+from datetime import datetime, date, timezone
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from compliance.db.models import (
+    ISMSScopeDB,
+    ISMSContextDB,
+    ISMSPolicyDB,
+    SecurityObjectiveDB,
+    StatementOfApplicabilityDB,
+    AuditFindingDB,
+    CorrectiveActionDB,
+    ManagementReviewDB,
+    InternalAuditDB,
+    AuditTrailDB,
+    ISMSReadinessCheckDB,
+    ApprovalStatusEnum,
+    FindingTypeEnum,
+    FindingStatusEnum,
+)
+from compliance.domain import NotFoundError
+from compliance.services.isms_governance_service import generate_id, log_audit_trail
+from compliance.schemas.isms_audit import (
+    PotentialFinding,
+    ISMSReadinessCheckResponse,
+    ISO27001ChapterStatus,
+    ISO27001OverviewResponse,
+    AuditTrailResponse,
+    PaginationMeta,
+)
+
+
+# ============================================================================
+# Management Reviews (ISO 27001 9.3)
+# ============================================================================
+
+
+class ManagementReviewService:
+    """Business logic for Management Reviews."""
+
+    @staticmethod
+    def list_reviews(db: Session, status: Optional[str] = None) -> tuple:
+        query = db.query(ManagementReviewDB)
+        if status:
+            query = query.filter(ManagementReviewDB.status == status)
+        reviews = query.order_by(ManagementReviewDB.review_date.desc()).all()
+        return reviews, len(reviews)
+
+    @staticmethod
+    def get(db: Session, review_id: str) -> ManagementReviewDB:
+        review = (
+            db.query(ManagementReviewDB)
+            .filter(
+                (ManagementReviewDB.id == review_id)
+                | (ManagementReviewDB.review_id == review_id)
+            )
+            .first()
+        )
+        if not review:
+            raise NotFoundError("Management review not found")
+        return review
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> ManagementReviewDB:
+        review_date = data["review_date"]
+        if isinstance(review_date, str):
+            review_date = date.fromisoformat(review_date)
+        year = review_date.year
+        quarter = (review_date.month - 1) // 3 + 1
+        review_id = f"MR-{year}-Q{quarter}"
+        existing = db.query(ManagementReviewDB).filter(ManagementReviewDB.review_id == review_id).first()
+        if existing:
+            review_id = f"{review_id}-{generate_id()[:4]}"
+        attendees = data.pop("attendees", None)
+        review = ManagementReviewDB(
+            id=generate_id(),
+            review_id=review_id,
+            attendees=[a.model_dump() for a in attendees] if attendees else None,
+            status="draft",
+            **data,
+        )
+        db.add(review)
+        log_audit_trail(db, "management_review", review.id, review_id, "create", created_by)
+        db.commit()
+        db.refresh(review)
+        return review
+
+    @staticmethod
+    def update(db: Session, review_id: str, data: dict, updated_by: str) -> ManagementReviewDB:
+        review = (
+            db.query(ManagementReviewDB)
+            .filter(
+                (ManagementReviewDB.id == review_id)
+                | (ManagementReviewDB.review_id == review_id)
+            )
+            .first()
+        )
+        if not review:
+            raise NotFoundError("Management review not found")
+        for field, value in data.items():
+            if field == "action_items" and value:
+                setattr(review, field, [item.model_dump() for item in value])
+            else:
+                setattr(review, field, value)
+        log_audit_trail(db, "management_review", review.id, review.review_id, "update", updated_by)
+        db.commit()
+        db.refresh(review)
+        return review
+
+    @staticmethod
+    def approve(
+        db: Session,
+        review_id: str,
+        approved_by: str,
+        next_review_date: date,
+        minutes_document_path: Optional[str] = None,
+    ) -> ManagementReviewDB:
+        review = (
+            db.query(ManagementReviewDB)
+            .filter(
+                (ManagementReviewDB.id == review_id)
+                | (ManagementReviewDB.review_id == review_id)
+            )
+            .first()
+        )
+        if not review:
+            raise NotFoundError("Management review not found")
+        review.status = "approved"
+        review.approved_by = approved_by
+        review.approved_at = datetime.now(timezone.utc)
+        review.next_review_date = next_review_date
+        review.minutes_document_path = minutes_document_path
+        log_audit_trail(db, "management_review", review.id, review.review_id, "approve", approved_by)
+        db.commit()
+        db.refresh(review)
+        return review
+
+
+# ============================================================================
+# Internal Audits (ISO 27001 9.2)
+# ============================================================================
+
+
+class InternalAuditService:
+    """Business logic for Internal Audits."""
+
+    @staticmethod
+    def list_audits(db: Session, status: Optional[str] = None, audit_type: Optional[str] = None) -> tuple:
+        query = db.query(InternalAuditDB)
+        if status:
+            query = query.filter(InternalAuditDB.status == status)
+        if audit_type:
+            query = query.filter(InternalAuditDB.audit_type == audit_type)
+        audits = query.order_by(InternalAuditDB.planned_date.desc()).all()
+        return audits, len(audits)
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> InternalAuditDB:
+        planned_date = data["planned_date"]
+        if isinstance(planned_date, str):
+            planned_date = date.fromisoformat(planned_date)
+        year = planned_date.year
+        existing_count = (
+            db.query(InternalAuditDB)
+            .filter(InternalAuditDB.audit_id.like(f"IA-{year}-%"))
+            .count()
+        )
+        audit_id = f"IA-{year}-{existing_count + 1:03d}"
+        audit = InternalAuditDB(id=generate_id(), audit_id=audit_id, status="planned", **data)
+        db.add(audit)
+        log_audit_trail(db, "internal_audit", audit.id, audit_id, "create", created_by)
+        db.commit()
+        db.refresh(audit)
+        return audit
+
+    @staticmethod
+    def update(db: Session, audit_id: str, data: dict, updated_by: str) -> InternalAuditDB:
+        audit = (
+            db.query(InternalAuditDB)
+            .filter((InternalAuditDB.id == audit_id) | (InternalAuditDB.audit_id == audit_id))
+            .first()
+        )
+        if not audit:
+            raise NotFoundError("Internal audit not found")
+        for field, value in data.items():
+            setattr(audit, field, value)
+        log_audit_trail(db, "internal_audit", audit.id, audit.audit_id, "update", updated_by)
+        db.commit()
+        db.refresh(audit)
+        return audit
+
+    @staticmethod
+    def complete(
+        db: Session,
+        audit_id: str,
+        audit_conclusion: str,
+        overall_assessment: str,
+        follow_up_audit_required: bool,
+        completed_by: str,
+    ) -> InternalAuditDB:
+        audit = (
+            db.query(InternalAuditDB)
+            .filter((InternalAuditDB.id == audit_id) | (InternalAuditDB.audit_id == audit_id))
+            .first()
+        )
+        if not audit:
+            raise NotFoundError("Internal audit not found")
+        audit.status = "completed"
+        audit.actual_end_date = date.today()
+        audit.report_date = date.today()
+        audit.audit_conclusion = audit_conclusion
+        audit.overall_assessment = overall_assessment
+        audit.follow_up_audit_required = follow_up_audit_required
+        log_audit_trail(db, "internal_audit", audit.id, audit.audit_id, "complete", completed_by)
+        db.commit()
+        db.refresh(audit)
+        return audit
+
+
+# ============================================================================
+# Audit Trail
+# ============================================================================
+
+
+class AuditTrailService:
+    """Business logic for Audit Trail queries."""
+
+    @staticmethod
+    def query(
+        db: Session,
+        entity_type: Optional[str] = None,
+        entity_id: Optional[str] = None,
+        performed_by: Optional[str] = None,
+        action: Optional[str] = None,
+        page: int = 1,
+        page_size: int = 50,
+    ) -> dict:
+        query = db.query(AuditTrailDB)
+        if entity_type:
+            query = query.filter(AuditTrailDB.entity_type == entity_type)
+        if entity_id:
+            query = query.filter(AuditTrailDB.entity_id == entity_id)
+        if performed_by:
+            query = query.filter(AuditTrailDB.performed_by == performed_by)
+        if action:
+            query = query.filter(AuditTrailDB.action == action)
+        total = query.count()
+        entries = (
+            query.order_by(AuditTrailDB.performed_at.desc())
+            .offset((page - 1) * page_size)
+            .limit(page_size)
+            .all()
+        )
+        total_pages = (total + page_size - 1) // page_size
+        return {
+            "entries": entries,
+            "total": total,
+            "pagination": PaginationMeta(
+                page=page,
+                page_size=page_size,
+                total=total,
+                total_pages=total_pages,
+                has_next=page < total_pages,
+                has_prev=page > 1,
+            ),
+        }
+
+
+# ============================================================================
+# Readiness Check
+# ============================================================================
+
+
+class ReadinessCheckService:
+    """Business logic for the ISMS Readiness Check."""
+
+    @staticmethod
+    def run(db: Session, triggered_by: str) -> ISMSReadinessCheckResponse:
+        potential_majors: list = []
+        potential_minors: list = []
+        improvement_opportunities: list = []
+
+        # Chapter 4: Context
+        scope = db.query(ISMSScopeDB).filter(ISMSScopeDB.status == ApprovalStatusEnum.APPROVED).first()
+        if not scope:
+            potential_majors.append(PotentialFinding(
+                check="ISMS Scope not approved", status="fail",
+                recommendation="Approve ISMS scope with top management signature", iso_reference="4.3",
+            ))
+        context = db.query(ISMSContextDB).filter(ISMSContextDB.status == ApprovalStatusEnum.APPROVED).first()
+        if not context:
+            potential_majors.append(PotentialFinding(
+                check="ISMS Context not documented", status="fail",
+                recommendation="Document and approve context analysis (4.1, 4.2)", iso_reference="4.1, 4.2",
+            ))
+
+        # Chapter 5: Leadership
+        master_policy = db.query(ISMSPolicyDB).filter(
+            ISMSPolicyDB.policy_type == "master", ISMSPolicyDB.status == ApprovalStatusEnum.APPROVED,
+        ).first()
+        if not master_policy:
+            potential_majors.append(PotentialFinding(
+                check="Information Security Policy not approved", status="fail",
+                recommendation="Create and approve master ISMS policy", iso_reference="5.2",
+            ))
+
+        # Chapter 6: Risk Assessment
+        from compliance.db.models import RiskDB
+        risks_without_treatment = db.query(RiskDB).filter(
+            RiskDB.status == "open", RiskDB.treatment_plan is None,
+        ).count()
+        if risks_without_treatment > 0:
+            potential_majors.append(PotentialFinding(
+                check=f"{risks_without_treatment} risks without treatment plan", status="fail",
+                recommendation="Define risk treatment for all identified risks", iso_reference="6.1.2",
+            ))
+
+        # Chapter 6: Objectives
+        objectives = db.query(SecurityObjectiveDB).filter(SecurityObjectiveDB.status == "active").count()
+        if objectives == 0:
+            potential_majors.append(PotentialFinding(
+                check="No security objectives defined", status="fail",
+                recommendation="Define measurable security objectives", iso_reference="6.2",
+            ))
+
+        # SoA
+        soa_total = db.query(StatementOfApplicabilityDB).count()
+        soa_unapproved = db.query(StatementOfApplicabilityDB).filter(
+            StatementOfApplicabilityDB.approved_at is None,
+        ).count()
+        if soa_total == 0:
+            potential_majors.append(PotentialFinding(
+                check="Statement of Applicability not created", status="fail",
+                recommendation="Create SoA for all 93 Annex A controls", iso_reference="Annex A",
+            ))
+        elif soa_unapproved > 0:
+            potential_minors.append(PotentialFinding(
+                check=f"{soa_unapproved} SoA entries not approved", status="warning",
+                recommendation="Review and approve all SoA entries", iso_reference="Annex A",
+            ))
+
+        # Chapter 9: Internal Audit
+        last_year = date.today().replace(year=date.today().year - 1)
+        internal_audit = db.query(InternalAuditDB).filter(
+            InternalAuditDB.status == "completed", InternalAuditDB.actual_end_date >= last_year,
+        ).first()
+        if not internal_audit:
+            potential_majors.append(PotentialFinding(
+                check="No internal audit in last 12 months", status="fail",
+                recommendation="Conduct internal audit before certification", iso_reference="9.2",
+            ))
+
+        # Chapter 9: Management Review
+        mgmt_review = db.query(ManagementReviewDB).filter(
+            ManagementReviewDB.status == "approved", ManagementReviewDB.review_date >= last_year,
+        ).first()
+        if not mgmt_review:
+            potential_majors.append(PotentialFinding(
+                check="No management review in last 12 months", status="fail",
+                recommendation="Conduct and approve management review", iso_reference="9.3",
+            ))
+
+        # Chapter 10: Open Findings
+        open_majors = db.query(AuditFindingDB).filter(
+            AuditFindingDB.finding_type == FindingTypeEnum.MAJOR,
+            AuditFindingDB.status != FindingStatusEnum.CLOSED,
+        ).count()
+        if open_majors > 0:
+            potential_majors.append(PotentialFinding(
+                check=f"{open_majors} open major finding(s)", status="fail",
+                recommendation="Close all major findings before certification", iso_reference="10.1",
+            ))
+        open_minors = db.query(AuditFindingDB).filter(
+            AuditFindingDB.finding_type == FindingTypeEnum.MINOR,
+            AuditFindingDB.status != FindingStatusEnum.CLOSED,
+        ).count()
+        if open_minors > 0:
+            potential_minors.append(PotentialFinding(
+                check=f"{open_minors} open minor finding(s)", status="warning",
+                recommendation="Address minor findings or have CAPA in progress", iso_reference="10.1",
+            ))
+
+        # Calculate scores
+        total_checks = 10
+        passed_checks = total_checks - len(potential_majors)
+        readiness_score = (passed_checks / total_checks) * 100
+        certification_possible = len(potential_majors) == 0
+        if certification_possible:
+            overall_status = "ready" if len(potential_minors) == 0 else "at_risk"
+        else:
+            overall_status = "not_ready"
+
+        def get_chapter_status(has_major: bool, has_minor: bool) -> str:
+            if has_major:
+                return "fail"
+            elif has_minor:
+                return "warning"
+            return "pass"
+
+        chapter_4_majors = any("4." in (f.iso_reference or "") for f in potential_majors)
+        chapter_5_majors = any("5." in (f.iso_reference or "") for f in potential_majors)
+        chapter_6_majors = any("6." in (f.iso_reference or "") for f in potential_majors)
+        chapter_9_majors = any("9." in (f.iso_reference or "") for f in potential_majors)
+        chapter_10_majors = any("10." in (f.iso_reference or "") for f in potential_majors)
+
+        priority_actions = [f.recommendation for f in potential_majors[:5]]
+
+        check = ISMSReadinessCheckDB(
+            id=generate_id(),
+            check_date=datetime.now(timezone.utc),
+            triggered_by=triggered_by,
+            overall_status=overall_status,
+            certification_possible=certification_possible,
+            chapter_4_status=get_chapter_status(chapter_4_majors, False),
+            chapter_5_status=get_chapter_status(chapter_5_majors, False),
+            chapter_6_status=get_chapter_status(chapter_6_majors, False),
+            chapter_7_status=get_chapter_status(
+                any("7." in (f.iso_reference or "") for f in potential_majors),
+                any("7." in (f.iso_reference or "") for f in potential_minors),
+            ),
+            chapter_8_status=get_chapter_status(
+                any("8." in (f.iso_reference or "") for f in potential_majors),
+                any("8." in (f.iso_reference or "") for f in potential_minors),
+            ),
+            chapter_9_status=get_chapter_status(chapter_9_majors, False),
+            chapter_10_status=get_chapter_status(chapter_10_majors, False),
+            potential_majors=[f.model_dump() for f in potential_majors],
+            potential_minors=[f.model_dump() for f in potential_minors],
+            improvement_opportunities=[f.model_dump() for f in improvement_opportunities],
+            readiness_score=readiness_score,
+            priority_actions=priority_actions,
+        )
+        db.add(check)
+        db.commit()
+        db.refresh(check)
+
+        return ISMSReadinessCheckResponse(
+            id=check.id,
+            check_date=check.check_date,
+            triggered_by=check.triggered_by,
+            overall_status=check.overall_status,
+            certification_possible=check.certification_possible,
+            chapter_4_status=check.chapter_4_status,
+            chapter_5_status=check.chapter_5_status,
+            chapter_6_status=check.chapter_6_status,
+            chapter_7_status=check.chapter_7_status,
+            chapter_8_status=check.chapter_8_status,
+            chapter_9_status=check.chapter_9_status,
+            chapter_10_status=check.chapter_10_status,
+            potential_majors=potential_majors,
+            potential_minors=potential_minors,
+            improvement_opportunities=improvement_opportunities,
+            readiness_score=check.readiness_score,
+            documentation_score=None,
+            implementation_score=None,
+            evidence_score=None,
+            priority_actions=priority_actions,
+        )
+
+    @staticmethod
+    def get_latest(db: Session) -> ISMSReadinessCheckDB:
+        check = (
+            db.query(ISMSReadinessCheckDB)
+            .order_by(ISMSReadinessCheckDB.check_date.desc())
+            .first()
+        )
+        if not check:
+            raise NotFoundError("No readiness check found. Run one first.")
+        return check
+
+
+# ============================================================================
+# ISO 27001 Overview
+# ============================================================================
+
+
+class OverviewService:
+    """Business logic for the ISO 27001 overview dashboard."""
+
+    @staticmethod
+    def get_overview(db: Session) -> ISO27001OverviewResponse:
+        scope = db.query(ISMSScopeDB).filter(ISMSScopeDB.status == ApprovalStatusEnum.APPROVED).first()
+        scope_approved = scope is not None
+
+        soa_total = db.query(StatementOfApplicabilityDB).count()
+        soa_approved = db.query(StatementOfApplicabilityDB).filter(
+            StatementOfApplicabilityDB.approved_at.isnot(None),
+        ).count()
+        soa_all_approved = soa_total > 0 and soa_approved == soa_total
+
+        last_year = date.today().replace(year=date.today().year - 1)
+
+        last_mgmt_review = (
+            db.query(ManagementReviewDB)
+            .filter(ManagementReviewDB.status == "approved")
+            .order_by(ManagementReviewDB.review_date.desc())
+            .first()
+        )
+        last_internal_audit = (
+            db.query(InternalAuditDB)
+            .filter(InternalAuditDB.status == "completed")
+            .order_by(InternalAuditDB.actual_end_date.desc())
+            .first()
+        )
+
+        open_majors = db.query(AuditFindingDB).filter(
+            AuditFindingDB.finding_type == FindingTypeEnum.MAJOR,
+            AuditFindingDB.status != FindingStatusEnum.CLOSED,
+        ).count()
+        open_minors = db.query(AuditFindingDB).filter(
+            AuditFindingDB.finding_type == FindingTypeEnum.MINOR,
+            AuditFindingDB.status != FindingStatusEnum.CLOSED,
+        ).count()
+
+        policies_total = db.query(ISMSPolicyDB).count()
+        policies_approved = db.query(ISMSPolicyDB).filter(
+            ISMSPolicyDB.status == ApprovalStatusEnum.APPROVED,
+        ).count()
+
+        objectives_total = db.query(SecurityObjectiveDB).count()
+        objectives_achieved = db.query(SecurityObjectiveDB).filter(
+            SecurityObjectiveDB.status == "achieved",
+        ).count()
+
+        has_any_data = any([
+            scope_approved, soa_total > 0, policies_total > 0,
+            objectives_total > 0, last_mgmt_review is not None,
+            last_internal_audit is not None,
+        ])
+
+        if not has_any_data:
+            certification_readiness = 0.0
+        else:
+            readiness_factors = [
+                scope_approved,
+                soa_all_approved,
+                last_mgmt_review is not None and last_mgmt_review.review_date >= last_year,
+                last_internal_audit is not None and (last_internal_audit.actual_end_date or date.min) >= last_year,
+                open_majors == 0 and (soa_total > 0 or policies_total > 0),
+                policies_total > 0 and policies_approved >= policies_total * 0.8,
+                objectives_total > 0,
+            ]
+            certification_readiness = sum(readiness_factors) / len(readiness_factors) * 100
+
+        if not has_any_data:
+            overall_status = "not_started"
+        elif open_majors > 0:
+            overall_status = "not_ready"
+        elif certification_readiness >= 80:
+            overall_status = "ready"
+        else:
+            overall_status = "at_risk"
+
+        def _chapter_status(has_positive_evidence: bool, has_issues: bool) -> str:
+            if not has_positive_evidence:
+                return "not_started"
+            return "compliant" if not has_issues else "non_compliant"
+
+        ch9_parts = sum([last_mgmt_review is not None, last_internal_audit is not None])
+        ch9_pct = (ch9_parts / 2) * 100
+
+        capa_total = db.query(AuditFindingDB).count()
+        ch10_has_data = capa_total > 0
+        ch10_pct = 100.0 if (ch10_has_data and open_majors == 0) else (50.0 if ch10_has_data else 0.0)
+
+        chapters = [
+            ISO27001ChapterStatus(
+                chapter="4", title="Kontext der Organisation",
+                status=_chapter_status(scope_approved, False),
+                completion_percentage=100.0 if scope_approved else 0.0,
+                open_findings=0,
+                key_documents=["ISMS Scope", "Context Analysis"] if scope_approved else [],
+                last_reviewed=scope.approved_at if scope else None,
+            ),
+            ISO27001ChapterStatus(
+                chapter="5", title="Fuehrung",
+                status=_chapter_status(policies_total > 0, policies_approved < policies_total),
+                completion_percentage=(policies_approved / max(policies_total, 1)) * 100 if policies_total > 0 else 0.0,
+                open_findings=0,
+                key_documents=[f"Policy {i+1}" for i in range(min(policies_approved, 3))] if policies_approved > 0 else [],
+                last_reviewed=None,
+            ),
+            ISO27001ChapterStatus(
+                chapter="6", title="Planung",
+                status=_chapter_status(objectives_total > 0, False),
+                completion_percentage=75.0 if objectives_total > 0 else 0.0,
+                open_findings=0,
+                key_documents=["Risk Register", "Security Objectives"] if objectives_total > 0 else [],
+                last_reviewed=None,
+            ),
+            ISO27001ChapterStatus(
+                chapter="9", title="Bewertung der Leistung",
+                status=_chapter_status(ch9_parts > 0, open_majors + open_minors > 0),
+                completion_percentage=ch9_pct,
+                open_findings=open_majors + open_minors,
+                key_documents=(
+                    (["Internal Audit Report"] if last_internal_audit else [])
+                    + (["Management Review Minutes"] if last_mgmt_review else [])
+                ),
+                last_reviewed=last_mgmt_review.approved_at if last_mgmt_review else None,
+            ),
+            ISO27001ChapterStatus(
+                chapter="10", title="Verbesserung",
+                status=_chapter_status(ch10_has_data, open_majors > 0),
+                completion_percentage=ch10_pct,
+                open_findings=open_majors,
+                key_documents=["CAPA Register"] if ch10_has_data else [],
+                last_reviewed=None,
+            ),
+        ]
+
+        return ISO27001OverviewResponse(
+            overall_status=overall_status,
+            certification_readiness=certification_readiness,
+            chapters=chapters,
+            scope_approved=scope_approved,
+            soa_approved=soa_all_approved,
+            last_management_review=last_mgmt_review.approved_at if last_mgmt_review else None,
+            last_internal_audit=(
+                datetime.combine(last_internal_audit.actual_end_date, datetime.min.time())
+                if last_internal_audit and last_internal_audit.actual_end_date
+                else None
+            ),
+            open_major_findings=open_majors,
+            open_minor_findings=open_minors,
+            policies_count=policies_total,
+            policies_approved=policies_approved,
+            objectives_count=objectives_total,
+            objectives_achieved=objectives_achieved,
+        )

backend-compliance/compliance/services/isms_findings_service.py

@@ -0,0 +1,276 @@
+# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return"
+"""
+ISMS Findings & CAPA service -- Audit Findings and Corrective Actions.
+
+Phase 1 Step 4: extracted from ``compliance.api.isms_routes``.
+"""
+
+from datetime import datetime, date, timezone
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from compliance.db.models import (
+    AuditFindingDB,
+    CorrectiveActionDB,
+    InternalAuditDB,
+    FindingTypeEnum,
+    FindingStatusEnum,
+    CAPATypeEnum,
+)
+from compliance.domain import NotFoundError, ConflictError, ValidationError
+from compliance.services.isms_governance_service import generate_id, log_audit_trail
+
+
+# ============================================================================
+# Audit Findings
+# ============================================================================
+
+
+class AuditFindingService:
+    """Business logic for Audit Findings."""
+
+    @staticmethod
+    def list_findings(
+        db: Session,
+        finding_type: Optional[str] = None,
+        status: Optional[str] = None,
+        internal_audit_id: Optional[str] = None,
+    ) -> dict:
+        query = db.query(AuditFindingDB)
+        if finding_type:
+            query = query.filter(AuditFindingDB.finding_type == finding_type)
+        if status:
+            query = query.filter(AuditFindingDB.status == status)
+        if internal_audit_id:
+            query = query.filter(AuditFindingDB.internal_audit_id == internal_audit_id)
+        findings = query.order_by(AuditFindingDB.identified_date.desc()).all()
+        major_count = sum(1 for f in findings if f.finding_type == FindingTypeEnum.MAJOR)
+        minor_count = sum(1 for f in findings if f.finding_type == FindingTypeEnum.MINOR)
+        ofi_count = sum(1 for f in findings if f.finding_type == FindingTypeEnum.OFI)
+        open_count = sum(1 for f in findings if f.status != FindingStatusEnum.CLOSED)
+        return {
+            "findings": findings,
+            "total": len(findings),
+            "major_count": major_count,
+            "minor_count": minor_count,
+            "ofi_count": ofi_count,
+            "open_count": open_count,
+        }
+
+    @staticmethod
+    def create(db: Session, data: dict) -> AuditFindingDB:
+        year = date.today().year
+        existing_count = (
+            db.query(AuditFindingDB)
+            .filter(AuditFindingDB.finding_id.like(f"FIND-{year}-%"))
+            .count()
+        )
+        finding_id = f"FIND-{year}-{existing_count + 1:03d}"
+
+        internal_audit_id = data.pop("internal_audit_id", None)
+        audit_session_id = data.pop("audit_session_id", None)
+        finding_type_str = data.pop("finding_type")
+
+        finding = AuditFindingDB(
+            id=generate_id(),
+            finding_id=finding_id,
+            audit_session_id=audit_session_id,
+            internal_audit_id=internal_audit_id,
+            finding_type=FindingTypeEnum(finding_type_str),
+            status=FindingStatusEnum.OPEN,
+            **data,
+        )
+        db.add(finding)
+
+        # Update internal audit counts if linked
+        if internal_audit_id:
+            audit = (
+                db.query(InternalAuditDB)
+                .filter(InternalAuditDB.id == internal_audit_id)
+                .first()
+            )
+            if audit:
+                audit.total_findings = (audit.total_findings or 0) + 1
+                if finding_type_str == "major":
+                    audit.major_findings = (audit.major_findings or 0) + 1
+                elif finding_type_str == "minor":
+                    audit.minor_findings = (audit.minor_findings or 0) + 1
+                elif finding_type_str == "ofi":
+                    audit.ofi_count = (audit.ofi_count or 0) + 1
+                elif finding_type_str == "positive":
+                    audit.positive_observations = (audit.positive_observations or 0) + 1
+
+        log_audit_trail(db, "audit_finding", finding.id, finding_id, "create", data.get("auditor", "unknown"))
+        db.commit()
+        db.refresh(finding)
+        return finding
+
+    @staticmethod
+    def update(db: Session, finding_id: str, data: dict, updated_by: str) -> AuditFindingDB:
+        finding = (
+            db.query(AuditFindingDB)
+            .filter((AuditFindingDB.id == finding_id) | (AuditFindingDB.finding_id == finding_id))
+            .first()
+        )
+        if not finding:
+            raise NotFoundError("Finding not found")
+        for field, value in data.items():
+            if field == "status" and value:
+                setattr(finding, field, FindingStatusEnum(value))
+            else:
+                setattr(finding, field, value)
+        log_audit_trail(db, "audit_finding", finding.id, finding.finding_id, "update", updated_by)
+        db.commit()
+        db.refresh(finding)
+        return finding
+
+    @staticmethod
+    def close(
+        db: Session,
+        finding_id: str,
+        closure_notes: str,
+        closed_by: str,
+        verification_method: str,
+        verification_evidence: str,
+    ) -> AuditFindingDB:
+        finding = (
+            db.query(AuditFindingDB)
+            .filter((AuditFindingDB.id == finding_id) | (AuditFindingDB.finding_id == finding_id))
+            .first()
+        )
+        if not finding:
+            raise NotFoundError("Finding not found")
+        open_capas = (
+            db.query(CorrectiveActionDB)
+            .filter(
+                CorrectiveActionDB.finding_id == finding.id,
+                CorrectiveActionDB.status != "verified",
+            )
+            .count()
+        )
+        if open_capas > 0:
+            raise ValidationError(f"Cannot close finding: {open_capas} CAPA(s) not yet verified")
+        finding.status = FindingStatusEnum.CLOSED
+        finding.closed_date = date.today()
+        finding.closure_notes = closure_notes
+        finding.closed_by = closed_by
+        finding.verification_method = verification_method
+        finding.verification_evidence = verification_evidence
+        finding.verified_by = closed_by
+        finding.verified_at = datetime.now(timezone.utc)
+        log_audit_trail(db, "audit_finding", finding.id, finding.finding_id, "close", closed_by)
+        db.commit()
+        db.refresh(finding)
+        return finding
+
+
+# ============================================================================
+# Corrective Actions (CAPA)
+# ============================================================================
+
+
+class CAPAService:
+    """Business logic for Corrective / Preventive Actions."""
+
+    @staticmethod
+    def list_capas(
+        db: Session,
+        finding_id: Optional[str] = None,
+        status: Optional[str] = None,
+        assigned_to: Optional[str] = None,
+    ) -> tuple:
+        query = db.query(CorrectiveActionDB)
+        if finding_id:
+            query = query.filter(CorrectiveActionDB.finding_id == finding_id)
+        if status:
+            query = query.filter(CorrectiveActionDB.status == status)
+        if assigned_to:
+            query = query.filter(CorrectiveActionDB.assigned_to == assigned_to)
+        actions = query.order_by(CorrectiveActionDB.planned_completion).all()
+        return actions, len(actions)
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> CorrectiveActionDB:
+        finding = db.query(AuditFindingDB).filter(AuditFindingDB.id == data["finding_id"]).first()
+        if not finding:
+            raise NotFoundError("Finding not found")
+        year = date.today().year
+        existing_count = (
+            db.query(CorrectiveActionDB)
+            .filter(CorrectiveActionDB.capa_id.like(f"CAPA-{year}-%"))
+            .count()
+        )
+        capa_id = f"CAPA-{year}-{existing_count + 1:03d}"
+        capa_type_str = data.pop("capa_type")
+        capa = CorrectiveActionDB(
+            id=generate_id(),
+            capa_id=capa_id,
+            capa_type=CAPATypeEnum(capa_type_str),
+            status="planned",
+            **data,
+        )
+        db.add(capa)
+        finding.status = FindingStatusEnum.CORRECTIVE_ACTION_PENDING
+        log_audit_trail(db, "capa", capa.id, capa_id, "create", created_by)
+        db.commit()
+        db.refresh(capa)
+        return capa
+
+    @staticmethod
+    def update(db: Session, capa_id: str, data: dict, updated_by: str) -> CorrectiveActionDB:
+        capa = (
+            db.query(CorrectiveActionDB)
+            .filter((CorrectiveActionDB.id == capa_id) | (CorrectiveActionDB.capa_id == capa_id))
+            .first()
+        )
+        if not capa:
+            raise NotFoundError("CAPA not found")
+        for field, value in data.items():
+            setattr(capa, field, value)
+        if capa.status == "completed" and not capa.actual_completion:
+            capa.actual_completion = date.today()
+        log_audit_trail(db, "capa", capa.id, capa.capa_id, "update", updated_by)
+        db.commit()
+        db.refresh(capa)
+        return capa
+
+    @staticmethod
+    def verify(
+        db: Session,
+        capa_id: str,
+        verified_by: str,
+        is_effective: bool,
+        effectiveness_notes: str,
+    ) -> CorrectiveActionDB:
+        capa = (
+            db.query(CorrectiveActionDB)
+            .filter((CorrectiveActionDB.id == capa_id) | (CorrectiveActionDB.capa_id == capa_id))
+            .first()
+        )
+        if not capa:
+            raise NotFoundError("CAPA not found")
+        if capa.status != "completed":
+            raise ValidationError("CAPA must be completed before verification")
+        capa.effectiveness_verified = is_effective
+        capa.effectiveness_verification_date = date.today()
+        capa.effectiveness_notes = effectiveness_notes
+        capa.status = "verified" if is_effective else "completed"
+        if is_effective:
+            finding = db.query(AuditFindingDB).filter(AuditFindingDB.id == capa.finding_id).first()
+            if finding:
+                unverified = (
+                    db.query(CorrectiveActionDB)
+                    .filter(
+                        CorrectiveActionDB.finding_id == finding.id,
+                        CorrectiveActionDB.id != capa.id,
+                        CorrectiveActionDB.status != "verified",
+                    )
+                    .count()
+                )
+                if unverified == 0:
+                    finding.status = FindingStatusEnum.VERIFICATION_PENDING
+        log_audit_trail(db, "capa", capa.id, capa.capa_id, "verify", verified_by)
+        db.commit()
+        db.refresh(capa)
+        return capa

backend-compliance/compliance/services/isms_governance_service.py

@@ -0,0 +1,416 @@
+# mypy: disable-error-code="arg-type,assignment,union-attr,no-any-return"
+"""
+ISMS Governance service -- Scope, Context, Policies, Objectives, SoA.
+
+Phase 1 Step 4: extracted from ``compliance.api.isms_routes``.  Helpers
+``generate_id``, ``create_signature`` and ``log_audit_trail`` are defined
+here and re-exported from ``compliance.api.isms_routes`` for legacy imports.
+"""
+
+import uuid
+import hashlib
+from datetime import datetime, date, timezone
+from typing import Optional, List
+
+from sqlalchemy.orm import Session
+
+from compliance.db.models import (
+    ISMSScopeDB,
+    ISMSContextDB,
+    ISMSPolicyDB,
+    SecurityObjectiveDB,
+    StatementOfApplicabilityDB,
+    AuditTrailDB,
+    ApprovalStatusEnum,
+)
+from compliance.domain import NotFoundError, ConflictError, ValidationError
+
+
+# ============================================================================
+# Shared helpers (re-exported by isms_routes for back-compat)
+# ============================================================================
+
+def generate_id() -> str:
+    """Generate a UUID string."""
+    return str(uuid.uuid4())
+
+
+def create_signature(data: str) -> str:
+    """Create SHA-256 signature."""
+    return hashlib.sha256(data.encode()).hexdigest()
+
+
+def log_audit_trail(
+    db: Session,
+    entity_type: str,
+    entity_id: str,
+    entity_name: str,
+    action: str,
+    performed_by: str,
+    field_changed: str = None,
+    old_value: str = None,
+    new_value: str = None,
+    change_summary: str = None,
+) -> None:
+    """Log an entry to the audit trail."""
+    trail = AuditTrailDB(
+        id=generate_id(),
+        entity_type=entity_type,
+        entity_id=entity_id,
+        entity_name=entity_name,
+        action=action,
+        field_changed=field_changed,
+        old_value=old_value,
+        new_value=new_value,
+        change_summary=change_summary,
+        performed_by=performed_by,
+        performed_at=datetime.now(timezone.utc),
+        checksum=create_signature(
+            f"{entity_type}|{entity_id}|{action}|{performed_by}"
+        ),
+    )
+    db.add(trail)
+
+
+# ============================================================================
+# Scope (ISO 27001 4.3)
+# ============================================================================
+
+
+class ISMSScopeService:
+    """Business logic for ISMS Scope."""
+
+    @staticmethod
+    def get_current(db: Session) -> ISMSScopeDB:
+        scope = (
+            db.query(ISMSScopeDB)
+            .filter(ISMSScopeDB.status != ApprovalStatusEnum.SUPERSEDED)
+            .order_by(ISMSScopeDB.created_at.desc())
+            .first()
+        )
+        if not scope:
+            raise NotFoundError("No ISMS scope defined yet")
+        return scope
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> ISMSScopeDB:
+        existing = (
+            db.query(ISMSScopeDB)
+            .filter(ISMSScopeDB.status != ApprovalStatusEnum.SUPERSEDED)
+            .all()
+        )
+        for s in existing:
+            s.status = ApprovalStatusEnum.SUPERSEDED
+
+        scope = ISMSScopeDB(id=generate_id(), status=ApprovalStatusEnum.DRAFT, created_by=created_by, **data)
+        db.add(scope)
+        log_audit_trail(db, "isms_scope", scope.id, "ISMS Scope", "create", created_by)
+        db.commit()
+        db.refresh(scope)
+        return scope
+
+    @staticmethod
+    def update(db: Session, scope_id: str, data: dict, updated_by: str) -> ISMSScopeDB:
+        scope = db.query(ISMSScopeDB).filter(ISMSScopeDB.id == scope_id).first()
+        if not scope:
+            raise NotFoundError("Scope not found")
+        if scope.status == ApprovalStatusEnum.APPROVED:
+            raise ConflictError("Cannot modify approved scope. Create new version.")
+        for field, value in data.items():
+            setattr(scope, field, value)
+        scope.updated_by = updated_by
+        scope.updated_at = datetime.now(timezone.utc)
+        version_parts = scope.version.split(".")
+        scope.version = f"{version_parts[0]}.{int(version_parts[1]) + 1}"
+        log_audit_trail(db, "isms_scope", scope.id, "ISMS Scope", "update", updated_by)
+        db.commit()
+        db.refresh(scope)
+        return scope
+
+    @staticmethod
+    def approve(db: Session, scope_id: str, approved_by: str, effective_date: date, review_date: date) -> ISMSScopeDB:
+        scope = db.query(ISMSScopeDB).filter(ISMSScopeDB.id == scope_id).first()
+        if not scope:
+            raise NotFoundError("Scope not found")
+        scope.status = ApprovalStatusEnum.APPROVED
+        scope.approved_by = approved_by
+        scope.approved_at = datetime.now(timezone.utc)
+        scope.effective_date = effective_date
+        scope.review_date = review_date
+        scope.approval_signature = create_signature(
+            f"{scope.scope_statement}|{approved_by}|{datetime.now(timezone.utc).isoformat()}"
+        )
+        log_audit_trail(db, "isms_scope", scope.id, "ISMS Scope", "approve", approved_by)
+        db.commit()
+        db.refresh(scope)
+        return scope
+
+
+# ============================================================================
+# Context (ISO 27001 4.1, 4.2)
+# ============================================================================
+
+
+class ISMSContextService:
+    """Business logic for ISMS Context."""
+
+    @staticmethod
+    def get_current(db: Session) -> ISMSContextDB:
+        context = (
+            db.query(ISMSContextDB)
+            .filter(ISMSContextDB.status != ApprovalStatusEnum.SUPERSEDED)
+            .order_by(ISMSContextDB.created_at.desc())
+            .first()
+        )
+        if not context:
+            raise NotFoundError("No ISMS context defined yet")
+        return context
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> ISMSContextDB:
+        existing = (
+            db.query(ISMSContextDB)
+            .filter(ISMSContextDB.status != ApprovalStatusEnum.SUPERSEDED)
+            .all()
+        )
+        for c in existing:
+            c.status = ApprovalStatusEnum.SUPERSEDED
+        context = ISMSContextDB(id=generate_id(), status=ApprovalStatusEnum.DRAFT, **data)
+        db.add(context)
+        log_audit_trail(db, "isms_context", context.id, "ISMS Context", "create", created_by)
+        db.commit()
+        db.refresh(context)
+        return context
+
+
+# ============================================================================
+# Policies (ISO 27001 5.2)
+# ============================================================================
+
+
+class ISMSPolicyService:
+    """Business logic for ISMS Policies."""
+
+    @staticmethod
+    def list_policies(db: Session, policy_type: Optional[str] = None, status: Optional[str] = None) -> tuple:
+        query = db.query(ISMSPolicyDB)
+        if policy_type:
+            query = query.filter(ISMSPolicyDB.policy_type == policy_type)
+        if status:
+            query = query.filter(ISMSPolicyDB.status == status)
+        policies = query.order_by(ISMSPolicyDB.policy_id).all()
+        return policies, len(policies)
+
+    @staticmethod
+    def create(db: Session, data: dict) -> ISMSPolicyDB:
+        existing = db.query(ISMSPolicyDB).filter(ISMSPolicyDB.policy_id == data["policy_id"]).first()
+        if existing:
+            raise ConflictError(f"Policy {data['policy_id']} already exists")
+        authored_by = data.pop("authored_by")
+        policy = ISMSPolicyDB(
+            id=generate_id(), authored_by=authored_by, status=ApprovalStatusEnum.DRAFT, **data,
+        )
+        db.add(policy)
+        log_audit_trail(db, "isms_policy", policy.id, policy.policy_id, "create", authored_by)
+        db.commit()
+        db.refresh(policy)
+        return policy
+
+    @staticmethod
+    def get(db: Session, policy_id: str) -> ISMSPolicyDB:
+        policy = (
+            db.query(ISMSPolicyDB)
+            .filter((ISMSPolicyDB.id == policy_id) | (ISMSPolicyDB.policy_id == policy_id))
+            .first()
+        )
+        if not policy:
+            raise NotFoundError("Policy not found")
+        return policy
+
+    @staticmethod
+    def update(db: Session, policy_id: str, data: dict, updated_by: str) -> ISMSPolicyDB:
+        policy = (
+            db.query(ISMSPolicyDB)
+            .filter((ISMSPolicyDB.id == policy_id) | (ISMSPolicyDB.policy_id == policy_id))
+            .first()
+        )
+        if not policy:
+            raise NotFoundError("Policy not found")
+        if policy.status == ApprovalStatusEnum.APPROVED:
+            version_parts = policy.version.split(".")
+            policy.version = f"{int(version_parts[0]) + 1}.0"
+            policy.status = ApprovalStatusEnum.DRAFT
+        for field, value in data.items():
+            setattr(policy, field, value)
+        log_audit_trail(db, "isms_policy", policy.id, policy.policy_id, "update", updated_by)
+        db.commit()
+        db.refresh(policy)
+        return policy
+
+    @staticmethod
+    def approve(db: Session, policy_id: str, reviewed_by: str, approved_by: str, effective_date: date) -> ISMSPolicyDB:
+        policy = (
+            db.query(ISMSPolicyDB)
+            .filter((ISMSPolicyDB.id == policy_id) | (ISMSPolicyDB.policy_id == policy_id))
+            .first()
+        )
+        if not policy:
+            raise NotFoundError("Policy not found")
+        policy.reviewed_by = reviewed_by
+        policy.approved_by = approved_by
+        policy.approved_at = datetime.now(timezone.utc)
+        policy.effective_date = effective_date
+        policy.next_review_date = date(
+            effective_date.year + (policy.review_frequency_months // 12),
+            effective_date.month,
+            effective_date.day,
+        )
+        policy.status = ApprovalStatusEnum.APPROVED
+        policy.approval_signature = create_signature(
+            f"{policy.policy_id}|{approved_by}|{datetime.now(timezone.utc).isoformat()}"
+        )
+        log_audit_trail(db, "isms_policy", policy.id, policy.policy_id, "approve", approved_by)
+        db.commit()
+        db.refresh(policy)
+        return policy
+
+
+# ============================================================================
+# Security Objectives (ISO 27001 6.2)
+# ============================================================================
+
+
+class SecurityObjectiveService:
+    """Business logic for Security Objectives."""
+
+    @staticmethod
+    def list_objectives(db: Session, category: Optional[str] = None, status: Optional[str] = None) -> tuple:
+        query = db.query(SecurityObjectiveDB)
+        if category:
+            query = query.filter(SecurityObjectiveDB.category == category)
+        if status:
+            query = query.filter(SecurityObjectiveDB.status == status)
+        objectives = query.order_by(SecurityObjectiveDB.objective_id).all()
+        return objectives, len(objectives)
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> SecurityObjectiveDB:
+        objective = SecurityObjectiveDB(id=generate_id(), status="active", **data)
+        db.add(objective)
+        log_audit_trail(db, "security_objective", objective.id, objective.objective_id, "create", created_by)
+        db.commit()
+        db.refresh(objective)
+        return objective
+
+    @staticmethod
+    def update(db: Session, objective_id: str, data: dict, updated_by: str) -> SecurityObjectiveDB:
+        objective = (
+            db.query(SecurityObjectiveDB)
+            .filter((SecurityObjectiveDB.id == objective_id) | (SecurityObjectiveDB.objective_id == objective_id))
+            .first()
+        )
+        if not objective:
+            raise NotFoundError("Objective not found")
+        for field, value in data.items():
+            setattr(objective, field, value)
+        if objective.progress_percentage >= 100 and objective.status == "active":
+            objective.status = "achieved"
+            objective.achieved_date = date.today()
+        log_audit_trail(db, "security_objective", objective.id, objective.objective_id, "update", updated_by)
+        db.commit()
+        db.refresh(objective)
+        return objective
+
+
+# ============================================================================
+# Statement of Applicability (SoA)
+# ============================================================================
+
+
+class SoAService:
+    """Business logic for Statement of Applicability."""
+
+    @staticmethod
+    def list_entries(
+        db: Session,
+        is_applicable: Optional[bool] = None,
+        implementation_status: Optional[str] = None,
+        category: Optional[str] = None,
+    ) -> dict:
+        query = db.query(StatementOfApplicabilityDB)
+        if is_applicable is not None:
+            query = query.filter(StatementOfApplicabilityDB.is_applicable == is_applicable)
+        if implementation_status:
+            query = query.filter(StatementOfApplicabilityDB.implementation_status == implementation_status)
+        if category:
+            query = query.filter(StatementOfApplicabilityDB.annex_a_category == category)
+        entries = query.order_by(StatementOfApplicabilityDB.annex_a_control).all()
+        applicable_count = sum(1 for e in entries if e.is_applicable)
+        implemented_count = sum(1 for e in entries if e.implementation_status == "implemented")
+        planned_count = sum(1 for e in entries if e.implementation_status == "planned")
+        return {
+            "entries": entries,
+            "total": len(entries),
+            "applicable_count": applicable_count,
+            "not_applicable_count": len(entries) - applicable_count,
+            "implemented_count": implemented_count,
+            "planned_count": planned_count,
+        }
+
+    @staticmethod
+    def create(db: Session, data: dict, created_by: str) -> StatementOfApplicabilityDB:
+        existing = (
+            db.query(StatementOfApplicabilityDB)
+            .filter(StatementOfApplicabilityDB.annex_a_control == data["annex_a_control"])
+            .first()
+        )
+        if existing:
+            raise ConflictError(f"SoA entry for {data['annex_a_control']} already exists")
+        entry = StatementOfApplicabilityDB(id=generate_id(), **data)
+        db.add(entry)
+        log_audit_trail(db, "soa", entry.id, entry.annex_a_control, "create", created_by)
+        db.commit()
+        db.refresh(entry)
+        return entry
+
+    @staticmethod
+    def update(db: Session, entry_id: str, data: dict, updated_by: str) -> StatementOfApplicabilityDB:
+        entry = (
+            db.query(StatementOfApplicabilityDB)
+            .filter(
+                (StatementOfApplicabilityDB.id == entry_id)
+                | (StatementOfApplicabilityDB.annex_a_control == entry_id)
+            )
+            .first()
+        )
+        if not entry:
+            raise NotFoundError("SoA entry not found")
+        for field, value in data.items():
+            setattr(entry, field, value)
+        version_parts = entry.version.split(".")
+        entry.version = f"{version_parts[0]}.{int(version_parts[1]) + 1}"
+        log_audit_trail(db, "soa", entry.id, entry.annex_a_control, "update", updated_by)
+        db.commit()
+        db.refresh(entry)
+        return entry
+
+    @staticmethod
+    def approve(db: Session, entry_id: str, reviewed_by: str, approved_by: str) -> StatementOfApplicabilityDB:
+        entry = (
+            db.query(StatementOfApplicabilityDB)
+            .filter(
+                (StatementOfApplicabilityDB.id == entry_id)
+                | (StatementOfApplicabilityDB.annex_a_control == entry_id)
+            )
+            .first()
+        )
+        if not entry:
+            raise NotFoundError("SoA entry not found")
+        entry.reviewed_by = reviewed_by
+        entry.reviewed_at = datetime.now(timezone.utc)
+        entry.approved_by = approved_by
+        entry.approved_at = datetime.now(timezone.utc)
+        log_audit_trail(db, "soa", entry.id, entry.annex_a_control, "approve", approved_by)
+        db.commit()
+        db.refresh(entry)
+        return entry