Spaces:
Running
Running
| import hashlib | |
| import secrets | |
| import time | |
| from dataclasses import dataclass, field | |
| SESSION_TTL_SECONDS = 8 * 60 * 60 | |
| def _password_hash(password: str, salt: bytes) -> bytes: | |
| return hashlib.scrypt( | |
| password.encode("utf-8"), | |
| salt=salt, | |
| n=2**14, | |
| r=8, | |
| p=1, | |
| dklen=32, | |
| ) | |
| class RuntimeState: | |
| """Ephemeral setup/admin state. It intentionally resets with the container.""" | |
| configured: bool = False | |
| bootstrap_code: str = "" | |
| password_salt: bytes = b"" | |
| password_digest: bytes = b"" | |
| sessions: dict[str, float] = field(default_factory=dict) | |
| def create_bootstrap_code(self) -> str: | |
| self.bootstrap_code = secrets.token_urlsafe(12) | |
| return self.bootstrap_code | |
| def verify_bootstrap_code(self, provided: str) -> bool: | |
| if not self.bootstrap_code: | |
| return False | |
| return secrets.compare_digest(provided.strip(), self.bootstrap_code) | |
| def consume_bootstrap_code(self, provided: str) -> bool: | |
| valid = self.verify_bootstrap_code(provided) | |
| if valid: | |
| self.bootstrap_code = "" | |
| return valid | |
| def set_password(self, password: str) -> None: | |
| self.password_salt = secrets.token_bytes(16) | |
| self.password_digest = _password_hash(password, self.password_salt) | |
| self.sessions.clear() | |
| def verify_password(self, password: str) -> bool: | |
| if not self.password_salt or not self.password_digest: | |
| return False | |
| candidate = _password_hash(password, self.password_salt) | |
| return secrets.compare_digest(candidate, self.password_digest) | |
| def create_session(self) -> str: | |
| self.prune_sessions() | |
| token = secrets.token_urlsafe(32) | |
| self.sessions[token] = time.time() + SESSION_TTL_SECONDS | |
| return token | |
| def valid_session(self, token: str | None) -> bool: | |
| if not token: | |
| return False | |
| self.prune_sessions() | |
| expiry = self.sessions.get(token) | |
| return bool(expiry and expiry > time.time()) | |
| def revoke_session(self, token: str | None) -> None: | |
| if token: | |
| self.sessions.pop(token, None) | |
| def prune_sessions(self) -> None: | |
| now = time.time() | |
| expired = [token for token, expiry in self.sessions.items() if expiry <= now] | |
| for token in expired: | |
| self.sessions.pop(token, None) | |