diff --git a/scripts/upstream.py b/scripts/upstream.py index f7429f06..a4a4a9a5 100644 --- a/scripts/upstream.py +++ b/scripts/upstream.py @@ -20,6 +20,9 @@ from pathlib import Path USER_AGENT = "retrobios-profile-sync/1.0" ABSENT = "\0absent\0" GITHUB_COMPARE_CAP = 300 +GITHUB_HOSTS = frozenset( + {"github.com", "api.github.com", "raw.githubusercontent.com"} +) _HOSTS: dict[str, tuple[str, str, str]] = { "github.com": ( @@ -125,19 +128,20 @@ def raw_url(repo: Repo, sha: str, path: str) -> str: return f"{repo.raw_base}/{repo.owner}/{repo.name}/raw/commit/{sha}/{quoted}" -def _headers(accept_json: bool = False) -> dict[str, str]: +def _headers(url: str, accept_json: bool = False) -> dict[str, str]: + """Request headers. GITHUB_TOKEN is only ever sent to GitHub.""" headers = {"User-Agent": USER_AGENT} if accept_json: headers["Accept"] = "application/json" token = os.environ.get("GITHUB_TOKEN", "") - if token: + if token and urllib.parse.urlsplit(url).netloc in GITHUB_HOSTS: headers["Authorization"] = f"token {token}" return headers def _http_text(url: str) -> str | None: """Body of a GET, or None on 404. Replaced in tests.""" - req = urllib.request.Request(url, headers=_headers()) + req = urllib.request.Request(url, headers=_headers(url)) try: with urllib.request.urlopen(req, timeout=30) as resp: return resp.read().decode("utf-8", errors="replace") @@ -153,7 +157,7 @@ def _http_text(url: str) -> str | None: def _http_json(url: str) -> object | None: """Parsed JSON body of a GET, or None on 404. Replaced in tests.""" - req = urllib.request.Request(url, headers=_headers(accept_json=True)) + req = urllib.request.Request(url, headers=_headers(url, accept_json=True)) try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode()) diff --git a/tests/test_upstream.py b/tests/test_upstream.py index 7e43b650..d33fa3b2 100644 --- a/tests/test_upstream.py +++ b/tests/test_upstream.py @@ -93,6 +93,50 @@ class TestMakeRepo(unittest.TestCase): self.assertIsNone(make_repo("example.invalid", "o", "n")) +class TestTokenScope(unittest.TestCase): + """GITHUB_TOKEN must never reach a forge other than GitHub.""" + + def setUp(self): + self._orig = os.environ.get("GITHUB_TOKEN") + os.environ["GITHUB_TOKEN"] = "gho_secret" + + def tearDown(self): + if self._orig is None: + os.environ.pop("GITHUB_TOKEN", None) + else: + os.environ["GITHUB_TOKEN"] = self._orig + + def test_sent_to_github_api(self): + h = upstream._headers("https://api.github.com/repos/o/n/commits") + self.assertEqual(h["Authorization"], "token gho_secret") + + def test_sent_to_github_raw(self): + h = upstream._headers("https://raw.githubusercontent.com/o/n/sha/a.c") + self.assertIn("Authorization", h) + + def test_withheld_from_codeberg(self): + h = upstream._headers("https://codeberg.org/api/v1/repos/o/n/commits") + self.assertNotIn("Authorization", h) + + def test_withheld_from_gitlab(self): + h = upstream._headers("https://gitlab.com/api/v4/projects/x/repository/commits") + self.assertNotIn("Authorization", h) + + def test_withheld_from_forgejo_instances(self): + for host in ("git.citron-emu.org", "git.eden-emu.dev"): + h = upstream._headers(f"https://{host}/api/v1/repos/o/n/commits") + self.assertNotIn("Authorization", h, host) + + def test_withheld_from_a_lookalike_host(self): + h = upstream._headers("https://github.com.evil.example/repos/o/n") + self.assertNotIn("Authorization", h) + + def test_absent_token_adds_no_header(self): + os.environ.pop("GITHUB_TOKEN", None) + h = upstream._headers("https://api.github.com/repos/o/n") + self.assertNotIn("Authorization", h) + + class TestCache(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory()