46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
import time
|
|
|
|
from . import audit
|
|
from .crypto import SecretBox
|
|
from .db import Database
|
|
|
|
MASK = "***"
|
|
|
|
|
|
class SettingsStore:
|
|
def __init__(self, db: Database, box: SecretBox):
|
|
self._db = db
|
|
self._box = box
|
|
|
|
def get(self, key: str, default: str | None = None) -> str | None:
|
|
rows = self._db.query("SELECT value, is_secret FROM settings WHERE key=?", (key,))
|
|
if not rows:
|
|
return default
|
|
value, is_secret = rows[0]["value"], rows[0]["is_secret"]
|
|
return self._box.decrypt(value, key) if is_secret else value
|
|
|
|
def get_int(self, key: str, default: int, minimum: int = 0) -> int:
|
|
try:
|
|
value = int(float(self.get(key) or default))
|
|
except ValueError:
|
|
value = default
|
|
return max(value, minimum)
|
|
|
|
def is_set(self, key: str) -> bool:
|
|
return bool(self._db.query("SELECT 1 FROM settings WHERE key=?", (key,)))
|
|
|
|
def set(self, key: str, value: str, user: str, secret: bool = False) -> None:
|
|
existed = self.is_set(key)
|
|
old = self.get(key)
|
|
stored = self._box.encrypt(value, key) if secret else value
|
|
self._db.execute(
|
|
"INSERT INTO settings(key, value, is_secret, updated_at) VALUES (?,?,?,?) "
|
|
"ON CONFLICT(key) DO UPDATE SET value=excluded.value, "
|
|
"is_secret=excluded.is_secret, updated_at=excluded.updated_at",
|
|
(key, stored, int(secret), int(time.time())),
|
|
)
|
|
if secret:
|
|
audit.record(self._db, user, "setting.set", key, MASK if existed else None, MASK)
|
|
else:
|
|
audit.record(self._db, user, "setting.set", key, old, value)
|