Security best practices for plugin development.
Core Principles
- Tenant Isolation: Never access another tenant's data
- Least Privilege: Request only needed permissions
- Defense in Depth: Validate at every boundary
- Secrets Hygiene: Never log or expose credentials
- Input Validation: Trust nothing from external sources
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
- No hardcoded secrets in code
- All database queries parameterized
- External input validated and sanitized
- Tenant isolation verified in all operations
- Error messages don't expose internals
- Security events logged
- Rate limiting implemented for external calls
Code Review
ctx.tenant_idused for all data accesssecrets.require()orsecrets.get_secret()for credentials- No f-string SQL queries
- URLs validated before fetching
- Exceptions don't contain sensitive data
- No logging of secrets or PII