ollama / app /runtime.py
charlielin0121's picture
Upload 13 files
340ee2f verified
Raw
History Blame Contribute Delete
2.41 kB
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,
)
@dataclass
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)