86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
|
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
|