TOTP auth (#11288)

mekarpeles · web-flow · commit dd3ddb27bae5 · 2026-03-25T11:59:57.000-07:00
TOTP Service
* docs(ai): add OTP service subsystem doc; fix broken doctest
- Move OTP/TOTP docs into docs/ai/subsystems/otp_service.md
- Remove network-dependent doctest from otp_service_issue.POST
  (test env blocks network requests; doctest had no assertions anyway)
diff --git a/docs/ai/subsystems/otp_service.md b/docs/ai/subsystems/otp_service.md
@@ -0,0 +1,65 @@
+# OTP Auth Service
+
+Open Library acts as a TOTP (Timed One-Time Password) provider for Lenny, the Internet Archive book-lending service. Lenny requests an OTP, Open Library generates one and emails it to the patron, and the patron enters it in Lenny, which forwards it back to Open Library for verification.
+
+**Relevant files:**
+- `openlibrary/core/auth.py` — `TimedOneTimePassword` class (generate, validate, rate-limit)
+- `openlibrary/plugins/upstream/account.py` — `otp_service_issue` and `otp_service_redeem` endpoints
+
+## Endpoints
+
+| Method | Path | Purpose |
+|--------|------|---------|
+| POST | `/account/otp/issue` | Generate and email an OTP to a patron |
+| POST | `/account/otp/redeem` | Verify an OTP submitted by a patron |
+
+Both endpoints read `service_ip` from the `X-Forwarded-For` request header (set automatically by nginx in production/docker). They return JSON.
+
+## Local Docker Testing
+
+**1. Add `otp_seed` to the dev config** (`conf/openlibrary.yml`):
+
+```yaml
+otp_seed: "dev-secret-seed"
+```
+
+**2. Start the stack:**
+
+```bash
+docker compose up
+```
+
+**3. Issue an OTP** (pass `sendmail=false` to skip SMTP in local dev):
+
+```bash
+curl -s -X POST http://localhost:8080/account/otp/issue \
+  -H 'X-Forwarded-For: 1.2.3.4' \
+  -d 'email=patron@example.com&ip=5.6.7.8&sendmail=false'
+# → {"success": "issued"}
+```
+
+**4. Compute the expected OTP** (since email is not sent locally):
+
+```bash
+docker compose exec web python3 - <<'EOF'
+from openlibrary.core.auth import TimedOneTimePassword as OTP
+# Use the same service_ip, email, ip as the issue request
+print(OTP.generate("1.2.3.4", "patron@example.com", "5.6.7.8"))
+EOF
+```
+
+**5. Redeem the OTP:**
+
+```bash
+curl -s -X POST http://localhost:8080/account/otp/redeem \
+  -H 'X-Forwarded-For: 1.2.3.4' \
+  -d 'email=patron@example.com&ip=5.6.7.8&otp=<OTP_FROM_STEP_4>'
+# → {"success": "redeemed"}
+```
+
+## Notes
+
+- `service_ip` (from `X-Forwarded-For`) must match between issue and redeem requests.
+- OTPs are valid for `VALID_MINUTES = 10` minutes (checked across rolling 1-minute windows).
+- Rate limiting uses memcache: 1 request per TTL per client, max 3 attempts per email/ip globally.
+- `verify_service` (challenge URL verification) is intentionally disabled — Open Library cannot make outbound requests to verify endpoints due to WAF/proxy restrictions.
diff --git a/openlibrary/core/auth.py b/openlibrary/core/auth.py
@@ -1,9 +1,15 @@
 import datetime
 import hashlib
 import hmac
+import socket
+import string
 import time
+from urllib.parse import urlparse
+
+import requests
 
 from infogami import config
+from openlibrary.core import cache
 
 
 class ExpiredTokenError(Exception):
@@ -93,3 +99,79 @@ def verify(
         if err:
             raise err
         return result
+
+
+class TimedOneTimePassword:
+
+    VALID_MINUTES = 10
+
+    @staticmethod
+    def shorten(digest: bytes, length=6) -> str:
+        """
+        Convert an HMAC digest (bytes) into a short alphanumeric code.
+        """
+        alphabet = string.digits + string.ascii_uppercase
+        num = int.from_bytes(digest, "big")
+        base36 = ""
+        while num > 0:
+            num, i = divmod(num, 36)
+            base36 = alphabet[i] + base36
+        return base36[:length].lower()
+
+    @classmethod
+    def generate(
+        cls, service_ip: str, client_email: str, client_ip: str, ts: int | None = None
+    ) -> str:
+        seed = config.get("otp_seed")
+        ts = ts or int(time.time() // 60)
+        payload = f"{service_ip}:{client_email}:{client_ip}:{ts}".encode()
+        digest = hmac.new(seed.encode('utf-8'), payload, hashlib.sha256).digest()
+        return cls.shorten(digest)
+
+    @staticmethod
+    def verify_service(service_ip: str, service_url: str) -> bool:
+        """Doesn't work because of VPN"""
+        parsed = urlparse(service_url)
+        if not parsed.hostname:
+            return False
+        resolved_ip = socket.gethostbyname(parsed.hostname)
+        r = requests.get(service_url, timeout=5)
+        return (
+            service_url.startswith("https://")
+            and resolved_ip == service_ip
+            and bool(r.json())
+        )
+
+    @classmethod
+    def is_ratelimited(cls, ttl=60, service_ip="", **kwargs):
+        def ratelimit_error(key, ttl):
+            return {"error": "ratelimit", "ratelimit": {"ttl": ttl, "key": key}}
+
+        mc = cache.get_memcache()
+        # Limit requests to 1 / ttl per client
+        for key, value in kwargs.items():
+            cache_key = f"otp-client:{service_ip}:{key}:{value}"
+            if not mc.add(cache_key, 1, expires=ttl):
+                return ratelimit_error(cache_key, ttl)
+
+        # Limit globally to 3 attempts per email and ip per / ttl
+        for key, value in kwargs.items():
+            cache_key = f"otp-global:{key}:{value}"
+            count = (mc.get(cache_key) or 0) + 1
+            mc.set(cache_key, count, expires=ttl)
+            if count > 3:
+                return ratelimit_error(cache_key, ttl)
+
+    @classmethod
+    def validate(cls, otp, service_ip, client_email, client_ip, ts):
+        expected_otp = cls.generate(service_ip, client_email, client_ip, ts)
+        return hmac.compare_digest(otp.lower(), expected_otp.lower())
+
+    @classmethod
+    def is_valid(cls, client_email, client_ip, service_ip, otp):
+        now_minute = int(time.time() // 60)
+        for delta in range(cls.VALID_MINUTES):
+            minute_ts = now_minute - delta
+            if cls.validate(otp, service_ip, client_email, client_ip, minute_ts):
+                return True
+        return False
diff --git a/openlibrary/plugins/upstream/account.py b/openlibrary/plugins/upstream/account.py
@@ -34,6 +34,7 @@
 from openlibrary.core import helpers as h
 from openlibrary.core import lending, stats
 from openlibrary.core.auth import ExpiredTokenError, HMACToken, MissingKeyError
+from openlibrary.core.auth import TimedOneTimePassword as OTP
 from openlibrary.core.booknotes import Booknotes
 from openlibrary.core.bookshelves import Bookshelves
 from openlibrary.core.follows import PubSub
@@ -445,6 +446,55 @@ def POST(self):
             infogami_login().POST()
 
 
+class otp_service_issue(delegate.page):
+    path = "/account/otp/issue"
+
+    def POST(self):
+        web.header('Content-Type', 'application/json')
+        i = web.input(email="", ip="", challenge_url="", sendmail='true')
+        required_keys = ("email", "ip", "service_ip")
+        i.email = i.email.replace(" ", "+").lower()
+        i.service_ip = web.ctx.env.get('HTTP_X_FORWARDED_FOR')
+        if missing_fields := [k for k in required_keys if not getattr(i, k)]:
+            return delegate.RawText(
+                json.dumps({"error": "missing_keys", "missing_keys": missing_fields})
+            )
+
+        # Challenge currently does not work due to Firewall/Proxy limitations
+        if i.challenge_url and not OTP.verify_service(i.service_ip, i.challenge_url):
+            return delegate.RawText(json.dumps({"error": "challenge_failed"}))
+        if error := OTP.is_ratelimited(service_ip=i.service_ip, email=i.email, ip=i.ip):
+            return delegate.RawText(json.dumps(error))
+
+        otp = OTP.generate(i.service_ip, i.email, i.ip)
+        if i.sendmail.lower() == 'true':
+            web.sendmail(
+                config.from_address,
+                i.email,
+                subject="Your One Time Password",
+                message=web.safestr(f"Your one time password is: {otp.upper()}"),
+            )
+        return delegate.RawText(json.dumps({"success": "issued"}))
+
+
+class otp_service_redeem(delegate.page):
+    path = "/account/otp/redeem"
+
+    def POST(self):
+        web.header('Content-Type', 'application/json')
+        required_keys = ("email", "ip", "service_ip", "otp")
+        i = web.input(email="", ip="", otp="")
+        i.email = i.email.replace(" ", "+").lower()
+        i.service_ip = web.ctx.env.get('HTTP_X_FORWARDED_FOR')
+        if missing_fields := [k for k in required_keys if not getattr(i, k)]:
+            return delegate.RawText(
+                json.dumps({"error": "missing_keys", "missing_keys": missing_fields})
+            )
+        if OTP.is_valid(i.email, i.ip, i.service_ip, i.otp):
+            return delegate.RawText(json.dumps({"success": "redeemed"}))
+        return delegate.RawText(json.dumps({"error": "otp_mismatch"}))
+
+
 class account_login(delegate.page):
     """Account login.
 
