granthi-sync/client/granthi_sync_client.py

#!/usr/bin/env python3
"""granthi-sync client daemon.

Commands:
  link  [--server URL] [--token ZITADEL_TOKEN] [--device NAME]
        Zitadel OAuth device flow against https://id.shre.ai (or --token for
        the headless path), then POST /v1/link on the granthi-link service.
        Stores {server, gitea_base, login, token, token_name} in
        ~/.granthi-sync/config.json (0600).
  list
        Table of every repo the linked token can see on the forge, with the
        local folder each one is already synced to (if any).
  get   <repo|owner/repo> [--into DIR]
        Clone a forge repo and register it for `watch` -- the download half
        of `add`. Remote is named 'granthi' at clone time and the token is
        supplied by the credential helper, never embedded in the URL.
  add   <folder> [--name N] [--private/--public]
        git init (branch main) if needed, create the cloud repo via
        /v1/repos, add remote 'granthi', initial commit + push. The token
        is supplied via a git credential helper (this script's hidden
        `git-credential` subcommand), never embedded in the remote URL.
  watch [--interval 30] [--once]
        Poll linked folders: autocommit local changes, fetch, ff-pull when
        the remote is strictly ahead, push when local is strictly ahead.
        DIVERGED branches are logged + recorded and SKIPPED -- never force,
        never merge (same policy as the estate's gitea_sync.py mesh).
        SIGTERM-clean.
  status
        Table of linked folders, last sync, divergence flags.

Stdlib + git CLI only.
"""

import argparse
import base64
import json
import os
import re
import shlex
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

VERSION = "1.1.0"

CONFIG_DIR = os.path.expanduser(os.environ.get("GRANTHI_SYNC_HOME", "~/.granthi-sync"))
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")

ZITADEL_BASE = "https://id.shre.ai"
DEVICE_CLIENT_ID = "386909715541590022"
# ^ Zitadel native app "granthi-sync-device" (appId 386909715541524486) in
#   project granthi-forge (386906525790109702); device-code + refresh grants.
DEFAULT_SERVER = "http://100.111.127.127:3042"
DEVICE_SCOPE = "openid profile email"

FORGE_PAGE_LIMIT = 50
FORGE_MAX_PAGES = 40  # 2000 repos; a guard against an unbounded paging loop


def log(msg):
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    print(f"[{ts}] {msg}", flush=True)


# --------------------------------------------------------------------------
# HTTP helpers (patchable in tests)
# --------------------------------------------------------------------------

def http_json(method, url, headers=None, body=None, form=None, timeout=30):
    """Returns (status, parsed-json-or-{'raw': text})."""
    data = None
    hdrs = dict(headers or {})
    # Cloudflare in front of id.shre.ai 403s the default Python-urllib UA.
    hdrs.setdefault("User-Agent", f"granthi-sync/{VERSION}")
    if form is not None:
        data = urllib.parse.urlencode(form).encode()
        hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
    elif body is not None:
        data = json.dumps(body).encode()
        hdrs.setdefault("Content-Type", "application/json")
    req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw, status = resp.read(), resp.status
    except urllib.error.HTTPError as e:
        raw, status = e.read(), e.code
    except (urllib.error.URLError, OSError) as e:
        return 599, {"error": str(e)}
    try:
        return status, json.loads(raw) if raw else {}
    except ValueError:
        return status, {"raw": raw.decode(errors="replace")}


# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------

