Replace requests library default headers with python-sdk headers

Author: iamalwaysuncomfortable

sdk/python/aleo/_client_common.py

@@ -60,6 +60,18 @@ def package_version() -> str:
         return "0.0.0"
 
 
+def user_agent() -> str:
+    """The SDK's ``User-Agent`` string, sent on every call.
+
+    Identifies the Python SDK (and its version) in the standard, always-logged
+    header, overriding the underlying ``python-requests`` / ``python-httpx``
+    default.  Only injected on the default transport (see :func:`method_headers`
+    and the scanners' header builders); when the caller supplies their own
+    transport they own the headers, so the SDK does not set it.
+    """
+    return f"aleo-python-sdk/{package_version()}"
+
+
 def make_default_headers() -> dict[str, str]:
     return {
         "X-Aleo-SDK-Version": package_version(),
@@ -79,7 +91,7 @@ def method_headers(
 ) -> dict[str, str]:
     if has_custom_transport:
         return user_headers(headers)
-    return {**headers, "X-ALEO-METHOD": method}
+    return {**headers, "X-ALEO-METHOD": method, "User-Agent": user_agent()}
 
 
 def jwt_origin(host: str) -> str:

sdk/python/aleo/async_record_scanner.py

@@ -6,7 +6,7 @@
 from typing import Any
 from urllib.parse import urlparse
 
-from ._client_common import jwt_expired
+from ._client_common import jwt_expired, user_agent
 from ._scanner_common import (
     DecryptionNotEnabledError,
     OwnedFilter,
@@ -104,6 +104,7 @@ def __init__(
         self._account: Any | None = None
 
         # Build httpx client
+        self._has_custom_transport: bool = transport is not None
         if transport is not None:
             self._client: Any = httpx.AsyncClient(transport=transport)
         else:
@@ -174,6 +175,11 @@ def set_account(self, account: Any) -> None:
     async def _build_headers(self) -> dict[str, str]:
         """Build authentication headers, refreshing JWT if needed."""
         hdrs: dict[str, str] = {"Content-Type": "application/json"}
+        # Identify the SDK on every call — unless a custom transport owns the
+        # HTTP layer, in which case the caller controls headers.
+        sdk_ua = None if self._has_custom_transport else user_agent()
+        if sdk_ua:
+            hdrs["User-Agent"] = sdk_ua
 
         if self._api_key:
             hdrs[self._api_key["header"]] = self._api_key["value"]
@@ -183,6 +189,8 @@ async def _build_headers(self) -> dict[str, str]:
             if self._api_key and self.consumer_id:
                 jwt_url = f"{self._origin}/jwts/{self.consumer_id}"
                 jwt_hdrs = {self._api_key["header"]: self._api_key["value"]}
+                if sdk_ua:
+                    jwt_hdrs["User-Agent"] = sdk_ua
                 resp = await self._client.post(jwt_url, headers=jwt_hdrs)
                 if resp.is_success:
                     auth = resp.headers.get("Authorization") or resp.headers.get("authorization")

sdk/python/aleo/record_scanner.py

@@ -7,7 +7,7 @@
 
 import requests
 
-from ._client_common import jwt_expired
+from ._client_common import jwt_expired, user_agent
 from ._scanner_common import (
     DecryptionNotEnabledError,
     OwnedFilter,
@@ -120,6 +120,11 @@ def _http(self, method: str, url: str, **kwargs: Any) -> requests.Response:
     def _build_headers(self) -> dict[str, str]:
         """Build authentication headers, refreshing JWT if needed."""
         hdrs: dict[str, str] = {"Content-Type": "application/json"}
+        # Identify the SDK on every call — unless a custom transport owns the
+        # HTTP layer, in which case the caller controls headers.
+        sdk_ua = None if callable(self._transport) else user_agent()
+        if sdk_ua:
+            hdrs["User-Agent"] = sdk_ua
 
         # Always attach api_key header if set
         if self._api_key:
@@ -132,6 +137,8 @@ def _build_headers(self) -> dict[str, str]:
                 # Refresh JWT
                 jwt_url = f"{self._origin}/jwts/{self.consumer_id}"
                 jwt_hdrs = {self._api_key["header"]: self._api_key["value"]}
+                if sdk_ua:
+                    jwt_hdrs["User-Agent"] = sdk_ua
                 resp = self._http("POST", jwt_url, headers=jwt_hdrs)
                 if resp.ok:
                     auth = resp.headers.get("Authorization") or resp.headers.get("authorization")

sdk/python/tests/test_network_client.py

@@ -233,6 +233,9 @@ def test_default_sdk_headers_present() -> None:
     assert "X-Aleo-SDK-Version" in req_headers
     assert "X-Aleo-environment" in req_headers
     assert req_headers["X-Aleo-environment"] == "python"
+    # The standard User-Agent identifies the SDK (overriding requests' default).
+    from aleo._client_common import package_version
+    assert req_headers["User-Agent"] == f"aleo-python-sdk/{package_version()}"
 
 
 @resp_lib.activate
@@ -244,6 +247,11 @@ def test_per_method_header() -> None:
     assert req_headers.get("X-ALEO-METHOD") == "getBlock"
 
 
+def test_user_agent_value() -> None:
+    from aleo._client_common import package_version, user_agent
+    assert user_agent() == f"aleo-python-sdk/{package_version()}"
+
+
 def test_custom_transport_callable_used_for_requests() -> None:
     """A callable transport is invoked for every HTTP request."""
     import requests as _requests
@@ -273,6 +281,30 @@ def test_custom_transport_suppresses_sdk_headers() -> None:
     req_headers = resp_lib.calls[0].request.headers
     assert "X-Aleo-SDK-Version" not in req_headers
     assert "X-ALEO-METHOD" not in req_headers
+    # UA is suppressed under a custom transport: the SDK does not set its own
+    # (requests' own default python-requests/... may still be present).
+    assert not req_headers.get("User-Agent", "").startswith("aleo-python-sdk/")
+
+
+def test_custom_transport_preserves_user_supplied_user_agent() -> None:
+    """Under a custom transport the caller owns headers — a User-Agent they set
+    is passed through untouched (not stripped as an SDK header)."""
+    import requests as _requests
+
+    captured: dict[str, Any] = {}
+
+    def my_transport(method: str, url: str, **kwargs: Any) -> _requests.Response:
+        captured["headers"] = dict(kwargs.get("headers") or {})
+        r = _requests.Response()
+        r.status_code = 200
+        r._content = b"{}"
+        return r
+
+    c = AleoNetworkClient(
+        BASE, network=NET, transport=my_transport, headers={"User-Agent": "myapp/1.0"}
+    )
+    c.get_latest_block()
+    assert captured["headers"].get("User-Agent") == "myapp/1.0"
 
 
 @resp_lib.activate

sdk/python/tests/test_network_client_async.py

@@ -249,6 +249,8 @@ def handler(req: httpx.Request) -> httpx.Response:
     await c.get_latest_block()
     assert "X-Aleo-SDK-Version" in captured[0].headers
     assert captured[0].headers.get("X-Aleo-environment") == "python"
+    from aleo._client_common import package_version
+    assert captured[0].headers.get("User-Agent") == f"aleo-python-sdk/{package_version()}"
 
 
 @pytest.mark.asyncio

sdk/python/tests/test_record_scanner.py

@@ -662,3 +662,43 @@ def test_find_record_returns_first() -> None:
 
     record = scanner.find_record({"uuid": GOLDEN_UUID})  # type: ignore[arg-type]
     assert record == OWNED_RECORDS[0]
+
+
+# ---------------------------------------------------------------------------
+# User-Agent header
+# ---------------------------------------------------------------------------
+
+@resp_lib.activate
+def test_scanner_sends_user_agent() -> None:
+    """Every scanner call carries the SDK User-Agent."""
+    from aleo.mainnet import Field
+    from aleo._client_common import package_version
+
+    resp_lib.add(resp_lib.POST, f"{HOST}/records/owned", json=[])
+    scanner = _make_scanner()
+    scanner._uuid = Field.from_string(GOLDEN_UUID)
+    scanner.owned({"uuid": GOLDEN_UUID, "unspent": True})  # type: ignore[arg-type]
+
+    hdrs = resp_lib.calls[0].request.headers
+    assert hdrs["User-Agent"] == f"aleo-python-sdk/{package_version()}"
+
+
+def test_scanner_custom_transport_suppresses_user_agent() -> None:
+    """A custom transport owns the HTTP layer — the SDK sets no User-Agent."""
+    import requests as _requests
+    from aleo.mainnet import Field
+
+    captured: dict[str, Any] = {}
+
+    def transport(method: str, url: str, **kwargs: Any) -> _requests.Response:
+        captured["headers"] = dict(kwargs.get("headers") or {})
+        r = _requests.Response()
+        r.status_code = 200
+        r._content = b"[]"
+        return r
+
+    scanner = _make_scanner(transport=transport)
+    scanner._uuid = Field.from_string(GOLDEN_UUID)
+    scanner.owned({"uuid": GOLDEN_UUID, "unspent": True})  # type: ignore[arg-type]
+
+    assert "User-Agent" not in captured["headers"]

sdk/python/tests/test_record_scanner_async.py

@@ -429,3 +429,48 @@ async def test_async_find_credits_records_success() -> None:
 
     result3 = await scanner3.find_credits_records([999], {})  # type: ignore[arg-type]
     assert result3 == []
+
+
+# ---------------------------------------------------------------------------
+# User-Agent header
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_async_scanner_sends_user_agent() -> None:
+    """Async scanner carries the SDK User-Agent when it owns the transport."""
+    from aleo.mainnet import Field
+    from aleo._client_common import package_version
+
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["ua"] = request.headers.get("user-agent")
+        return httpx.Response(200, json=[])
+
+    # No transport passed to __init__ (so it is NOT treated as a custom
+    # transport); swap the client for a mock one, mirroring the network-client
+    # test pattern.
+    scanner = AsyncRecordScanner(BASE_URL)
+    scanner._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    scanner._uuid = Field.from_string(GOLDEN_UUID)
+    await scanner.owned({"uuid": GOLDEN_UUID, "unspent": True})  # type: ignore[arg-type]
+
+    assert captured["ua"] == f"aleo-python-sdk/{package_version()}"
+
+
+@pytest.mark.asyncio
+async def test_async_scanner_custom_transport_suppresses_user_agent() -> None:
+    """A custom transport (passed to __init__) suppresses the SDK User-Agent."""
+    from aleo.mainnet import Field
+
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["ua"] = request.headers.get("user-agent")
+        return httpx.Response(200, json=[])
+
+    scanner = AsyncRecordScanner(BASE_URL, transport=httpx.MockTransport(handler))
+    scanner._uuid = Field.from_string(GOLDEN_UUID)
+    await scanner.owned({"uuid": GOLDEN_UUID, "unspent": True})  # type: ignore[arg-type]
+
+    assert not (captured["ua"] or "").startswith("aleo-python-sdk/")