feat: replace sequential IDs with opaque tokens in public URLs

Sequential IDs in /planner/export/<id> and /leads/<id>/unlock leaked
business volume (e.g. export_id=47 reveals ~47 PDFs sold). Replace with
22-char URL-safe tokens that carry no countable information.

- Migration 0017: adds `token TEXT` to business_plan_exports and
  lead_requests, backfills existing rows with secrets.token_urlsafe(16),
  creates unique indexes for fast lookups
- billing/routes.py: INSERT into business_plan_exports includes token
- leads/routes.py: INSERT into lead_requests includes token; enqueue
  payload includes lead_token; verify_quote() looks up by token
- planner/routes.py: /export/<token> route (was /export/<int:export_id>)
- suppliers/routes.py: /leads/<token>/unlock (was /leads/<int:lead_id>)
- worker.py: email links use token for both export and verify URLs
- Templates: url_for() calls use token= param
- test_phase0.py: _submit_guest_quote() returns (lead_id, auth_token,
  lead_token); verify URL tests use opaque lead token

Integer PKs unchanged; admin routes unchanged.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-22 17:03:13 +01:00
parent 35fe934fec
commit 815fb7d98a
11 changed files with 103 additions and 38 deletions

View File

@@ -4,6 +4,7 @@ Payment provider: paddle
"""
import json
import secrets
from datetime import datetime
from pathlib import Path
@@ -453,10 +454,10 @@ async def _handle_transaction_completed(data: dict, custom_data: dict) -> None:
transaction_id = data.get("id", "")
export_id = await execute(
"""INSERT INTO business_plan_exports
(user_id, scenario_id, paddle_transaction_id, language, status, created_at)
VALUES (?, ?, ?, ?, 'pending', ?)""",
(user_id, scenario_id, paddle_transaction_id, language, status, token, created_at)
VALUES (?, ?, ?, ?, 'pending', ?, ?)""",
(int(user_id), int(scenario_id) if scenario_id else 0,
transaction_id, language, now),
transaction_id, language, secrets.token_urlsafe(16), now),
)
# Enqueue PDF generation
from ..worker import enqueue

View File

