Add linehaul telemetry data to User-Agent header for PyPI statistics

pyproject.toml

@@ -40,6 +40,7 @@ classifiers = [
 ]
 dependencies = [
   "click>=8.0.6",
+  "distro>=1.0.0; sys_platform == 'linux'",
   "hatchling>=1.27.0",
   "httpx>=0.22.0",
   "hyperlink>=21.0.0",

src/hatch/index/core.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import sys
 from functools import cached_property
 from typing import TYPE_CHECKING
 
@@ -52,13 +51,10 @@ def __init__(self, repo: str, *, user="", auth="", ca_cert=None, client_cert=Non
     def client(self) -> httpx.Client:
         import httpx
 
+        from hatch.utils.linehaul import get_linehaul_component
         from hatch.utils.network import DEFAULT_TIMEOUT
 
-        user_agent = (
-            f"Hatch/{__version__} "
-            f"{sys.implementation.name}/{'.'.join(map(str, sys.version_info[:3]))} "
-            f"HTTPX/{httpx.__version__}"
-        )
+        user_agent = f"Hatch/{__version__} {get_linehaul_component()} HTTPX/{httpx.__version__}"
         return httpx.Client(
             headers={"User-Agent": user_agent},
             transport=httpx.HTTPTransport(retries=3, verify=self.__verify, cert=self.__cert),

src/hatch/utils/linehaul.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import json
+import os
+import platform
+import sys
+from functools import lru_cache
+from typing import Any
+
+from hatch._version import __version__
+
+
+def get_linehaul_data() -> dict[str, Any]:
+    data: dict[str, Any] = {
+        "installer": {"name": "hatch", "version": __version__},
+        "python": platform.python_version(),
+        "implementation": {
+            "name": platform.python_implementation(),
+            "version": _get_implementation_version(),
+        },
+        "system": {
+            "name": platform.system(),
+            "release": platform.release(),
+        },
+        "cpu": platform.machine(),
+        "ci": _looks_like_ci() or None,
+    }
+
+    if sys.platform.startswith("linux"):
+        distro_info = _get_linux_distro_info()
+        if distro_info:
+            data["distro"] = distro_info
+    elif sys.platform == "darwin":
+        mac_version = platform.mac_ver()[0]
+        if mac_version:
+            data["distro"] = {"name": "macOS", "version": mac_version}
+
+    openssl_version = _get_openssl_version()
+    if openssl_version:
+        data["openssl_version"] = openssl_version
+
+    return data
+
+
+def _get_implementation_version() -> str:
+    implementation_name = platform.python_implementation()
+
+    if implementation_name == "PyPy":
+        pypy_version_info = sys.pypy_version_info  # type: ignore[attr-defined]
+        if pypy_version_info.releaselevel == "final":
+            pypy_version_info = pypy_version_info[:3]
+        return ".".join(str(x) for x in pypy_version_info)
+
+    return platform.python_version()
+
+
+def _get_linux_distro_info() -> dict[str, Any] | None:
+    import distro
+
+    name = distro.name()
+    if not name:
+        return None
+
+    distro_info: dict[str, Any] = {"name": name}
+
+    version = distro.version()
+    if version:
+        distro_info["version"] = version
+
+    codename = distro.codename()
+    if codename:
+        distro_info["id"] = codename
+
+    libc_info = _get_libc_info()
+    if libc_info:
+        distro_info["libc"] = libc_info
+
+    return distro_info
+
+
+def _get_libc_info() -> dict[str, str] | None:
+    libc, version = _get_libc_version()
+    if libc and version:
+        return {"lib": libc, "version": version}
+    return None
+
+
+def _get_libc_version() -> tuple[str | None, str | None]:
+    import ctypes
+    import ctypes.util
+
+    libc_name = ctypes.util.find_library("c")
+    if not libc_name:
+        return None, None
+
+    try:
+        libc = ctypes.CDLL(libc_name)
+        libc.gnu_get_libc_version.restype = ctypes.c_char_p
+        version = libc.gnu_get_libc_version()
+        if version:
+            return "glibc", version.decode()
+    except (OSError, AttributeError, UnicodeDecodeError):
+        pass
+
+    return None, None
+
+
+def _get_openssl_version() -> str | None:
+    try:
+        import ssl
+    except (ImportError, AttributeError):
+        return None
+    else:
+        return ssl.OPENSSL_VERSION
+
+
+def _looks_like_ci() -> bool:
+    # INFO: Matches pip's CI detection for linehaul statistics.
+    # Uses existence check (not value check) because variables like BUILD_BUILDID
+    # and BUILD_ID are set to build IDs, not boolean strings.
+    ci_env_vars = (
+        "BUILD_BUILDID",
+        "BUILD_ID",
+        "CI",
+        "GITHUB_ACTIONS",
+        "GITLAB_CI",
+        "PIP_IS_CI",
+        "TRAVIS",
+    )
+    return any(env_var in os.environ for env_var in ci_env_vars)
+
+
+@lru_cache(maxsize=1)
+def get_linehaul_component() -> str:
+    data = get_linehaul_data()
+    return json.dumps(data, separators=(",", ":"), sort_keys=True)

tests/index/test_core.py

@@ -1,11 +1,12 @@
+import json
 import platform
-import sys
 
 import httpx
 import pytest
 
 from hatch._version import __version__
 from hatch.index.core import PackageIndex
+from hatch.utils.linehaul import get_linehaul_component
 
 
 class TestRepo:
@@ -80,7 +81,50 @@ def test_user_agent_header_format(self):
 
         user_agent = client.headers["User-Agent"]
 
-        expected = (
-            f"Hatch/{__version__} {sys.implementation.name}/{platform.python_version()} HTTPX/{httpx.__version__}"
-        )
-        assert user_agent == expected
+        assert user_agent.startswith(f"Hatch/{__version__} ")
+        assert user_agent.endswith(f" HTTPX/{httpx.__version__}")
+
+    def _extract_json_from_user_agent(self, user_agent: str) -> dict:
+        json_start = user_agent.index("{")
+        json_end = user_agent.rindex("}") + 1
+        return json.loads(user_agent[json_start:json_end])
+
+    def test_user_agent_contains_linehaul_json(self):
+        index = PackageIndex("https://foo.internal/a/b/")
+        client = index.client
+
+        user_agent = client.headers["User-Agent"]
+        data = self._extract_json_from_user_agent(user_agent)
+
+        assert data["installer"]["name"] == "hatch"
+        assert data["installer"]["version"] == __version__
+        assert data["python"] == platform.python_version()
+        assert data["implementation"]["name"] == platform.python_implementation()
+        assert data["system"]["name"] == platform.system()
+        assert data["cpu"] == platform.machine()
+
+    def test_user_agent_linehaul_ci_detection(self, monkeypatch):
+        get_linehaul_component.cache_clear()
+        monkeypatch.setenv("CI", "true")
+
+        index = PackageIndex("https://foo.internal/a/b/")
+        client = index.client
+
+        user_agent = client.headers["User-Agent"]
+        data = self._extract_json_from_user_agent(user_agent)
+
+        assert data["ci"] is True
+
+    def test_user_agent_linehaul_macos_distro(self, mocker):
+        get_linehaul_component.cache_clear()
+        mocker.patch("hatch.utils.linehaul.sys.platform", "darwin")
+        mocker.patch("hatch.utils.linehaul.platform.mac_ver", return_value=("14.0", "", "arm64"))
+
+        index = PackageIndex("https://foo.internal/a/b/")
+        client = index.client
+
+        user_agent = client.headers["User-Agent"]
+        data = self._extract_json_from_user_agent(user_agent)
+
+        assert data["distro"]["name"] == "macOS"
+        assert data["distro"]["version"] == "14.0"