diff --git a/src/hiero_sdk_python/channels.py b/src/hiero_sdk_python/channels.py index 8963303a3..0b9287a96 100644 --- a/src/hiero_sdk_python/channels.py +++ b/src/hiero_sdk_python/channels.py @@ -1,5 +1,10 @@ from __future__ import annotations +from collections import namedtuple +from importlib.metadata import PackageNotFoundError, version + +import grpc + from hiero_sdk_python.hapi.services import ( address_book_service_pb2_grpc, consensus_service_pb2_grpc, @@ -14,6 +19,84 @@ ) +class _UserAgentInterceptor(grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor): + """ + gRPC interceptor that appends an x-user-agent header to all outgoing requests. + """ + + _HEADER_KEY = "x-user-agent" + _SDK_NAME = "hiero-sdk-python" + _CallDetails = namedtuple( + "_CallDetails", + ("method", "timeout", "metadata", "credentials", "wait_for_ready", "compression"), + ) + + def __init__(self) -> None: + """ + Initialize the interceptor and compute the user agent value. + The user agent is computed once during initialization to avoid repeated package metadata lookups on every request. + """ + + try: + sdk_version = version(self._SDK_NAME) + except PackageNotFoundError: + sdk_version = "dev" + self._user_agent = f"{self._SDK_NAME}/{sdk_version}" + + def _with_user_agent(self, details: grpc.ClientCallDetails) -> grpc.ClientCallDetails: + """ + Append the user agent header to the call details. + + Args: + details: The original gRPC call details. + + Returns: + A new ClientCallDetails object with the x-user-agent header included in the metadata. + """ + metadata = [] if details.metadata is None else list(details.metadata) + metadata = [entry for entry in metadata if entry[0] != self._HEADER_KEY] + metadata.append((self._HEADER_KEY, self._user_agent)) + + return self._CallDetails( + details.method, + details.timeout, + metadata, + getattr(details, "credentials", None), + getattr(details, "wait_for_ready", None), + getattr(details, "compression", None), + ) + + def intercept_unary_unary(self, continuation, client_call_details, request): + """ + Intercept unary-unary calls and append the user agent header. + + Args: + continuation: The gRPC continuation function to call the next interceptor or actual RPC. + client_call_details: The details of the gRPC call, including method, timeout, metadata, etc. + request: The request object being sent. + + Returns: + The result of the gRPC call after appending the user agent header. + """ + + return continuation(self._with_user_agent(client_call_details), request) + + def intercept_unary_stream(self, continuation, client_call_details, request): + """ + Intercept unary-stream calls and append the user agent header. + + Args: + continuation: The gRPC continuation function to call the next interceptor or actual RPC. + client_call_details: The details of the gRPC call, including method, timeout, metadata, etc. + request: The request object being sent. + + Returns: + The result of the gRPC call after appending the user agent header. + """ + + return continuation(self._with_user_agent(client_call_details), request) + + class _Channel: """ The _Channel class is a wrapper around gRPC channels that provides access to various diff --git a/src/hiero_sdk_python/client/client.py b/src/hiero_sdk_python/client/client.py index 068f006c9..c3f95ff21 100644 --- a/src/hiero_sdk_python/client/client.py +++ b/src/hiero_sdk_python/client/client.py @@ -12,6 +12,7 @@ from dotenv import load_dotenv from hiero_sdk_python.account.account_id import AccountId +from hiero_sdk_python.channels import _UserAgentInterceptor from hiero_sdk_python.crypto.private_key import PrivateKey from hiero_sdk_python.hapi.mirror import ( consensus_service_pb2_grpc as mirror_consensus_grpc, @@ -163,6 +164,8 @@ def _init_mirror_stub(self) -> None: self.mirror_channel = grpc.secure_channel(mirror_address, grpc.ssl_channel_credentials()) else: self.mirror_channel = grpc.insecure_channel(mirror_address) + + self.mirror_channel = grpc.intercept_channel(self.mirror_channel, _UserAgentInterceptor()) self.mirror_stub = mirror_consensus_grpc.ConsensusServiceStub(self.mirror_channel) def set_operator(self, account_id: AccountId, private_key: PrivateKey) -> None: diff --git a/src/hiero_sdk_python/node.py b/src/hiero_sdk_python/node.py index a83918508..9fba5cf6c 100644 --- a/src/hiero_sdk_python/node.py +++ b/src/hiero_sdk_python/node.py @@ -9,7 +9,7 @@ from hiero_sdk_python.account.account_id import AccountId from hiero_sdk_python.address_book.node_address import NodeAddress -from hiero_sdk_python.channels import _Channel +from hiero_sdk_python.channels import _Channel, _UserAgentInterceptor from hiero_sdk_python.managed_node_address import _ManagedNodeAddress @@ -147,6 +147,8 @@ def _get_channel(self): else: channel = grpc.insecure_channel(str(self._address)) + channel = grpc.intercept_channel(channel, _UserAgentInterceptor()) + self._channel = _Channel(channel) return self._channel diff --git a/tests/unit/channel_interceptor_test.py b/tests/unit/channel_interceptor_test.py new file mode 100644 index 000000000..d467e21f8 --- /dev/null +++ b/tests/unit/channel_interceptor_test.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import grpc +import pytest + +from hiero_sdk_python import channels +from hiero_sdk_python.channels import _UserAgentInterceptor + + +pytestmark = pytest.mark.unit + + +class _DummyCallDetails: + def __init__( + self, + metadata=None, + method: str = "/proto.Service/Method", + timeout: float | None = None, + credentials=None, + wait_for_ready: bool | None = None, + compression=None, + ): + self.method = method + self.timeout = timeout + self.metadata = metadata + self.credentials = credentials + self.wait_for_ready = wait_for_ready + self.compression = compression + + +def test_user_agent_value_from_installed_version(monkeypatch): + """Interceptor should build x-user-agent from package version when available.""" + + monkeypatch.setattr(channels, "version", lambda _name: "0.0.0") + + interceptor = _UserAgentInterceptor() + assert interceptor._user_agent == "hiero-sdk-python/0.0.0" + + +def test_user_agent_value_falls_back_to_dev(monkeypatch): + """Interceptor should fall back to dev when package metadata is unavailable.""" + + def _raise_not_found(_name): + raise channels.PackageNotFoundError + + monkeypatch.setattr(channels, "version", _raise_not_found) + + interceptor = _UserAgentInterceptor() + assert interceptor._user_agent == "hiero-sdk-python/dev" + + +def test_intercept_unary_unary_appends_user_agent_metadata(monkeypatch): + """Unary calls should forward metadata including x-user-agent.""" + + monkeypatch.setattr(channels, "version", lambda _name: "0.0.0") + + interceptor = _UserAgentInterceptor() + request = object() + captured = {} + + def continuation(client_call_details, req): + captured["details"] = client_call_details + captured["request"] = req + return "ok" + + details = _DummyCallDetails(metadata=[("existing", "value")]) + result = interceptor.intercept_unary_unary(continuation, details, request) + + assert result == "ok" + assert captured["request"] is request + assert ("existing", "value") in captured["details"].metadata + expected_header = (interceptor._HEADER_KEY, f"{interceptor._SDK_NAME}/0.0.0") + assert expected_header in captured["details"].metadata + + +def test_intercept_unary_stream_adds_metadata_when_none(monkeypatch): + """Unary-stream calls should work even when original metadata is None.""" + + monkeypatch.setattr(channels, "version", lambda _name: "0.0.0") + + interceptor = _UserAgentInterceptor() + request = object() + captured = {} + + def continuation(client_call_details, req): + captured["details"] = client_call_details + captured["request"] = req + return "stream" + + details = _DummyCallDetails(metadata=None) + result = interceptor.intercept_unary_stream(continuation, details, request) + + assert result == "stream" + assert captured["request"] is request + expected_header = (interceptor._HEADER_KEY, f"{interceptor._SDK_NAME}/0.0.0") + assert captured["details"].metadata == [expected_header] + + +def test_interceptor_implements_required_grpc_interfaces(): + """Guard against accidental loss of gRPC interceptor interface inheritance.""" + interceptor = _UserAgentInterceptor() + assert isinstance(interceptor, grpc.UnaryUnaryClientInterceptor) + assert isinstance(interceptor, grpc.UnaryStreamClientInterceptor)