feat(backend): add ControlVerificationTemplate table and CRUD endpoints

backend-api/alembic/env.py

@@ -20,6 +20,7 @@
 from app.models.gcp_connection import GCPConnection  # noqa
 from app.models.aws_connection import AWSConnection  # noqa
 from app.models.user_settings import UserSettings  # noqa
+from app.models.control_verification_template import ControlVerificationTemplate  # noqa
 from app.core.config import get_settings
 
 # this is the Alembic Config object, which provides

backend-api/alembic/versions/8a7b91ea95d9_add_control_verification_template.py

@@ -0,0 +1,60 @@
+"""add control verification template
+
+Revision ID: 8a7b91ea95d9
+Revises: j1k2l3m4n567
+Create Date: 2026-05-08 09:22:25.327975
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '8a7b91ea95d9'
+down_revision: Union[str, Sequence[str], None] = 'j1k2l3m4n567'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        'control_verification_template',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('framework', sa.String(length=50), nullable=False),
+        sa.Column('benchmark', sa.String(length=100), nullable=False),
+        sa.Column('version', sa.String(length=20), nullable=False),
+        sa.Column('control_id', sa.String(length=50), nullable=False),
+        sa.Column('title', sa.String(length=200), nullable=False),
+        sa.Column('instructions', sa.Text(), nullable=False),
+        sa.Column('keywords', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column('severity', sa.String(length=20), nullable=False),
+        sa.Column('evidence_type', sa.String(length=50), nullable=True),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint(
+            'framework',
+            'benchmark',
+            'version',
+            'control_id',
+            name='uq_template_framework_benchmark_version_control',
+        ),
+    )
+    op.create_index(
+        op.f('ix_control_verification_template_control_id'),
+        'control_verification_template',
+        ['control_id'],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index(
+        op.f('ix_control_verification_template_control_id'),
+        table_name='control_verification_template',
+    )
+    op.drop_table('control_verification_template')

backend-api/app/api/v1/router.py

@@ -9,6 +9,7 @@
     scans,
     settings,
     test,
+    verification_templates,
 )
 
 api_router = APIRouter()
@@ -39,3 +40,6 @@
 
 # User settings routes
 api_router.include_router(settings.router)
+
+# Manual control verification template routes
+api_router.include_router(verification_templates.router)

backend-api/app/api/v1/verification_templates.py

@@ -0,0 +1,213 @@
+"""CRUD endpoints for ControlVerificationTemplate.
+
+Powers the manual control verification workflow: admins maintain the
+templates that auditors see when verifying pending manual controls.
+
+Identity:
+    Each template is uniquely identified by the
+    (framework, benchmark, version, control_id) tuple. This matches the
+    same tuple Scan already uses (see app/api/v1/scans.py) so the lookup
+    pattern is consistent end-to-end and the same control_id can carry
+    different instructions across benchmark versions and across frameworks.
+
+Authorisation:
+- POST / PATCH / DELETE: admin only (templates are GRC content; auditors
+  consume them but do not author them).
+- GET endpoints: any authenticated user (auditors and viewers need read
+  access to surface instructions in the UI).
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.auth import get_current_user
+from app.core.permissions import require_admin
+from app.db.session import get_async_session
+from app.models.control_verification_template import ControlVerificationTemplate
+from app.models.user import User
+from app.schemas.control_verification_template import (
+    ControlVerificationTemplateCreate,
+    ControlVerificationTemplateRead,
+    ControlVerificationTemplateUpdate,
+)
+
+router = APIRouter(
+    prefix="/verification-templates",
+    tags=["Verification Templates"],
+)
+
+
+# ── Internal helpers ────────────────────────────────────────────────────
+
+
+async def _get_template_or_404(
+    db: AsyncSession,
+    framework: str,
+    benchmark: str,
+    version: str,
+    control_id: str,
+) -> ControlVerificationTemplate:
+    """Fetch the template for the given tuple or raise 404."""
+    result = await db.execute(
+        select(ControlVerificationTemplate).where(
+            ControlVerificationTemplate.framework == framework,
+            ControlVerificationTemplate.benchmark == benchmark,
+            ControlVerificationTemplate.version == version,
+            ControlVerificationTemplate.control_id == control_id,
+        )
+    )
+    template = result.scalar_one_or_none()
+    if template is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=(
+                f"Template for {framework}/{benchmark}/{version}/{control_id} not found"
+            ),
+        )
+    return template
+
+
+# ── Endpoints ───────────────────────────────────────────────────────────
+
+
+@router.post(
+    "/",
+    response_model=ControlVerificationTemplateRead,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create a verification template (admin only)",
+)
+async def create_template(
+    data: ControlVerificationTemplateCreate,
+    db: AsyncSession = Depends(get_async_session),
+    _: User = Depends(require_admin),
+) -> ControlVerificationTemplate:
+    """Create a new verification template for a manual control.
+
+    Returns 409 Conflict if a template for the given
+    (framework, benchmark, version, control_id) tuple already exists.
+    """
+    result = await db.execute(
+        select(ControlVerificationTemplate).where(
+            ControlVerificationTemplate.framework == data.framework,
+            ControlVerificationTemplate.benchmark == data.benchmark,
+            ControlVerificationTemplate.version == data.version,
+            ControlVerificationTemplate.control_id == data.control_id,
+        )
+    )
+    if result.scalar_one_or_none() is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                f"Template for {data.framework}/{data.benchmark}/"
+                f"{data.version}/{data.control_id} already exists"
+            ),
+        )
+
+    template = ControlVerificationTemplate(**data.model_dump())
+    db.add(template)
+    await db.commit()
+    await db.refresh(template)
+    return template
+
+
+@router.get(
+    "/",
+    response_model=list[ControlVerificationTemplateRead],
+    summary="List verification templates",
+)
+async def list_templates(
+    db: AsyncSession = Depends(get_async_session),
+    _: User = Depends(get_current_user),
+    framework: str | None = None,
+    benchmark: str | None = None,
+    version: str | None = None,
+) -> list[ControlVerificationTemplate]:
+    """List verification templates, optionally filtered by framework/benchmark/version.
+
+    Sorted by (framework, benchmark, version, control_id) so output is stable
+    across calls. Phase 2 seeding scripts can pass the tuple to fetch only
+    the templates for a single benchmark version.
+    """
+    query = select(ControlVerificationTemplate)
+    if framework is not None:
+        query = query.where(ControlVerificationTemplate.framework == framework)
+    if benchmark is not None:
+        query = query.where(ControlVerificationTemplate.benchmark == benchmark)
+    if version is not None:
+        query = query.where(ControlVerificationTemplate.version == version)
+    query = query.order_by(
+        ControlVerificationTemplate.framework,
+        ControlVerificationTemplate.benchmark,
+        ControlVerificationTemplate.version,
+        ControlVerificationTemplate.control_id,
+    )
+
+    result = await db.execute(query)
+    return list(result.scalars().all())
+
+
+@router.get(
+    "/{framework}/{benchmark}/{version}/{control_id}",
+    response_model=ControlVerificationTemplateRead,
+    summary="Get a verification template by (framework, benchmark, version, control_id)",
+)
+async def get_template(
+    framework: str,
+    benchmark: str,
+    version: str,
+    control_id: str,
+    db: AsyncSession = Depends(get_async_session),
+    _: User = Depends(get_current_user),
+) -> ControlVerificationTemplate:
+    """Fetch the verification template for the given tuple."""
+    return await _get_template_or_404(db, framework, benchmark, version, control_id)
+
+
+@router.patch(
+    "/{framework}/{benchmark}/{version}/{control_id}",
+    response_model=ControlVerificationTemplateRead,
+    summary="Partially update a verification template (admin only)",
+)
+async def update_template(
+    framework: str,
+    benchmark: str,
+    version: str,
+    control_id: str,
+    data: ControlVerificationTemplateUpdate,
+    db: AsyncSession = Depends(get_async_session),
+    _: User = Depends(require_admin),
+) -> ControlVerificationTemplate:
+    """Partially update a template. Only fields provided in the body are applied.
+
+    Explicit null values for non-nullable fields are rejected by the schema
+    validator (422), not allowed through to the DB.
+    """
+    template = await _get_template_or_404(db, framework, benchmark, version, control_id)
+
+    update_fields = data.model_dump(exclude_unset=True)
+    for field, value in update_fields.items():
+        setattr(template, field, value)
+
+    await db.commit()
+    await db.refresh(template)
+    return template
+
+
+@router.delete(
+    "/{framework}/{benchmark}/{version}/{control_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="Delete a verification template (admin only)",
+)
+async def delete_template(
+    framework: str,
+    benchmark: str,
+    version: str,
+    control_id: str,
+    db: AsyncSession = Depends(get_async_session),
+    _: User = Depends(require_admin),
+) -> None:
+    """Delete the verification template for the given tuple."""
+    template = await _get_template_or_404(db, framework, benchmark, version, control_id)
+    await db.delete(template)
+    await db.commit()

