fix: address 3 codex [P2] findings on list/get
- parse_repo_arg(): validate <name> / <owner>/<name> 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 <login>/<name>. - 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 <[email protected]> Claude-Session: https://claude.ai/code/session_01LTARYHX7GPepi3CH3tp5pg
This commit is contained in:
Nirav Patel
co-authored by
Claude Opus 5
Claude Opus 5
parent
88604a90f6
commit
be11e319f5
@@ -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 <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_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):
|
||||
"""'<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 = 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)
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user