Security

Reference Best Practices

DVP CMS is a truth distillation system for AI-generated content. Plugins are evidence suppliers—they verify facts, pull live data, and make content more trustworthy over time. Learn more →

Security best practices for plugin development.

Core Principles

Tenant Isolation

Always Use HookContext

# Wrong - hardcoded or missing tenant
def dvp_before_content_create(self, ctx, content, metadata=None):
    data = self.db.query("SELECT * FROM content")  # All tenants!
    return content

# Right - scope to tenant from context
def dvp_before_content_create(self, ctx, content, metadata=None):
    data = self.db.query(
        "SELECT * FROM content WHERE tenant_id = $1",
        ctx.tenant_id,
    )
    return content

Secrets Management

Never Log Secrets

# WRONG - exposes secret in logs
async def initialize(self):
    api_key = await self.secrets.get_secret("api_key")
    self.logger.info(f"Using API key: {api_key}")  # DANGER!

# RIGHT - log without exposing
async def initialize(self):
    api_key = await self.secrets.get_secret("api_key")
    if api_key:
        self.logger.info("API key configured")
    else:
        self.logger.warning("API key not found")

Use Secrets Manager, Never Hardcode

# WRONG - hardcoded credentials
class MyPlugin(Plugin):
    API_KEY = "sk-abc123..."  # DANGER! Will be in git history

# RIGHT - use secrets manager
class MyPlugin(Plugin):
    async def get_api_key(self, tenant_id: UUID) -> str:
        secrets = self.get_secrets(tenant_id)
        return await secrets.require("api_key")

Input Validation

Validate All External Input

async def dvp_seo_metrics_received(
    self,
    ctx: HookContext,
    payload: SEOMetricsPayload,
) -> SEOMetricsPayload:
    # Validate source is expected
    allowed_sources = {"gsc", "ahrefs", "semrush"}
    if payload.source not in allowed_sources:
        self.logger.warning(f"Unknown source: {payload.source}")
        return payload  # Pass through but don't process

    # Validate domain format
    if not self._is_valid_domain(payload.domain):
        raise ValueError(f"Invalid domain format: {payload.domain}")

    return payload

SQL Injection Prevention

Always Use Parameterized Queries

# WRONG - SQL injection vulnerability
async def get_content(self, content_id: str) -> dict:
    query = f"SELECT * FROM content WHERE id = '{content_id}'"  # DANGER!
    return await self.db.fetchone(query)

# RIGHT - parameterized query
async def get_content(self, content_id: str) -> dict:
    query = "SELECT * FROM content WHERE id = $1"
    return await self.db.fetchone(query, content_id)

Error Handling Security

Don't Expose Internal Details

# WRONG - exposes internal structure
async def dvp_execute_health_check(self, check, ctx, execution_ctx):
    try:
        return await self._check_domain(execution_ctx.domain)
    except Exception as e:
        # Exposes stack trace, internal paths, etc.
        raise RuntimeError(f"Check failed: {e}\\n{traceback.format_exc()}")

# RIGHT - sanitized error message
async def dvp_execute_health_check(self, check, ctx, execution_ctx):
    try:
        return await self._check_domain(execution_ctx.domain)
    except Exception as e:
        # Log full details internally
        self.logger.exception("Health check failed")

        # Return sanitized message
        return HealthCheckResult(
            domain=execution_ctx.domain,
            overall_status=HealthStatus.UNHEALTHY,
            details=[HealthCheckDetail(message="Health check failed")],
        )

Security Checklist

Before Deployment

Code Review

See Also