A user completes an analysis without signing in. The next screen offers two actions: “email this result” and “save it to my account.” These actions look similar in a product flow, but they require different evidence.
An email field tells the server where someone wants a message delivered. A verified session identifies an account. Neither, by itself, proves that the account controls a particular anonymous result.
This matters for any product with private reports, document conversions, exports or assessments. A mistyped destination must not put one person's result into another person's account history.
The anonymous-to-account flow in AI Doctor Scan motivates the example below. This article uses newly written, standalone demonstration code. It does not publish the private application's implementation, describe an observed production incident, or make a claim about medical accuracy.
Start with three separate facts
| Fact | Evidence | What it does not establish |
|---|---|---|
| Delivery destination | An address entered into a form | The sender owns that address or the result |
| Account identity | A verified server-side session | The account controls every anonymous result |
| Control of one anonymous result | A high-entropy capability bound to that result | The holder is a clinician or may grant access to unrelated results |
The ownership transition should consume both a verified session and the anonymous result's capability. The mail path should require control of the result too, while keeping the destination separate from ownership.
Make the ownership transition conditional
A “read, check, then write” sequence can let two concurrent requests pass the same check. Put the condition in the update itself:
UPDATE results
SET owner_id = :verified_user,
claim_hash = NULL,
claim_expires_at = NULL
WHERE id = :result_id
AND owner_id IS NULL
AND claim_hash = :supplied_token_hash
AND claim_expires_at > :now;
One affected row means this request acquired ownership. Zero means it did not. Database errors are failures, not permission to continue. The authenticated user must come from server-verified authentication; never take that identity directly from a request body or query string.
This demonstration generates a token using secrets.token_urlsafe(32) and stores only its SHA-256 hash. Python's secrets documentation describes this as 32 random bytes encoded for URLs. The high entropy is essential: fast hashing is not a substitute for a password-hashing function when the input is a human-chosen password.
The expiry, single-use behavior and protected storage follow the same broad token-handling principles described in OWASP's Forgot Password Cheat Sheet. Result claiming has its own threat model; a password-reset checklist is not a complete design for it.
A runnable example with no real data or email
Save the following as claim_demo.py. It needs Python 3.10+ and only the standard library. It creates no network connections. The delivery function records a request locally; it does not send mail.
The delivery function holds a SQLite write transaction while checking authorization and recording the request. Its only insert is into delivery_requests; it never changes results.owner_id.
"""Standalone ownership demo. No web server, real email, health data or app imports.
authenticated_user_id must come from a verified server-side session. Never pass
an identity directly from an HTTP body, header or query parameter into this API.
"""
import hashlib
import hmac
import secrets
import sqlite3
import time
import uuid
class AccessDenied(Exception):
pass
def connect(path):
db = sqlite3.connect(path, timeout=10, isolation_level=None)
db.execute("PRAGMA foreign_keys = ON")
return db
def initialize(db):
db.executescript("""
CREATE TABLE IF NOT EXISTS results (
id TEXT PRIMARY KEY,
owner_id TEXT,
claim_hash TEXT,
claim_expires_at INTEGER
);
CREATE TABLE IF NOT EXISTS delivery_requests (
id TEXT PRIMARY KEY,
result_id TEXT NOT NULL REFERENCES results(id),
destination TEXT NOT NULL
);
""")
def token_hash(raw_token):
return hashlib.sha256(raw_token.encode("utf-8")).hexdigest()
def create_anonymous_result(db, *, now=None, ttl_seconds=3600):
if ttl_seconds <= 0:
raise ValueError("ttl_seconds must be positive")
now = int(time.time()) if now is None else now
result_id, token = str(uuid.uuid4()), secrets.token_urlsafe(32)
db.execute(
"INSERT INTO results VALUES (?, NULL, ?, ?)",
(result_id, token_hash(token), now + ttl_seconds),
)
return result_id, token
def claim_result(db, result_id, raw_token, *, authenticated_user_id, now=None):
"""Atomically consume a capability and assign the verified session owner."""
if not authenticated_user_id or not authenticated_user_id.strip():
raise AccessDenied("A verified session is required")
now = int(time.time()) if now is None else now
changed = db.execute(
"""UPDATE results
SET owner_id = ?, claim_hash = NULL, claim_expires_at = NULL
WHERE id = ? AND owner_id IS NULL
AND claim_hash = ? AND claim_expires_at > ?""",
(authenticated_user_id, result_id, token_hash(raw_token), now),
).rowcount
return changed == 1
def request_delivery(db, result_id, destination, *,
authenticated_user_id=None, raw_token=None, now=None):
"""Record a delivery request; this example never sends a real email.
Result control is required even for delivery. A typed destination is not
identity proof and this function never updates results.owner_id.
"""
if not destination or "@" not in destination or any(c in destination for c in "\r\n"):
raise ValueError("Invalid destination")
now = int(time.time()) if now is None else now
db.execute("BEGIN IMMEDIATE")
try:
row = db.execute(
"SELECT owner_id, claim_hash, claim_expires_at FROM results WHERE id = ?",
(result_id,),
).fetchone()
if row is None:
raise AccessDenied("Result is unavailable")
owner, expected_hash, expires_at = row
session_controls_result = owner is not None and authenticated_user_id == owner
token_controls_result = (
owner is None and raw_token is not None and expected_hash is not None
and expires_at > now
and hmac.compare_digest(token_hash(raw_token), expected_hash)
)
if not (session_controls_result or token_controls_result):
raise AccessDenied("Result is unavailable")
request_id = str(uuid.uuid4())
db.execute(
"INSERT INTO delivery_requests VALUES (?, ?, ?)",
(request_id, result_id, destination),
)
db.execute("COMMIT")
return request_id
except Exception:
db.execute("ROLLBACK")
raise
Companion tests
import concurrent.futures
import tempfile
import threading
import unittest
from pathlib import Path
from claim_demo import (AccessDenied, claim_result, connect, create_anonymous_result,
initialize, request_delivery)
class OwnershipTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.path = Path(self.tmp.name) / "example.sqlite"
self.db = connect(self.path)
initialize(self.db)
self.result_id, self.token = create_anonymous_result(self.db, now=100, ttl_seconds=60)
def tearDown(self):
self.db.close()
self.tmp.cleanup()
def row(self):
return self.db.execute("SELECT owner_id, claim_hash, claim_expires_at FROM results WHERE id = ?",
(self.result_id,)).fetchone()
def test_anonymous_row_has_no_owner_and_stores_only_token_hash(self):
owner, stored, expiry = self.row()
self.assertIsNone(owner)
self.assertNotEqual(stored, self.token)
self.assertEqual(expiry, 160)
def test_typed_email_does_not_assign_ownership(self):
before = self.row()
request_delivery(self.db, self.result_id, "someone@example.invalid",
raw_token=self.token, now=101)
self.assertEqual(self.row(), before)
def test_delivery_without_result_control_is_rejected(self):
with self.assertRaises(AccessDenied):
request_delivery(self.db, self.result_id, "someone@example.invalid", now=101)
self.assertEqual(self.db.execute("SELECT COUNT(*) FROM delivery_requests").fetchone()[0], 0)
def test_valid_claim_consumes_capability_and_replay_fails(self):
self.assertTrue(claim_result(self.db, self.result_id, self.token,
authenticated_user_id="user-a", now=101))
self.assertEqual(self.row(), ("user-a", None, None))
self.assertFalse(claim_result(self.db, self.result_id, self.token,
authenticated_user_id="user-b", now=102))
def test_wrong_token_and_unrelated_session_cannot_claim(self):
self.assertFalse(claim_result(self.db, self.result_id, "incorrect-token",
authenticated_user_id="user-b", now=101))
self.assertIsNone(self.row()[0])
def test_expiry_is_exclusive_at_boundary(self):
self.assertFalse(claim_result(self.db, self.result_id, self.token,
authenticated_user_id="user-a", now=160))
with self.assertRaises(AccessDenied):
request_delivery(self.db, self.result_id, "someone@example.invalid",
raw_token=self.token, now=160)
def test_unverified_identity_cannot_claim(self):
with self.assertRaises(AccessDenied):
claim_result(self.db, self.result_id, self.token,
authenticated_user_id="", now=101)
def test_result_ids_bind_capabilities(self):
other_id, _ = create_anonymous_result(self.db, now=100, ttl_seconds=60)
self.assertFalse(claim_result(self.db, other_id, self.token,
authenticated_user_id="user-a", now=101))
def test_claimed_delivery_uses_owner_session_and_cannot_rebind(self):
claim_result(self.db, self.result_id, self.token, authenticated_user_id="user-a", now=101)
request_delivery(self.db, self.result_id, "different@example.invalid",
authenticated_user_id="user-a", now=102)
self.assertEqual(self.row(), ("user-a", None, None))
with self.assertRaises(AccessDenied):
request_delivery(self.db, self.result_id, "someone@example.invalid",
raw_token=self.token, now=102)
with self.assertRaises(AccessDenied):
request_delivery(self.db, self.result_id, "someone@example.invalid",
authenticated_user_id="user-b", now=102)
def test_eight_concurrent_claims_have_exactly_one_winner(self):
barrier = threading.Barrier(8)
def attempt(index):
db = connect(self.path)
try:
barrier.wait(timeout=10)
ok = claim_result(db, self.result_id, self.token,
authenticated_user_id=f"user-{index}", now=101)
return index, ok
finally:
db.close()
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
results = list(pool.map(attempt, range(8)))
winners = [index for index, ok in results if ok]
self.assertEqual(len(winners), 1)
self.assertEqual(self.row(), (f"user-{winners[0]}", None, None))
if __name__ == "__main__":
unittest.main(verbosity=2)
Verify the boundary, not just the happy path
Save the companion test block above as test_claim_demo.py beside the example and run:
python3 -m unittest -v test_claim_demo.py
The local validation for this example ran 10 tests successfully, including eight simultaneous claim attempts through separate SQLite connections. Exactly one acquired ownership. These are tests of this demonstration, not a production audit or a benchmark of all databases.
| Case | Expected result |
|---|---|
| New anonymous result | No owner; only the token hash is stored |
| Valid capability and a typed destination | Delivery request recorded; owner unchanged |
| Delivery without result control | Rejected; no delivery row created |
| Valid capability and verified session | One owner assigned; capability removed |
| Replay of the consumed capability | Rejected |
| Wrong token or token for another result | Rejected |
| Token at its exact expiry boundary | Rejected |
| No verified identity | Rejected |
| Claimed result and a different typed destination | Current owner unchanged; unrelated sessions rejected |
| Eight concurrent claims | Exactly one winner |
What this example deliberately leaves to the real application
There is no HTTP server or authentication implementation here. In a deployed service, add the applicable session checks, CSRF protection, authorization for result reads, delivery consent, rate limits, retention and abuse controls. An email request must not become an arbitrary mail relay.
A capability is a bearer secret. Keep it out of analytics, page metadata, third-party error reporting, referrers and public share links. Choose client-side storage and its expiry with the shared-device and XSS threat models in mind. A hash in the database does not protect a raw capability that leaks elsewhere.
The mail transaction only records work. A real outbox worker needs its own idempotency and permission rules. Avoid placing the raw claim token in an outbox payload. If access expires or changes between queuing and dispatch, decide explicitly whether the worker must recheck it; this example does not implement that worker.
SQLite serializes the write transactions used in this example. Verify the corresponding transaction and conditional-write behavior in the database you actually deploy; see SQLite's UPDATE documentation for the statement used here. This is not evidence about a PostgreSQL deployment.
The central invariant is small enough to audit: only the claim operation may assign an owner, and a delivery destination never supplies that identity. Tests should make that boundary visible under wrong inputs, expiry, replay and concurrency.
AI disclosure: This article and its standalone example were prepared primarily by an AI coding agent for the AI Doctor Scan team. The example's tests were executed locally. It is not presented as a first-person incident report.
Top comments (0)