318 lines
13 KiB
Python
318 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Reference client for the dsh sync server — adapt this into the DeepSeek Harness.
|
|
|
|
Implements the sync protocol from ARCHITECTURE.md:
|
|
* layered settings pull with ETag caching (offline-tolerant)
|
|
* optimistic-concurrency push (base_version, 409 on conflict)
|
|
* plugin install with sha256 verification and atomic directory replace
|
|
|
|
Usage:
|
|
python dsh_sync_client.py login https://sync.example.com <invite-code> [--username alice]
|
|
python dsh_sync_client.py pull
|
|
python dsh_sync_client.py push <key> <json-value>
|
|
python dsh_sync_client.py plugins [--harness-version 1.0.0]
|
|
python dsh_sync_client.py install <plugin-id> [--version 1.0.0] [--harness-version 1.0.0]
|
|
python dsh_sync_client.py status
|
|
|
|
State lives in ~/.dsh/ by default (override with DSH_STATE_DIR).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tarfile
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
try:
|
|
from .util import semver_tuple # works when vendored inside a package
|
|
except ImportError:
|
|
def semver_tuple(version: str) -> tuple[int, int, int]:
|
|
parts: list[int] = []
|
|
for p in version.split("."):
|
|
digits = ""
|
|
for ch in p:
|
|
if ch.isdigit():
|
|
digits += ch
|
|
else:
|
|
break
|
|
parts.append(int(digits) if digits else 0)
|
|
while len(parts) < 3:
|
|
parts.append(0)
|
|
return tuple(parts[:3])
|
|
|
|
|
|
class SyncError(Exception):
|
|
pass
|
|
|
|
|
|
class DshSync:
|
|
def __init__(self, state_dir: Path | None = None):
|
|
self.state_dir = Path(state_dir or os.environ.get("DSH_STATE_DIR", "~/.dsh")).expanduser()
|
|
self.state_dir.mkdir(parents=True, exist_ok=True)
|
|
self.state_file = self.state_dir / "sync-state.json"
|
|
self.state: dict = self._load_state()
|
|
|
|
# ------------------------------------------------------------- plumbing
|
|
|
|
def _load_state(self) -> dict:
|
|
if self.state_file.exists():
|
|
return json.loads(self.state_file.read_text(encoding="utf-8"))
|
|
return {}
|
|
|
|
def _save_state(self) -> None:
|
|
text = json.dumps(self.state, indent=2, ensure_ascii=False)
|
|
self.state_file.write_text(text, encoding="utf-8")
|
|
try:
|
|
os.chmod(self.state_file, 0o600)
|
|
except OSError:
|
|
pass
|
|
|
|
def _require_auth(self) -> tuple[str, str]:
|
|
base_url, token = self.state.get("base_url"), self.state.get("token")
|
|
if not base_url or not token:
|
|
raise SyncError("not logged in — run the 'login' command first")
|
|
return base_url.rstrip("/"), token
|
|
|
|
def _request(self, method: str, path: str, body: dict | None = None,
|
|
headers: dict | None = None, raw: bool = False):
|
|
base_url, token = self._require_auth()
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
hdrs = {"Authorization": f"Bearer {token}"}
|
|
if data is not None:
|
|
hdrs["Content-Type"] = "application/json"
|
|
hdrs.update(headers or {})
|
|
req = urllib.request.Request(base_url + path, data=data, method=method, headers=hdrs)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
payload = resp.read()
|
|
resp_headers = {k.lower(): v for k, v in resp.headers.items()}
|
|
return resp.status, payload if raw else (json.loads(payload) if payload else {}), resp_headers
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 304: # not modified — expected by pull() caching
|
|
return 304, b"", {k.lower(): v for k, v in e.headers.items()}
|
|
detail = e.read().decode("utf-8", "replace")
|
|
raise SyncError(f"{method} {path} -> HTTP {e.code}: {detail}") from e
|
|
except urllib.error.URLError as e:
|
|
raise SyncError(f"server unreachable ({e.reason}) — using local cache if available") from e
|
|
|
|
# ------------------------------------------------------------- commands
|
|
|
|
def login(self, base_url: str, invite_code: str, username: str) -> dict:
|
|
req = urllib.request.Request(
|
|
base_url.rstrip("/") + "/v1/tokens",
|
|
data=json.dumps({"username": username, "invite_code": invite_code}).encode(),
|
|
method="POST", headers={"Content-Type": "application/json"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
out = json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
raise SyncError(f"login failed: HTTP {e.code} {e.read().decode('utf-8', 'replace')}") from e
|
|
self.state.update(base_url=base_url.rstrip("/"), token=out["token"], user=out["user"])
|
|
self._save_state()
|
|
return out
|
|
|
|
def pull(self) -> dict:
|
|
"""Fetch merged team+user settings; returns cached snapshot when offline/304."""
|
|
headers = {}
|
|
if self.state.get("settings_etag"):
|
|
headers["If-None-Match"] = self.state["settings_etag"]
|
|
try:
|
|
status, payload, resp_headers = self._request("GET", "/v1/settings", headers=headers)
|
|
except SyncError as e:
|
|
self.state["stale"] = True
|
|
self._save_state()
|
|
cached = self.state.get("settings", {})
|
|
if cached:
|
|
print(f"[warn] {e}; using cached settings", file=sys.stderr)
|
|
return cached
|
|
raise
|
|
if status == 304:
|
|
self.state["stale"] = False
|
|
self._save_state()
|
|
return self.state.get("settings", {})
|
|
settings = {k: v["value"] for k, v in payload["settings"].items()}
|
|
self.state.update(settings=settings, settings_versions=payload["settings"],
|
|
settings_etag=resp_headers.get("etag"), stale=False, last_pull=_now())
|
|
self._save_state()
|
|
return settings
|
|
|
|
def push(self, key: str, value) -> dict:
|
|
"""Push a user setting with optimistic concurrency; surfaces 409 conflicts."""
|
|
cached = self.state.get("settings_versions", {}).get(key, {})
|
|
body = {"value": value}
|
|
if cached:
|
|
body["base_version"] = cached["version"]
|
|
try:
|
|
status, payload, _ = self._request("PUT", f"/v1/settings/user/{key}", body)
|
|
except SyncError as e:
|
|
if "HTTP 409" in str(e):
|
|
raise SyncError(
|
|
f"conflict on '{key}': server has a newer version — pull and re-apply"
|
|
) from e
|
|
raise
|
|
# The write already succeeded; a failed refresh only means the cache is
|
|
# stale, which the next pull fixes. Reporting it as a push failure would
|
|
# make callers retry a write that already landed.
|
|
try:
|
|
self.pull()
|
|
except SyncError as e:
|
|
print(f"[warn] pushed '{key}' but could not refresh cache: {e}", file=sys.stderr)
|
|
self.state["stale"] = True
|
|
self._save_state()
|
|
return payload
|
|
|
|
def plugins(self, harness_version: str | None = None) -> list[dict]:
|
|
q = f"?harness_version={harness_version}" if harness_version else ""
|
|
_, payload, _ = self._request("GET", f"/v1/plugins{q}")
|
|
return payload["plugins"]
|
|
|
|
def install(self, plugin_id: str, version: str | None = None,
|
|
harness_version: str | None = None, target_root: Path | None = None) -> Path:
|
|
"""Download a plugin, verify sha256, atomically replace its directory."""
|
|
listing = self.plugins(harness_version)
|
|
plugin = next((p for p in listing if p["id"] == plugin_id), None)
|
|
if plugin is None:
|
|
raise SyncError(f"plugin '{plugin_id}' not found (or excluded by harness_version)")
|
|
chosen = plugin["latest"] if version is None else next(
|
|
(v for v in plugin["versions"] if v["version"] == version), None)
|
|
if chosen is None:
|
|
raise SyncError(f"version {version!r} of '{plugin_id}' not available")
|
|
if harness_version and semver_tuple(chosen["min_harness_version"]) > semver_tuple(harness_version):
|
|
raise SyncError(f"plugin requires harness >= {chosen['min_harness_version']}")
|
|
|
|
_, data, _ = self._request(
|
|
"GET", f"/v1/plugins/{plugin_id}/{chosen['version']}/download", raw=True)
|
|
if len(data) != chosen["size_bytes"]:
|
|
raise SyncError(f"size mismatch: got {len(data)}, expected {chosen['size_bytes']}")
|
|
import hashlib
|
|
digest = hashlib.sha256(data).hexdigest()
|
|
if digest != chosen["sha256"]:
|
|
raise SyncError(f"sha256 mismatch: got {digest}, expected {chosen['sha256']}")
|
|
|
|
target_root = Path(target_root or self.state_dir / "plugins")
|
|
target = target_root / plugin_id
|
|
tmp = target_root / f".{plugin_id}.tmp"
|
|
bak = target_root / f".{plugin_id}.bak"
|
|
if tmp.exists():
|
|
shutil.rmtree(tmp)
|
|
tmp.mkdir(parents=True)
|
|
with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tf:
|
|
members = []
|
|
for m in tf.getmembers():
|
|
parts = [p for p in PurePosixPath(m.name).parts if p != "."]
|
|
if not parts or parts[0] == ".." or ".." in parts or m.name.startswith("/"):
|
|
raise SyncError(f"unsafe path in archive: {m.name}")
|
|
if not m.isfile():
|
|
continue
|
|
members.append(("/".join(parts), m))
|
|
|
|
# Archives come both ways: manifest.json at the root, or wrapped in a
|
|
# single directory named after the plugin. Detect the wrapper instead
|
|
# of blindly dropping the first segment — doing that unconditionally
|
|
# flattened `files/main.py` into `main.py`.
|
|
strip = _wrapped_top_level([rel for rel, _ in members])
|
|
|
|
for rel, m in members:
|
|
if strip:
|
|
rel = rel.split("/", 1)[1]
|
|
dest = tmp / rel
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_bytes(tf.extractfile(m).read())
|
|
if bak.exists():
|
|
shutil.rmtree(bak)
|
|
if target.exists():
|
|
os.replace(target, bak)
|
|
os.replace(tmp, target)
|
|
if bak.exists():
|
|
shutil.rmtree(bak, ignore_errors=True)
|
|
print(f"installed {plugin_id} {chosen['version']} -> {target}")
|
|
return target
|
|
|
|
def status(self) -> dict:
|
|
return {k: self.state.get(k) for k in
|
|
("base_url", "user", "stale", "last_pull", "settings_etag")}
|
|
|
|
|
|
def _wrapped_top_level(names: list[str]) -> bool:
|
|
"""True when every entry sits under one directory that is not the payload root.
|
|
|
|
`plugin/manifest.json` + `plugin/files/x` is a wrapped archive and the
|
|
wrapper should go. `manifest.json` + `files/x` is already flat, and stripping
|
|
there would rename `files/x` to `x`.
|
|
"""
|
|
tops = {n.split("/", 1)[0] for n in names}
|
|
if len(tops) != 1:
|
|
return False
|
|
top = tops.pop()
|
|
# A wrapper contains no manifest at the archive root, by construction.
|
|
return not any(n == "manifest.json" for n in names) and top != "manifest.json"
|
|
|
|
|
|
def _now() -> str:
|
|
import datetime
|
|
return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def _parse_value(text: str):
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return text
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(description="dsh sync client (reference)")
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
p = sub.add_parser("login")
|
|
p.add_argument("base_url")
|
|
p.add_argument("invite_code")
|
|
p.add_argument("--username", required=True)
|
|
|
|
sub.add_parser("pull")
|
|
|
|
p = sub.add_parser("push")
|
|
p.add_argument("key")
|
|
p.add_argument("value")
|
|
|
|
p = sub.add_parser("plugins")
|
|
p.add_argument("--harness-version")
|
|
|
|
p = sub.add_parser("install")
|
|
p.add_argument("plugin_id")
|
|
p.add_argument("--version")
|
|
p.add_argument("--harness-version")
|
|
|
|
sub.add_parser("status")
|
|
|
|
args = ap.parse_args(argv)
|
|
client = DshSync()
|
|
if args.cmd == "login":
|
|
print(json.dumps(client.login(args.base_url, args.invite_code, args.username), indent=2))
|
|
elif args.cmd == "pull":
|
|
print(json.dumps(client.pull(), indent=2, ensure_ascii=False))
|
|
elif args.cmd == "push":
|
|
print(json.dumps(client.push(args.key, _parse_value(args.value)), indent=2))
|
|
elif args.cmd == "plugins":
|
|
print(json.dumps(client.plugins(args.harness_version), indent=2, ensure_ascii=False))
|
|
elif args.cmd == "install":
|
|
client.install(args.plugin_id, args.version, args.harness_version)
|
|
elif args.cmd == "status":
|
|
print(json.dumps(client.status(), indent=2, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main())
|
|
except SyncError as e:
|
|
print(f"error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|