RigShare 1.3.1: refuse folder paths outside the workspace (Windows)

tests/python/test_jwt.py

@@ -0,0 +1,60 @@
+"""HS256 tokens: round trip, tampering, expiry, issuer, algorithm confusion."""
+
+import base64
+import json
+import time
+
+import pytest
+
+from rigshare_core import jwt
+
+SECRET = b"s3cret"
+
+
+def claims(**extra):
+    return {"iss": "rig", "exp": time.time() + 60, **extra}
+
+
+def b64(obj):
+    return base64.urlsafe_b64encode(json.dumps(obj).encode()).rstrip(b"=").decode()
+
+
+def test_round_trip():
+    token = jwt.encode(claims(sub="bob"), SECRET)
+    assert jwt.decode(token, SECRET, "rig")["sub"] == "bob"
+
+
+def test_wrong_secret_and_tampered_payload():
+    token = jwt.encode(claims(sub="bob"), SECRET)
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(token, b"other", "rig")
+    head, _, sig = token.split(".")
+    forged = f"{head}.{b64(claims(sub='admin'))}.{sig}"
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(forged, SECRET, "rig")
+
+
+def test_expiry_leeway_and_not_before():
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 3600}, SECRET), SECRET, "rig")
+    jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 5}, SECRET), SECRET, "rig")  # inside leeway
+    assert jwt.decode(jwt.encode({"iss": "rig", "exp": time.time() - 3600}, SECRET), SECRET, "rig", verify_exp=False)
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode(claims(nbf=time.time() + 3600), SECRET), SECRET, "rig")
+
+
+def test_wrong_issuer():
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(jwt.encode(claims(), SECRET), SECRET, "someone-else")
+
+
+def test_alg_none_is_refused():
+    token = f"{b64({'alg': 'none', 'typ': 'JWT'})}.{b64(claims())}."
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(token, SECRET, "rig")
+
+
+@pytest.mark.parametrize("junk", ["", "a.b", "a.b.c.d", "!!!.???.***", "e30.e30.e30"])
+def test_malformed_tokens(junk):
+    with pytest.raises(jwt.InvalidToken):
+        jwt.decode(junk, SECRET, "rig")

tests/python/test_server.py

@@ -646,3 +646,23 @@ def test_performance_run_log_eta_history_and_clear(run):
             assert (await r.json())["cleared"] == 3
             assert hub.etas() == {"video": {"ms": 90000, "runs": 1}}, "the ETA starts over"
     run(scenario())
+
+
+def test_folder_routes_refuse_paths_outside_the_workspace(run, tmp_path):
+    async def scenario():
+        async with rig_server() as rig:
+            await rig.setup_admin()
+            await rig.add_user("alice", edit=True)
+            outside = tmp_path / "models"
+            outside.mkdir()
+            (outside / "keep.bin").write_bytes(b"x")
+            H, h = rig.headers, rig.http
+            targets = [str(outside), "C:" + str(outside)[2:], r"..\..\models", "shared/C:/evil"]
+            for route, extra in (("delete-folder", {}), ("rename-folder", {"name": "x"}), ("move-folder", {"dest": "@shared"})):
+                for target in targets:
+                    r = await h.post(rig.url(f"/rigshare/api/workspace/{route}"), json={"path": target, **extra}, headers=H["alice"])
+                    assert r.status in (400, 403, 404), (route, target, r.status)
+            r = await h.post(rig.url("/rigshare/api/workspace/mkdir"), json={"parent": "C:/", "name": "x"}, headers=H["alice"])
+            assert r.status in (400, 403, 404)
+            assert (outside / "keep.bin").exists()
+    run(scenario())

tests/python/test_stats.py

