initial prototype of OL TOTP auth service

Author: mekarpeles

openlibrary/core/auth.py

@@ -0,0 +1,86 @@
+import hashlib
+import hmac
+import string
+import time
+import requests
+import socket, ssl, json
+from urllib.parse import urlparse
+from infogami import config
+from openlibrary.core import cache
+
+class TimedOneTimePassword:
+
+    VALID_MINUTES = 10
+
+    @staticmethod
+    def shorten(digest: str, 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)
+        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

openlibrary/plugins/upstream/account.py

@@ -33,6 +33,7 @@
 )
 from openlibrary.core import helpers as h
 from openlibrary.core import lending, stats
+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
@@ -454,6 +455,59 @@ def POST(self):
             infogami_login().POST()
 
 
+class otp_service_issue(delegate.page):
+    path = "/account/otp/issue"
+
+    def POST(self):
+        """
+        >>> requests.get("https://staging.openlibrary.org/account/otp/issue", data={"email": "email@example.com", "ip": "127.0.0.1"})
+        """
+        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.