"""Auto-extracted from models.py — do not edit manually."""
import json
import os
import secrets
import sqlite3
import uuid
from datetime import UTC, datetime, timezone
import bcrypt
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 DB_PATH, _table_exists, get_db
# ─── Spam detection constants ──────────────────────────────────────────────────
_SPAM_KEYWORDS = [
"nigeria",
"prince",
"lottery",
"inheritance",
"wire money",
"click here",
"free money",
"bitcoin wallet",
"crypto",
"viagra",
"casino",
"poker",
"weight loss",
"pharmacy",
"cialis",
"buy now",
"cheap",
"discount",
"work from home",
"make money",
"earn cash",
"limited time offer",
"act now",
"guaranteed approval",
"no credit check",
]
_SPAM_URL_PATTERNS = [
"bit.ly",
"tinyurl",
"t.co",
"goo.gl",
"shorte.st",
"adfoc.us",
"linkvertise",
"ouo.io",
"sh.st",
"rbl.us",
]
def resolve_geo(ip_address):
"""Look up city and country for an IP address using GeoLite2.
Returns (geo_city, geo_country) tuple. Both may be None if DB missing,
IP is private/local, or lookup fails.
The GeoLite2 database lives at /app/data/geoip/GeoLite2-City.mmdb
(mounted from data/geoip/ in the container).
"""
if not ip_address or ip_address in ("unknown", "127.0.0.1", "::1"):
return None, None
# Skip private IPs (LAN, localhost, etc.)
import ipaddress
try:
addr = ipaddress.ip_address(ip_address)
if addr.is_private or addr.is_loopback or addr.is_reserved:
return None, None
except (ValueError, TypeError):
return None, None
db_path = os.path.join(os.path.dirname(DB_PATH), "geoip", "GeoLite2-City.mmdb")
if not os.path.exists(db_path):
return None, None
try:
import maxminddb
reader = maxminddb.open_database(db_path, maxminddb.MODE_MMAP)
try:
result = reader.city(ip_address)
city = result.get("city", {}).get("names", {}).get("en", None)
country = result.get("country", {}).get("iso_code", None)
return city, country
finally:
reader.close()
except Exception:
return None, None
def parse_user_agent(ua_string):
"""Parse a User-Agent string into device_type, browser, and os.
Returns dict with keys: device_type, browser, os.
All values default to 'unknown' if parsing fails.
"""
if not ua_string:
return {"device_type": "unknown", "browser": "Unknown", "os": "Unknown"}
try:
from user_agents import parse as parse_ua
ua = parse_ua(ua_string)
# Device type classification
if ua.is_bot:
device_type = "bot"
elif ua.is_mobile:
device_type = "mobile"
elif ua.is_tablet:
device_type = "tablet"
else:
device_type = "desktop"
# Browser — use the browser family name
browser = (ua.browser.family or "Unknown").strip()
# OS — combine OS family with major version
os_family = (ua.os.family or "Unknown").strip()
os_version = ua.os.version
if os_version and os_version[0]:
os_str = f"{os_family} {os_version[0]}"
else:
os_str = os_family
return {
"device_type": device_type,
"browser": browser,
"os": os_str,
}
except Exception:
return {"device_type": "unknown", "browser": "Unknown", "os": "Unknown"}
# ─── Phase 10: Advanced Analytics ──────────────────────────────────────────────
def _decrypt_invoice_schedule(row_dict, schedule_id):
"""Decrypt PII columns in an invoice schedule row dict."""
for col, plain_key in (
("customer_name_encrypted", "customer_name"),
("customer_email_encrypted", "customer_email"),
):
if row_dict.get(col):
dec = try_decrypt_entity("invoice_schedule", schedule_id, row_dict[col])
if dec:
row_dict[plain_key] = dec
return row_dict
def _decrypt_webhook_dest(row_dict):
"""Decrypt url in a webhook_destination row dict."""
if row_dict.get("url_encrypted"):
dec = try_decrypt_entity("webhook_dest", row_dict["id"], row_dict["url_encrypted"])
if dec:
row_dict["url"] = dec
return row_dict
def _decrypt_webhook_log(row_dict):
"""Decrypt webhook_url in a webhook_log row dict."""
if row_dict.get("webhook_url_encrypted"):
dec = try_decrypt_entity("webhook_log", row_dict["id"], row_dict["webhook_url_encrypted"])
if dec:
row_dict["webhook_url"] = dec
return row_dict
def _decrypt_team_member(row_dict):
"""Decrypt invited_email in a team_member row dict."""
if row_dict.get("invited_email_encrypted"):
dec = try_decrypt_entity("team_member", row_dict["id"], row_dict["invited_email_encrypted"])
if dec:
row_dict["invited_email"] = dec
return row_dict
def _is_encrypted_value(value: str) -> bool:
"""Check if a value appears to be encrypted (base64-encoded with valid length)."""
if not value:
return False
import base64
try:
decoded = base64.b64decode(value, validate=True)
return len(decoded) >= 12 + 1 + 16
except Exception:
return False
def _encrypt_submission_field(site_id, value, password_hash=None):
"""Encrypt a submission field value using the site owner's password-derived key.
If password_hash is provided, uses user-scoped encryption (only owner can decrypt).
Falls back to site-scoped encryption for backwards compatibility.
"""
if not value:
return value
if password_hash:
enc = encrypt_user_submission(password_hash, value)
if enc:
return enc
# Fallback: site-scoped (legacy)
return encrypt_value(site_id, value)
def _decrypt_submission_fields(site_id, row_dict, password_hash=None):
"""Decrypt PII fields in a submission row dict. Returns modified dict.
If password_hash is provided, uses user-scoped decryption (owner view).
If password_hash is None, returns the row UNCHANGED (admin view — raw encrypted).
"""
if not row_dict or not password_hash:
return row_dict
pii_fields = ["customer_name", "customer_phone", "customer_email", "customer_equipment", "customer_message"]
for field in pii_fields:
row_dict[field] = try_decrypt_user_submission(password_hash, row_dict.get(field), site_id=site_id)
# Also decrypt the dynamic data JSON
raw_data = row_dict.get("data")
if raw_data:
decrypted = try_decrypt_user_submission(password_hash, raw_data, site_id=site_id)
if decrypted and decrypted != raw_data:
row_dict["data"] = decrypted
return row_dict
def _get_site_owner_password_hash(site_id):
"""Look up the site owner's password hash for user-scoped encryption."""
conn = None
try:
conn = get_db()
site = conn.execute("SELECT user_id FROM sites WHERE id = ?", (site_id,)).fetchone()
if not site:
return None
user_id = site["user_id"]
user = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user_id,)).fetchone()
return user["password_hash"] if user else None
finally:
if conn:
conn.close()
def _decrypt_user_row(row):
"""Decrypt PII columns in a user row dict. Mutates in place and returns the row."""
if not row:
return row
uid = row.get("id")
encrypted_cols = {
"email_encrypted": "email",
"name_encrypted": "name",
"display_name_encrypted": "display_name",
}
for col, plain_col in encrypted_cols.items():
if row.get(col):
decrypted = try_decrypt_user_value(uid, row[col])
if decrypted:
row[plain_col] = decrypted
return row
def _current_month():
"""Return current month as 'YYYY-MM' string."""
return datetime.now(UTC).strftime("%Y-%m")
def _check_spam_content(data):
"""Check submission data against spam heuristics.
Returns reason string if spam detected, None otherwise.
"""
text_fields = ["name", "phone", "email", "equipment", "message"]
combined = " ".join(str(data.get(f, "")) for f in text_fields).lower()
# Keyword check
for keyword in _SPAM_KEYWORDS:
if keyword in combined:
return "keyword"
# URL pattern check
for pattern in _SPAM_URL_PATTERNS:
if pattern in combined:
return "suspicious_url"
# Excessive links (3+ URLs in combined text)
import re
url_count = len(re.findall(r"https?://\S+", combined))
if url_count >= 3:
return "excessive_links"
# Very long message with very short name (common bot pattern)
msg_len = len(str(data.get("message", "")))
name_len = len(str(data.get("name", "")))
if msg_len > 500 and name_len < 3:
return "bot_pattern"
return None
# ─── Analytics functions ──────────────────────────────────────────────────────