better secrets admin
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
import base64
|
||||
import secrets
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
def _fernet_key():
|
||||
return base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii")
|
||||
|
||||
|
||||
def _default_secrets():
|
||||
return {
|
||||
"runtime": {
|
||||
"secret_key": secrets.token_urlsafe(48),
|
||||
"application_key": secrets.token_urlsafe(48),
|
||||
"database_uri": "sqlite:///app.db",
|
||||
},
|
||||
"initial_data": {
|
||||
"admin_password": secrets.token_urlsafe(18),
|
||||
"admin_group": secrets.token_urlsafe(18),
|
||||
"user_password": secrets.token_urlsafe(18),
|
||||
"user_group": secrets.token_urlsafe(18),
|
||||
},
|
||||
"fernet": {
|
||||
"key": _fernet_key(),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _merge_missing(target, defaults):
|
||||
changed = False
|
||||
for key, value in defaults.items():
|
||||
if isinstance(value, dict):
|
||||
current = target.get(key)
|
||||
if not isinstance(current, dict):
|
||||
target[key] = {}
|
||||
current = target[key]
|
||||
changed = True
|
||||
if _merge_missing(current, value):
|
||||
changed = True
|
||||
elif not target.get(key):
|
||||
target[key] = value
|
||||
changed = True
|
||||
return changed
|
||||
|
||||
|
||||
def load_secrets(instance_path):
|
||||
instance_dir = Path(instance_path)
|
||||
instance_dir.mkdir(parents=True, exist_ok=True)
|
||||
secrets_path = instance_dir / "secrets.yaml"
|
||||
existed = secrets_path.exists()
|
||||
if existed:
|
||||
secrets_path.chmod(0o600)
|
||||
else:
|
||||
secrets_path.touch(mode=0o600)
|
||||
|
||||
if existed:
|
||||
loaded = yaml.safe_load(secrets_path.read_text(encoding="utf-8")) or {}
|
||||
if not isinstance(loaded, dict):
|
||||
raise ValueError(f"{secrets_path} must contain a YAML mapping")
|
||||
else:
|
||||
loaded = {}
|
||||
|
||||
defaults = _default_secrets()
|
||||
legacy_files = {
|
||||
"secret_key": instance_dir / "secret_key",
|
||||
"application_key": instance_dir / "application_key",
|
||||
}
|
||||
migrated_paths = []
|
||||
for key, legacy_path in legacy_files.items():
|
||||
if legacy_path.exists() and not loaded.get("runtime", {}).get(key):
|
||||
defaults["runtime"][key] = legacy_path.read_text(encoding="utf-8").strip()
|
||||
migrated_paths.append(legacy_path)
|
||||
|
||||
changed = _merge_missing(loaded, defaults)
|
||||
if changed or not existed:
|
||||
secrets_path.write_text(
|
||||
yaml.safe_dump(loaded, sort_keys=False),
|
||||
encoding="utf-8",
|
||||
)
|
||||
secrets_path.chmod(0o600)
|
||||
for legacy_path in migrated_paths:
|
||||
legacy_path.unlink()
|
||||
return loaded
|
||||
Reference in New Issue
Block a user