133 lines
6.0 KiB
Python
133 lines
6.0 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location("settings_guard", Path(__file__).with_name("check-settings.py"))
|
|
guard = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(guard)
|
|
|
|
SAFE = {"localAuthBypassEnabled": False, "apiRestricted": True, "autoUpdateEnabled": False}
|
|
KEY = "ab" * 32
|
|
|
|
|
|
class SettingsTests(unittest.TestCase):
|
|
def test_safe_escaped_names_are_semantically_validated(self):
|
|
raw = b'{"localAuthBypass\\u0045nabled":false,"api\\u0052estricted":true,"autoUpdateEnabled":false}'
|
|
self.assertEqual(guard.validate(raw), SAFE)
|
|
|
|
def test_unsafe_flags_defaults_and_wrong_types_refused(self):
|
|
for field, bad_values in {
|
|
"localAuthBypassEnabled": [True, None, 0, 1, "false"],
|
|
"apiRestricted": [False, None, 0, 1, "true"],
|
|
"autoUpdateEnabled": [True, None, 0, 1, "false"],
|
|
}.items():
|
|
for value in bad_values:
|
|
with self.subTest(field=field, value=value):
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.validate(json.dumps({**SAFE, field: value}).encode())
|
|
absent = dict(SAFE)
|
|
del absent[field]
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.validate(json.dumps(absent).encode())
|
|
|
|
def test_original_escaped_bypasses_refused(self):
|
|
for raw in [
|
|
b'{"localAuthBypass\\u0045nabled":true,"apiRestricted":true,"autoUpdateEnabled":false}',
|
|
b'{"localAuthBypassEnabled":false,"api\\u0052estricted":false,"autoUpdateEnabled":false}',
|
|
]:
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.validate(raw)
|
|
|
|
def test_redirect_and_legacy_key_properties_refused_even_null(self):
|
|
for key in ("userPath", "apiKey", "apiKeyPath"):
|
|
for value in (None, "", "/another-location"):
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.validate(json.dumps({**SAFE, key: value}).encode())
|
|
|
|
def test_ambiguous_non_json_and_oversized_inputs_refused(self):
|
|
for raw in [b'{}', b'[]', b'null', b'{"x":NaN}', b'{"x":Infinity}',
|
|
b'{"x":1,"\\u0078":2}', b'{"x":{},"x":{}}',
|
|
b'{"x":1,}', b'# comment\n{}', b'{}{}', b'\xff',
|
|
b' ' * (guard.MAX_SETTINGS_BYTES + 1)]:
|
|
with self.subTest(raw=raw[:40]):
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.validate(raw)
|
|
|
|
def test_snapshot_is_private_independent_and_preserves_original(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source, dest = Path(tmp) / "operator.json", Path(tmp) / "runtime.json"
|
|
original = json.dumps({**SAFE, "maxStorageCapacity": 21474836480}).encode()
|
|
source.write_bytes(original)
|
|
guard.snapshot(source, dest)
|
|
self.assertEqual(source.read_bytes(), original)
|
|
self.assertEqual(stat.S_IMODE(dest.stat().st_mode), 0o600)
|
|
self.assertEqual(json.loads(dest.read_bytes())["maxStorageCapacity"], 21474836480)
|
|
source.write_text('{"localAuthBypassEnabled":true}')
|
|
self.assertEqual(json.loads(dest.read_bytes())["localAuthBypassEnabled"], False)
|
|
with self.assertRaises(FileExistsError):
|
|
guard.snapshot(dest, dest)
|
|
|
|
def test_source_symlinks_and_output_collisions_refused(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source, dest, link = [Path(tmp) / x for x in ("source", "dest", "link")]
|
|
source.write_text(json.dumps(SAFE))
|
|
link.symlink_to(source)
|
|
with self.assertRaises(OSError):
|
|
guard.snapshot(link, dest)
|
|
dest.symlink_to(source)
|
|
with self.assertRaises(OSError):
|
|
guard.snapshot(source, dest)
|
|
self.assertEqual(json.loads(source.read_bytes()), SAFE)
|
|
|
|
def test_invalid_settings_do_not_create_snapshot(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source, dest = Path(tmp) / "source", Path(tmp) / "dest"
|
|
source.write_text('{"apiRestricted":false}')
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.snapshot(source, dest)
|
|
self.assertFalse(dest.exists())
|
|
|
|
def test_key_is_private_idempotent_and_never_rotated(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
key = Path(tmp) / "apikey.txt"
|
|
guard.ensure_api_key(key, KEY)
|
|
self.assertEqual(key.read_text(), KEY)
|
|
self.assertEqual(stat.S_IMODE(key.stat().st_mode), 0o600)
|
|
before = key.stat().st_mtime_ns
|
|
guard.ensure_api_key(key, KEY)
|
|
self.assertEqual(key.stat().st_mtime_ns, before)
|
|
for value in ("", "x" * 64, KEY + "\n", "cd" * 32):
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.ensure_api_key(key, value)
|
|
self.assertEqual(key.read_text(), KEY)
|
|
|
|
def test_existing_key_permissions_are_not_silently_repaired(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
key = Path(tmp) / "key"
|
|
key.write_text(KEY)
|
|
key.chmod(0o644)
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.ensure_api_key(key, KEY)
|
|
self.assertEqual(key.read_text(), KEY)
|
|
self.assertEqual(stat.S_IMODE(key.stat().st_mode), 0o644)
|
|
|
|
def test_key_symlink_and_fifo_refused_without_blocking(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
target, link = Path(tmp) / "target", Path(tmp) / "link"
|
|
target.write_text(KEY)
|
|
link.symlink_to(target)
|
|
with self.assertRaises(OSError):
|
|
guard.ensure_api_key(link, KEY)
|
|
fifo = Path(tmp) / "fifo"
|
|
os.mkfifo(fifo)
|
|
with self.assertRaises(guard.UnsafeSettings):
|
|
guard.ensure_api_key(fifo, KEY)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|