From be11e319f55df9ad57b02d90c69fa9d17f7c0445 Mon Sep 17 00:00:00 2001 From: Nirav Patel Date: Sat, 22 Aug 2026 14:29:05 -0400 Subject: [PATCH] fix: address 3 codex [P2] findings on list/get MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - parse_repo_arg(): validate / / against a strict segment pattern. Not shell injection (argv list, no shell), but '?', '#', '..', an encoded slash or an extra path component could redirect the clone URL and the remote that gets persisted. Validate rather than quote — the forge's own naming rules are this narrow anyway. - list now keys local folders on full_name, not bare name: an account that can see alice/cloud and bob/cloud showed BOTH as local when one was. `get` and `add` both record full_name; older entries fall back to /. - list_repos truncation was off by one page: a repo total that is an exact multiple of the page size ends on a full page and was reported as truncated. One sentinel fetch past the cap separates complete from truncated. Tests 65 -> 69, including hostile repo arguments and the exact-multiple case. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01LTARYHX7GPepi3CH3tp5pg --- client/granthi_sync_client.py | 51 ++++++++++++++++++++++++++++------- tests/test_client.py | 46 +++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 10 deletions(-) diff --git a/client/granthi_sync_client.py b/client/granthi_sync_client.py index 5e467fd..7ecaac0 100644 --- a/client/granthi_sync_client.py +++ b/client/granthi_sync_client.py @@ -326,18 +326,24 @@ def list_repos(cfg): 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'. """ - repos, page = [], 1 - while page <= FORGE_MAX_PAGES: + 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}") - batch = resp if isinstance(resp, list) else resp.get("data", []) + 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 - return repos, True + # 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): @@ -348,14 +354,19 @@ def cmd_list(args): return 0 # Which of them are already on this machine, so the table answers # "what can I pull down?" and not just "what exists?". - local_by_name = {m.get("name"): folder - for folder, m in cfg.get("folders", {}).items()} + # 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 /. + 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_name.get(r.get("name"), "-"))) + 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))) @@ -365,9 +376,29 @@ def cmd_list(args): return 0 +_SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") + + +def parse_repo_arg(repo, login): + """'' or '/' -> 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 or / using " + f"letters, digits, '.', '_' or '-'") + return "/".join(parts) + + def cmd_get(args): cfg = require_linked(load_config()) - full_name = args.repo if "/" in args.repo else f"{cfg['login']}/{args.repo}" + 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): @@ -390,7 +421,7 @@ def cmd_get(args): if rc != 0 or not branch: branch = "main" cfg.setdefault("folders", {})[dest] = { - "name": name, "branch": branch, + "name": name, "full_name": full_name, "branch": branch, "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"), "diverged": False} save_config(cfg) @@ -425,7 +456,7 @@ def cmd_add(args): autocommit(folder) git(folder, "push", "-u", "granthi", "main") cfg.setdefault("folders", {})[folder] = { - "name": name, "branch": "main", + "name": name, "full_name": f"{cfg['login']}/{name}", "branch": "main", "last_sync": datetime.now(timezone.utc).isoformat(timespec="seconds"), "diverged": False} save_config(cfg) diff --git a/tests/test_client.py b/tests/test_client.py index 678211e..aa3159d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -2,6 +2,7 @@ config handling, device-flow polling (mocked HTTP). Stdlib unittest only.""" import argparse +import io import json import os import shlex @@ -287,6 +288,20 @@ class TestListRepos(unittest.TestCase): self.assertTrue(truncated) self.assertEqual(len(repos), client.FORGE_MAX_PAGES * 50) + def test_exact_multiple_of_page_size_is_not_truncated(self): + """Every page full up to the cap does not imply more exist -- a total + that is an exact multiple ends on a full page. The sentinel fetch + past the cap is what tells the two apart.""" + def fake_http(method, url, headers=None, **kw): + page = int(url.split("page=")[1].split("&")[0]) + return 200, self._page(page, 0 if page > client.FORGE_MAX_PAGES + else 50) + + with mock.patch.object(client, "http_json", fake_http): + repos, truncated = client.list_repos(self.cfg) + self.assertFalse(truncated) + self.assertEqual(len(repos), client.FORGE_MAX_PAGES * 50) + def test_http_error_is_fatal_not_silent_empty(self): with mock.patch.object(client, "http_json", lambda *a, **k: (401, {"error": "bad token"})): @@ -378,6 +393,37 @@ class TestGet(GitScenarioBase): meta = client.load_config()["folders"][os.path.abspath(dest)] self.assertEqual(meta["branch"], "main") + def test_get_rejects_hostile_repo_arguments(self): + for bad in ["../../etc/passwd", "alice/cloud?x=1", "alice/cloud#frag", + "a/b/c", "..", "-flag", "alice/../bob", "cloud%2f..", + "", "alice/"]: + with self.subTest(repo=bad): + with self.assertRaises(SystemExit): + client.cmd_get(argparse.Namespace(repo=bad, into=None)) + + def test_get_records_full_name_so_list_matches_the_right_owner(self): + dest = os.path.join(self.tmp, "pulled4") + client.cmd_get(argparse.Namespace(repo="cloud", into=dest)) + meta = client.load_config()["folders"][os.path.abspath(dest)] + self.assertEqual(meta["full_name"], "alice/cloud") + + def test_list_does_not_mark_a_same_named_other_owner_repo_as_local(self): + dest = os.path.join(self.tmp, "pulled5") + client.cmd_get(argparse.Namespace(repo="cloud", into=dest)) + cfg = client.load_config() + repos = [{"name": "cloud", "full_name": "alice/cloud", "private": True, + "updated_at": "2026-08-20T00:00:00Z"}, + {"name": "cloud", "full_name": "bob/cloud", "private": False, + "updated_at": "2026-08-20T00:00:00Z"}] + with mock.patch.object(client, "list_repos", + lambda c: (repos, False)), \ + mock.patch("sys.stdout", new_callable=io.StringIO) as out: + client.cmd_list(argparse.Namespace()) + lines = {l.split()[0]: l for l in out.getvalue().splitlines() + if l.strip()} + self.assertIn(os.path.abspath(dest), lines["alice/cloud"]) + self.assertNotIn(os.path.abspath(dest), lines["bob/cloud"]) + def test_get_never_puts_token_in_remote_url(self): dest = os.path.join(self.tmp, "pulled3") client.cmd_get(argparse.Namespace(repo="cloud", into=dest))