345 lines
14 KiB
Python
345 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import tarfile
|
|
|
|
from conftest import ADMIN_AUTH, INVITE_CODE
|
|
|
|
|
|
def make_plugin_tar(plugin_id="demo", version="1.0.0", extra=None, unsafe_path=None):
|
|
manifest = {
|
|
"id": plugin_id,
|
|
"version": version,
|
|
"name": "Demo Plugin",
|
|
"description": "test plugin",
|
|
"channel": "stable",
|
|
"min_harness_version": "0.0.0",
|
|
}
|
|
if extra:
|
|
manifest.update(extra)
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
|
|
mjson = json.dumps(manifest).encode()
|
|
ti = tarfile.TarInfo("manifest.json")
|
|
ti.size = len(mjson)
|
|
tf.addfile(ti, io.BytesIO(mjson))
|
|
content = b"print('hello')\n"
|
|
ti2 = tarfile.TarInfo("files/main.py")
|
|
ti2.size = len(content)
|
|
tf.addfile(ti2, io.BytesIO(content))
|
|
if unsafe_path:
|
|
ti3 = tarfile.TarInfo(unsafe_path)
|
|
ti3.size = len(content)
|
|
tf.addfile(ti3, io.BytesIO(content))
|
|
return buf.getvalue(), manifest
|
|
|
|
|
|
# ------------------------------------------------------------------ basics
|
|
|
|
def test_healthz(client):
|
|
r = client.get("/healthz")
|
|
assert r.status_code == 200
|
|
assert r.json()["status"] == "ok"
|
|
|
|
|
|
def test_me_requires_token(client):
|
|
assert client.get("/v1/me").status_code == 401
|
|
assert client.get("/v1/me", headers={"Authorization": "Bearer nope"}).status_code == 401
|
|
|
|
|
|
def test_me_bootstrap_admin(client):
|
|
r = client.get("/v1/me", headers=ADMIN_AUTH)
|
|
assert r.status_code == 200
|
|
assert r.json()["role"] == "admin"
|
|
|
|
|
|
# ------------------------------------------------------------------ tokens
|
|
|
|
def test_invite_flow(client):
|
|
r = client.post("/v1/tokens", json={"username": "bob", "invite_code": INVITE_CODE})
|
|
assert r.status_code == 200
|
|
token = r.json()["token"]
|
|
me = client.get("/v1/me", headers={"Authorization": f"Bearer {token}"})
|
|
assert me.status_code == 200
|
|
assert me.json()["username"] == "bob"
|
|
assert me.json()["role"] == "member"
|
|
|
|
|
|
def test_invite_rejects_wrong_code(client):
|
|
r = client.post("/v1/tokens", json={"username": "bob", "invite_code": "wrong"})
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_invite_does_not_hijack_existing_user(client):
|
|
r = client.post("/v1/tokens", json={"username": "admin", "invite_code": INVITE_CODE})
|
|
assert r.status_code == 409
|
|
|
|
|
|
def test_self_token_and_revoke(client, member_auth):
|
|
second = client.post("/v1/tokens/self", json={"token_name": "laptop2"}, headers=member_auth)
|
|
assert second.status_code == 200
|
|
second_auth = {"Authorization": f"Bearer {second.json()['token']}"}
|
|
assert client.get("/v1/me", headers=second_auth).status_code == 200
|
|
|
|
listed = client.get("/v1/tokens", headers=member_auth).json()["tokens"]
|
|
assert len(listed) == 2
|
|
target = next(t["token_hash"] for t in listed if t["name"] == "laptop2")
|
|
assert client.delete(f"/v1/tokens/{target}", headers=member_auth).json()["revoked"] is True
|
|
assert client.get("/v1/me", headers=second_auth).status_code == 401
|
|
|
|
|
|
def test_admin_creates_user(client, member_auth):
|
|
r = client.post("/v1/users", json={"username": "carol", "role": "member"}, headers=ADMIN_AUTH)
|
|
assert r.status_code == 200
|
|
assert client.get("/v1/me", headers={"Authorization": f"Bearer {r.json()['token']}"}).json()["username"] == "carol"
|
|
assert client.post("/v1/users", json={"username": "eve"}, headers=member_auth).status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------- settings
|
|
|
|
def test_team_setting_roundtrip_and_merge(client, member_auth):
|
|
assert client.put("/v1/settings/team/api.base_url",
|
|
json={"value": "https://api.deepseek.com"}, headers=ADMIN_AUTH).json()["version"] == 1
|
|
assert client.put("/v1/settings/user/theme",
|
|
json={"value": "dark"}, headers=member_auth).status_code == 200
|
|
|
|
r = client.get("/v1/settings", headers=member_auth)
|
|
s = r.json()["settings"]
|
|
assert s["api.base_url"]["value"] == "https://api.deepseek.com"
|
|
assert s["api.base_url"]["scope"] == "team"
|
|
assert s["theme"]["value"] == "dark"
|
|
assert s["theme"]["scope"] == "user"
|
|
|
|
etag = r.headers["ETag"]
|
|
cached = client.get("/v1/settings", headers={**member_auth, "If-None-Match": etag})
|
|
assert cached.status_code == 304
|
|
|
|
|
|
def test_user_setting_overrides_team(client, member_auth):
|
|
client.put("/v1/settings/team/theme", json={"value": "dark"}, headers=ADMIN_AUTH)
|
|
client.put("/v1/settings/user/theme", json={"value": "light"}, headers=member_auth)
|
|
s = client.get("/v1/settings", headers=member_auth).json()["settings"]
|
|
assert s["theme"]["value"] == "light"
|
|
|
|
|
|
def test_setting_conflict(client, member_auth):
|
|
first = client.put("/v1/settings/user/pref",
|
|
json={"value": {"a": 1}, "base_version": 0}, headers=member_auth)
|
|
assert first.json()["version"] == 1
|
|
stale = client.put("/v1/settings/user/pref",
|
|
json={"value": {"a": 2}, "base_version": 0}, headers=member_auth)
|
|
assert stale.status_code == 409
|
|
assert stale.json()["detail"]["current_version"] == 1
|
|
ok = client.put("/v1/settings/user/pref",
|
|
json={"value": {"a": 2}, "base_version": 1}, headers=member_auth)
|
|
assert ok.json()["version"] == 2
|
|
lww = client.put("/v1/settings/user/pref", json={"value": {"a": 3}}, headers=member_auth)
|
|
assert lww.json()["version"] == 3
|
|
|
|
|
|
def test_team_setting_admin_only(client, member_auth):
|
|
assert client.put("/v1/settings/team/x", json={"value": 1}, headers=member_auth).status_code == 403
|
|
assert client.put("/v1/settings/team/x", json={"value": 1}, headers=ADMIN_AUTH).status_code == 200
|
|
|
|
|
|
def test_invalid_setting_key_rejected(client, member_auth):
|
|
r = client.put("/v1/settings/user/bad key!", json={"value": 1}, headers=member_auth)
|
|
assert r.status_code == 400
|
|
|
|
|
|
# ----------------------------------------------------------------- plugins
|
|
|
|
def test_plugin_upload_and_download(client, member_auth):
|
|
data, manifest = make_plugin_tar()
|
|
up = client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert up.status_code == 200, up.text
|
|
sha = up.json()["sha256"]
|
|
assert up.json()["size_bytes"] == len(data)
|
|
|
|
assert client.post("/v1/plugins/other",
|
|
files={"file": ("x.tar.gz", data, "application/gzip")},
|
|
headers=member_auth).status_code == 403
|
|
|
|
listing = client.get("/v1/plugins", headers=member_auth).json()["plugins"]
|
|
assert listing[0]["id"] == "demo"
|
|
assert listing[0]["latest"]["version"] == "1.0.0"
|
|
assert listing[0]["latest"]["sha256"] == sha
|
|
|
|
detail = client.get("/v1/plugins/demo/1.0.0", headers=member_auth)
|
|
assert detail.json()["manifest"]["id"] == "demo"
|
|
assert detail.json()["sha256"] == sha
|
|
|
|
dl = client.get("/v1/plugins/demo/1.0.0/download", headers=member_auth)
|
|
assert dl.status_code == 200
|
|
assert dl.content == data
|
|
assert dl.headers["X-Sha256"] == sha
|
|
|
|
|
|
def test_plugin_duplicate_version_rejected(client):
|
|
data, _ = make_plugin_tar()
|
|
files = {"file": ("demo-1.0.0.tar.gz", data, "application/gzip")}
|
|
assert client.post("/v1/plugins/demo", files=files, headers=ADMIN_AUTH).status_code == 200
|
|
assert client.post("/v1/plugins/demo", files=files, headers=ADMIN_AUTH).status_code == 409
|
|
|
|
|
|
def test_plugin_manifest_id_mismatch_rejected(client):
|
|
data, _ = make_plugin_tar(plugin_id="other")
|
|
r = client.post("/v1/plugins/demo",
|
|
files={"file": ("other-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_plugin_unsafe_path_rejected(client):
|
|
data, _ = make_plugin_tar(unsafe_path="../evil.txt")
|
|
r = client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_plugin_harness_version_filter(client):
|
|
data, _ = make_plugin_tar(extra={"min_harness_version": "1.2.0"})
|
|
assert client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH).status_code == 200
|
|
old = client.get("/v1/plugins", params={"harness_version": "1.1.0"}, headers=ADMIN_AUTH)
|
|
assert old.json()["plugins"][0]["versions"] == []
|
|
new = client.get("/v1/plugins", params={"harness_version": "1.2.0"}, headers=ADMIN_AUTH)
|
|
assert new.json()["plugins"][0]["versions"][0]["version"] == "1.0.0"
|
|
|
|
|
|
def test_delete_plugin_version(client, member_auth):
|
|
data, _ = make_plugin_tar()
|
|
assert client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH).status_code == 200
|
|
assert client.delete("/v1/plugins/demo/1.0.0", headers=member_auth).status_code == 403
|
|
assert client.delete("/v1/plugins/demo/1.0.0", headers=ADMIN_AUTH).json()["deleted"] is True
|
|
assert client.get("/v1/plugins/demo/1.0.0", headers=ADMIN_AUTH).status_code == 404
|
|
|
|
|
|
def test_plugin_latest_prefers_stable(client):
|
|
v1, _ = make_plugin_tar(version="1.0.0")
|
|
v2, _ = make_plugin_tar(version="1.1.0", extra={"channel": "beta"})
|
|
up = lambda d, v: client.post(f"/v1/plugins/demo",
|
|
files={"file": (f"demo-{v}.tar.gz", d, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert up(v1, "1.0.0").status_code == 200
|
|
assert up(v2, "1.1.0").status_code == 200
|
|
p = client.get("/v1/plugins", headers=ADMIN_AUTH).json()["plugins"][0]
|
|
assert p["latest"]["version"] == "1.0.0"
|
|
assert [v["version"] for v in p["versions"]] == ["1.1.0", "1.0.0"]
|
|
|
|
|
|
# ------------------------------------------------------------------- audit
|
|
|
|
def test_audit_log_admin_only(client, member_auth):
|
|
client.put("/v1/settings/team/x", json={"value": 1}, headers=ADMIN_AUTH)
|
|
data, _ = make_plugin_tar()
|
|
client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert client.get("/v1/audit", headers=member_auth).status_code == 403
|
|
events = client.get("/v1/audit", headers=ADMIN_AUTH).json()["events"]
|
|
actions = {e["action"] for e in events}
|
|
assert "settings.team.update" in actions
|
|
assert "plugin.publish" in actions
|
|
|
|
|
|
def test_plugin_row_stores_relative_path(client):
|
|
"""artifact_path must stay machine-independent so a moved data dir still works."""
|
|
data, _ = make_plugin_tar()
|
|
up = client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert up.status_code == 200, up.text
|
|
|
|
from dsh_sync.db import connect
|
|
from dsh_sync.config import Config
|
|
|
|
cfg = Config.from_env()
|
|
conn = connect(cfg)
|
|
try:
|
|
stored = conn.execute(
|
|
"SELECT artifact_path FROM plugin_versions WHERE plugin_id = 'demo'"
|
|
).fetchone()["artifact_path"]
|
|
finally:
|
|
conn.close()
|
|
|
|
assert stored == "demo/1.0.0.tar.gz", stored
|
|
assert ":" not in stored and not stored.startswith("/")
|
|
|
|
|
|
def test_plugin_download_resolves_relative_row(client, member_auth):
|
|
"""A relative row must resolve against plugins_dir, not the process CWD.
|
|
|
|
This is the regression: the row was written relative while the download
|
|
handler read it as a plain path, so it 404'd once CWD differed.
|
|
"""
|
|
data, _ = make_plugin_tar()
|
|
assert client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH).status_code == 200
|
|
|
|
from dsh_sync.config import Config
|
|
from dsh_sync.db import connect
|
|
|
|
cfg = Config.from_env()
|
|
conn = connect(cfg)
|
|
try:
|
|
conn.execute("UPDATE plugin_versions SET artifact_path = 'demo/1.0.0.tar.gz'")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
dl = client.get("/v1/plugins/demo/1.0.0/download", headers=member_auth)
|
|
assert dl.status_code == 200, dl.text
|
|
assert dl.content == data
|
|
|
|
|
|
def test_plugin_download_reports_missing_file(client, member_auth):
|
|
"""Row without its artifact (volume not mounted) is a 404, not a 500."""
|
|
data, _ = make_plugin_tar()
|
|
assert client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", data, "application/gzip")},
|
|
headers=ADMIN_AUTH).status_code == 200
|
|
|
|
from dsh_sync.config import Config
|
|
from dsh_sync.db import connect
|
|
|
|
cfg = Config.from_env()
|
|
conn = connect(cfg)
|
|
try:
|
|
conn.execute("UPDATE plugin_versions SET artifact_path = 'demo/gone.tar.gz'")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
dl = client.get("/v1/plugins/demo/1.0.0/download", headers=member_auth)
|
|
assert dl.status_code == 404
|
|
assert "missing on server" in dl.json()["detail"]
|
|
|
|
|
|
def test_archive_root_dot_entry_is_tolerated(client):
|
|
"""`tar -czf x.tar.gz .` emits a bare `./` member; it must not fail the upload."""
|
|
import io
|
|
import tarfile
|
|
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
|
|
for name, blob in (("./", b""), ("./manifest.json", b'{"id":"demo","version":"1.0.0"}'),
|
|
("./files/main.py", b"print(1)\n")):
|
|
info = tarfile.TarInfo(name)
|
|
info.type = tarfile.DIRTYPE if name.endswith("/") else tarfile.REGTYPE
|
|
info.size = len(blob)
|
|
tf.addfile(info, io.BytesIO(blob) if blob else None)
|
|
|
|
r = client.post("/v1/plugins/demo",
|
|
files={"file": ("demo-1.0.0.tar.gz", buf.getvalue(), "application/gzip")},
|
|
headers=ADMIN_AUTH)
|
|
assert r.status_code == 200, r.text
|