123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate a strict-JSON configuration and write the exact runtime snapshot.
|
|
|
|
Never repair the operator's file. Refuse ambiguous or unsupported configuration
|
|
rather than checking text that Core can interpret differently.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import sys
|
|
|
|
MAX_SETTINGS_BYTES = 1024 * 1024
|
|
|
|
|
|
class UnsafeSettings(ValueError):
|
|
pass
|
|
|
|
|
|
def unique_object(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
if key in result:
|
|
raise UnsafeSettings("duplicate JSON property")
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def reject_constant(_value):
|
|
raise UnsafeSettings("non-finite JSON number")
|
|
|
|
|
|
def validate(raw):
|
|
if len(raw) > MAX_SETTINGS_BYTES:
|
|
raise UnsafeSettings("settings exceed 1 MiB")
|
|
try:
|
|
settings = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object,
|
|
parse_constant=reject_constant)
|
|
except (UnicodeError, json.JSONDecodeError, RecursionError) as exc:
|
|
raise UnsafeSettings("settings must be strict UTF-8 JSON") from exc
|
|
if not isinstance(settings, dict):
|
|
raise UnsafeSettings("settings must be a JSON object")
|
|
# Explicit values avoid default differences between mainnet/test chains.
|
|
if settings.get("localAuthBypassEnabled") is not False:
|
|
raise UnsafeSettings("localAuthBypassEnabled must be explicitly false")
|
|
if settings.get("apiRestricted") is not True:
|
|
raise UnsafeSettings("apiRestricted must be explicitly true")
|
|
# The launched document must be the document checked here. Core's userPath
|
|
# can otherwise redirect settings loading to a different document.
|
|
for key in ("userPath", "apiKey", "apiKeyPath"):
|
|
if key in settings:
|
|
raise UnsafeSettings("settings contain a managed or redirected property")
|
|
if settings.get("autoUpdateEnabled") is not False:
|
|
raise UnsafeSettings("autoUpdateEnabled must be explicitly false for the pinned package")
|
|
return settings
|
|
|
|
|
|
def snapshot(source, destination):
|
|
fd = os.open(source, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
with os.fdopen(fd, "rb") as stream:
|
|
if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
|
|
raise UnsafeSettings("settings must be a regular file")
|
|
raw = stream.read(MAX_SETTINGS_BYTES + 1)
|
|
settings = validate(raw)
|
|
encoded = (json.dumps(settings, ensure_ascii=True, allow_nan=False) + "\n").encode("ascii")
|
|
# Caller creates a private temporary directory. Never follow/overwrite an
|
|
# existing output; failure leaves the operator's original untouched.
|
|
fd = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(encoded)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except BaseException:
|
|
os.unlink(destination)
|
|
raise
|
|
|
|
|
|
def ensure_api_key(path, value):
|
|
if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
|
|
raise UnsafeSettings("missing or invalid managed API key")
|
|
try:
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
except FileNotFoundError:
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, "wb") as stream:
|
|
stream.write(value.encode("ascii"))
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
return
|
|
with os.fdopen(fd, "rb") as stream:
|
|
mode = os.fstat(stream.fileno()).st_mode
|
|
if not stat.S_ISREG(mode) or stat.S_IMODE(mode) & 0o077:
|
|
raise UnsafeSettings("API key must be a private regular file")
|
|
if stream.read(65) != value.encode("ascii"):
|
|
raise UnsafeSettings("existing API key differs; explicit recovery required")
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 4:
|
|
print("usage: check-settings.py SOURCE NEW_SNAPSHOT API_KEY_FILE", file=sys.stderr)
|
|
return 64
|
|
snapshot_created = False
|
|
try:
|
|
snapshot(sys.argv[1], sys.argv[2])
|
|
snapshot_created = True
|
|
ensure_api_key(sys.argv[3], os.environ.get("QORTAL_API_KEY", ""))
|
|
except (OSError, ValueError, RecursionError):
|
|
if snapshot_created:
|
|
try:
|
|
os.unlink(sys.argv[2])
|
|
except FileNotFoundError:
|
|
pass
|
|
# Never print source values, parser snippets, API keys or environment.
|
|
print("Qortal settings refused: require strict JSON, unique keys, explicit safe flags, "
|
|
"no settings/API-key redirection and a matching managed key; see the package README.", file=sys.stderr)
|
|
return 78
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|