feat: seed 10 canonical controls + CRUD endpoints + frontend editor
All checks were successful
CI/CD / go-lint (push) Has been skipped
CI/CD / python-lint (push) Has been skipped
CI/CD / nodejs-lint (push) Has been skipped
CI/CD / test-go-ai-compliance (push) Successful in 39s
CI/CD / test-python-backend-compliance (push) Successful in 39s
CI/CD / test-python-document-crawler (push) Successful in 30s
CI/CD / test-python-dsms-gateway (push) Successful in 20s
CI/CD / validate-canonical-controls (push) Successful in 12s
CI/CD / deploy-hetzner (push) Successful in 1m37s

- Migration 045: Seed 10 controls (AUTH, NET, SUP, LOG, WEB, DATA, CRYP, REL)
  with 39 open-source anchors into the database
- Backend: POST/PUT/DELETE endpoints for canonical controls CRUD
- Frontend proxy: PUT and DELETE methods added to canonical route
- Frontend: Control Library with create/edit/delete UI, full form with
  open anchor management, scope, requirements, evidence, test procedures

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-13 00:28:21 +01:00
parent 453eec9ed8
commit c87f07c99a
4 changed files with 878 additions and 47 deletions

View File

@@ -1,19 +1,21 @@
"""
FastAPI routes for the Canonical Control Library.
Provides read-only access to independently authored security controls.
All controls are formulated without proprietary nomenclature and anchored
in open-source frameworks (OWASP, NIST, ENISA).
Independently authored security controls anchored in open-source frameworks
(OWASP, NIST, ENISA). No proprietary nomenclature.
Endpoints:
GET /v1/canonical/frameworks — All frameworks
GET /v1/canonical/frameworks/{framework_id} — Framework details
GET /v1/canonical/frameworks/{framework_id}/controls — Controls of a framework
GET /v1/canonical/controls — All controls (filterable)
GET /v1/canonical/controls/{control_id} — Single control by control_id
GET /v1/canonical/sourcesSource registry
GET /v1/canonical/licenses — License matrix
POST /v1/canonical/controls/{control_id}/similarity-check — Too-close check
GET /v1/canonical/frameworks — All frameworks
GET /v1/canonical/frameworks/{framework_id} — Framework details
GET /v1/canonical/frameworks/{framework_id}/controls — Controls of a framework
GET /v1/canonical/controls — All controls (filterable)
GET /v1/canonical/controls/{control_id} — Single control
POST /v1/canonical/controls Create a control
PUT /v1/canonical/controls/{control_id} — Update a control
DELETE /v1/canonical/controls/{control_id} — Delete a control
GET /v1/canonical/sources — Source registry
GET /v1/canonical/licenses — License matrix
POST /v1/canonical/controls/{control_id}/similarity-check — Too-close check
"""
from __future__ import annotations
@@ -72,6 +74,42 @@ class ControlResponse(BaseModel):
updated_at: str
class ControlCreateRequest(BaseModel):
framework_id: str # e.g. 'bp_security_v1'
control_id: str # e.g. 'AUTH-003'
title: str
objective: str
rationale: str
scope: dict = {}
requirements: list = []
test_procedure: list = []
evidence: list = []
severity: str = "medium"
risk_score: Optional[float] = None
implementation_effort: Optional[str] = None
evidence_confidence: Optional[float] = None
open_anchors: list = []
release_state: str = "draft"
tags: list = []
class ControlUpdateRequest(BaseModel):
title: Optional[str] = None
objective: Optional[str] = None
rationale: Optional[str] = None
scope: Optional[dict] = None
requirements: Optional[list] = None
test_procedure: Optional[list] = None
evidence: Optional[list] = None
severity: Optional[str] = None
risk_score: Optional[float] = None
implementation_effort: Optional[str] = None
evidence_confidence: Optional[float] = None
open_anchors: Optional[list] = None
release_state: Optional[str] = None
tags: Optional[list] = None
class SimilarityCheckRequest(BaseModel):
source_text: str
candidate_text: str
@@ -266,6 +304,150 @@ async def get_control(control_id: str):
return _control_row(row)
# =============================================================================
# CONTROL CRUD (CREATE / UPDATE / DELETE)
# =============================================================================
@router.post("/controls", status_code=201)
async def create_control(body: ControlCreateRequest):
"""Create a new canonical control."""
import json as _json
import re
# Validate control_id format
if not re.match(r"^[A-Z]{2,6}-[0-9]{3}$", body.control_id):
raise HTTPException(status_code=400, detail="control_id must match DOMAIN-NNN (e.g. AUTH-001)")
if body.severity not in ("low", "medium", "high", "critical"):
raise HTTPException(status_code=400, detail="severity must be low/medium/high/critical")
if body.risk_score is not None and not (0 <= body.risk_score <= 10):
raise HTTPException(status_code=400, detail="risk_score must be 0..10")
with SessionLocal() as db:
# Resolve framework
fw = db.execute(
text("SELECT id FROM canonical_control_frameworks WHERE framework_id = :fid"),
{"fid": body.framework_id},
).fetchone()
if not fw:
raise HTTPException(status_code=404, detail=f"Framework '{body.framework_id}' not found")
# Check duplicate
existing = db.execute(
text("SELECT id FROM canonical_controls WHERE framework_id = :fid AND control_id = :cid"),
{"fid": str(fw.id), "cid": body.control_id},
).fetchone()
if existing:
raise HTTPException(status_code=409, detail=f"Control '{body.control_id}' already exists")
row = db.execute(
text("""
INSERT INTO canonical_controls (
framework_id, control_id, title, objective, rationale,
scope, requirements, test_procedure, evidence,
severity, risk_score, implementation_effort, evidence_confidence,
open_anchors, release_state, tags
) VALUES (
:fw_id, :cid, :title, :objective, :rationale,
:scope::jsonb, :requirements::jsonb, :test_procedure::jsonb, :evidence::jsonb,
:severity, :risk_score, :effort, :confidence,
:anchors::jsonb, :release_state, :tags::jsonb
)
RETURNING id, framework_id, control_id, title, objective, rationale,
scope, requirements, test_procedure, evidence,
severity, risk_score, implementation_effort,
evidence_confidence, open_anchors, release_state, tags,
created_at, updated_at
"""),
{
"fw_id": str(fw.id),
"cid": body.control_id,
"title": body.title,
"objective": body.objective,
"rationale": body.rationale,
"scope": _json.dumps(body.scope),
"requirements": _json.dumps(body.requirements),
"test_procedure": _json.dumps(body.test_procedure),
"evidence": _json.dumps(body.evidence),
"severity": body.severity,
"risk_score": body.risk_score,
"effort": body.implementation_effort,
"confidence": body.evidence_confidence,
"anchors": _json.dumps(body.open_anchors),
"release_state": body.release_state,
"tags": _json.dumps(body.tags),
},
).fetchone()
db.commit()
return _control_row(row)
@router.put("/controls/{control_id}")
async def update_control(control_id: str, body: ControlUpdateRequest):
"""Update an existing canonical control (partial update)."""
import json as _json
updates = body.dict(exclude_none=True)
if not updates:
raise HTTPException(status_code=400, detail="No fields to update")
if "severity" in updates and updates["severity"] not in ("low", "medium", "high", "critical"):
raise HTTPException(status_code=400, detail="severity must be low/medium/high/critical")
if "risk_score" in updates and updates["risk_score"] is not None and not (0 <= updates["risk_score"] <= 10):
raise HTTPException(status_code=400, detail="risk_score must be 0..10")
# Build dynamic SET clause
set_parts = []
params: dict[str, Any] = {"cid": control_id.upper()}
json_fields = {"scope", "requirements", "test_procedure", "evidence", "open_anchors", "tags"}
for key, val in updates.items():
col = "implementation_effort" if key == "implementation_effort" else key
col = "evidence_confidence" if key == "evidence_confidence" else col
if key in json_fields:
set_parts.append(f"{col} = :{key}::jsonb")
params[key] = _json.dumps(val)
else:
set_parts.append(f"{col} = :{key}")
params[key] = val
set_parts.append("updated_at = NOW()")
with SessionLocal() as db:
row = db.execute(
text(f"""
UPDATE canonical_controls
SET {', '.join(set_parts)}
WHERE control_id = :cid
RETURNING id, framework_id, control_id, title, objective, rationale,
scope, requirements, test_procedure, evidence,
severity, risk_score, implementation_effort,
evidence_confidence, open_anchors, release_state, tags,
created_at, updated_at
"""),
params,
).fetchone()
if not row:
raise HTTPException(status_code=404, detail="Control not found")
db.commit()
return _control_row(row)
@router.delete("/controls/{control_id}", status_code=204)
async def delete_control(control_id: str):
"""Delete a canonical control."""
with SessionLocal() as db:
result = db.execute(
text("DELETE FROM canonical_controls WHERE control_id = :cid"),
{"cid": control_id.upper()},
)
if result.rowcount == 0:
raise HTTPException(status_code=404, detail="Control not found")
db.commit()
return None
# =============================================================================
# SIMILARITY CHECK
# =============================================================================