backend-api/app/models/__init__.py

@@ -12,6 +12,7 @@
 from app.models.evidence_validation import EvidenceValidation
 from app.models.contact import ContactSubmission, SubmissionNote, SubmissionHistory
 from app.models.user_settings import UserSettings
+from app.models.control_verification_template import ControlVerificationTemplate
 
 __all__ = [
     "User",
@@ -29,4 +30,5 @@
     "SubmissionNote",
     "SubmissionHistory",
     "UserSettings",
+    "ControlVerificationTemplate",
 ]

backend-api/app/models/control_verification_template.py

@@ -0,0 +1,78 @@
+"""ControlVerificationTemplate model.
+
+Stores per-control auditor instructions, keywords, and severity for the manual
+controls that AutoAudit cannot automate via the M365 collectors. Each row is
+uniquely identified by the (framework, benchmark, version, control_id) tuple
+so that the same control_id can carry different instructions across CIS
+benchmark versions (e.g. "1.1.2" in v3.1.0 vs v6.0.0) and across frameworks
+(e.g. CIS M365 vs future Azure benchmarks).
+
+Powers the semi-automated manual verification workflow: the auditor opens a
+pending manual control, sees the instructions, uploads evidence, and the
+validator matches the keywords to suggest a verdict.
+
+Keywords are stored as JSONB to match the patterns already used by
+EvidenceValidation.matches_json and ScanResult.evidence.
+"""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import DateTime, Integer, String, Text, UniqueConstraint
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.sql import func
+
+from app.db.base import Base
+
+
+class ControlVerificationTemplate(Base):
+    """Verification template for a single manual control within a framework/benchmark/version."""
+
+    __tablename__ = "control_verification_template"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+
+    # Framework/benchmark/version tuple — mirrors Scan.framework/benchmark/version
+    # so the lookup pattern (and column types) are consistent across the schema.
+    framework: Mapped[str] = mapped_column(String(50), nullable=False)
+    benchmark: Mapped[str] = mapped_column(String(100), nullable=False)
+    version: Mapped[str] = mapped_column(String(20), nullable=False)
+
+    # CIS control identifier within the (framework, benchmark, version) above,
+    # e.g. "1.1.2". Must match scan_result.control_id for the same tuple.
+    control_id: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
+
+    # Human-readable control title from the benchmark.
+    title: Mapped[str] = mapped_column(String(200), nullable=False)
+
+    # Numbered, portal-specific auditor instructions.
+    instructions: Mapped[str] = mapped_column(Text, nullable=False)
+
+    # List of keywords expected to appear in compliant evidence.
+    # JSONB so we can index/query individual keywords later if needed.
+    keywords: Mapped[list] = mapped_column(JSONB, nullable=False)
+
+    # Risk severity: "high" | "medium" | "low". Enforced at the API layer via
+    # the Pydantic Literal type on the Create/Update schemas.
+    severity: Mapped[str] = mapped_column(String(20), nullable=False)
+
+    # Expected evidence format: "screenshot" | "pdf_export" | "comment_only".
+    evidence_type: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "framework",
+            "benchmark",
+            "version",
+            "control_id",
+            name="uq_template_framework_benchmark_version_control",
+        ),
+    )