Spot spam in a contact form

"""
Reject spam in a contact form: honeypot, submission delay, link cap, banned phrases.

Rung N0. Four checks, no dependency, no training data, and a rejection that
comes with a reason you can show to whoever asks why a message was lost.

Two of the checks look at the sender rather than the text. A honeypot field,
kept out of sight in the form, and the time spent on the page catch the
scripts that post to the endpoint without rendering it, as long as they fill
in every field or post faster than a person types. Neither needs to read the
message.

The delay is the one that has to be built right, and it is where this kind of
check is usually got wrong. A hidden field holding the time the page was
rendered is a number the sender writes, and a script writes whatever it likes
in it. So the server issues that timestamp itself, signed with a secret only it
holds, and recomputes the signature on submission: a timestamp that was edited,
or that nobody signed, is not a delay at all, and counts as too fast. Thirty
lines below, and no dependency. Ned Batchelder's spinner, which this follows,
signs the visitor's address and the page identifier along with the time, so
that a token issued for one form cannot be replayed on another.

The two text checks are the weak half, and the entry says so.
"""

import hmac
import re
import time
import unicodedata
from hashlib import sha256

# The field is present in the form and hidden by the stylesheet — not with
# `type="hidden"`, which a script knows to skip — and named after something a
# naive form filler will want to complete.
HONEYPOT_FIELD = "website"

MINIMUM_SECONDS = 3.0
MAXIMUM_LINKS = 2

# One match per link, not one per part: a bare `https?://` alternative would
# count `http://example.com` twice and reject the customer who sends two.
# A bare domain must start a word that follows neither `@` nor a dot, so the
# domain of an email address is not a link, and a run like `a-a-a-...` gives
# the pattern one place to start instead of one per letter.
LINK = re.compile(
    r"(?:https?://|www\.)\S+"
    r"|(?<![\w@.-])[\w-]+(?:\.[\w-]+)*\.(?:com|net|org|ru|xyz|top)\b"
)

# Whole words only, so `cryptographie` or a shop called `Casino` is not
# rejected. Every word on such a list is one a customer may write some day.
BANNED = re.compile(r"\b(backlink|guest post|seo (?:services|ranking)|online casino|viagra)s?\b")


def issue_token(secret: bytes, issued_at: float | None = None) -> str:
    """
    The timestamp to put in the form, signed: `<seconds>.<signature>`.

    `secret` never leaves the server. Anything else the form carries about
    time is the sender's word.
    """
    issued = int(time.time() if issued_at is None else issued_at)
    return f"{issued}.{_signature(secret, issued)}"


def seconds_on_page(token: str, secret: bytes, now: float | None = None) -> float:
    """
    How long the form was really on the page, or zero when the token is not ours.

    Zero is the safe answer: `reasons` reads it as too fast. A token that was
    edited, replayed with a different time, or made up, lands here.
    """
    issued_text, _, signature = str(token).partition(".")
    if not issued_text.isdigit():
        return 0.0
    issued = int(issued_text)
    if not hmac.compare_digest(signature, _signature(secret, issued)):
        return 0.0
    return max(0.0, (time.time() if now is None else now) - issued)


def _signature(secret: bytes, issued: int) -> str:
    return hmac.new(secret, str(issued).encode(), sha256).hexdigest()


def fold(text: str) -> str:
    """Lowercase and strip accents, so `Rétrolien` and `RETROLIEN` match alike."""
    stripped = unicodedata.normalize("NFKD", text)
    return "".join(c for c in stripped if not unicodedata.combining(c)).casefold()


def reasons(fields: dict, seconds: float) -> list[str]:
    """
    Every reason to reject this submission. An empty list means: accept it.

    `seconds` is what `seconds_on_page` returned, not a number the form
    carried: this function trusts its caller, and the caller must not trust
    the sender.

    Returning reasons rather than a boolean is what makes the rule reviewable:
    a rejection you cannot explain is a rejection you cannot tune.
    """
    found = []
    if fields.get(HONEYPOT_FIELD, "").strip():
        found.append("honeypot filled")
    # Written as "not at least", so a missing or unreadable delay (NaN) counts
    # as too fast instead of slipping through.
    if not seconds >= MINIMUM_SECONDS:
        found.append("submitted too fast")

    message = fold(fields.get("message", ""))
    if len(LINK.findall(message)) > MAXIMUM_LINKS:
        found.append("too many links")
    found.extend(f"banned phrase: {phrase}" for phrase in sorted(set(BANNED.findall(message))))
    return found


