#!/usr/bin/env python3
"""Webhook / submission route tests (Phase 2 — 2c)
Tests for form submission endpoint:
- POST /api/submit — valid token, invalid token, missing token
- POST /api/submit — CSRF exemption (no session required)
- POST /api/submit — rate limiting enforcement
- POST /api/submit — payload with malicious HTML (should sanitize)
- POST /api/submit — oversized payload rejected
- Content type validation (JSON and form-data accepted)
- CORS preflight OPTIONS
- Honeypot bot detection
Uses shared flask_app from app.app (same pattern as test_auth/test_hardening).
"""
import json
import os
import random
import sqlite3
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Use shared test DB (singleton pattern — only initializes once)
import tests.test_shared_db # noqa: F401
# Set required env vars BEFORE importing app modules
os.environ.setdefault("AGENTFORMS_SECRET_KEY", "webhook-test-secret")
os.environ.setdefault("ENCRYPTION_KEY", "5c321ae59453d7bae2e05a9e19e97347411a033d9858a0456d5cc63acfcb0369")
os.environ.setdefault("SMTP_HOST", "smtp.test.invalid")
os.environ.setdefault("SMTP_PORT", "587")
# Import shared Flask app — same instance used by all test files
from app.app import app as flask_app
import app.models
# Ensure test mode
flask_app.config["TESTING"] = True
flask_app.config["SESSION_COOKIE_SECURE"] = False
def _create_site(client, name="Test Site"):
"""Create a test user + site and return the site token."""
email = f"site-webhook{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Site Owner",
}, follow_redirects=True)
# Mark email as verified (bypass email verification for tests)
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
conn.execute("UPDATE users SET email_verified = 1 WHERE email_hash = ?",
(hash_value(email.lower()),))
conn.commit()
conn.close()
resp = client.post("/sites/new", data={
"name": name,
"owner_email": email,
}, follow_redirects=True)
assert resp.status_code == 200
# Get the new site token
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
row = conn.execute(
"SELECT token FROM sites WHERE owner_email = ? ORDER BY created_at DESC LIMIT 1",
(email.lower(),)
).fetchone()
conn.close()
return row[0]
class TestSubmitEndpoint:
"""Test POST /api/submit — basic functionality."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_submit_missing_token(self, client):
resp = client.post("/api/submit", json={"name": "test"})
assert resp.status_code == 400
data = resp.get_json()
assert "token" in data.get("error", "").lower() or "Missing" in data.get("error", "")
def test_submit_invalid_token(self, client):
resp = client.post("/api/submit?token=invalidtoken123", json={"name": "test"})
assert resp.status_code == 404
def test_submit_valid_token_json(self, client):
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", json={
"name": "Test Submission",
"email": "test@example.com",
})
assert resp.status_code in (200, 201)
data = resp.get_json()
assert data.get("success") is True
def test_submit_valid_token_form_data(self, client):
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", data={
"name": "Form Submission",
"email": "form@example.com",
})
assert resp.status_code in (200, 201)
def test_submit_token_in_body(self, client):
"""Test token can be passed in JSON body instead of query param."""
email = f"bodytok-webhook{random.randint(10000, 99999)}@example.com"
client.post("/auth/register", data={
"email": email,
"password": "SecurePass123",
"name": "Body Token Owner",
}, follow_redirects=True)
from app.models import hash_value
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
conn.execute("UPDATE users SET email_verified = 1 WHERE email_hash = ?",
(hash_value(email.lower()),))
conn.commit()
conn.close()
resp = client.post("/sites/new", data={
"name": "Body Token Site",
"owner_email": email,
}, follow_redirects=True)
assert resp.status_code == 200
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
row = conn.execute(
"SELECT token FROM sites WHERE owner_email = ? ORDER BY created_at DESC LIMIT 1",
(email.lower(),)
).fetchone()
conn.close()
token = row[0]
resp = client.post("/api/submit", json={
"token": token,
"name": "Body Token Test",
})
assert resp.status_code in (200, 201)
class TestSubmitSecurity:
"""Test /api/submit — security features."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_submit_csrf_exempt(self, client):
"""Test that /api/submit is CSRF exempt (no session required)."""
token = _create_site(client)
fresh_client = flask_app.test_client()
resp = fresh_client.post(f"/api/submit?token={token}", json={
"name": "No Session",
})
assert resp.status_code != 403 # Should not be CSRF blocked
def test_submit_malicious_html_sandboxed(self, client):
"""Test that XSS payloads in submissions don't break the endpoint."""
token = _create_site(client)
xss_payload = {
"name": "<script>alert('xss')</script>",
"message": "<img src=x onerror=alert(1)>",
}
resp = client.post(f"/api/submit?token={token}", json=xss_payload)
assert resp.status_code in (200, 201)
def test_submit_oversized_payload_rejected(self, client):
"""Test that payloads over 100KB are rejected."""
token = _create_site(client)
large_payload = {"data": "A" * (200 * 1024)} # 200KB
resp = client.post(f"/api/submit?token={token}", json=large_payload)
assert resp.status_code == 413
def test_submit_cors_preflight(self, client):
"""Test that OPTIONS preflight returns 204."""
token = _create_site(client)
resp = client.options(f"/api/submit?token={token}")
assert resp.status_code == 204
def test_security_headers_present(self, client):
"""Verify security headers on API responses."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", json={
"name": "Security Header Test",
})
assert resp.headers.get("X-Frame-Options") == "DENY"
assert resp.headers.get("X-Content-Type-Options") == "nosniff"
assert "default-src 'self'" in resp.headers.get("Content-Security-Policy", "")
def test_csp_blocks_frames(self, client):
"""Verify Content-Security-Policy includes frame-ancestors 'none'."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", json={
"name": "CSP Test",
})
csp = resp.headers.get("Content-Security-Policy", "")
assert "frame-ancestors 'none'" in csp
class TestSubmitRateLimiting:
"""Test /api/submit — rate limiting."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_submit_rate_limit_per_ip(self, client):
"""Test that excessive submissions from same IP are rate limited."""
token = _create_site(client)
statuses = []
for i in range(40): # 40 > both IP limit (30) and site limit (20)
resp = client.post(f"/api/submit?token={token}", json={
"name": f"Rate Test {i}",
})
statuses.append(resp.status_code)
# At least some requests should be rate limited
assert 429 in statuses, f"Expected rate limiting, got statuses: {set(statuses)}"
class TestHoneypot:
"""Test honeypot bot detection."""
@pytest.fixture
def client(self):
return flask_app.test_client()
def test_honeypot_catches_bot(self, client):
"""Test that filled honeypot fields are silently accepted."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", json={
"name": "Bot Name",
"_fr_email": "filled@bot.com", # Honeypot field
})
assert resp.status_code == 200
data = resp.get_json()
assert data.get("success") is True
def test_honeypot_empty_allowed(self, client):
"""Test that empty honeypot fields allow submission."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}", json={
"name": "Human Name",
"_fr_email": "", # Empty honeypot — human
})
assert resp.status_code in (200, 201)
def test_submit_unsupported_content_type(self, client):
"""Test that unsupported content types are rejected with 415."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}",
data=b"<xml>not allowed</xml>",
content_type="application/xml"
)
assert resp.status_code == 415
data = resp.get_json()
assert "content type" in data.get("error", "").lower()
def test_submit_text_plain_rejected(self, client):
"""Test that text/plain content type is rejected."""
token = _create_site(client)
resp = client.post(f"/api/submit?token={token}",
data=b"raw text submission",
content_type="text/plain"
)
assert resp.status_code == 415
def test_submit_html_sanitized_in_db(self, client):
"""Test that HTML tags are stripped from submission data."""
token = _create_site(client)
xss_data = {
"name": "<script>alert('xss')</script>",
"message": "<img src=x onerror=alert(1)>test<b>bold</b>",
}
resp = client.post(f"/api/submit?token={token}", json=xss_data)
assert resp.status_code in (200, 201)
# Verify stored data is sanitized
with flask_app.app_context():
conn = sqlite3.connect(app.models.DB_PATH)
row = conn.execute(
"SELECT customer_name, data FROM submissions ORDER BY id DESC LIMIT 1"
).fetchone()
conn.close()
name_value = row[0] if row else ""
assert "<script>" not in name_value
assert "alert" not in name_value.lower() or "<script>" not in name_value
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])