1 files changed, 73 insertions, 72 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/captcha.py b/packages/meshbay-hub/src/meshbay_hub/captcha.py
index cf33d52..e6faf1a 100644
--- a/packages/meshbay-hub/src/meshbay_hub/captcha.py
+++ b/packages/meshbay-hub/src/meshbay_hub/captcha.py
@@ -1,85 +1,86 @@
-"""reCAPTCHA v2 server-side verification."""
+"""Proof-of-work challenge gating registration and password reset.
 
-import logging
+The hub hands out a challenge; the page finds the number behind it by trying
+them in turn and sends the answer with the form. Nobody else is asked: no
+script, cookie or address leaves for a third party, which is what the
+reCAPTCHA this replaced could not say (docs/MESHBAY_DESIGN.md §7.7).
 
-import httpx
+What it buys is cost, not a verdict on who is human. A person never sees it —
+the page solves while the form is being filled in — and a script pays it once
+per attempt. The bounds that hold against a script with machines to spare are
+the per-IP limits on the two routes and, above all, the mail budget
+(`mail.py`): the sign-up and the reset are what make this hub send mail.
 
-log = logging.getLogger(__name__)
+The scheme is ALTCHA's, keyed to this hub. A challenge is the SHA-256 of
+`<nonce>.<expires>.<number>`, with `number` drawn below `max_number`; its HMAC
+is the hub's word that it issued it. The answer carries the nonce, the expiry,
+the number and that HMAC, so the hub recomputes the challenge rather than
+remembering it — and every worker, and the next restart, agrees on it, since
+the key is derived from the hub's own private key and not drawn per process.
+What is remembered is an answer once spent, so one solve is one form.
+"""
 
-VERIFY_URL = "https://www.google.com/recaptcha/api/siteverify"
+import hashlib
+import hmac
+import secrets
+import time
 
+from meshbay_hub.auth import hub_private_key_pem
 
-async def verify_captcha(
-    secret_key: str,
-    token: str,
-    remote_ip: str | None = None,
-    allowed_hosts: frozenset[str] | set[str] | None = None,
-    allow_unattributed: bool = False,
-) -> bool:
-    """True when Google accepts the token, and it came from a host we expect.
+# How long a challenge stays good. Long enough for somebody filling in the form
+# slowly; the page fetches a fresh one if it outlived this before submitting.
+CHALLENGE_TTL = 1800
 
-    `allowed_hosts` is the answer to a reCAPTCHA key whose own origin check is
-    turned off. That check compares the *page's hostname* against the key's
-    domain list, and the desktop client's page is served from `app://meshbay`
-    — Chromium refuses a service worker on a custom scheme and the interface
-    ships inside the package, so it cannot be `https://<hub>` and the hostname
-    cannot be the hub's. Google answers "Invalid domain for site key" inside
-    its own widget, which is not something the client can configure away.
+_KEY_LABEL = b"meshbay captcha v1"
+# Answers already accepted, until their challenge would have expired anyway.
+_spent: dict[str, int] = {}
 
-    Turning the key's origin check off moves that decision here, where the hub
-    knows which hosts are its own. `siteverify` reports the hostname it
-    observed, not one the caller asserts, so this is a real check and not a
-    formality — a bot embedding the site key on a page of its own is refused
-    on the hostname it actually served from.
 
-    `None` (the default) skips it, which is what a deployment leaving the
-    origin check with Google wants: it is then already done, one layer up.
+def _key() -> bytes:
+    return hmac.new(hub_private_key_pem(), _KEY_LABEL, hashlib.sha256).digest()
 
-    `allow_unattributed` covers the case that made the desktop client fail
-    anyway. A solve from a page Google cannot attribute to a domain comes back
-    with an **empty** hostname, not the host part of the origin — measured
-    live: `app://meshbay` reports `''`, never `'meshbay'`. So an allowlist
-    entry can never match it, and an empty string cannot be an allowlist entry
-    either: a blank in a TOML list is a typo far more often than it is an
-    intention, and the config parser drops blanks for that reason.
 
-    What it admits is every non-web client, not only ours — a `file://` page or
-    somebody else's Electron application report the same nothing. That is the
-    same bar the desktop client's own origin would have been (`app://meshbay`
-    is not a credential; any application can claim it), and it is a bar: the
-    captcha still has to be *solved*, per token. What is given up is the origin
-    restriction for non-web clients, not the captcha.
+def _sign(challenge: str) -> str:
+    return hmac.new(_key(), challenge.encode(), hashlib.sha256).hexdigest()
+
+
+def issue(max_number: int) -> dict:
+    """A fresh challenge, in the shape the page solves (`static/captcha.js`)."""
+    prefix = f"{secrets.token_hex(12)}.{int(time.time()) + CHALLENGE_TTL}."
+    number = secrets.randbelow(max_number)
+    challenge = hashlib.sha256(f"{prefix}{number}".encode()).hexdigest()
+    return {
+        "prefix": prefix,
+        "challenge": challenge,
+        "max_number": max_number,
+        "signature": _sign(challenge),
+        "expires_in": CHALLENGE_TTL,
+    }
+
+
+def verify(token: str) -> bool:
+    """True for an answer this hub issued, still in date and not spent before.
+
+    The token is `<nonce>.<expires>.<number>.<signature>`. Nothing in it is
+    trusted on its own: the challenge is recomputed from the first three, and
+    only this hub's key turns that into the signature.
     """
-    payload: dict[str, str] = {"secret": secret_key, "response": token}
-    if remote_ip:
-        payload["remoteip"] = remote_ip
-    try:
-        async with httpx.AsyncClient(timeout=5) as client:
-            resp = await client.post(VERIFY_URL, data=payload)
-            resp.raise_for_status()
-            result = resp.json()
-            if not result.get("success"):
-                log.info("captcha rejected: %s", result.get("error-codes", []))
-                return False
-            if allowed_hosts is not None:
-                # Logged at warning with the hostname spelled out: this is also
-                # how an operator finds what to allow after adding a client
-                # whose origin they have not seen before. It is how the empty
-                # one was found.
-                host = result.get("hostname") or ""
-                if not host:
-                    if not allow_unattributed:
-                        log.warning(
-                            "captcha solved on a host Google did not attribute; "
-                            "set captcha.allow_unattributed_host to admit "
-                            "non-web clients such as the desktop application")
-                        return False
-                elif host not in allowed_hosts:
-                    log.warning(
-                        "captcha solved on an unexpected host %r; allowed: %s",
-                        host, sorted(allowed_hosts))
-                    return False
-            return True
-    except Exception:
-        log.exception("captcha verification request failed")
+    parts = token.split(".")
+    if len(parts) != 4:
+        return False
+    nonce, expires_s, number_s, signature = parts
+    if not (expires_s.isdigit() and number_s.isdigit()):
+        return False
+    now = int(time.time())
+    if not now < int(expires_s) <= now + CHALLENGE_TTL:
+        return False
+    challenge = hashlib.sha256(f"{nonce}.{expires_s}.{number_s}".encode()).hexdigest()
+    if not hmac.compare_digest(_sign(challenge), signature):
+        return False
+    for spent, until in list(_spent.items()):
+        if until <= now:
+            del _spent[spent]
+    if signature in _spent:
         return False
+    _spent[signature] = int(expires_s)
+    return True