# AgentForms Production Hardening Plan
**Date:** June 15, 2026
**Status:** ✅ **COMPLETED** — all 12 phases executed
**Audit basis:** Full security + infrastructure deep dive
---
## Summary
All 12 hardening phases completed. 327 tests passing, Docker Compose production-ready, no CVEs, solid security fundamentals (CSP nonces, bcrypt, SSRF, Bleach sanitization, encrypted PII). All 4 HIGH items fixed. Remaining items are MEDIUM/LOW — address in order of impact.
---
## Phase 1 — Database Connection Leaks (HIGH)
**Problem:** 47 unprotected `get_db()` calls in `app/models.py` lack try/finally guards. Exceptions between `get_db()` and `conn.close()` leak file descriptors. Under sustained load → `OperationalError: unable to open database file`.
**Impact:** Service crashes under load. Slow-moving but guaranteed.
**Scope:** ~47 locations across `app/models.py`. Previous fix attempt resolved 58 of 72; 14 remained. Now 47 still unprotected.
**Approach:**
1. Write a Python script that scans `app/models.py` for all `get_db()` calls lacking try/finally within 30 lines
2. For each location, wrap in:
```python
conn = None
try:
conn = get_db()
# ... existing logic ...
finally:
if conn:
conn.close()
```
3. Run tests after each batch of 10 functions to catch regressions
4. Verify with grep: `get_db()` count ≈ `try:` + `finally:` count
**Risk:** Low — try/finally is non-destructive. Tests should catch any issues.
**Estimate:** 1 session. Mechanical but requires care with nested try blocks and early returns.
---
## Phase 2 — Global Payload Size Limit (HIGH)
**Problem:** No `MAX_CONTENT_LENGTH` configured globally. Only `/api/submit` has manual 100KB checking. All other endpoints (API v2, admin, auth) accept unlimited payloads.
**Impact:** DoS via oversized requests to any endpoint.
**Fix:**
```python
# In app/app.py create_app():
app.config["MAX_CONTENT_LENGTH"] = 1 * 1024 * 1024 # 1MB
```
**Add error handler:**
```python
@app.errorhandler(413)
def payload_too_large(e):
if request.is_json:
return jsonify({"error": "Payload too large (max 1MB)"}), 413
return "Payload too large", 413
```
**Estimate:** 15 minutes. Trivial change, verify with test that oversized request returns 413.
---
## Phase 3 — Encryption Key Rotation (HIGH)
**Problem:** Single `ENCRYPTION_KEY` env var. No key versioning. If compromised, all historical PII exposed with no recovery path.
**Impact:** Catastrophic if key is leaked (core dump, env var exposure, log leak).
**Approach:**
1. Add `ENCRYPTION_KEY_V1` and `ENCRYPTION_KEY_V2` env vars
2. In `app/crypto.py`:
- `encrypt_value()` uses current key (V2)
- `decrypt_value()` tries V2 first, falls back to V1
- Store key version prefix in ciphertext (Fernet tokens encode this)
3. Migration script: re-encrypt all PII under V2 key
4. DB schema: add `encryption_version` column to track which key was used per row
5. `app/models.py` migration function iterates all encrypted columns, re-encrypts under V2
**Implementation:**
```python
# app/crypto.py additions:
import os
# Key resolution order: V2 > V1 > legacy
CURRENT_KEY = os.environ.get("ENCRYPTION_KEY_V2") or os.environ.get("ENCRYPTION_KEY")
LEGACY_KEY = os.environ.get("ENCRYPTION_KEY_V1")
def encrypt_value(value):
"""Encrypt with current key."""
from cryptography.fernet import Fernet
f = Fernet(base64.urlsafe_b64encode(hashlib.sha256(CURRENT_KEY.encode()).digest()))
return f.encrypt(value.encode()).decode()
def decrypt_value(token):
"""Try current key first, fall back to legacy."""
from cryptography.fernet import Fernet, InvalidToken
try:
f = Fernet(base64.urlsafe_b64encode(hashlib.sha256(CURRENT_KEY.encode()).digest()))
return f.decrypt(token.encode()).decode()
except InvalidToken:
if LEGACY_KEY:
f = Fernet(base64.urlsafe_b64encode(hashlib.sha256(LEGACY_KEY.encode()).digest()))
return f.decrypt(token.encode()).decode()
raise
```
**Estimate:** 1-2 sessions. Requires careful testing with existing encrypted data to verify fallback works.
---
## Phase 4 — DNS Rebinding on SSRF (MEDIUM)
**Problem:** Webhook URL validated at registration time, delivered async in worker. Attacker can rebind DNS between validation and delivery.
**Impact:** Internal network scanning, cloud metadata access if DNS rebinding succeeds.
**Fix:** Re-resolve and validate immediately before `requests.post()` in the webhook delivery code.
**Location:** `app/services/webhook.py` — delivery function
**Change:**
```python
def _deliver_webhook(url, payload, headers=None):
# Re-validate immediately before sending (prevents DNS rebinding)
is_safe, error = validate_webhook_url(url)
if not is_safe:
logger.warning("SSRF blocked at delivery: %s (%s)", url, error)
return False, f"SSRF blocked: {error}"
# ... proceed with requests.post() ...
```
**Estimate:** 30 minutes. Single function change. Add test case for DNS rebinding scenario (mock DNS to change between validation and delivery).
---
## Phase 5 — IPv6 SSRF Validation (MEDIUM)
**Problem:** DNS validation only checks `socket.AF_INET` (IPv4). IPv6 addresses bypass SSRF checks.
**Impact:** IPv6 private/unique-local addresses can be targeted.
**Fix:** Also resolve with `socket.AF_INET6` in `validate_webhook_url()`:
```python
# In app/services/webhook.py validate_webhook_url():
for af in (socket.AF_INET, socket.AF_INET6):
addr_info = socket.getaddrinfo(hostname, None, af)
for info in addr_info:
ip = ipaddress.ip_address(info[4][0])
for net in PRIVATE_RANGES:
if ip in net:
return False, f"private IP blocked: {ip}"
```
**Also add IPv6 private ranges:**
```python
PRIVATE_RANGES.extend([
ipaddress.ip_network("fc00::/7"), # Unique local
ipaddress.ip_network("fe80::/10"), # Link-local
ipaddress.ip_network("::1/128"), # Loopback
])
```
**Estimate:** 20 minutes. Add IPv6 test cases to test_hardening.py.
---
## Phase 6 — Webhook Alert Deduplication (MEDIUM)
**Problem:** Permanently failed webhooks re-alerted every retry cycle. Same 2 submissions logged 6+ times per minute.
**Impact:** Log spam, alert fatigue, operator ignores real issues.
**Fix:** Track alerted submission IDs. Once alerted, mark as `alerted` so it's not re-alerted.
**Approach:**
```python
# In webhook retry thread:
_ALERTED_SUBMISSIONS = set()
_ALERTED_MAX_AGE = 86400 # 24 hours
def process_webhook_retries(max_retries=5):
# ... normal retry logic ...
permanently_failed = get_permanently_failed_webhooks(max_retries=max_retries)
now = time.time()
for entry in permanently_failed:
sub_key = f"{entry['site_id']}:{entry['submission_id']}"
if sub_key in _ALERTED_SUBMISSIONS:
continue # already alerted
alert.warning("Webhook delivery permanently failed", ...)
_ALERTED_SUBMISSIONS.add(sub_key)
# Cleanup old entries
stale = [k for k in _ALERTED_SUBMISSIONS if now - _ALERTED_TIMES[k] > _ALERTED_MAX_AGE]
for k in stale:
del _ALERTED_SUBMISSIONS[k]
```
**Estimate:** 15 minutes. Add test for alert deduplication.
---
## Phase 7 — datetime.utcnow() Deprecation (MEDIUM)
**Problem:** 80+ deprecation warnings. `datetime.utcnow()` deprecated in Python 3.12, removed in 3.13+.
**Impact:** Warning noise now, broken code on Python 3.13 upgrade.
**Fix:** Replace `datetime.utcnow()` → `datetime.now(datetime.UTC)` throughout:
```bash
# In app/models.py:
sed 's/datetime\.utcnow()/datetime.now(datetime.UTC)/g'
# Also add: from datetime import datetime, UTC (Python 3.11+)
# For Python 3.11 compatibility: datetime.now(datetime.timezone.utc)
```
**Location:** `app/models.py` lines 7134, 7624 + RQ utils (external — upgrade RQ instead).
**Estimate:** 20 minutes for app code. RQ warnings require upgrading the RQ package — check `requirements.txt` for latest version.
---
## Phase 8 — Admin Session Timeout (LOW)
**Problem:** Admin session valid for 24 hours. Long for privileged role.
**Fix:** Reduce from 86400s to 3600s (1 hour) in `app/routes/admin.py:82`.
**Estimate:** 5 minutes. Single line change.
---
## Phase 9 — Admin Username Default (LOW)
**Problem:** `ADMIN_USER` defaults to `"admin"` — most commonly guessed username.
**Fix:** Remove default in `app/routes/admin.py:12`:
```python
ADMIN_USER = os.environ.get("ADMIN_USER")
if not ADMIN_USER:
raise RuntimeError("ADMIN_USER environment variable not set")
```
**Estimate:** 5 minutes.
---
## Phase 10 — Stripe Webhook Secret Validation (LOW)
**Problem:** `STRIPE_WEBHOOK_SECRET` defaults to `""` — silent pass/fail on webhook verification.
**Fix:** Validate at startup if Stripe integration is enabled:
```python
if stripe_enabled and not os.environ.get("STRIPE_WEBHOOK_SECRET"):
logger.warning("STRIPE_WEBHOOK_SECRET not set — Stripe webhooks will not be verified")
```
**Estimate:** 5 minutes.
---
## Phase 11 — API Key Expiration (LOW)
**Problem:** API keys valid forever until revoked.
**Approach:**
1. Add `expires_at` column to `api_keys` table (nullable)
2. Alembic migration: `ALTER TABLE api_keys ADD COLUMN expires_at TIMESTAMP NULL`
3. In `create_api_key()`: accept optional `expires_in_days` parameter
4. In `validate_api_key()`: check `expires_at` before bcrypt verification
5. API: add `expires_in_days` field to key creation endpoint
6. Admin UI: show expiration date, allow extending
**Estimate:** 1 session. Non-breaking — column is nullable, existing keys remain valid forever.
---
## Execution Order
```
[✓] Phase 1 (DB connection leaks) — COMPLETED (1 leak fixed in app.py health_ready, 0 remaining)
[✓] Phase 2 (MAX_CONTENT_LENGTH) — COMPLETED
[✓] Phase 3 (encryption key rotation) — COMPLETED
[✓] Phase 4 (DNS rebinding SSRF) — COMPLETED
[✓] Phase 5 (IPv6 SSRF validation) — COMPLETED
[✓] Phase 6 (webhook dedup) — COMPLETED
[✓] Phase 7 (datetime.utcnow deprec) — COMPLETED
[✓] Phase 8 (admin timeout) — COMPLETED
[✓] Phase 9 (admin username) — COMPLETED
[✓] Phase 10 (stripe secret) — COMPLETED
[✓] Phase 11 (API key expiration) — COMPLETED
[✓] Phase 12 (CSP + HSTS headers) — COMPLETED (already in place)
```
**Total: 12/12 completed. All phases done.**
## Verification Protocol
After each phase:
1. Run full test suite: `docker compose exec relay python -m pytest tests/ -v`
2. Check for new warnings/errors
3. Verify no regression in existing functionality
Before declaring production-ready:
1. All 327+ tests passing
2. `pip-audit -r requirements.txt` clean
3. `grep 'get_db()' app/models.py | wc -l` ≈ `grep 'finally:' app/models.py | wc -l`
4. `grep 'MAX_CONTENT_LENGTH' app/app.py` returns a result
5. SSL cert expiry check (if public-facing)