def load_config():
    try:
        with open(CONFIG_PATH) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def save_config(cfg):
    os.makedirs(CONFIG_DIR, mode=0o700, exist_ok=True)
    tmp = CONFIG_PATH + ".tmp"
    # O_CREAT with mode 0600 -- the file is never observable with wider
    # permissions (a write-then-chmod sequence leaves a umask-sized window
    # in which the token is world-readable).
    fd = os.open(tmp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(cfg, f, indent=2)
    os.replace(tmp, CONFIG_PATH)


# --------------------------------------------------------------------------
# git helpers
# --------------------------------------------------------------------------

def git(folder, *args, check=True):
    """Run git in folder. Returns (rc, stdout). Never uses --force."""
    proc = subprocess.run(["git", "-C", folder] + list(args),
                          capture_output=True, text=True)
    if check and proc.returncode != 0:
        raise RuntimeError(
            f"git {' '.join(args)} failed in {folder}: {proc.stderr.strip()}")
    return proc.returncode, proc.stdout.strip()


def ensure_repo(folder):
    if os.path.isdir(os.path.join(folder, ".git")):
        return
    rc, _ = git(folder, "init", "-b", "main", check=False)
    if rc != 0:  # git < 2.28 fallback
        git(folder, "init")
        git(folder, "symbolic-ref", "HEAD", "refs/heads/main")


def autocommit(folder):
    """Commit all local changes as 'sync: <ISO ts>'. Returns True if a
    commit was made."""
    _, status = git(folder, "status", "--porcelain")
    if not status:
        return False
    git(folder, "add", "-A")
    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    git(folder, "commit", "-m", f"sync: {ts}")
    return True


def sync_state(folder, remote="granthi", branch="main"):
    """Classify local vs remote after a fetch.

    Returns one of: 'no-remote-branch', 'in-sync', 'local-ahead',
    'remote-ahead-ff', 'diverged'.
    """
    rc, _ = git(folder, "rev-parse", "--verify", "--quiet",
                f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "no-remote-branch"
    _, local = git(folder, "rev-parse", branch)
    _, remote_sha = git(folder, "rev-parse", f"refs/remotes/{remote}/{branch}")
    if local == remote_sha:
        return "in-sync"
    rc, base = git(folder, "merge-base", branch,
                   f"refs/remotes/{remote}/{branch}", check=False)
    if rc != 0:
        return "diverged"  # unrelated histories: treat as divergence
    base = base.strip()
    if base == remote_sha:
        return "local-ahead"
    if base == local:
        return "remote-ahead-ff"
    return "diverged"


def sync_folder(folder, remote="granthi", branch="main"):
    """One sync pass for one folder. Returns (outcome, detail).

    outcome in {'synced', 'pushed', 'clean', 'diverged', 'error'}.
    Policy: ff-only pulls, plain pushes. NEVER --force, NEVER merge.
    """
    try:
        committed = autocommit(folder)
        git(folder, "fetch", remote)
        state = sync_state(folder, remote, branch)
        ff_pulled = False
        if state == "remote-ahead-ff":
            git(folder, "merge", "--ff-only", f"refs/remotes/{remote}/{branch}")
            ff_pulled = True
            state = sync_state(folder, remote, branch)
        if state == "diverged":
            return "diverged", "local and remote both advanced; skipping (no force, no merge)"
        if state in ("local-ahead", "no-remote-branch"):
            git(folder, "push", "-u", remote, branch)
            return "pushed", "committed+pushed" if committed else "pushed"
        # state == "in-sync"
        if ff_pulled:
            return "synced", "ff-pulled"
        return "clean", "in-sync"
    except RuntimeError as e:
        return "error", str(e)


# --------------------------------------------------------------------------
# credential helper (git calls back into this script; token never in URL)
# --------------------------------------------------------------------------

def cmd_git_credential(argv):
    op = argv[0] if argv else "get"
    if op != "get":
        return 0  # ignore store/erase
    attrs = {}
    for line in sys.stdin:
        line = line.strip()
        if not line:
            break
        k, _, v = line.partition("=")
        attrs[k] = v
    cfg = load_config()
    gitea = cfg.get("gitea_base", "")
    want = urllib.parse.urlparse(gitea)
    if attrs.get("host") == want.netloc:
        print(f"username={cfg.get('login', '')}")
        print(f"password={cfg.get('token', '')}")
    return 0


def credential_helper_value():
    """Shell command git runs for credentials. Both paths are shlex-quoted:
    a Python or script path containing spaces (or shell metacharacters)
    must neither break the helper nor inject into the shell."""
    return "!{} {} git-credential".format(
        shlex.quote(sys.executable),
        shlex.quote(os.path.abspath(__file__)))


def install_credential_helper(folder):
    git(folder, "config", "credential.helper", credential_helper_value())


# --------------------------------------------------------------------------
# commands
# --------------------------------------------------------------------------

def device_flow():
    """Zitadel OAuth 2.0 device authorization grant. Returns access token."""
    status, resp = http_json(
        "POST", f"{ZITADEL_BASE}/oauth/v2/device_authorization",
        form={"client_id": DEVICE_CLIENT_ID, "scope": DEVICE_SCOPE})
    if status != 200:
        raise SystemExit(f"device authorization failed (HTTP {status}): {resp}")
    print(f"\nTo link this device, open:\n\n  {resp.get('verification_uri_complete') or resp.get('verification_uri')}\n")
    print(f"and enter code: {resp['user_code']}\n")
    interval = int(resp.get("interval", 5))
    deadline = time.time() + int(resp.get("expires_in", 300))
    while time.time() < deadline:
        time.sleep(interval)
        status, tok = http_json(
            "POST", f"{ZITADEL_BASE}/oauth/v2/token",
            form={"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
                  "device_code": resp["device_code"],
                  "client_id": DEVICE_CLIENT_ID})
        if status == 200:
            return tok["access_token"]
        err = tok.get("error", "")
        if err == "authorization_pending":
            continue
        if err == "slow_down":
            interval += 5
            continue
        raise SystemExit(f"device flow failed: {tok}")
    raise SystemExit("device flow timed out (code expired)")


def cmd_link(args):
    zitadel_token = args.token or device_flow()
    status, resp = http_json(
        "POST", f"{args.server.rstrip('/')}/v1/link",
        body={"zitadel_access_token": zitadel_token,
              "device_name": args.device})
    if status != 200:
        raise SystemExit(f"link failed (HTTP {status}): {resp}")
    cfg = load_config()
    cfg.update({"server": args.server.rstrip("/"),
                "gitea_base": resp["gitea_base"],
                "login": resp["login"],
                "token": resp["token"],
                "token_name": resp["token_name"]})
    cfg.setdefault("folders", {})
    save_config(cfg)
    log(f"linked as {resp['login']} on {resp['gitea_base']} "
        f"(token {resp['token_name']}); config: {CONFIG_PATH}")
    return 0


def require_linked(cfg):
    """Every forge-touching command fails the same way on an unlinked box."""
    if "token" not in cfg:
        raise SystemExit("not linked yet -- run: granthi-sync link")
    return cfg


def forge_get(cfg, path, params=None):
    """Authenticated GET against the forge the link handed us. The client
    already holds a scoped user token, so read paths need no granthi-link
    round-trip."""
    url = f"{cfg['gitea_base'].rstrip('/')}{path}"
    if params:
        url = f"{url}?{urllib.parse.urlencode(params)}"
    return http_json("GET", url,
                     headers={"Authorization": f"token {cfg['token']}"})


def list_repos(cfg):
    """Every repo the linked token can see, following pagination.

    Returns (repos, truncated). `truncated` is True when FORGE_MAX_PAGES was
    hit -- a bounded page must never be presented as 'that is all of them'.
    """
    def fetch(page):
        status, resp = forge_get(cfg, "/api/v1/user/repos",
                                 {"page": page, "limit": FORGE_PAGE_LIMIT})
        if status != 200:
            raise SystemExit(f"listing repos failed (HTTP {status}): {resp}")
        return resp if isinstance(resp, list) else resp.get("data", [])

    repos, page = [], 1
    while page <= FORGE_MAX_PAGES:
        batch = fetch(page)
        repos.extend(batch)
        if len(batch) < FORGE_PAGE_LIMIT:
            return repos, False
        page += 1
    # Every page up to the cap was full, which does not by itself mean more
    # exist: a total that is an exact multiple of the page size ends on a
    # full page. One sentinel fetch tells "complete" from "truncated".
    return repos, bool(fetch(FORGE_MAX_PAGES + 1))


def cmd_list(args):
    cfg = require_linked(load_config())
    repos, truncated = list_repos(cfg)
    if not repos:
        print("no repos on the forge for this account")
        return 0
    # Which of them are already on this machine, so the table answers
    # "what can I pull down?" and not just "what exists?".
    # Keyed on full_name, not name: an account that can see both alice/cloud
    # and bob/cloud would otherwise show both as local when only one is.
    # Folders written before full_name was recorded fall back to <login>/<name>.
    local_by_full = {}
    for folder, m in cfg.get("folders", {}).items():
        full = m.get("full_name") or f"{cfg.get('login')}/{m.get('name')}"
        local_by_full[full] = folder
    rows = [("REPO", "VIS", "UPDATED", "LOCAL FOLDER")]
    for r in sorted(repos, key=lambda r: r.get("full_name") or ""):
        rows.append((r.get("full_name") or "?",
                     "private" if r.get("private") else "public",
                     (r.get("updated_at") or "")[:10],
                     local_by_full.get(r.get("full_name"), "-")))
    widths = [max(len(row[i]) for row in rows) for i in range(4)]
    for row in rows:
        print("  ".join(c.ljust(w) for c, w in zip(row, widths)))
    if truncated:
        print(f"\n... more repos exist: stopped after {FORGE_MAX_PAGES} pages "
              f"of {FORGE_PAGE_LIMIT}. This list is NOT complete.")
    return 0


_SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")


def parse_repo_arg(repo, login):
    """'<name>' or '<owner>/<name>' -> full_name. Rejects anything else.

    The result is concatenated into a URL and a filesystem path, so a segment
    carrying '?', '#', '..', an encoded slash, or an extra path component
    could redirect the clone or the remote that gets persisted. Validate
    rather than quote: the forge's own naming rules are this narrow anyway.
    """
    parts = repo.split("/") if "/" in repo else [login, repo]
    if len(parts) != 2 or not all(_SEGMENT_RE.match(p) and p not in (".", "..")
                                  for p in parts):
        raise SystemExit(
            f"invalid repo {repo!r}: expected <name> or <owner>/<name> using "
            f"letters, digits, '.', '_' or '-'")
    return "/".join(parts)


def cmd_get(args):
    cfg = require_linked(load_config())
    full_name = parse_repo_arg(args.repo, cfg["login"])
    name = full_name.rsplit("/", 1)[-1]
    dest = os.path.abspath(args.into or name)
    if os.path.exists(dest) and os.listdir(dest):
        raise SystemExit(f"refusing to clone into a non-empty path: {dest}")
    clone_url = f"{cfg['gitea_base'].rstrip('/')}/{full_name}.git"
    # -c supplies the helper *during* the clone -- install_credential_helper
    # cannot run first because the repo does not exist yet -- and git also
    # persists it into the new repo's config. --origin names the remote
    # 'granthi' up front so `watch` picks the folder up without a rename.
    proc = subprocess.run(
        ["git", "clone",
         "-c", f"credential.helper={credential_helper_value()}",
         "--origin", "granthi", clone_url, dest],
        capture_output=True, text=True)
    if proc.returncode != 0:
        raise SystemExit(f"clone failed: {proc.stderr.strip()}")
    install_credential_helper(dest)  # idempotent; guarantees persistence
    # symbolic-ref, not rev-parse: an empty repo has an unborn HEAD.
    rc, branch = git(dest, "symbolic-ref", "--short", "HEAD", check=False)
    if rc != 0 or not branch:
        branch = "main"
    cfg.setdefault("folders", {})[dest] = {
        "name": name, "full_name": full_name, "branch": branch,
        "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "diverged": False}
    save_config(cfg)
    log(f"cloned {full_name} -> {dest} (branch {branch}); "
        f"`granthi-sync watch` will keep it synced")
    return 0


def cmd_add(args):
    cfg = require_linked(load_config())
    folder = os.path.abspath(args.folder)
    if not os.path.isdir(folder):
        raise SystemExit(f"no such folder: {folder}")
    name = args.name or re.sub(r"[^a-zA-Z0-9._-]+", "-", os.path.basename(folder))
    ensure_repo(folder)
    install_credential_helper(folder)
    status, resp = http_json(
        "POST", f"{cfg['server']}/v1/repos",
        body={"token": cfg["token"], "name": name, "private": args.private})
    if status == 200:
        clone_url = resp["clone_url"]
    elif status == 409:
        clone_url = f"{cfg['gitea_base']}/{cfg['login']}/{name}.git"
        log(f"cloud repo {name} already exists; reusing")
    else:
        raise SystemExit(f"repo create failed (HTTP {status}): {resp}")
    rc, _ = git(folder, "remote", "get-url", "granthi", check=False)
    if rc == 0:
        git(folder, "remote", "set-url", "granthi", clone_url)
    else:
        git(folder, "remote", "add", "granthi", clone_url)
    autocommit(folder)
    git(folder, "push", "-u", "granthi", "main")
    cfg.setdefault("folders", {})[folder] = {
        "name": name, "full_name": f"{cfg['login']}/{name}", "branch": "main",
        "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "diverged": False}
    save_config(cfg)
    log(f"linked folder {folder} -> {clone_url}")
    return 0


_STOP = False


def _sigterm(signum, frame):
    global _STOP
    _STOP = True
    log(f"signal {signum} received; finishing current pass then exiting")


def watch_pass():
    cfg = load_config()
    for folder, meta in sorted(cfg.get("folders", {}).items()):
        if not os.path.isdir(folder):
            log(f"SKIP {folder}: missing")
            continue
        outcome, detail = sync_folder(folder, branch=meta.get("branch", "main"))
        meta["diverged"] = outcome == "diverged"
        if outcome in ("pushed", "synced"):
            meta["last_sync"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
        if outcome == "diverged":
            log(f"DIVERGED {folder}: {detail}")
        elif outcome == "error":
            log(f"ERROR {folder}: {detail}")
        else:
            log(f"{outcome} {folder}: {detail}")
    save_config(cfg)


def cmd_watch(args):
    signal.signal(signal.SIGTERM, _sigterm)
    signal.signal(signal.SIGINT, _sigterm)
    log(f"granthi-sync watch starting (interval {args.interval}s)")
    while True:
        watch_pass()
        if args.once or _STOP:
            break
        deadline = time.time() + args.interval
        while time.time() < deadline and not _STOP:
            time.sleep(1)
        if _STOP:
            break
    log("watch stopped cleanly")
    return 0


def cmd_status(args):
    cfg = load_config()
    if "login" in cfg:
        print(f"linked: {cfg['login']} @ {cfg.get('gitea_base')} "
              f"(token {cfg.get('token_name')})")
    else:
        print("not linked")
    folders = cfg.get("folders", {})
    if not folders:
        print("no folders linked")
        return 0
    rows = [("FOLDER", "REPO", "LAST SYNC", "DIVERGED")]
    for folder, meta in sorted(folders.items()):
        rows.append((folder, meta.get("name", "?"),
                     meta.get("last_sync", "never"),
                     "YES" if meta.get("diverged") else "no"))
    widths = [max(len(r[i]) for r in rows) for i in range(4)]
    for r in rows:
        print("  ".join(c.ljust(w) for c, w in zip(r, widths)))
    return 0


def main(argv=None):
    p = argparse.ArgumentParser(prog="granthi-sync",
                                description="Granthi folder-to-cloud sync client")
    p.add_argument("--version", action="version", version=VERSION)
    sub = p.add_subparsers(dest="cmd", required=True)

    sp = sub.add_parser("link", help="link this device to your granthi account")
    sp.add_argument("--server", default=DEFAULT_SERVER)
    sp.add_argument("--token", help="ready Zitadel access token (headless path)")
    sp.add_argument("--device", default=os.uname().nodename.split(".")[0])
    sp.set_defaults(fn=cmd_link)

    sp = sub.add_parser("list", help="list forge repos this account can see")
    sp.set_defaults(fn=cmd_list)

    sp = sub.add_parser("get", help="clone a forge repo and keep it synced")
    sp.add_argument("repo", help="repo name, or owner/repo")
    sp.add_argument("--into", help="target folder (default: ./<repo>)")
    sp.set_defaults(fn=cmd_get)

    sp = sub.add_parser("add", help="link a folder and push it to the cloud")
    sp.add_argument("folder")
    sp.add_argument("--name")
    grp = sp.add_mutually_exclusive_group()
    grp.add_argument("--private", dest="private", action="store_true", default=True)
    grp.add_argument("--public", dest="private", action="store_false")
    sp.set_defaults(fn=cmd_add)

    sp = sub.add_parser("watch", help="sync loop over linked folders")
    sp.add_argument("--interval", type=int, default=30)
    sp.add_argument("--once", action="store_true", help="single pass then exit")
    sp.set_defaults(fn=cmd_watch)

    sp = sub.add_parser("status", help="show linked folders")
    sp.set_defaults(fn=cmd_status)

    argv = list(sys.argv[1:] if argv is None else argv)
    if argv and argv[0] == "git-credential":  # hidden helper protocol
        return cmd_git_credential(argv[1:])
    args = p.parse_args(argv)
    return args.fn(args)


if __name__ == "__main__":
    sys.exit(main())