def is_spam(fields: dict, seconds: float) -> bool:
    """The same decision, for callers that only want the verdict."""
    return bool(reasons(fields, seconds))

/**
 * Reject spam in a contact form: honeypot, submission delay, link cap, banned
 * phrases.
 *
 * Rung N0. Four checks, no dependency, no training data, and a rejection that
 * comes with a reason you can show to whoever asks why a message was lost.
 *
 * Two of the checks look at the sender rather than the text. A honeypot field,
 * kept out of sight in the form, and the time spent on the page catch the
 * scripts that post to the endpoint without rendering it, as long as they fill
 * in every field or post faster than a person types. Neither needs to read the
 * message.
 *
 * The delay is the one that has to be built right, and it is where this kind
 * of check is usually got wrong. A hidden field holding the time the page was
 * rendered is a number the sender writes, and a script writes whatever it
 * likes in it. So the server issues that timestamp itself, signed with a
 * secret only it holds, and recomputes the signature on submission: a
 * timestamp that was edited, or that nobody signed, is not a delay at all, and
 * counts as too fast. Thirty lines below, and nothing but `node:crypto`. Ned
 * Batchelder's spinner, which this follows, signs the visitor's address and the
 * page identifier along with the time, so that a token issued for one form
 * cannot be replayed on another.
 *
 * The two text checks are the weak half, and the entry says so.
 */
import { createHmac, timingSafeEqual } from 'node:crypto';

// The field is present in the form and hidden by the stylesheet — not with
// `type="hidden"`, which a script knows to skip — and named after something a
// naive form filler will want to complete.
export const HONEYPOT_FIELD = 'website';

export const MINIMUM_SECONDS = 3;
export const MAXIMUM_LINKS = 2;

// One match per link, not one per part: a bare `https?://` alternative would
// count `http://example.com` twice and reject the customer who sends two.
// A bare domain must start a word that follows neither `@` nor a dot, so the
// domain of an email address is not a link, and a run like `a-a-a-...` gives
// the pattern one place to start instead of one per letter.
const LINK = /(?:https?:\/\/|www\.)\S+|(?<![\w@.-])[\w-]+(?:\.[\w-]+)*\.(?:com|net|org|ru|xyz|top)\b/g;

// Whole words only, so `cryptographie` or a shop called `Casino` is not
// rejected. Every word on such a list is one a customer may write some day.
const BANNED = /\b(backlink|guest post|seo (?:services|ranking)|online casino|viagra)s?\b/g;

/**
 * The timestamp to put in the form, signed: `<seconds>.<signature>`.
 *
 * `secret` never leaves the server. Anything else the form carries about time
 * is the sender's word.
 */
export function issueToken(secret, issuedAt = Date.now() / 1000) {
  const issued = Math.floor(issuedAt);
  return `${issued}.${signature(secret, issued)}`;
}

/**
 * How long the form was really on the page, or zero when the token is not ours.
 *
 * Zero is the safe answer: `reasons` reads it as too fast. A token that was
 * edited, replayed with a different time, or made up, lands here.
 */
export function secondsOnPage(token, secret, now = Date.now() / 1000) {
  const [issuedText = '', given = ''] = String(token).split('.');
  if (!/^\d+$/.test(issuedText)) return 0;
  const expected = Buffer.from(signature(secret, Number(issuedText)));
  const offered = Buffer.from(given);
  if (offered.length !== expected.length || !timingSafeEqual(offered, expected)) return 0;
  return Math.max(0, now - Number(issuedText));
}

function signature(secret, issued) {
  return createHmac('sha256', secret).update(String(issued)).digest('hex');
}

/** Lowercase and strip accents, so `Rétrolien` and `RETROLIEN` match alike. */
export function fold(text) {
  return text.normalize('NFKD').replace(/\p{Diacritic}/gu, '').toLowerCase();
}

/**
 * Every reason to reject this submission. An empty list means: accept it.
 *
 * Returning reasons rather than a boolean is what makes the rule reviewable:
 * a rejection you cannot explain is a rejection you cannot tune.
 */
