feat: add OAuth2 CSRF state management

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-08 13:28:18 -04:00
parent ca908ee967
commit dbb779b013
2 changed files with 55 additions and 0 deletions
+23
View File
@@ -26,6 +26,29 @@ STATE_TTL = 600 # 10 minutes
_states: dict[str, float] = {}
def make_state() -> str:
"""Generate a CSRF state token, store it with TTL, and return it."""
_purge_states()
token = secrets.token_hex(32)
_states[token] = time.time() + STATE_TTL
return token
def validate_state(state: str) -> bool:
"""Return True if *state* is known and unexpired; always removes it."""
expiry = _states.pop(state, None)
if expiry is None:
return False
return time.time() < expiry
def _purge_states() -> None:
now = time.time()
expired = [k for k, exp in list(_states.items()) if exp < now]
for k in expired:
del _states[k]
class OAuthError(Exception):
"""Raised when the OAuth2 flow fails for any reason."""