1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
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
|