export function reasons(fields, seconds) {
  const found = [];
  if ((fields[HONEYPOT_FIELD] ?? '').trim()) found.push('honeypot filled');
  // Written as "not at least", so a missing or unreadable delay (NaN) counts
  // as too fast instead of slipping through.
  if (!(seconds >= MINIMUM_SECONDS)) found.push('submitted too fast');

  const message = fold(fields.message ?? '');
  if ((message.match(LINK) ?? []).length > MAXIMUM_LINKS) found.push('too many links');
  const phrases = new Set(Array.from(message.matchAll(BANNED), (match) => match[1]));
  for (const phrase of [...phrases].sort()) found.push(`banned phrase: ${phrase}`);
  return found;
}

/** The same decision, for callers that only want the verdict. */
export function isSpam(fields, seconds) {
  return reasons(fields, seconds).length > 0;
}

"""
Sort a contact form submission by asking a general-purpose model.

Rung N3. It is here so you can see what it costs, not because this entry
recommends it.

Note what the code has to do that N0 did not: refuse an empty or oversized
input before paying for a call, ask again when the provider fails or answers
something unusable, every attempt being billed, parse an answer that is only
probably valid JSON, and refuse to guess when none of the attempts is usable.
That plumbing is the real cost of this rung, and it is the part your tests have
to cover, because a test against a double says nothing of how the model judges.

Note too what it cannot do. The message goes into the same prompt as the
instructions, and nothing in the protocol tells the model which of the two to
obey.
"""

from __future__ import annotations

import json

PROMPT = (
    "You moderate the contact form of a small company.\n"
    "Decide whether the submission below is unsolicited commercial spam.\n"
    'Answer with JSON only: {{"spam": true or false, "reason": "one short sentence"}}\n\n'
    "Submission:\n{message}"
)

MAX_CHARACTERS = 4000

# The provider named here is an example, not a recommendation: the reasoning
# holds for any general-purpose model API, and the client is swappable. Pass
# any object with a `complete(prompt=..., temperature=...)` method.
MODEL = "gpt-4.1-mini"  # an example id: check the parameters your model accepts


class ProviderClient:
    """The one call this snippet makes, on top of the provider's SDK."""

    def __init__(self, sdk=None, model: str = MODEL):
        if sdk is None:  # pragma: no cover - needs a key and a network
            from openai import OpenAI

            sdk = OpenAI()
        self.sdk, self.model = sdk, model

    def complete(self, *, prompt: str, temperature: float) -> str:
        response = self.sdk.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
        )
        return response.choices[0].message.content


class ClassificationUnavailable(Exception):
    """The provider could not be reached, or answered something unusable."""


def classify(message: str, client=None, *, attempts: int = 3) -> dict:
    """
    Return `{"spam": bool, "reason": str}` for one form submission.

    `client` is injected so this function can be tested without a network call.
    In production it defaults to a real provider client.
    """
    if not message.strip():
        raise ValueError("empty submission: there is nothing to classify")
    # A model charges by the token, and a form field is a place where anyone
    # can paste a novel. Refusing oversized input is not an optimisation, it is
    # a cost control.
    if len(message) > MAX_CHARACTERS:
        raise ValueError(f"submission longer than {MAX_CHARACTERS} characters")

    client = client or ProviderClient()
    verdict = _ask(client, message, attempts)
    reason = verdict.get("reason")
    return {"spam": verdict["spam"], "reason": "" if reason is None else str(reason)}


def _ask(client, message: str, attempts: int) -> dict:
    last_error: Exception | None = None
    for _ in range(attempts):
        try:
            # Temperature zero, the low end of the range, which the provider
            # documents as more focused and deterministic. It does not promise
            # that two identical calls agree.
            answer = client.complete(prompt=PROMPT.format(message=message), temperature=0)
        except Exception as error:  # noqa: BLE001 - any provider failure is retried
            last_error = error
            continue
        verdict = _parse(answer)
        if verdict is not None:
            return verdict
        last_error = ValueError("the model answered without a usable verdict")
    raise ClassificationUnavailable(str(last_error))


def _parse(answer) -> dict | None:
    """The verdict in `answer`, or None: no text, not JSON, or no boolean `spam`."""
    if not isinstance(answer, str):  # the SDK types the content as optional
        return None
    # A JSON answer wrapped whole in one code fence is read; nothing else is:
    # a fence opened and never closed, or prose around it, is a failed answer.
    text = answer.strip()
    if text.startswith("```") and text.endswith("```") and text.count("```") == 2:
        text = text[3:-3].removeprefix("json")
    try:
        parsed = json.loads(text)
    except (ValueError, RecursionError):
        return None
    return parsed if isinstance(parsed, dict) and isinstance(parsed.get("spam"), bool) else None

