WEB-APP/ Optimization/ Build fix/ Hotkey edit mode/ Log rotation/ Form a11y/ E2E non-blocking

utils/web_password_reset_code.py

@@ -0,0 +1,92 @@
+import hmac
+
+from config import LOGIN_CODE_TTL_SEC
+from core.redis_cache import (
+    cache_delete,
+    cache_get,
+    cache_incr,
+    cache_key,
+    cache_set,
+    cache_setnx,
+    redis_connection_ok,
+)
+
+_RESEND_COOLDOWN_SEC = 60.0
+_IP_WINDOW_SEC = 3600.0
+_IP_MAX_SENDS = 40
+_EMAIL_WINDOW_SEC = 3600.0
+_EMAIL_MAX_SENDS = 5
+_EMAIL_VERIFY_WINDOW_SEC = 600.0
+_EMAIL_MAX_VERIFY_ATTEMPTS = 10
+
+
+def _code_key(email_norm: str) -> str:
+    return cache_key("web_pwd_reset_code", email_norm)
+
+
+def _cooldown_key(email_norm: str) -> str:
+    return cache_key("web_pwd_reset_cooldown", email_norm)
+
+
+def _ip_key(ip: str) -> str:
+    return cache_key("web_pwd_reset_send_ip", ip)
+
+
+def _email_send_key(email_norm: str) -> str:
+    return cache_key("web_pwd_reset_email_sends", email_norm)
+
+
+def _email_verify_key(email_norm: str) -> str:
+    return cache_key("web_pwd_reset_email_verify", email_norm)
+
+
+async def redis_ready() -> bool:
+    return await redis_connection_ok()
+
+
+async def try_consume_ip_budget(ip: str) -> bool:
+    if not ip:
+        return True
+    n = await cache_incr(_ip_key(ip), _IP_WINDOW_SEC)
+    return n <= _IP_MAX_SENDS
+
+
+async def try_consume_email_send_budget(email_norm: str) -> bool:
+    if not email_norm:
+        return True
+    n = await cache_incr(_email_send_key(email_norm), _EMAIL_WINDOW_SEC)
+    return n <= _EMAIL_MAX_SENDS
+
+
+async def try_consume_email_verify_budget(email_norm: str) -> bool:
+    if not email_norm:
+        return True
+    n = await cache_incr(_email_verify_key(email_norm), _EMAIL_VERIFY_WINDOW_SEC)
+    return n <= _EMAIL_MAX_VERIFY_ATTEMPTS
+
+
+async def try_acquire_cooldown(email_norm: str) -> bool:
+    return await cache_setnx(_cooldown_key(email_norm), 1, _RESEND_COOLDOWN_SEC)
+
+
+async def release_cooldown(email_norm: str) -> None:
+    await cache_delete(_cooldown_key(email_norm))
+
+
+async def store_code(email_norm: str, code: str) -> bool:
+    return await cache_set(_code_key(email_norm), code, float(LOGIN_CODE_TTL_SEC))
+
+
+async def delete_code(email_norm: str) -> None:
+    await cache_delete(_code_key(email_norm))
+
+
+async def verify_and_consume_code(email_norm: str, code: str) -> bool:
+    key = _code_key(email_norm)
+    stored = await cache_get(key)
+    if not isinstance(stored, str):
+        return False
+    if not hmac.compare_digest(stored.strip(), (code or "").strip()):
+        return False
+    await cache_delete(key)
+    return True