Add OTP sendmail

[mekarpeles]

The OTP must first be sanitized by a regex.

from openlibrary.core import cache

class thirdparty_otp(delegate.page):
    path = "/account/otp/send"

    def POST(self):
        import time
        i = web.input(otp=None, email=None, ts=None, ip=None)
        if not (i.ip and i.email and i.ts and i.otp):
            return web.badrequest()

        mc = cache.get_memcache()

        # IP-based throttle                                                                                                                                                       
        ip_key = f"otp:ip:{i.ip}"
        if not mc.add(ip_key, 1, expires=60):  # 60s TTL                                                                                                                          
            return web.forbidden("This IP must wait before requesting again")

        # Email-based throttle                                                                                                                                                    
        email_key = f"otp:email:{i.email}"
        if not mc.add(email_key, 1, expires=60):
            return web.forbidden("This email must wait before requesting again")

        web.sendmail(                                                                                                                                                            
            config.from_address,                                                                                                                                                 
            i.email,                                                                                                                                                             
            subject="Your One Time Password",                                                                                                                                    
            message=web.safestr(f"Your one time password is: {i.otp}"),                                                                                                          
        )                                                                                                                                                                        
        return delegate.RawText(f"OTP Successfully Sent")

[mekarpeles]

We may be able to move all of the auth portion to Open Library with something like:

class thirdparty_otp(delegate.page):
    path = "/account/otp/send"

    OTP_VALID_MINUTES = 10

    @staticmethod
    def generate(email: str, ip_address: str, seed: str, issued_minute=None) -> str:
        """Generates an OTP for a given email, IP address, and timestamp."""
        ts = issued_minute or int(time.time() // 60)
        print(f"ts {ts}")
        payload = f"{email}:{ip_address}:{ts}".encode()
        return hmac.new(str(seed).encode('utf-8'), payload, hashlib.sha256).hexdigest()

    def GET(self):
        i = web.input(email=None, ip=None, seed=None, otp=None)
        # XXX seed needs to be POSTed or a header!                                                                                                                                       
        if not all([i.email, i.ip, i.seed, i.otp]):
            return web.badrequest("Missing required parameters")
        email = i.email.replace(" ", "+")
        now_minute = int(time.time() // 60)
        for delta in range(self.OTP_VALID_MINUTES):
            ts = now_minute - delta
            expected_otp = self.generate(email, i.ip, i.seed, ts)
            print(f"Expecting {expected_otp} for {ts}")
            if hmac.compare_digest(i.otp, expected_otp):
                return delegate.RawText(f"Success! {i.otp} {expected_otp}")
        return delegate.RawText(f"Failure! {i.otp} {expected_otp}")

    def POST(self):
        i = web.input(email=None, ip=None, seed=None)
        if not (i.ip and i.email and i.seed):
            return web.badrequest()
        print(f"email: {i.email}")
        mc = cache.get_memcache()

        # IP-based throttle                                                                                                                                                              
        ip_key = f"otp:ip:{i.ip}"
        if not mc.add(ip_key, 1, expires=60):  # 60s TTL                                                                                                                                 
            return web.forbidden("This IP must wait before requesting again")

        # Email-based throttle                                                                                                                                                           
        email_key = f"otp:email:{i.email}"
        if not mc.add(email_key, 1, expires=60):
            return web.forbidden("This email must wait before requesting again")

        otp = self.generate(i.email, i.ip, i.seed)
        web.sendmail(                                                                                                                                                                   
            config.from_address,                                                                                                                                                        
            i.email,                                                                                                                                                                    
            subject="Your One Time Password",                                                                                                                                           
            message=web.safestr(f"Your one time password is: {i.otp}"),                                                                                                                 
        )                                                                                                                                                                               
        return delegate.RawText(f"Success! {otp} {mc.get(email_key)} {mc.get(ip_key)}")

Testable with:

fetch("/account/otp/send", {
  method: "POST",
  headers: { "Content-Type": "application/x-www-form-urlencoded" },
  body: new URLSearchParams({
    seed: "123456",
    email: "michael.karpeles+test1234@gmail.com",
    ip: "203.0.113.7"
  })
}).then(r => r.text());

[ronibhakta1]

This issue has been solved in PR #91