diff options
Diffstat (limited to 'packages/meshbay-common/tests')
4 files changed, 225 insertions, 29 deletions
diff --git a/packages/meshbay-common/tests/test_admin_subject_parity.py b/packages/meshbay-common/tests/test_admin_subject_parity.py new file mode 100644 index 0000000..7a229a9 --- /dev/null +++ b/packages/meshbay-common/tests/test_admin_subject_parity.py @@ -0,0 +1,145 @@ +""" +The subjects of multi-value admin operations are byte-identical in the browser and +in Python. + +The subject is what the operator's signature covers of a request, and each side +builds it on its own — the node from the request it stored, the client from what +the person asked for. A one-byte disagreement does not weaken anything (the client +refuses to sign), but it makes the operation impossible from a browser, and nothing +else in the suite crosses this boundary. + +Skipped when node is unavailable; that is a coverage gap, not a pass. +""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest +from meshbay_common.adminop import ( + group_attach_subject, + invite_create_subject, + root_add_subject, + secret_digest, + structured_subject, + tmdb_config_subject, +) + +CRYPTO_JS = (Path(__file__).resolve().parents[2] + / "meshbay-hub" / "src" / "meshbay_hub" / "static" / "crypto.js") + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not CRYPTO_JS.exists(), + reason="node or crypto.js unavailable — parity cannot be checked", +) + +ROOT_ADD = [ + ("/srv/Films", "", "generic", False, False), + ("/srv/Films", "Films", "video", True, True), + ("C:\\Users\\me\\Share", "Partagé", "photo", True, False), + ('/srv/a "quoted", odd:name|x', "名前", "audio", False, True), + ("/srv/tab\there\nnewline\x01ctl", "é", "generic", True, False), +] +GROUP_ATTACH = [ + ("photos", "/srv/photos", True), + ("famille-été", "/mnt/disque externe/Photos", False), +] +INVITE_CREATE = [ + ("0f8fad5b-d9cb-469f-a165-70867728950e", ""), + ("0f8fad5b-d9cb-469f-a165-70867728950e", "Élodie \"E\" 🙂"), +] +TMDB_CONFIG = [ + (None, None), ("", None), (None, ""), ("", ""), + ("eyJhbGciOiJIUzI1NiJ9.token", "fr-FR"), + ("abc", "keep"), +] + +_HARNESS = r""" +const fs = require('fs'); +globalThis.window = {}; +const src = fs.readFileSync(process.argv[2], 'utf8'); +const M = new Function(src + '\nreturn { rootAddSubject, groupAttachSubject, ' + + 'inviteCreateSubject, tmdbConfigSubject };')(); +const v = JSON.parse(fs.readFileSync(process.argv[3], 'utf8')); +(async () => { + const out = { + root_add: v.root_add.map((a) => M.rootAddSubject(...a)), + group_attach: v.group_attach.map((a) => M.groupAttachSubject(...a)), + invite_create: v.invite_create.map((a) => M.inviteCreateSubject(...a)), + tmdb_config: [], + }; + for (const a of v.tmdb_config) out.tmdb_config.push(await M.tmdbConfigSubject(...a)); + process.stdout.write(JSON.stringify(out)); +})(); +""" + + +@pytest.fixture(scope="module") +def js(tmp_path_factory): + d = tmp_path_factory.mktemp("subject-parity") + (d / "harness.js").write_text(_HARNESS, encoding="utf-8") + (d / "vectors.json").write_text(json.dumps({ + "root_add": ROOT_ADD, "group_attach": GROUP_ATTACH, + "invite_create": INVITE_CREATE, "tmdb_config": TMDB_CONFIG, + }), encoding="utf-8") + proc = subprocess.run( + ["node", str(d / "harness.js"), str(CRYPTO_JS), str(d / "vectors.json")], + capture_output=True, text=True, encoding="utf-8", timeout=60) + if proc.returncode != 0: + pytest.fail(f"node harness failed:\n{proc.stderr}") + return json.loads(proc.stdout) + + +def _bytes(s: str) -> bytes: + return s.encode("utf-8") + + +@pytest.mark.parametrize("i,args", list(enumerate(ROOT_ADD))) +def test_root_add_subject_parity(i, args, js): + assert _bytes(js["root_add"][i]) == _bytes(root_add_subject(*args)) + + +@pytest.mark.parametrize("i,args", list(enumerate(GROUP_ATTACH))) +def test_group_attach_subject_parity(i, args, js): + assert _bytes(js["group_attach"][i]) == _bytes(group_attach_subject(*args)) + + +@pytest.mark.parametrize("i,args", list(enumerate(INVITE_CREATE))) +def test_invite_create_subject_parity(i, args, js): + assert _bytes(js["invite_create"][i]) == _bytes(invite_create_subject(*args)) + + +@pytest.mark.parametrize("i,args", list(enumerate(TMDB_CONFIG))) +def test_tmdb_config_subject_parity(i, args, js): + assert _bytes(js["tmdb_config"][i]) == _bytes(tmdb_config_subject(*args)) + + +def test_every_value_changes_the_subject(): + base = ("/srv/Films", "Films", "video", False, False) + variants = {root_add_subject(*base)} + for i, other in enumerate(("/srv/Other", "Other", "audio", True, True)): + args = list(base) + args[i] = other + variants.add(root_add_subject(*args)) + assert len(variants) == 6 + + +def test_unchanged_cleared_and_set_are_three_subjects(): + assert len({tmdb_config_subject(None, None), tmdb_config_subject("", None), + tmdb_config_subject("t", None)}) == 3 + assert len({tmdb_config_subject(None, None), tmdb_config_subject(None, ""), + tmdb_config_subject(None, "fr-FR")}) == 3 + + +def test_the_token_is_never_written_into_the_subject(): + token = "eyJhbGciOiJIUzI1NiJ9.a-real-looking-secret" + assert token not in tmdb_config_subject(token, "fr-FR") + assert secret_digest(token).startswith("sha256:") + + +def test_a_crafted_field_cannot_impersonate_another(): + # Under a naive "path|name" join these two would collide. + a = structured_subject({"path": "/a|name=b", "name": ""}) + b = structured_subject({"path": "/a", "name": "b"}) + assert a != b diff --git a/packages/meshbay-common/tests/test_groupbox.py b/packages/meshbay-common/tests/test_groupbox.py index 65d8ce6..6e687ea 100644 --- a/packages/meshbay-common/tests/test_groupbox.py +++ b/packages/meshbay-common/tests/test_groupbox.py @@ -45,8 +45,8 @@ def test_purposes_are_separate_key_spaces(gek): The reason there are two info strings rather than one key reused. An ack sealed under the index subkey would otherwise be openable by anything - holding the index subkey, which is the confusion `GroupIndex.serialize()`'s - "the index as chunk 0 of a virtual index file" creates for chunk keys. + holding the index subkey — the confusion that treating "the index as chunk 0 of + a virtual index file" would create for chunk keys. """ assert group_key(gek, PURPOSE_INDEX) != group_key(gek, PURPOSE_ACK) sealed = seal(gek, PURPOSE_INDEX, "index_sync", "g1", PAYLOAD) diff --git a/packages/meshbay-common/tests/test_portable_name_parity.py b/packages/meshbay-common/tests/test_portable_name_parity.py new file mode 100644 index 0000000..3a491a7 --- /dev/null +++ b/packages/meshbay-common/tests/test_portable_name_parity.py @@ -0,0 +1,78 @@ +""" +The browser makes a name writable everywhere exactly as Python does. + +`static/portable-name.js` renames a file at the moment a member saves it; +`meshbay_common.paths.sanitize_for_download` is the same rule in Python. Two +copies of a rule that differ decide differently which files get renamed, so the +real module runs under node here against the real Python. + +Skipped when node is unavailable; that is a coverage gap, not a pass. +""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest +from meshbay_common.paths import is_portable_name, sanitize_for_download + +PORTABLE_JS = (Path(__file__).resolve().parents[2] + / "meshbay-hub" / "src" / "meshbay_hub" / "static" / "portable-name.js") + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not PORTABLE_JS.exists(), + reason="node or portable-name.js unavailable — parity cannot be checked", +) + +NAMES = [ + "plain.txt", "Réunion 12:30.pdf", 'a<b>c:d"e/f\\g|h?i*j.txt', "tab\tnew\nline", + "ends with dot.", "ends with space ", "trailing . . ", "CON", "con.txt", "aux.tar.gz", + "COM1", "com10.txt", "LPT9.log", "nul.", ".", "..", "", " ", ".bashrc", "...", + "名前:ファイル.mkv", "emoji 🙂?.png", "\x01\x1f.bin", "prn .txt", "Con", + "a.b.c", "COM1.", "normal name (2).mp4", +] + +_HARNESS = r""" +const fs = require('fs'); +const src = fs.readFileSync(process.argv[2], 'utf8').replace(/^export /gm, ''); +const M = new Function(src + '\nreturn { portableName, portablePath };')(); +const names = JSON.parse(fs.readFileSync(process.argv[3], 'utf8')); +process.stdout.write(JSON.stringify({ + names: names.map((n) => M.portableName(n)), + path: M.portablePath('Top:Folder/sub//aux.txt'), +})); +""" + + +@pytest.fixture(scope="module") +def js(tmp_path_factory): + d = tmp_path_factory.mktemp("portable") + (d / "harness.js").write_text(_HARNESS, encoding="utf-8") + (d / "names.json").write_text(json.dumps(NAMES), encoding="utf-8") + proc = subprocess.run( + ["node", str(d / "harness.js"), str(PORTABLE_JS), str(d / "names.json")], + capture_output=True, text=True, encoding="utf-8", timeout=60) + if proc.returncode != 0: + pytest.fail(f"node harness failed:\n{proc.stderr}") + return json.loads(proc.stdout) + + +@pytest.mark.parametrize("i,name", list(enumerate(NAMES))) +def test_the_browser_renames_as_python_does(i, name, js): + assert js["names"][i] == sanitize_for_download(name) + + +@pytest.mark.parametrize("name", NAMES) +def test_what_comes_out_can_be_written_everywhere(name): + out = sanitize_for_download(name) + assert is_portable_name(out), (name, out) + assert sanitize_for_download(out) == out + + +def test_a_portable_name_is_left_alone(): + assert sanitize_for_download("plain.txt") == "plain.txt" + + +def test_a_path_is_made_portable_segment_by_segment(js): + assert js["path"] == "Top_Folder/sub//aux_.txt" diff --git a/packages/meshbay-common/tests/test_webcrypto.py b/packages/meshbay-common/tests/test_webcrypto.py index fbffdc1..ffe3f36 100644 --- a/packages/meshbay-common/tests/test_webcrypto.py +++ b/packages/meshbay-common/tests/test_webcrypto.py @@ -17,17 +17,6 @@ def test_aes_roundtrip(): assert decrypt_chunk_aes(key, nonce, ct) == data -def test_aes_key_distinct_from_chacha_key(): - """AES and ChaCha20 keys for the same chunk must differ.""" - from meshbay_common.crypto import chunk_key as chacha_key - gek = generate_gek() - data = os.urandom(100) - fh = blake3.blake3(data).digest() - aes_k = chunk_key_aes(gek, fh, 0) - chacha_k = chacha_key(gek, fh, 0) - assert aes_k != chacha_k - - def test_aes_wrong_key_rejected(): gek = generate_gek() data = b"private content" @@ -76,19 +65,3 @@ def test_aes_gek_wrap_wrong_key_rejected(): bundle = wrap_gek_aes(gek, pk_to_raw(sk_a.public_key())) with pytest.raises(Exception): unwrap_gek_aes(bundle, sk_to_raw(sk_b), pk_to_raw(sk_b.public_key())) - - -def test_aes_gek_wrap_differs_from_chacha_wrap(): - from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey - from meshbay_common.crypto import ( - pk_to_raw, - wrap_gek, - wrap_gek_aes, - ) - gek = generate_gek() - sk = X25519PrivateKey.generate() - pk_raw = pk_to_raw(sk.public_key()) - - bundle_aes = wrap_gek_aes(gek, pk_raw) - bundle_chacha = wrap_gek(gek, pk_raw) - assert bundle_aes["wrapped_b64"] != bundle_chacha["wrapped_b64"] |