Spaces:
Running
Running
File size: 1,624 Bytes
2f1e7ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | """Opaque Redis sessions contain identities, never submitted access credentials."""
import json
import os
import secrets
from dataclasses import asdict
from functools import lru_cache
from src.services.auth import AuthError, Principal, token_digest
SESSION_COOKIE = "__Host-brainapi-session"
class ConsoleSessions:
def __init__(self, redis):
self.redis = redis
def _key(self, cookie):
return "system:console-session:" + token_digest(cookie)
def create(self, principal):
cookie = secrets.token_urlsafe(32)
csrf = secrets.token_urlsafe(32)
ttl = max(300, min(int(os.getenv("CONSOLE_SESSION_TTL_SECONDS", "28800")), 86400))
record = asdict(principal)
record["permissions"] = sorted(principal.permissions)
self.redis.set(self._key(cookie), json.dumps({"principal": record, "csrf": csrf}), ex=ttl)
return cookie, csrf, ttl
def resolve(self, cookie, service):
if not cookie:
raise AuthError()
raw = self.redis.get(self._key(cookie))
if not raw:
raise AuthError()
record = json.loads(raw)
snapshot = record["principal"]
snapshot["permissions"] = frozenset(snapshot["permissions"])
principal = Principal(**snapshot)
principal = service.revalidate(principal)
return principal, record["csrf"]
def revoke(self, cookie):
if cookie:
self.redis.delete(self._key(cookie))
@lru_cache(maxsize=1)
def get_console_sessions():
from src.lib.redis.client import RedisClient
return ConsoleSessions(RedisClient().client)
|