# 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)