@@ -0,0 +1,45 @@
+"""Server stats: every metric is optional and never raises."""
+
+from rigshare_core import stats
+
+
+class Queue:
+    def __init__(self, running, pending):
+        self.running, self.pending = running, pending
+
+    def get_current_queue_volatile(self):
+        return self.running, self.pending
+
+
+class Server:
+    def __init__(self, queue):
+        self.prompt_queue = queue
+
+
+def test_queue_items_order_and_node_count():
+    server = Server(Queue([(1, "r1", {"1": {}, "2": {}})], [(5, "p5", {"1": {}}), (3, "p3", {})]))
+    items = stats.queue_items(server)
+    assert [(i["id"], i["status"], i["nodes"]) for i in items] == [("r1", "running", 2), ("p3", "pending", 0), ("p5", "pending", 1)]
+
+
+def test_queue_items_survive_odd_entries_and_missing_queue():
+    assert stats.queue_items(Server(Queue([(1,)], [(9,)]))) == []
+    assert stats.queue_items(object()) is None
+
+
+def test_queue_counts_with_fallback():
+    assert stats._queue(Server(Queue([1], [2, 3]))) == {"running": 1, "pending": 2}
+
+    class Old:
+        def get_tasks_remaining(self):
+            return 4
+    assert stats._queue(Server(Old())) == {"running": 0, "pending": 4}
+    assert stats._queue(object()) is None
+
+
+def test_collect_stats_without_psutil_or_gpu(monkeypatch):
+    monkeypatch.setattr(stats, "psutil", None)
+    monkeypatch.setattr(stats, "pynvml", None)
+    out = stats.collect_stats(Server(Queue([], [])))
+    assert out["gpus"] == [] and "cpu" not in out and out["queue"] == {"running": 0, "pending": 0}
+    assert stats.collect_stats() == {"gpus": []}

tests/python/test_workspace.py

@@ -131,3 +131,55 @@ def test_rename_user_moves_private_folder(ws):
     ws.rename_user("alice", "alicia")
     assert os.path.exists(os.path.join(ws.root, "users", "alicia", "keep.json"))
     assert ws.folders["shared/Team"]["owner"] == "alicia"
+
+
+@pytest.mark.parametrize("raw", ["C:/models", "C:", "a/D:x/b", r"..\..\x", "c%3A%5Cmodels"])
+def test_norm_drops_drive_segments(raw):
+    assert ":" not in norm(raw) and ".." not in norm(raw).split("/")
+
+
+@pytest.mark.parametrize("bad", ["C:/models", "C:", "..", "a/../..", "x:y"])
+def test_safe_join_refuses_escapes(ws, bad):
+    from rigshare_core.workspace import safe_join
+    with pytest.raises(PermissionError):
+        safe_join(ws.root, bad)
+
+
+def test_folder_ops_cannot_leave_the_workspace(ws, tmp_path):
+    outside = tmp_path / "models"
+    outside.mkdir()
+    (outside / "keep.txt").write_text("x")
+    for bad in (str(outside), "C:" + str(outside)[2:], "shared/C:evil"):
+        with pytest.raises(PermissionError):
+            ws.delete_folder(bad)
+        with pytest.raises(PermissionError):
+            ws.rename_folder(bad, "renamed")
+    assert (outside / "keep.txt").exists()
+
+
+@pytest.mark.parametrize("raw", ["users./bob/a.json", "users /bob", "USERS~1/bob", "users/bob./a.json"])
+def test_norm_drops_windows_aliases(raw):
+    if os.name != "nt" and ("." in raw.split("/")[0] or " " in raw.split("/")[0]):
+        pytest.skip("trailing dots and spaces only alias on Windows")
+    assert "~1" not in norm(raw)
+
+
+@pytest.mark.skipif(os.name != "nt", reason="case-insensitive filesystem")
+def test_case_variants_cannot_bypass_private_folders(ws):
+    touch(ws, "users/alice/a.json")
+    assert norm("Users/Alice/a.json") == "users/alice/a.json"
+    assert norm("Workflows/USERS/ALICE") == "workflows/users/alice"
+    assert ws.role(norm("USERS/alice/a.json"), "user", "bob", EDITOR) is None
+
+
+def test_name_rules():
+    from rigshare_core.workspace import FOLDER_NAME_RE
+    from rigshare_core.store import USERNAME_RE
+    for bad in ("con", "NUL", "com1.txt", "art.", "art ", "bad\n"):
+        assert not FOLDER_NAME_RE.match(bad), bad
+    for bad in ("..", "bob.", ".bob", "con", "aux", "ab\n", "a"):
+        assert not USERNAME_RE.match(bad), bad
+    for ok in ("Art (v2)", "a", "my-folder"):
+        assert FOLDER_NAME_RE.match(ok), ok
+    for ok in ("bob", "al.ice", "x_y-z", "ab"):
+        assert USERNAME_RE.match(ok), ok