Spaces:
Sleeping
Sleeping
File size: 2,917 Bytes
990895d | 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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """Persist bootstrap credentials and session-version state.
Configuration updates are serialized and atomically replace `config.json`.
"""
import json
from datetime import datetime, timezone
from app.fsutil import atomic_write_text, file_lock, read_json
from app.models import StoredConfig
from app.paths import Paths
from app.security_passwords import hash_password, verify_password
def utc_now() -> datetime:
"""Return an aware UTC timestamp."""
return datetime.now(timezone.utc)
class ConfigStore:
"""Read and mutate the durable single-user configuration."""
def __init__(self, paths: Paths) -> None:
self.paths = paths
def bootstrap(self, initial_password: str | None) -> StoredConfig | None:
"""Create initial configuration only when a password is available."""
with file_lock(self.paths.config):
existing = read_json(self.paths.config)
if existing is not None:
return StoredConfig.model_validate(existing)
if not initial_password:
return None
now = utc_now()
config = StoredConfig(
password_hash=hash_password(initial_password),
session_version=1,
created_at=now,
updated_at=now,
)
self._write_unlocked(config)
return config
def load(self) -> StoredConfig | None:
"""Load configuration, or None when setup has not occurred."""
value = read_json(self.paths.config)
return StoredConfig.model_validate(value) if value is not None else None
def password_matches(self, password: str) -> bool:
"""Check a candidate password against current configuration."""
config = self.load()
return bool(config and verify_password(password, config.password_hash))
def change_password(self, old_password: str, new_password: str) -> bool:
"""Rotate the password and invalidate all prior sessions."""
with file_lock(self.paths.config):
value = read_json(self.paths.config)
if value is None:
return False
config = StoredConfig.model_validate(value)
if not verify_password(old_password, config.password_hash):
return False
updated = config.model_copy(
update={
"password_hash": hash_password(new_password),
"session_version": config.session_version + 1,
"updated_at": utc_now(),
}
)
self._write_unlocked(updated)
return True
def _write_unlocked(self, config: StoredConfig) -> None:
payload = json.dumps(
config.model_dump(mode="json"),
ensure_ascii=False,
indent=2,
)
atomic_write_text(self.paths.config, payload + "\n")
|