gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(compliance): add GDPR, HIPAA, SOC2 compliance templates with one-click apply and dashboard

Browse Source 
- Add ComplianceTemplate model in app/models.py
- Create database migration 027_add_compliance_templates
- Add compliance_enabled feature flag to config and settings metadata
- Create compliance_service.py with pre-built template definitions and evaluation
- Create compliance API endpoints (list, get, apply, status, summary)
- Create compliance admin view and dashboard template
- Add compliance link to admin navigation (desktop and mobile)
- Seed compliance templates at application startup
- Add comprehensive tests (29 passing)

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-09 23:55:29 +00:00
parent 341dad643f
commit 666f739f4e
14 changed files with 1573 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+2
View File
@@ -11,6 +11,7 @@ from app.api.api_tokens import router as api_tokens_router
from app.api.azure import router as azure_router
from app.api.backup import router as backup_router
from app.api.billing import router as billing_router
from app.api.compliance import router as compliance_router
from app.api.database import router as database_router
from app.api.diagnostic import router as diagnostic_router
from app.api.dropbox import router as dropbox_router
@@ -82,3 +83,4 @@ router.include_router(imap_accounts_router)
router.include_router(integrations_router)
router.include_router(notifications_router)
router.include_router(scheduled_jobs_router)
router.include_router(compliance_router)
app/api/compliance.py
+179
View File
@@ -0,0 +1,179 @@
"""API endpoints for managing compliance templates (GDPR, HIPAA, SOC2).
All endpoints require admin privileges.
Available routes:
GET /api/compliance/templates – list all compliance templates
GET /api/compliance/templates/{name} – get a single template with checks
POST /api/compliance/templates/{name}/apply – one-click apply a template
GET /api/compliance/templates/{name}/status – evaluate compliance status
GET /api/compliance/summary – overall compliance dashboard data
"""
import logging
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.database import get_db
from app.utils.compliance_service import (
apply_template,
evaluate_template_status,
get_all_templates,
get_compliance_summary,
get_template_by_name,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/compliance", tags=["compliance"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Authorisation helper
# ---------------------------------------------------------------------------
def _require_admin(request: Request) -> dict:
"""Ensure the caller is an admin; raises HTTP 403 otherwise."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
# ---------------------------------------------------------------------------
# Pydantic response models
# ---------------------------------------------------------------------------
class CheckResult(BaseModel):
"""Individual compliance check result."""
key: str
label: str
description: str
expected: str
actual: str
passing: bool
class TemplateStatusResponse(BaseModel):
"""Status evaluation for a compliance template."""
status: str
total: int
passed: int
failed: int
check_results: list[CheckResult]
class TemplateResponse(BaseModel):
"""Full compliance template representation."""
id: int
name: str
display_name: str
description: str | None
enabled: bool
status: str
applied_at: str | None
applied_by: str | None
settings: dict[str, str]
checks: list[dict[str, Any]]
check_count: int
class ApplyResponse(BaseModel):
"""Result of applying a compliance template."""
success: bool
template: str | None = None
applied_settings: dict[str, str] | None = None
errors: list[str] | None = None
error: str | None = None
status: TemplateStatusResponse | None = None
class SummaryTemplateResponse(BaseModel):
"""Per-template summary for the compliance dashboard."""
name: str
display_name: str
enabled: bool
status: str
total: int
passed: int
failed: int
applied_at: str | None
applied_by: str | None
class ComplianceSummaryResponse(BaseModel):
"""Overall compliance dashboard summary."""
overall_status: str
total_checks: int
total_passed: int
total_failed: int
templates: list[SummaryTemplateResponse]
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/templates", response_model=list[TemplateResponse])
async def list_templates(db: DbSession, admin: AdminUser) -> list[dict[str, Any]]:
"""List all compliance templates with their current status."""
return get_all_templates(db)
@router.get("/templates/{name}", response_model=TemplateResponse)
async def get_template(name: str, db: DbSession, admin: AdminUser) -> dict[str, Any]:
"""Get a single compliance template by name."""
templates = get_all_templates(db)
for t in templates:
if t["name"] == name:
return t
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Template '{name}' not found")
@router.post("/templates/{name}/apply", response_model=ApplyResponse)
async def apply_compliance_template(name: str, db: DbSession, admin: AdminUser) -> dict[str, Any]:
"""Apply a compliance template (one-click).
Writes all template settings to the database and evaluates the resulting
compliance status.
"""
template = get_template_by_name(db, name)
if template is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Template '{name}' not found")
admin_email = admin.get("email", "admin")
result = apply_template(db, name, applied_by=admin_email)
if not result.get("success") and result.get("error"):
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"])
return result
@router.get("/templates/{name}/status", response_model=TemplateStatusResponse)
async def get_template_status(name: str, db: DbSession, admin: AdminUser) -> dict[str, Any]:
"""Evaluate the live compliance status of a template."""
template = get_template_by_name(db, name)
if template is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Template '{name}' not found")
return evaluate_template_status(db, name)
@router.get("/summary", response_model=ComplianceSummaryResponse)
async def compliance_summary(db: DbSession, admin: AdminUser) -> dict[str, Any]:
"""Overall compliance dashboard summary across all templates."""
return get_compliance_summary(db)
app/config.py
+8
View File
@@ -484,6 +484,14 @@ class Settings(BaseSettings):
# Feature flags
allow_file_delete: bool = True # Default to allowing file deletion from database
compliance_enabled: bool = Field(
default=True,
description=(
"Enable the compliance templates dashboard (GDPR, HIPAA, SOC 2). "
"When enabled, admins can view compliance status and apply "
"pre-built regulatory configurations. Default: True."
),
)
# PDF/A archival conversion settings
enable_pdfa_conversion: bool = Field(
app/main.py
+14
View File
@@ -146,6 +146,20 @@ async def lifespan(app: FastAPI):
except Exception:
logging.debug("Scheduled jobs seeding skipped — DB may not be ready yet") # noqa: S110
# Seed the built-in compliance templates (GDPR, HIPAA, SOC2) so they
# are available in the admin compliance dashboard on first startup.
try:
from app.database import SessionLocal as _SessionLocal # noqa: F811
from app.utils.compliance_service import seed_compliance_templates as _seed_compliance
_db_compliance = _SessionLocal()
try:
_seed_compliance(_db_compliance)
finally:
_db_compliance.close()
except Exception:
logging.debug("Compliance template seeding skipped — DB may not be ready yet") # noqa: S110
# Application is now running
yield
app/models.py
+23
View File
@@ -833,3 +833,26 @@ class ScheduledJob(Base):
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
class ComplianceTemplate(Base):
"""Pre-built compliance configuration templates (GDPR, HIPAA, SOC2).
Each row represents an applied compliance template. The ``settings_json``
column stores the concrete setting key/value pairs that were written when
the template was applied. ``status`` tracks the current compliance posture.
"""
__tablename__ = "compliance_templates"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), unique=True, nullable=False, index=True) # GDPR, HIPAA, SOC2
display_name = Column(String(100), nullable=False)
description = Column(Text, nullable=True)
settings_json = Column(Text, nullable=False, default="{}") # JSON of applied settings
enabled = Column(Boolean, nullable=False, default=False)
status = Column(String(20), nullable=False, default="not_applied") # not_applied, compliant, partial, non_compliant
applied_at = Column(DateTime(timezone=True), nullable=True)
applied_by = Column(String(255), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
app/utils/compliance_service.py
+433
View File
@@ -0,0 +1,433 @@
"""Compliance service for managing GDPR, HIPAA, and SOC2 compliance templates.
Provides pre-built compliance configurations that can be applied with one click
to ensure the DocuElevate instance meets regulatory requirements.
"""
import json
import logging
from datetime import datetime, timezone
from typing import Any
from sqlalchemy.orm import Session
from app.models import ComplianceTemplate
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Pre-built compliance template definitions
# ---------------------------------------------------------------------------
COMPLIANCE_TEMPLATES: dict[str, dict[str, Any]] = {
"gdpr": {
"display_name": "GDPR (General Data Protection Regulation)",
"description": (
"European Union regulation for data protection and privacy. "
"Enforces data minimisation, encryption at rest, audit logging, "
"and limits PII exposure in telemetry."
),
"settings": {
"auth_enabled": "True",
"sentry_send_default_pii": "False",
"security_headers_enabled": "True",
"security_header_hsts_enabled": "True",
"security_header_csp_enabled": "True",
"security_header_x_frame_options_enabled": "True",
"enable_deduplication": "True",
},
"checks": [
{
"key": "auth_enabled",
"expected": "True",
"label": "Authentication enabled",
"description": "User authentication must be enabled to control access to personal data.",
},
{
"key": "sentry_send_default_pii",
"expected": "False",
"label": "PII excluded from telemetry",
"description": "Personally identifiable information must not be sent to external monitoring services.",
},
{
"key": "security_headers_enabled",
"expected": "True",
"label": "Security headers enabled",
"description": "HTTP security headers protect against common web vulnerabilities.",
},
{
"key": "security_header_hsts_enabled",
"expected": "True",
"label": "HSTS enabled",
"description": "HTTP Strict Transport Security ensures encrypted connections.",
},
{
"key": "security_header_csp_enabled",
"expected": "True",
"label": "Content Security Policy enabled",
"description": "CSP headers prevent cross-site scripting and data injection attacks.",
},
{
"key": "security_header_x_frame_options_enabled",
"expected": "True",
"label": "Clickjacking protection enabled",
"description": "X-Frame-Options header prevents clickjacking attacks.",
},
{
"key": "enable_deduplication",
"expected": "True",
"label": "Deduplication enabled",
"description": "Data minimisation: avoid storing duplicate documents.",
},
],
},
"hipaa": {
"display_name": "HIPAA (Health Insurance Portability and Accountability Act)",
"description": (
"United States regulation for protecting health information. "
"Requires strong access controls, audit trails, encryption, "
"and strict session management."
),
"settings": {
"auth_enabled": "True",
"multi_user_enabled": "True",
"sentry_send_default_pii": "False",
"security_headers_enabled": "True",
"security_header_hsts_enabled": "True",
"security_header_csp_enabled": "True",
"security_header_x_frame_options_enabled": "True",
"enable_deduplication": "True",
},
"checks": [
{
"key": "auth_enabled",
"expected": "True",
"label": "Authentication enabled",
"description": "Access controls are required to protect electronic Protected Health Information (ePHI).",
},
{
"key": "multi_user_enabled",
"expected": "True",
"label": "Multi-user mode enabled",
"description": "Individual user accounts required for access accountability.",
},
{
"key": "sentry_send_default_pii",
"expected": "False",
"label": "PII excluded from telemetry",
"description": "Protected Health Information must not be sent to external services.",
},
{
"key": "security_headers_enabled",
"expected": "True",
"label": "Security headers enabled",
"description": "Security headers protect ePHI during transmission.",
},
{
"key": "security_header_hsts_enabled",
"expected": "True",
"label": "HSTS enabled",
"description": "Encrypted transport required for all ePHI transmissions.",
},
{
"key": "security_header_csp_enabled",
"expected": "True",
"label": "Content Security Policy enabled",
"description": "CSP prevents injection attacks that could expose ePHI.",
},
{
"key": "security_header_x_frame_options_enabled",
"expected": "True",
"label": "Clickjacking protection enabled",
"description": "Prevents embedding the application in unauthorized frames.",
},
{
"key": "enable_deduplication",
"expected": "True",
"label": "Deduplication enabled",
"description": "Minimise data footprint for ePHI.",
},
],
},
"soc2": {
"display_name": "SOC 2 (Service Organization Control 2)",
"description": (
"Trust Service Criteria framework for service organisations. "
"Focuses on security, availability, processing integrity, "
"confidentiality, and privacy."
),
"settings": {
"auth_enabled": "True",
"multi_user_enabled": "True",
"sentry_send_default_pii": "False",
"security_headers_enabled": "True",
"security_header_hsts_enabled": "True",
"security_header_csp_enabled": "True",
"security_header_x_frame_options_enabled": "True",
"enable_deduplication": "True",
},
"checks": [
{
"key": "auth_enabled",
"expected": "True",
"label": "Authentication enabled",
"description": "Logical access controls required (CC6.1).",
},
{
"key": "multi_user_enabled",
"expected": "True",
"label": "Multi-user mode enabled",
"description": "Individual user accounts for access management (CC6.2).",
},
{
"key": "sentry_send_default_pii",
"expected": "False",
"label": "PII excluded from telemetry",
"description": "Confidential information must not leak to external services (CC6.7).",
},
{
"key": "security_headers_enabled",
"expected": "True",
"label": "Security headers enabled",
"description": "Protection against common web threats (CC6.6).",
},
{
"key": "security_header_hsts_enabled",
"expected": "True",
"label": "HSTS enabled",
"description": "Encrypted transport in transit (CC6.7).",
},
{
"key": "security_header_csp_enabled",
"expected": "True",
"label": "Content Security Policy enabled",
"description": "Application-level security controls (CC6.6).",
},
{
"key": "security_header_x_frame_options_enabled",
"expected": "True",
"label": "Clickjacking protection enabled",
"description": "UI redress attack prevention (CC6.6).",
},
{
"key": "enable_deduplication",
"expected": "True",
"label": "Deduplication enabled",
"description": "Data integrity through deduplication (PI1.1).",
},
],
},
}
def seed_compliance_templates(db: Session) -> None:
"""Create or update the built-in compliance template rows.
Called once at application startup to ensure the ``compliance_templates``
table always contains the latest definitions.
"""
for name, defn in COMPLIANCE_TEMPLATES.items():
existing = db.query(ComplianceTemplate).filter(ComplianceTemplate.name == name).first()
if existing is None:
template = ComplianceTemplate(
name=name,
display_name=defn["display_name"],
description=defn["description"],
settings_json=json.dumps(defn["settings"]),
enabled=False,
status="not_applied",
)
db.add(template)
logger.info(f"Seeded compliance template: {name}")
else:
# Update display_name and description if changed, but preserve user state
existing.display_name = defn["display_name"]
existing.description = defn["description"]
try:
db.commit()
except Exception:
db.rollback()
logger.exception("Failed to seed compliance templates")
def get_all_templates(db: Session) -> list[dict[str, Any]]:
"""Return all compliance templates with their current status."""
templates = db.query(ComplianceTemplate).order_by(ComplianceTemplate.name).all()
result = []
for t in templates:
defn = COMPLIANCE_TEMPLATES.get(t.name, {})
checks = defn.get("checks", [])
result.append(
{
"id": t.id,
"name": t.name,
"display_name": t.display_name,
"description": t.description,
"enabled": t.enabled,
"status": t.status,
"applied_at": t.applied_at.isoformat() if t.applied_at else None,
"applied_by": t.applied_by,
"settings": json.loads(t.settings_json) if t.settings_json else {},
"checks": checks,
"check_count": len(checks),
}
)
return result
def get_template_by_name(db: Session, name: str) -> ComplianceTemplate | None:
"""Retrieve a single compliance template by name."""
return db.query(ComplianceTemplate).filter(ComplianceTemplate.name == name).first()
def evaluate_template_status(db: Session, name: str) -> dict[str, Any]:
"""Evaluate the compliance status of a template against live settings.
Returns a dict with ``status``, ``total``, ``passed``, ``failed``, and
a list of individual ``check_results``.
"""
from app.config import settings as app_settings
from app.utils.settings_service import get_all_settings_from_db
defn = COMPLIANCE_TEMPLATES.get(name)
if defn is None:
return {"status": "unknown", "total": 0, "passed": 0, "failed": 0, "check_results": []}
db_settings = get_all_settings_from_db(db)
checks = defn.get("checks", [])
results: list[dict[str, Any]] = []
passed = 0
for check in checks:
key = check["key"]
expected = check["expected"]
# Resolve effective value: DB > config object
if key in db_settings and db_settings[key] is not None:
actual = str(db_settings[key])
else:
actual = str(getattr(app_settings, key, ""))
is_passing = actual.lower() == expected.lower()
if is_passing:
passed += 1
results.append(
{
"key": key,
"label": check["label"],
"description": check["description"],
"expected": expected,
"actual": actual,
"passing": is_passing,
}
)
total = len(checks)
if passed == total:
status = "compliant"
elif passed > 0:
status = "partial"
else:
status = "non_compliant"
return {
"status": status,
"total": total,
"passed": passed,
"failed": total - passed,
"check_results": results,
}
def apply_template(db: Session, name: str, applied_by: str = "admin") -> dict[str, Any]:
"""Apply a compliance template by writing its settings to the database.
Returns a summary of what was applied.
"""
from app.utils.settings_service import save_setting_to_db
defn = COMPLIANCE_TEMPLATES.get(name)
if defn is None:
return {"success": False, "error": f"Unknown template: {name}"}
template = get_template_by_name(db, name)
if template is None:
return {"success": False, "error": f"Template not found in database: {name}"}
applied_settings: dict[str, str] = {}
errors: list[str] = []
for key, value in defn["settings"].items():
try:
save_setting_to_db(db, key, value, changed_by=f"compliance:{name}")
applied_settings[key] = value
except Exception as e:
errors.append(f"{key}: {e}")
logger.error(f"Failed to apply compliance setting {key}={value}: {e}")
# Update the template record
now = datetime.now(timezone.utc)
template.enabled = True
template.settings_json = json.dumps(applied_settings)
template.applied_at = now
template.applied_by = applied_by
# Evaluate and store status
eval_result = evaluate_template_status(db, name)
template.status = eval_result["status"]
try:
db.commit()
except Exception:
db.rollback()
logger.exception(f"Failed to update compliance template record: {name}")
return {"success": False, "error": "Database commit failed"}
logger.info(f"Applied compliance template '{name}' by {applied_by}: {len(applied_settings)} settings written")
return {
"success": len(errors) == 0,
"template": name,
"applied_settings": applied_settings,
"errors": errors,
"status": eval_result,
}
def get_compliance_summary(db: Session) -> dict[str, Any]:
"""Return a high-level compliance dashboard summary across all templates."""
templates = db.query(ComplianceTemplate).order_by(ComplianceTemplate.name).all()
summary: list[dict[str, Any]] = []
total_checks = 0
total_passed = 0
for t in templates:
eval_result = evaluate_template_status(db, t.name)
total_checks += eval_result["total"]
total_passed += eval_result["passed"]
summary.append(
{
"name": t.name,
"display_name": t.display_name,
"enabled": t.enabled,
"status": eval_result["status"],
"total": eval_result["total"],
"passed": eval_result["passed"],
"failed": eval_result["failed"],
"applied_at": t.applied_at.isoformat() if t.applied_at else None,
"applied_by": t.applied_by,
}
)
overall = "compliant" if total_checks > 0 and total_passed == total_checks else "non_compliant"
if 0 < total_passed < total_checks:
overall = "partial"
return {
"overall_status": overall,
"total_checks": total_checks,
"total_passed": total_passed,
"total_failed": total_checks - total_passed,
"templates": summary,
}
app/utils/settings_service.py
+12
View File
@@ -1595,6 +1595,18 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
"compliance_enabled": {
"category": "Feature Flags",
"description": (
"Enable the compliance templates dashboard (GDPR, HIPAA, SOC 2). "
"When enabled, admins can view compliance status and apply "
"pre-built regulatory configurations. Default: True."
),
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Backup / Restore
"backup_enabled": {
"category": "Backup",
app/views/__init__.py
+2
View File
@@ -7,6 +7,7 @@ from fastapi import APIRouter
from app.views.admin_users import router as admin_users_router
from app.views.api_tokens import router as api_tokens_router
from app.views.backup import router as backup_router
from app.views.compliance import router as compliance_router
from app.views.db_wizard import router as db_wizard_router
from app.views.dropbox import router as dropbox_router
from app.views.filemanager import router as filemanager_router
@@ -61,3 +62,4 @@ router.include_router(integrations_router) # Unified integrations dashboard
router.include_router(notifications_router) # User notification dashboard
router.include_router(scheduled_jobs_router) # Admin scheduled batch jobs
router.include_router(help_router) # Built-in help / How-To docs
router.include_router(compliance_router) # Compliance templates dashboard
app/views/compliance.py
+48
View File
@@ -0,0 +1,48 @@
"""Admin view: compliance templates dashboard page."""
import logging
from fastapi import HTTPException, Request, status
from fastapi.responses import RedirectResponse
from app.views.base import APIRouter, require_login, settings, templates
logger = logging.getLogger(__name__)
router = APIRouter()
def _require_admin(request: Request):
"""Return the session user if they are an admin, else None."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
logger.warning("Non-admin user attempted to access /admin/compliance")
return None
return user
@router.get("/admin/compliance")
@require_login
async def compliance_page(request: Request):
"""Admin compliance templates dashboard page.
Displays GDPR, HIPAA, and SOC2 compliance templates with their current
status and one-click apply functionality.
"""
user = _require_admin(request)
if user is None:
return RedirectResponse(url="/", status_code=status.HTTP_302_FOUND)
try:
return templates.TemplateResponse(
"compliance.html",
{
"request": request,
"app_version": settings.version,
},
)
except Exception as e:
logger.error(f"Error loading compliance page: {e}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to load compliance page",
)