feat: add gh auth

vinca/distro.py

@@ -1,10 +1,11 @@
 import os
-import urllib.request
 
 from rosdistro import get_cached_distribution, get_index, get_index_url
 from rosdistro.dependency_walker import DependencyWalker
 from rosdistro.manifest_provider import get_release_tag
 
+from vinca import http as vinca_http
+
 
 class Distro(object):
     def __init__(
@@ -209,9 +210,8 @@ def get_package_xml_for_additional_package(self, pkg_info):
             return self._additional_xml_cache[raw_url]
 
         try:
-            with urllib.request.urlopen(raw_url) as resp:
-                xml_content = resp.read().decode("utf-8")
-                self._additional_xml_cache[raw_url] = xml_content
-                return xml_content
+            xml_content = vinca_http.fetch(raw_url).text
+            self._additional_xml_cache[raw_url] = xml_content
+            return xml_content
         except Exception as e:
             raise RuntimeError(f"Failed to fetch package.xml from {raw_url}: {e}")

vinca/http.py

@@ -0,0 +1,40 @@
+import os
+import urllib.parse
+
+import requests
+
+GITHUB_HOSTS = {"github.com", "raw.githubusercontent.com", "api.github.com"}
+
+
+def _github_token():
+    return (
+        os.environ.get("GH_TOKEN")
+        or os.environ.get("GITHUB_TOKEN")
+        or os.environ.get("API_TOKEN_GITHUB")
+    )
+
+
+def _auth_headers(url):
+    host = urllib.parse.urlparse(url).hostname or ""
+    if host in GITHUB_HOSTS:
+        token = _github_token()
+        if token:
+            return {"Authorization": f"Bearer {token}"}
+    return {}
+
+
+def fetch(url, *, timeout=30.0):
+    resp = requests.get(url, headers=_auth_headers(url), timeout=timeout)
+    if resp.status_code in (401, 403):
+        host = urllib.parse.urlparse(url).hostname or ""
+        hint = ""
+        if host in GITHUB_HOSTS and not _github_token():
+            hint = (
+                " (set GH_TOKEN, GITHUB_TOKEN, or API_TOKEN_GITHUB"
+                " to access private repos)"
+            )
+        raise RuntimeError(
+            f"Auth failed for {url}: HTTP {resp.status_code}{hint}"
+        )
+    resp.raise_for_status()
+    return resp

vinca/main.py

@@ -251,19 +251,24 @@ def read_vinca_yaml(filepath):
 
 
 def read_snapshot(vinca_conf):
-    if "rosdistro_snapshot" not in vinca_conf:
+    snapshot_key = "rosdistro_snapshot"
+    additional_key = "rosdistro_additional_recipes"
+
+    if snapshot_key not in vinca_conf and not vinca_conf.get(additional_key):
         return None, None
 
     yaml = ruamel.yaml.YAML()
-    # load primary snapshot
-    snapshot = yaml.load(open(vinca_conf["rosdistro_snapshot"], "r")) or {}
-    # if additional snapshot file specified, load and merge
-    additional_key = "rosdistro_additional_recipes"
+
+    snapshot = None
+    if snapshot_key in vinca_conf:
+        snapshot = yaml.load(open(vinca_conf[snapshot_key], "r")) or {}
+
     additional = None
-    if additional_key in vinca_conf and vinca_conf[additional_key]:
+    if vinca_conf.get(additional_key):
         additional = yaml.load(open(vinca_conf[additional_key], "r")) or {}
-        # merge additional entries, overriding or adding
-        snapshot.update(additional)
+        if snapshot is not None:
+            # merge additional entries, overriding or adding
+            snapshot.update(additional)
 
     return snapshot, additional
 
@@ -1172,34 +1177,17 @@ def main():
 
             fns = list(fn)
             for fn in fns:
-                selected_bn = None
-
                 print(f"Fetching repodata: {fn}")
                 repodata = get_repodata(fn, get_conda_subdir())
-                # currently we don't check the build numbers of local repodatas,
-                # only URLs
-                if "://" in fn:
-                    selected_bn = vinca_conf.get("build_number", 0)
 
                 all_pkgs = repodata.get("packages", {})
                 all_pkgs.update(repodata.get("packages.conda", {}))
                 for _, pkg in all_pkgs.items():
-                    is_built = False
-                    if selected_bn is not None:
-                        pkg_build_number = get_pkg_build_number(
-                            selected_bn, pkg["name"], vinca_conf
-                        )
-                        if pkg["build_number"] == pkg_build_number:
-                            is_built = True
+                    print(f"Skipping {pkg['name']}")
+                    if vinca_conf["trigger_new_versions"]:
+                        skip_built_packages.add((pkg["name"], pkg["version"]))
                     else:
-                        is_built = True
-
-                    if is_built:
-                        print(f"Skipping {pkg['name']}")
-                        if vinca_conf["trigger_new_versions"]:
-                            skip_built_packages.add((pkg["name"], pkg["version"]))
-                        else:
-                            skip_built_packages.add(pkg["name"])
+                        skip_built_packages.add(pkg["name"])
 
                 vinca_conf["skip_built_packages"] = skip_built_packages
         else:

vinca/resolve.py

@@ -1,6 +1,6 @@
 import os
-from urllib.request import urlopen
 from vinca import config
+from vinca import http as vinca_http
 
 map_platform_python_to_conda = {
     "linux-64": "linux",
@@ -22,7 +22,7 @@ def get_conda_index(vinca_conf, base_dir):
         if os.path.isfile(ip):
             rawdata = yaml.load(open(ip, "r"))
         else:
-            rawdata = yaml.load(urlopen(i))
+            rawdata = yaml.load(vinca_http.fetch(i).text)
         conda_index.append(rawdata)
     return conda_index
 

vinca/utils.py

@@ -3,7 +3,8 @@
 import os
 import time
 import json
-import requests
+
+from vinca import http as vinca_http
 
 
 class folded_unicode(str):
@@ -60,7 +61,7 @@ def get_repodata(url_or_path, platform=None):
             with open(fn) as fi:
                 return json.load(fi)
 
-    repodata = requests.get(url_or_path)
+    repodata = vinca_http.fetch(url_or_path)
     content = repodata.content
     with open(fn, "w") as fcache:
         fcache.write(content.decode("utf-8"))