-
Notifications
You must be signed in to change notification settings - Fork 40
feat(backend): add ControlVerificationTemplate table and CRUD endpoints #229
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| """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('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') | ||
| ) | ||
| op.create_index(op.f('ix_control_verification_template_control_id'), 'control_verification_template', ['control_id'], unique=True) | ||
|
|
||
|
|
||
| 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') |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| """CRUD endpoints for ControlVerificationTemplate. | ||
|
|
||
| Powers the manual control verification workflow: admins maintain the | ||
| templates that auditors see when verifying pending manual controls. | ||
|
|
||
| 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"], | ||
| ) | ||
|
|
||
|
|
||
| @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 control_id already exists. | ||
| """ | ||
| result = await db.execute( | ||
| select(ControlVerificationTemplate).where( | ||
| 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 control_id '{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 all verification templates", | ||
| ) | ||
| async def list_templates( | ||
| db: AsyncSession = Depends(get_async_session), | ||
| _: User = Depends(get_current_user), | ||
| ) -> list[ControlVerificationTemplate]: | ||
| """List every verification template, ordered by control_id.""" | ||
| result = await db.execute( | ||
| select(ControlVerificationTemplate).order_by( | ||
| ControlVerificationTemplate.control_id | ||
| ) | ||
| ) | ||
| return list(result.scalars().all()) | ||
|
|
||
|
|
||
| @router.get( | ||
| "/{control_id}", | ||
| response_model=ControlVerificationTemplateRead, | ||
| summary="Get a verification template by control_id", | ||
| ) | ||
| async def get_template( | ||
| control_id: str, | ||
| db: AsyncSession = Depends(get_async_session), | ||
| _: User = Depends(get_current_user), | ||
| ) -> ControlVerificationTemplate: | ||
| """Fetch the verification template for the given control_id.""" | ||
| result = await db.execute( | ||
| select(ControlVerificationTemplate).where( | ||
| 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 control_id '{control_id}' not found", | ||
| ) | ||
| return template | ||
|
|
||
|
|
||
| @router.patch( | ||
| "/{control_id}", | ||
| response_model=ControlVerificationTemplateRead, | ||
| summary="Partially update a verification template (admin only)", | ||
| ) | ||
| async def update_template( | ||
| 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.""" | ||
| result = await db.execute( | ||
| select(ControlVerificationTemplate).where( | ||
| 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 control_id '{control_id}' not found", | ||
| ) | ||
|
|
||
| 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( | ||
| "/{control_id}", | ||
| status_code=status.HTTP_204_NO_CONTENT, | ||
| summary="Delete a verification template (admin only)", | ||
| ) | ||
| async def delete_template( | ||
| control_id: str, | ||
| db: AsyncSession = Depends(get_async_session), | ||
| _: User = Depends(require_admin), | ||
| ) -> None: | ||
| """Delete the verification template for the given control_id.""" | ||
| result = await db.execute( | ||
| select(ControlVerificationTemplate).where( | ||
| 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 control_id '{control_id}' not found", | ||
| ) | ||
|
|
||
| await db.delete(template) | ||
| await db.commit() | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,6 +12,8 @@ | |
| 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.user_settings import UserSettings | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This line is a duplicate of the one above and should be removed |
||
| from app.models.control_verification_template import ControlVerificationTemplate | ||
|
|
||
| __all__ = [ | ||
| "User", | ||
|
|
@@ -29,4 +31,5 @@ | |
| "SubmissionNote", | ||
| "SubmissionHistory", | ||
| "UserSettings", | ||
| "ControlVerificationTemplate", | ||
| ] | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| """ControlVerificationTemplate model. | ||
|
|
||
| Stores per-control auditor instructions, keywords, and severity for the 14 | ||
| manual controls that AutoAudit cannot automate via the M365 collectors. | ||
| Each row corresponds to one CIS control_id (e.g. "1.1.2") and 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 | ||
| 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 CIS control. | ||
|
|
||
| One row per control_id (enforced via unique constraint). | ||
| """ | ||
|
|
||
| __tablename__ = "control_verification_template" | ||
|
|
||
| id: Mapped[int] = mapped_column(Integer, primary_key=True) | ||
|
|
||
| # CIS control identifier, e.g. "1.1.2". Must match scan_result.control_id. | ||
| control_id: Mapped[str] = mapped_column( | ||
| String(50), unique=True, nullable=False, index=True | ||
| ) | ||
|
|
||
| # Human-readable control title from the CIS 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". | ||
| severity: Mapped[str] = mapped_column(String(20), nullable=False) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The comment above shows what the values should be, but there's no actual enforcement that what's coming in here matches those values. If this is what we expect to see, we should enforce this - Pydantic supports the concept of a |
||
|
|
||
| # 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() | ||
| ) | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| """Pydantic schemas for ControlVerificationTemplate.""" | ||
|
|
||
| from datetime import datetime | ||
|
|
||
| from pydantic import BaseModel, Field | ||
|
|
||
|
|
||
| class ControlVerificationTemplateCreate(BaseModel): | ||
| """Schema for creating a new verification template.""" | ||
|
|
||
| control_id: str = Field(..., max_length=50) | ||
| title: str = Field(..., max_length=200) | ||
| instructions: str | ||
| keywords: list[str] | ||
| severity: str = Field(..., max_length=20) | ||
| evidence_type: str | None = Field(None, max_length=50) | ||
|
|
||
|
|
||
| class ControlVerificationTemplateUpdate(BaseModel): | ||
| """Schema for partial update of a verification template. | ||
|
|
||
| All fields optional — only provided fields are applied. | ||
| control_id is intentionally not updatable (it's the resource key). | ||
| """ | ||
|
|
||
| title: str | None = Field(None, max_length=200) | ||
| instructions: str | None = None | ||
| keywords: list[str] | None = None | ||
| severity: str | None = Field(None, max_length=20) | ||
| evidence_type: str | None = Field(None, max_length=50) | ||
|
|
||
|
|
||
| class ControlVerificationTemplateRead(BaseModel): | ||
| """Schema for reading a verification template.""" | ||
|
|
||
| id: int | ||
| control_id: str | ||
| title: str | ||
| instructions: str | ||
| keywords: list[str] | ||
| severity: str | ||
| evidence_type: str | None | ||
| created_at: datetime | ||
| updated_at: datetime | ||
|
|
||
| class Config: | ||
| from_attributes = True |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When a PATCH request explicitly sends
nullfor non-nullable fields such astitle,instructions,keywords, orseverity, the update schema accepts it andexclude_unset=Truestill includes that field, so this loop writesNoneto columns declarednullable=False. In that scenario the subsequent commit raises a database integrity error and returns a 500 instead of a client error; either disallowNonein the update schema or reject provided nulls before assigning.Useful? React with 👍 / 👎.