"""Submission model functions — submissions, impressions, spam, accept_submission."""
import hashlib
import json
from datetime import UTC, datetime, timezone
from app.crypto import (
decrypt_entity,
decrypt_user_value,
encrypt_entity,
encrypt_user_submission,
encrypt_user_value,
encrypt_value,
hash_value,
is_encrypted,
key_is_configured,
try_decrypt,
try_decrypt_entity,
try_decrypt_user_submission,
try_decrypt_user_value,
)
from app.db import get_db
# Import helpers
from app.helpers import (
_check_spam_content,
_current_month,
_decrypt_submission_fields,
_encrypt_submission_field,
_get_site_owner_password_hash,
_is_encrypted_value,
parse_user_agent,
resolve_geo,
)
# Import tier config
from app.models_config import TIERS
def add_submission(
site_id,
data,
client_ip=None,
user_agent=None,
spam_flag=False,
geo_city=None,
geo_country=None,
device_type=None,
browser=None,
os_name=None,
):
"""Add a submission without tier enforcement.
Use `accept_submission` for the public API endpoint — it atomically
checks the limit AND inserts the submission in one transaction.
Submissions are encrypted with the site owner's password-derived key.
"""
# Get site owner's password hash for user-scoped encryption
password_hash = _get_site_owner_password_hash(site_id)
conn = None
try:
conn = get_db()
# Store dynamic fields in JSON (everything not captured by hardcoded columns)
hardcoded_keys = {"name", "phone", "email", "equipment", "message", "token", "_fr_hp", "recaptcha_token"}
dynamic_data = {k: v for k, v in data.items() if k not in hardcoded_keys}
data_json = json.dumps(dynamic_data) if dynamic_data else None
# Encrypt PII fields (user-scoped if we have password_hash)
cust_name = _encrypt_submission_field(site_id, data.get("name", ""), password_hash)
cust_phone = _encrypt_submission_field(site_id, data.get("phone", ""), password_hash)
cust_email = _encrypt_submission_field(site_id, data.get("email", ""), password_hash)
cust_equip = _encrypt_submission_field(site_id, data.get("equipment", ""), password_hash)
cust_msg = _encrypt_submission_field(site_id, data.get("message", ""), password_hash)
enc_data_json = (
encrypt_user_submission(password_hash, data_json)
if password_hash and data_json
else (encrypt_value(site_id, data_json) if data_json else None)
)
cursor = conn.execute(
"""INSERT INTO submissions
(site_id, customer_name, customer_phone, customer_email,
customer_equipment, customer_message, data, client_ip, user_agent, spam_flag,
geo_city, geo_country, device_type, browser, os)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
site_id,
cust_name,
cust_phone,
cust_email,
cust_equip,
cust_msg,
enc_data_json,
client_ip,
user_agent,
1 if spam_flag else 0,
geo_city,
geo_country,
device_type,
browser,
os_name,
),
)
conn.commit()
submission_id = cursor.lastrowid
submission = conn.execute("SELECT * FROM submissions WHERE id = ?", (submission_id,)).fetchone()
finally:
if conn:
conn.close()
if submission:
return _decrypt_submission_fields(site_id, dict(submission), password_hash)
return None
def add_form_impression(site_id, client_ip=None, user_agent=None):
"""Record a form impression (view) with geo/device enrichment.
Args:
site_id: The site/form ID.
client_ip: Client IP address (hashed for privacy).
user_agent: Client user-agent string.
Returns:
dict with impression record or None on failure.
"""
conn = None
try:
conn = get_db()
# Hash IP for privacy
ip_hash = None
if client_ip:
ip_hash = hashlib.sha256(client_ip.encode()).hexdigest()
# Enrich: GeoIP + User-Agent
geo_city, geo_country = resolve_geo(client_ip)
ua_info = parse_user_agent(user_agent)
cursor = conn.execute(
"""INSERT INTO form_impressions
(site_id, ip_hash, user_agent, geo_city, geo_country, device_type, browser, os)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
site_id,
ip_hash,
user_agent,
geo_city,
geo_country,
ua_info["device_type"],
ua_info["browser"],
ua_info["os"],
),
)
conn.commit()
impression_id = cursor.lastrowid
impression = conn.execute("SELECT * FROM form_impressions WHERE id = ?", (impression_id,)).fetchone()
return dict(impression) if impression else None
except Exception as e:
print(f"[model] add_form_impression error: {e}")
return None
finally:
if conn:
conn.close()
def list_submissions(limit=50, site_filter=None, password_hash=None):
"""List submissions, optionally decrypted with user-scoped key.
If password_hash is provided, uses user-scoped decryption.
Otherwise returns raw encrypted data (admin view).
"""
conn = None
try:
conn = get_db()
if site_filter:
rows = conn.execute(
"""SELECT s.*, si.name as site_name
FROM submissions s JOIN sites si ON s.site_id = si.id
WHERE si.id = ? ORDER BY s.submitted_at DESC LIMIT ?""",
(site_filter, limit),
).fetchall()
else:
rows = conn.execute(
"""SELECT s.*, si.name as site_name
FROM submissions s JOIN sites si ON s.site_id = si.id
ORDER BY s.submitted_at DESC LIMIT ?""",
(limit,),
).fetchall()
finally:
if conn:
conn.close()
results = []
for r in rows:
d = dict(r)
results.append(_decrypt_submission_fields(d["site_id"], d, password_hash))
return results
def list_submissions_raw(limit=50, site_filter=None):
"""List submissions without any decryption (admin view)."""
return list_submissions(limit=limit, site_filter=site_filter, password_hash=None)
def get_submission(sub_id, password_hash=None):
"""Get a single submission, optionally decrypted with user-scoped key.
If password_hash is provided, uses user-scoped decryption.
Otherwise returns raw encrypted data (admin view).
"""
conn = None
try:
conn = get_db()
row = conn.execute(
"""SELECT s.*, si.name as site_name, si.owner_email
FROM submissions s JOIN sites si ON s.site_id = si.id
WHERE s.id = ?""",
(sub_id,),
).fetchone()
finally:
if conn:
conn.close()
if row:
d = dict(row)
return _decrypt_submission_fields(d["site_id"], d, password_hash)
return None
def accept_submission(site_id, user_id, user_tier, data, site=None, client_ip=None, user_agent=None):
"""Atomically check submission limit, insert submission, and increment counter.
All three operations happen inside a single BEGIN IMMEDIATE transaction.
If any step fails, nothing is committed — the counter stays accurate.
Args:
site_id: The site to submit to.
user_id: Owner of the site (for tier enforcement).
user_tier: User's current tier string.
data: Submission data dict.
site: Site dict (optional, for honeypot/spam check config).
client_ip: Client IP address.
user_agent: Client user-agent string.
Returns:
(success: bool, submission: dict or None, current: int, max_count: int, reason: str or None)
"""
tier_config = TIERS.get(user_tier, TIERS["free"])
max_subs = tier_config["max_submissions"]
month = _current_month()
conn = None
try:
conn = get_db()
conn.execute("BEGIN IMMEDIATE")
# 1. Check limit under write lock
row = conn.execute(
"SELECT submission_count FROM monthly_usage WHERE user_id = ? AND month = ?",
(user_id, month),
).fetchone()
current = row["submission_count"] if row else 0
if current >= max_subs:
conn.execute("ROLLBACK")
return False, None, current, max_subs, "limit"
# 2. Insert the submission
hardcoded_keys = {"name", "phone", "email", "equipment", "message", "token", "_fr_hp", "recaptcha_token"}
dynamic_data = {k: v for k, v in data.items() if k not in hardcoded_keys}
data_json = json.dumps(dynamic_data) if dynamic_data else None
# Get owner's password hash for user-scoped encryption
owner_row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,)).fetchone()
owner_pw_hash = owner_row["password_hash"] if owner_row else None
# Encrypt PII fields (user-scoped if we have password_hash)
cust_name = _encrypt_submission_field(site_id, data.get("name", ""), owner_pw_hash)
cust_phone = _encrypt_submission_field(site_id, data.get("phone", ""), owner_pw_hash)
cust_email = _encrypt_submission_field(site_id, data.get("email", ""), owner_pw_hash)
cust_equip = _encrypt_submission_field(site_id, data.get("equipment", ""), owner_pw_hash)
cust_msg = _encrypt_submission_field(site_id, data.get("message", ""), owner_pw_hash)
enc_data_json = (
encrypt_user_submission(owner_pw_hash, data_json)
if owner_pw_hash and data_json
else (encrypt_value(site_id, data_json) if data_json else None)
)
# Determine spam flag
is_spam = 0
spam_reason = None
if site.get("spam_filter_enabled"):
spam_reason = _check_spam_content(data)
if spam_reason:
is_spam = 1
# Enrich: GeoIP + User-Agent parsing
geo_city, geo_country = resolve_geo(client_ip)
ua_info = parse_user_agent(user_agent)
cursor = conn.execute(
"""INSERT INTO submissions
(site_id, customer_name, customer_phone, customer_email,
customer_equipment, customer_message, data, client_ip, user_agent, spam_flag,
geo_city, geo_country, device_type, browser, os)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
site_id,
cust_name,
cust_phone,
cust_email,
cust_equip,
cust_msg,
enc_data_json,
client_ip,
user_agent,
is_spam,
geo_city,
geo_country,
ua_info["device_type"],
ua_info["browser"],
ua_info["os"],
),
)
submission_id = cursor.lastrowid
# 3. Increment the counter
conn.execute(
"""INSERT INTO monthly_usage (user_id, month, submission_count)
VALUES (?, ?, 1)
ON CONFLICT(user_id, month) DO UPDATE SET submission_count = submission_count + 1""",
(user_id, month),
)
# 4. Update site analytics (per-day)
day = datetime.now(UTC).strftime("%Y-%m-%d")
analytics_update = "submission_count = submission_count + 1"
if is_spam:
analytics_update += ", spam_count = spam_count + 1"
conn.execute(
f"""INSERT INTO site_analytics (site_id, day, submission_count, spam_count)
VALUES (?, ?, 1, {1 if is_spam else 0})
ON CONFLICT(site_id, day) DO UPDATE SET {analytics_update}""",
(site_id, day),
)
conn.commit()
# 5. Return the inserted submission
submission = conn.execute("SELECT * FROM submissions WHERE id = ?", (submission_id,)).fetchone()
decrypted = _decrypt_submission_fields(site_id, dict(submission), owner_pw_hash) if submission else None
return True, decrypted, current + 1, max_subs, spam_reason
except Exception:
conn.execute("ROLLBACK")
raise
finally:
if conn:
conn.close()