/**
 * Sort a contact form submission by asking a general-purpose model.
 *
 * Rung N3. It is here so you can see what it costs, not because this entry
 * recommends it.
 *
 * Note what the code has to do that N0 did not: refuse an empty or oversized
 * input before paying for a call, ask again when the provider fails or answers
 * something unusable, every attempt being billed, parse an answer that is only
 * probably valid JSON, and refuse to guess when none of the attempts is usable.
 * That plumbing is the real cost of this rung, and it is the part your tests
 * have to cover, because a test against a double says nothing of how the model
 * judges.
 *
 * Note too what it cannot do. The message goes into the same prompt as the
 * instructions, and nothing in the protocol tells the model which of the two
 * to obey.
 */

const PROMPT = [
  'You moderate the contact form of a small company.',
  'Decide whether the submission below is unsolicited commercial spam.',
  'Answer with JSON only: {"spam": true or false, "reason": "one short sentence"}',
  '',
  'Submission:',
].join('\n');

export const MAX_CHARACTERS = 4000;

// The provider named here is an example, not a recommendation: the reasoning
// holds for any general-purpose model API, and the client is swappable. Pass
// any object with a `complete({ prompt, temperature })` method.
export const MODEL = 'gpt-4.1-mini'; // an example id: check the parameters your model accepts

export async function providerClient(sdk, model = MODEL) {
  if (!sdk) {
    const { OpenAI } = await import('openai');
    sdk = new OpenAI();
  }
  return {
    async complete({ prompt, temperature }) {
      const response = await sdk.chat.completions.create({
        model,
        messages: [{ role: 'user', content: prompt }],
        temperature,
      });
      return response.choices[0].message.content;
    },
  };
}

export class ClassificationUnavailable extends Error {}

/**
 * Return `{ spam, reason }` for one form submission.
 *
 * @param {string} message
 * @param {object} options
 * @param {{complete: Function}} [options.client] injected so this can be
 *   tested without a network call; defaults to a real provider client
 * @param {number} [options.attempts]
 */
export async function classify(message, { client, attempts = 3 } = {}) {
  if (!message.trim()) throw new RangeError('empty submission: there is nothing to classify');
  // A model charges by the token, and a form field is a place where anyone can
  // paste a novel. Refusing oversized input is not an optimisation, it is a
  // cost control. Counted in code points, as Python counts characters.
  if ([...message].length > MAX_CHARACTERS) throw new RangeError(`submission longer than ${MAX_CHARACTERS} characters`);

  client ??= await providerClient();
  const verdict = await ask(client, message, attempts);
  return { spam: verdict.spam, reason: String(verdict.reason ?? '') };
}

async function ask(client, message, attempts) {
  let lastError;
  for (let i = 0; i < attempts; i += 1) {
    let answer;
    try {
      // Temperature zero, the low end of the range, which the provider
      // documents as more focused and deterministic. It does not promise
      // that two identical calls agree.
      answer = await client.complete({ prompt: `${PROMPT}\n${message}`, temperature: 0 });
    } catch (error) {
      lastError = error; // any provider failure is retried
      continue;
    }
    const verdict = parse(answer);
    if (verdict) return verdict;
    lastError = new Error('the model answered without a usable verdict');
  }
  throw new ClassificationUnavailable(String(lastError));
}

/** The verdict in `answer`, or null: no text, not JSON, or no boolean `spam`. */
function parse(answer) {
  if (typeof answer !== 'string') return null; // the SDK types the content as nullable
  // A JSON answer wrapped whole in one code fence is read; nothing else is: a
  // fence opened and never closed, or prose around it, is a failed answer.
  const trimmed = answer.trim();
  const fenced = trimmed.startsWith('```') && trimmed.endsWith('```') && trimmed.split('```').length === 3;
  const text = fenced ? trimmed.slice(3, -3).replace(/^json/, '') : trimmed;
  try {
    const parsed = JSON.parse(text);
    return parsed && typeof parsed.spam === 'boolean' ? parsed : null;
  } catch {
    return null;
  }
}