operator
242 lines
10 KiB
Python
242 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""test_muse_session_bind.py — Per-session credential isolation (P3).
|
|
|
|
Covers: session dir layout (symlinks + private auth copy), newest-wins
|
|
save-back that never touches the global auth.json, dead-only reap,
|
|
exec env, and the resume-pool session binding record.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
REPO_ROOT = Path("/home/super/Projects/NetVM")
|
|
BIN_DIR = REPO_ROOT / "bin"
|
|
sys.path.insert(0, str(BIN_DIR))
|
|
|
|
import muse_session_bind as b
|
|
|
|
|
|
def _mkconfig(root):
|
|
"""Fake global config: profile creds + global files."""
|
|
cfg = os.path.join(root, "config")
|
|
os.makedirs(os.path.join(cfg, "accounts", "alice"))
|
|
with open(os.path.join(cfg, "accounts", "alice", "auth.json"), "wb") as f:
|
|
f.write(b"TOKEN-ALICE")
|
|
with open(os.path.join(cfg, "auth.json"), "wb") as f:
|
|
f.write(b"GLOBAL-TOKEN")
|
|
with open(os.path.join(cfg, "settings.json"), "w") as f:
|
|
f.write("{}")
|
|
with open(os.path.join(cfg, "notes.txt"), "w") as f:
|
|
f.write("hi")
|
|
return cfg
|
|
|
|
|
|
class TestBuild(unittest.TestCase):
|
|
def test_layout_links_and_private_copy(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
parent = os.path.join(td, "run")
|
|
os.makedirs(parent)
|
|
sess = b.build_session_dir(
|
|
parent, 4242, cfg,
|
|
os.path.join(cfg, "accounts", "alice", "auth.json"),
|
|
"alice")
|
|
muse = os.path.join(sess, "muse")
|
|
# Everything but auth.json is a symlink to global.
|
|
self.assertTrue(os.path.islink(os.path.join(muse, "settings.json")))
|
|
self.assertTrue(os.path.islink(os.path.join(muse, "notes.txt")))
|
|
self.assertTrue(os.path.islink(os.path.join(muse, "accounts")))
|
|
# auth.json is a real file with the profile bytes, 0600.
|
|
auth = os.path.join(muse, "auth.json")
|
|
self.assertFalse(os.path.islink(auth))
|
|
with open(auth, "rb") as f:
|
|
self.assertEqual(f.read(), b"TOKEN-ALICE")
|
|
self.assertEqual(stat.S_IMODE(os.stat(auth).st_mode), 0o600)
|
|
bind = json.load(open(os.path.join(sess, "bind.json")))
|
|
self.assertEqual((bind["profile"], bind["pid"]),
|
|
("alice", 4242))
|
|
|
|
def test_missing_creds_refused(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
with self.assertRaises(ValueError):
|
|
b.build_session_dir(td, 1, cfg,
|
|
os.path.join(td, "nope.json"), "alice")
|
|
|
|
def test_live_slot_refused_stale_slot_wiped(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
auth = os.path.join(cfg, "accounts", "alice", "auth.json")
|
|
sess = b.build_session_dir(td, 99, cfg, auth, "alice")
|
|
with mock.patch.object(b, "session_liveness",
|
|
return_value="live"):
|
|
with self.assertRaises(RuntimeError):
|
|
b.build_session_dir(td, 99, cfg, auth, "alice")
|
|
with mock.patch.object(b, "session_liveness",
|
|
return_value="dead"):
|
|
sess2 = b.build_session_dir(td, 99, cfg, auth, "alice")
|
|
self.assertEqual(sess2, sess)
|
|
self.assertTrue(os.path.isfile(
|
|
os.path.join(sess2, "muse", "auth.json")))
|
|
|
|
|
|
class TestSaveBack(unittest.TestCase):
|
|
def _bound(self, td, pid=777):
|
|
cfg = _mkconfig(td)
|
|
parent = os.path.join(td, "run")
|
|
os.makedirs(parent)
|
|
sess = b.build_session_dir(
|
|
parent, pid, cfg,
|
|
os.path.join(cfg, "accounts", "alice", "auth.json"), "alice")
|
|
return cfg, sess
|
|
|
|
def test_newer_session_syncs_to_profile(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg, sess = self._bound(td)
|
|
prof = os.path.join(cfg, "accounts", "alice", "auth.json")
|
|
sauth = os.path.join(sess, "muse", "auth.json")
|
|
with open(sauth, "wb") as f:
|
|
f.write(b"TOKEN-REFRESHED")
|
|
now = time.time()
|
|
os.utime(prof, (now - 100, now - 100))
|
|
os.utime(sauth, (now, now))
|
|
res = b.save_session(sess, cfg)
|
|
self.assertEqual(res["status"], "synced")
|
|
with open(prof, "rb") as f:
|
|
self.assertEqual(f.read(), b"TOKEN-REFRESHED")
|
|
self.assertEqual(stat.S_IMODE(os.stat(prof).st_mode), 0o600)
|
|
# Global auth.json untouched.
|
|
with open(os.path.join(cfg, "auth.json"), "rb") as f:
|
|
self.assertEqual(f.read(), b"GLOBAL-TOKEN")
|
|
|
|
def test_stale_session_skipped(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg, sess = self._bound(td)
|
|
prof = os.path.join(cfg, "accounts", "alice", "auth.json")
|
|
sauth = os.path.join(sess, "muse", "auth.json")
|
|
now = time.time()
|
|
os.utime(prof, (now, now))
|
|
os.utime(sauth, (now - 100, now - 100))
|
|
res = b.save_session(sess, cfg)
|
|
self.assertEqual(res["status"], "skipped-stale")
|
|
with open(prof, "rb") as f:
|
|
self.assertEqual(f.read(), b"TOKEN-ALICE")
|
|
|
|
def test_unbound_dir_never_reaped(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
foreign = os.path.join(td, "muse-session-1")
|
|
os.makedirs(foreign)
|
|
with open(os.path.join(foreign, "keep.txt"), "w") as f:
|
|
f.write("x")
|
|
self.assertEqual(b.list_bound(td), [])
|
|
res = b.reap(parent=td, config_src=os.path.join(td, "cfg"))
|
|
self.assertEqual(res["reaped"], [])
|
|
self.assertTrue(os.path.isfile(
|
|
os.path.join(foreign, "keep.txt")))
|
|
|
|
def test_reap_dead_keeps_live(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
parent = os.path.join(td, "run")
|
|
os.makedirs(parent)
|
|
auth = os.path.join(cfg, "accounts", "alice", "auth.json")
|
|
dead = b.build_session_dir(parent, 11, cfg, auth, "alice")
|
|
live = b.build_session_dir(parent, 22, cfg, auth, "alice")
|
|
with mock.patch.object(
|
|
b, "session_liveness",
|
|
side_effect=lambda s: "live" if s == live else "dead"):
|
|
res = b.reap(parent=parent, config_src=cfg)
|
|
self.assertEqual([r["sessdir"] for r in res["reaped"]], [dead])
|
|
self.assertEqual(res["live"], [live])
|
|
self.assertFalse(os.path.exists(dead))
|
|
self.assertTrue(os.path.isdir(live))
|
|
|
|
|
|
class TestLaunch(unittest.TestCase):
|
|
def test_dry_run_plans_without_touching_disk(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
plan = b.launch(profile="alice", config_src=cfg,
|
|
cmd=["muse-code", "--foo"], parent=td,
|
|
dry_run=True)
|
|
self.assertIn("muse-session-", plan["sessdir"])
|
|
self.assertEqual(plan["xdg_config_home"], plan["sessdir"])
|
|
self.assertTrue(plan["auth_src"].endswith(
|
|
"accounts/alice/auth.json"))
|
|
self.assertFalse(os.path.exists(plan["sessdir"]))
|
|
|
|
def test_launch_execs_with_isolated_env(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
seen = {}
|
|
|
|
def fake_exec(path, argv, env):
|
|
seen.update(path=path, argv=argv, env=env)
|
|
|
|
with mock.patch.object(b.os, "getpid", return_value=555):
|
|
b.launch(profile="alice", config_src=cfg,
|
|
cmd=["muse-code", "chat"], parent=td,
|
|
_exec=fake_exec)
|
|
sess = os.path.join(td, "muse-session-555")
|
|
self.assertEqual(seen["path"], "muse-code")
|
|
self.assertEqual(seen["env"]["XDG_CONFIG_HOME"], sess)
|
|
self.assertEqual(seen["env"]["MUSE_SESSION_BIND_DIR"], sess)
|
|
self.assertTrue(os.path.isfile(
|
|
os.path.join(sess, "muse", "auth.json")))
|
|
|
|
def test_launch_records_session_profile(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
cfg = _mkconfig(td)
|
|
with mock.patch.object(b.os, "getpid", return_value=556):
|
|
b.launch(profile="alice", config_src=cfg,
|
|
session_id="sess-1", cmd=["muse-code"],
|
|
parent=td, _exec=lambda *a: None)
|
|
data = json.load(open(os.path.join(cfg, "session_profiles.json")))
|
|
self.assertEqual(data, {"sess-1": "alice"})
|
|
|
|
def test_launch_without_profile_or_auth_refused(self):
|
|
with self.assertRaises(ValueError):
|
|
b.launch(cmd=["muse-code"], dry_run=True)
|
|
|
|
|
|
class TestLiveness(unittest.TestCase):
|
|
def test_recycled_pid_is_dead(self):
|
|
import tempfile
|
|
with tempfile.TemporaryDirectory() as td:
|
|
sess = os.path.join(td, "muse-session-9")
|
|
os.makedirs(sess)
|
|
with open(os.path.join(sess, "bind.json"), "w") as f:
|
|
json.dump({"profile": "alice", "pid": 9}, f)
|
|
with mock.patch.object(b, "_pid_alive", return_value=True), \
|
|
mock.patch.object(b, "_pid_is_muse", return_value=False):
|
|
self.assertEqual(b.session_liveness(sess), "dead")
|
|
|
|
def test_pid_is_muse_matches_binary_names(self):
|
|
import io
|
|
with mock.patch("builtins.open",
|
|
mock.mock_open(read_data=b"muse-bin-1.4\x00--x\x00")):
|
|
self.assertTrue(b._pid_is_muse(123))
|
|
with mock.patch("builtins.open",
|
|
mock.mock_open(read_data=b"python3\x00foo\x00")):
|
|
self.assertFalse(b._pid_is_muse(123))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|