mirror of
https://github.com/Abdess/retroarch_system.git
synced 2026-10-10 13:33:24 -05:00
fix: send github token to github only
This commit is contained in:
1 parent
a3cb6c59ae
commit
8957f86d46
2 files changed
+52
-4
No files matched your search
+8
-4
@@ -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())
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user