@@ -333,8 +333,8 @@ async def quote_request():
previous_supplier_contact, services_needed, additional_info,
contact_name, contact_email, contact_phone, contact_company,
stakeholder_type,
heat_score, status, credit_cost, created_at)
VALUES (?, 'quote', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
heat_score, status, credit_cost, token, created_at)
VALUES (?, 'quote', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
user_id,
form.get("court_count", 0),
@@ -361,6 +361,7 @@ async def quote_request():
heat,
status,
credit_cost,
secrets.token_urlsafe(16),
datetime.utcnow().isoformat(),
),
)
@@ -422,11 +423,17 @@ async def quote_request():
token = secrets.token_urlsafe(32)
await create_auth_token(new_user_id, token, minutes=60)
lead_token_row = await fetch_one(
"SELECT token FROM lead_requests WHERE id = ?", (lead_id,)
)
lead_token = lead_token_row["token"]
from ..worker import enqueue
await enqueue("send_quote_verification", {
"email": contact_email,
"token": token,
"lead_id": lead_id,
"lead_token": lead_token,
"lang": g.get("lang", "en"),
"contact_name": form.get("contact_name", ""),
"facility_type": form.get("facility_type", ""),
@@ -466,10 +473,10 @@ async def quote_request():
async def verify_quote():
"""Verify email from quote submission — activates lead and logs user in."""
token_str = request.args.get("token")
lead_id = request.args.get("lead")
lead_token = request.args.get("lead")
_t = get_translations(g.get("lang", "en"))
if not token_str or not lead_id:
if not token_str or not lead_token:
await flash(_t["flash_verify_invalid"], "error")
return redirect(url_for("leads.quote_request"))
@@ -479,10 +486,10 @@ async def verify_quote():
await flash(_t["flash_verify_expired"], "error")
return redirect(url_for("leads.quote_request"))
# Validate lead exists and is pending
# Validate lead exists and is pending — look up by opaque token
lead = await fetch_one(
"SELECT * FROM lead_requests WHERE id = ? AND status = 'pending_verification'",
(lead_id,),
"SELECT * FROM lead_requests WHERE token = ? AND status = 'pending_verification'",
(lead_token,),
)
if not lead:
await flash(_t["flash_verify_invalid_lead"], "error")
@@ -497,7 +504,7 @@ async def verify_quote():
now = datetime.utcnow().isoformat()
await execute(
"UPDATE lead_requests SET status = 'new', verified_at = ?, credit_cost = ? WHERE id = ?",
(now, credit_cost, lead_id),
(now, credit_cost, lead["id"]),
)
# Set user name from contact_name if not already set

View File

@@ -0,0 +1,46 @@
"""Add opaque public tokens to business_plan_exports and lead_requests.
Sequential integer IDs in public URLs leak business volume (e.g. export_id=47
reveals ~47 PDFs sold). Opaque tokens hide this while keeping integer PKs for
internal use.
"""
import secrets
def up(conn):
# Add token columns (nullable to allow backfill before NOT NULL constraint)
conn.execute(
"ALTER TABLE business_plan_exports ADD COLUMN token TEXT"
)
conn.execute(
"ALTER TABLE lead_requests ADD COLUMN token TEXT"
)
# Backfill existing rows
export_ids = [row[0] for row in conn.execute(
"SELECT id FROM business_plan_exports WHERE token IS NULL"
).fetchall()]
for export_id in export_ids:
conn.execute(
"UPDATE business_plan_exports SET token = ? WHERE id = ?",
(secrets.token_urlsafe(16), export_id),
)
lead_ids = [row[0] for row in conn.execute(
"SELECT id FROM lead_requests WHERE token IS NULL"
).fetchall()]
for lead_id in lead_ids:
conn.execute(
"UPDATE lead_requests SET token = ? WHERE id = ?",
(secrets.token_urlsafe(16), lead_id),
)
# Unique indexes for fast token lookups
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS"
" idx_business_plan_exports_token ON business_plan_exports (token)"
)
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS"
" idx_lead_requests_token ON lead_requests (token)"
)

View File

@@ -534,13 +534,13 @@ async def export_success():
return await render_template("export_success.html", exports=exports)
@bp.route("/export/<int:export_id>")
@bp.route("/export/<token>")
@login_required
async def export_download(export_id: int):
async def export_download(token: str):
"""Download a generated PDF."""
export = await fetch_one(
"SELECT * FROM business_plan_exports WHERE id = ? AND user_id = ?",
(export_id, g.user["id"]),
"SELECT * FROM business_plan_exports WHERE token = ? AND user_id = ?",
(token, g.user["id"]),
)
if not export:
return jsonify({"error": "Export not found."}), 404
@@ -562,7 +562,7 @@ async def export_download(export_id: int):
pdf_bytes,
mimetype="application/pdf",
headers={
"Content-Disposition": f'attachment; filename="padel-business-plan-{export_id}.pdf"'
"Content-Disposition": f'attachment; filename="padel-business-plan-{export["id"]}.pdf"'
},
)

View File

@@ -85,7 +85,7 @@
</div>
<div>
{% if e.status == 'ready' %}
<a href="{{ url_for('planner.export_download', export_id=e.id) }}" class="btn-outline" style="font-size:0.75rem;padding:4px 12px">{{ t.export_download }}</a>
<a href="{{ url_for('planner.export_download', token=e.token) }}" class="btn-outline" style="font-size:0.75rem;padding:4px 12px">{{ t.export_download }}</a>
{% elif e.status == 'pending' or e.status == 'generating' %}
<span class="exp-status exp-status--pending">{{ t.export_generating }}</span>
{% else %}

View File

@@ -10,7 +10,7 @@
{% if exports %}
{% for e in exports %}
{% if e.status == 'ready' %}
<a href="{{ url_for('planner.export_download', export_id=e.id) }}" class="btn" style="display:inline-block;padding:12px 32px">{{ t.export_download }}</a>
<a href="{{ url_for('planner.export_download', token=e.token) }}" class="btn" style="display:inline-block;padding:12px 32px">{{ t.export_download }}</a>
{% else %}
<div style="background:#FEF3C7;border:1px solid #FDE68A;border-radius:10px;padding:1rem;margin-bottom:1rem;font-size:0.875rem;color:#92400E">
{{ t.export_success_status }}

View File

@@ -504,16 +504,22 @@ async def lead_feed():
)
@bp.route("/leads/<int:lead_id>/unlock", methods=["POST"])
@bp.route("/leads/<token>/unlock", methods=["POST"])
@_lead_tier_required
@csrf_protect
async def unlock_lead(lead_id: int):
async def unlock_lead(token: str):
"""Spend credits to unlock a lead. Returns full-details card via HTMX."""
from ..credits import InsufficientCredits
from ..credits import unlock_lead as do_unlock
supplier = g.supplier
# Resolve opaque token to integer ID for internal operations
lead_row = await fetch_one("SELECT id FROM lead_requests WHERE token = ?", (token,))
if not lead_row:
return jsonify({"error": "Lead not found."}), 404
lead_id = lead_row["id"]
try:
result = await do_unlock(supplier["id"], lead_id)
except InsufficientCredits as e:

View File

@@ -121,7 +121,7 @@
{% endif %}
<div class="lf-card__foot">
<div class="lf-card__cost"><strong>{{ lead.credit_cost or '?' }}</strong> {{ t.sd_leads_credits_to_unlock }}</div>
<form hx-post="{{ url_for('suppliers.unlock_lead', lead_id=lead.id) }}" hx-target="#lead-card-{{ lead.id }}" hx-swap="innerHTML">
<form hx-post="{{ url_for('suppliers.unlock_lead', token=lead.token) }}" hx-target="#lead-card-{{ lead.id }}" hx-swap="innerHTML">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="lf-unlock-btn">{{ t.sd_leads_unlock }}</button>
</form>

View File

@@ -25,7 +25,7 @@
<div class="lf-card__foot">
<span class="lf-card__cost"><strong>{{ lead.credit_cost or '?' }}</strong> {{ t.sd_card_credits }}</span>
<form hx-post="{{ url_for('suppliers.unlock_lead', lead_id=lead.id) }}"
<form hx-post="{{ url_for('suppliers.unlock_lead', token=lead.token) }}"
hx-target="#lead-card-{{ lead.id }}" hx-swap="innerHTML">
<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">
<button type="submit" class="lf-unlock-btn">{{ t.sd_card_unlock_btn }}</button>

View File

@@ -206,7 +206,7 @@ async def handle_send_quote_verification(payload: dict) -> None:
lang = payload.get("lang", "en")
link = (
f"{config.BASE_URL}/{lang}/leads/verify"
f"?token={payload['token']}&lead={payload['lead_id']}"
f"?token={payload['token']}&lead={payload['lead_token']}"
)
if config.DEBUG:
@@ -526,12 +526,16 @@ async def handle_generate_business_plan(payload: dict) -> None:
)
# Notify user via email
export_row = await fetch_one(
"SELECT token FROM business_plan_exports WHERE id = ?", (export_id,)
)
export_token = export_row["token"]
user = await fetch_one("SELECT email FROM users WHERE id = ?", (user_id,))
if user:
body = (
f'<h2 style="margin:0 0 16px;color:#0F172A;font-size:20px;">Your Business Plan is Ready</h2>'
f"<p>Your padel business plan PDF has been generated and is ready for download.</p>"
f'{_email_button(f"{config.BASE_URL}/planner/export/{export_id}", "Download PDF")}'
f'{_email_button(f"{config.BASE_URL}/planner/export/{export_token}", "Download PDF")}'
)
await send_email(
to=user["email"],