Repérer le spam dans un formulaire de contact

"""
Écarter le spam d'un formulaire de contact : pot de miel, délai de
soumission, plafond de liens, phrases interdites.

Niveau N0. Quatre contrôles, aucune dépendance, aucune donnée
d'entraînement, et un rejet qui vient avec un motif présentable à qui
demande pourquoi son message s'est perdu.

Deux des contrôles regardent l'expéditeur plutôt que le texte. Un champ
pot de miel, masqué par la feuille de style — et non par « type="hidden" »,
qu'un script sait ignorer —, et le temps passé sur
la page attrapent les scripts qui postent sur le point de terminaison sans
l'afficher, pour peu qu'ils remplissent tous les champs ou postent plus
vite qu'une personne ne tape. Aucun des deux n'a besoin de lire le message.

Le délai est celui qu'il faut bâtir correctement, et c'est là que ce genre de
contrôle se rate d'ordinaire. Un champ caché qui porte l'heure d'affichage de
la page est un nombre que l'expéditeur écrit, et un script y écrit ce qu'il
veut. Le serveur émet donc lui-même cet horodatage, signé avec un secret que
lui seul détient, et recalcule la signature à la réception : un horodatage
modifié, ou que personne n'a signé, n'est pas un délai du tout, et compte
comme trop rapide. Trente lignes plus bas, sans aucune dépendance. Le
« spinner » de Ned Batchelder, dont ceci s'inspire, signe aussi l'adresse du
visiteur et l'identifiant de la page, pour qu'un jeton émis pour un formulaire
ne puisse pas être rejoué sur un autre.

Les deux contrôles de texte sont la moitié faible, et la fiche le dit.
"""

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))

/**
 * Écarter le spam d'un formulaire de contact : pot de miel, délai de
 * soumission, plafond de liens, phrases interdites.
 *
 * Niveau N0. Quatre contrôles, aucune dépendance, aucune donnée
 * d'entraînement, et un rejet qui vient avec un motif présentable à qui
 * demande pourquoi son message s'est perdu.
 *
 * Deux des contrôles regardent l'expéditeur plutôt que le texte. Un champ
 * pot de miel, masqué par la feuille de style — et non par « type="hidden" »,
 * qu'un script sait ignorer —, et le temps passé sur
 * la page attrapent les scripts qui postent sur le point de terminaison sans
 * l'afficher, pour peu qu'ils remplissent tous les champs ou postent plus
 * vite qu'une personne ne tape. Aucun des deux n'a besoin de lire le message.
 *
 * Le délai est celui qu'il faut bâtir correctement, et c'est là que ce genre de
 * contrôle se rate d'ordinaire. Un champ caché qui porte l'heure d'affichage de
 * la page est un nombre que l'expéditeur écrit, et un script y écrit ce qu'il
 * veut. Le serveur émet donc lui-même cet horodatage, signé avec un secret que
 * lui seul détient, et recalcule la signature à la réception : un horodatage
 * modifié, ou que personne n'a signé, n'est pas un délai du tout, et compte
 * comme trop rapide. Trente lignes plus bas, sans aucune dépendance. Le
 * « spinner » de Ned Batchelder, dont ceci s'inspire, signe aussi l'adresse du
 * visiteur et l'identifiant de la page, pour qu'un jeton émis pour un formulaire
 * ne puisse pas être rejoué sur un autre.
 *
 * Les deux contrôles de texte sont la moitié faible, et la fiche le dit.
 */
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;
}

"""
Trier un envoi de formulaire de contact en interrogeant un modèle
généraliste.

Niveau N3. Il est ici pour qu'on voie ce qu'il coûte, pas parce que cette
fiche le recommande.

Voyez ce que le code doit faire et que N0 n'avait pas à faire : refuser une
entrée vide ou trop longue avant de payer un appel, redemander quand le
fournisseur échoue ou répond quelque chose d'inexploitable, chaque essai
étant facturé, analyser une réponse qui n'est que probablement du JSON
valide, et refuser de deviner quand aucun essai n'est exploitable. Cette
plomberie est le vrai coût du niveau, et c'est la part que vos tests
doivent couvrir, parce qu'un test contre un double ne dit rien de la façon
dont le modèle juge.

Voyez aussi ce qu'il ne peut pas faire. Le message entre dans la même
invite que les consignes, et rien dans le protocole ne dit au modèle auquel
des deux obéir.
"""

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

/**
 * Trier un envoi de formulaire de contact en interrogeant un modèle
 * généraliste.
 *
 * Niveau N3. Il est ici pour qu'on voie ce qu'il coûte, pas parce que cette
 * fiche le recommande.
 *
 * Voyez ce que le code doit faire et que N0 n'avait pas à faire : refuser une
 * entrée vide ou trop longue avant de payer un appel, redemander quand le
 * fournisseur échoue ou répond quelque chose d'inexploitable, chaque essai
 * étant facturé, analyser une réponse qui n'est que probablement du JSON
 * valide, et refuser de deviner quand aucun essai n'est exploitable. Cette
 * plomberie est le vrai coût du niveau, et c'est la part que vos tests
 * doivent couvrir, parce qu'un test contre un double ne dit rien de la façon
 * dont le modèle juge.
 *
 * Voyez aussi ce qu'il ne peut pas faire. Le message entre dans la même
 * invite que les consignes, et rien dans le protocole ne dit au modèle auquel
 * des deux obéir.
 */

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;
  }
}