python: Add support for fetching rights from a Daml 2.0 ledger.

Author: da-tanabe

.gitignore

@@ -18,6 +18,8 @@ python/dazl.egg-info
 python/dist
 python/pip-wheel-metadata
 app.log
+canton.log
+canton_errors.log
 navigator.log
 sandbox.log
 target

VERSION

@@ -1 +1 @@
-7.6.5
+7.7.0

_build/daml-connect/daml-connect.conf

@@ -3,7 +3,7 @@
 
 [tool.poetry]
 name = "dazl"
-version = "7.6.5"
+version = "7.7.0"
 description = "high-level Ledger API client for Daml ledgers"
 license = "Apache-2.0"
 authors = ["Davin K. Tanabe <davin.tanabe@digitalasset.com>"]

pyproject.toml

@@ -1,6 +1,8 @@
 # Copyright (c) 2017-2022 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
+import abc
 from datetime import datetime
+import sys
 from typing import (
     TYPE_CHECKING,
     AbstractSet,
@@ -19,7 +21,15 @@
 from ..prim import LEDGER_STRING_REGEX, ContractData, ContractId, Party, to_parties
 from ..util.typing import safe_cast
 
+if sys.version_info >= (3, 8):
+    from typing import final
+else:
+    from typing_extensions import final
+
+
 __all__ = [
+    "ActAs",
+    "Admin",
     "ApplicationMeteringReport",
     "ArchiveEvent",
     "Boundary",
@@ -35,6 +45,8 @@
     "ExerciseResponse",
     "ParticipantMeteringReport",
     "PartyInfo",
+    "ReadAs",
+    "Right",
     "SubmitResponse",
     "User",
 ]
@@ -603,6 +615,54 @@ def __init__(self, id: str, primary_party: Party):
         self.primary_party = primary_party
 
 
+class Right(abc.ABC):
+    def __setattr__(self, key, value):
+        """
+        Overridden to make Right objects read-only.
+        """
+        raise AttributeError
+
+
+@final
+class ReadAs(Right):
+    __slots__ = ("party",)
+    __match_args__ = ("party",)
+
+    party: Party
+
+    def __init__(self, __party: Party):
+        object.__setattr__(self, "party", __party)
+
+    def __repr__(self) -> str:
+        return f"ReadAs({self.party!r})"
+
+
+@final
+class ActAs(Right):
+    __slots__ = ("party",)
+    __match_args__ = ("party",)
+
+    party: Party
+
+    def __init__(self, __party: Party):
+        object.__setattr__(self, "party", __party)
+
+    def __repr__(self) -> str:
+        return f"ActAs({self.party!r})"
+
+
+@final
+class _Admin(Right):
+    __slots__ = ()
+    __match_args__ = ()
+
+    def __repr__(self) -> str:
+        return "Admin"
+
+
+Admin = _Admin()
+
+
 class PartyInfo:
     """
     Full information about a ``Party``.

python/dazl/ledger/api_types.py

@@ -19,6 +19,7 @@
     Optional,
     Union,
 )
+import warnings
 
 from ...prim import Party
 from .exc import ConfigError
@@ -189,7 +190,7 @@ def token(self) -> str:
         raise NotImplementedError
 
     @property
-    def token_version(self) -> "Optional[Literal[1]]":
+    def token_version(self) -> "Optional[Literal[1, 2]]":
         """
         The version of the token supplied at configuration time, as provided by a signing authority
         that is trusted by the server.
@@ -210,6 +211,8 @@ class TokenBasedAccessConfig(AccessConfig):
     party rights, the application name, and ledger ID are all derived off of the token.
     """
 
+    _token_version: Literal[1, 2]
+
     def __init__(self, oauth_token: str):
         """
         Initialize a token-based access configuration.
@@ -229,17 +232,34 @@ def token(self) -> str:
     @token.setter
     def token(self, value: str) -> None:
         self._token = value
-        claims = decode_token(self._token)
+        claims = decode_token_claims(self._token)
+
+        v1_claims = claims.get(DamlLedgerApiNamespace)
+        if v1_claims is not None:
+            self._set(
+                read_as=frozenset(claims.get("readAs", ())),
+                act_as=frozenset(claims.get("actAs", ())),
+                admin=bool(claims.get("admin", False)),
+            )
+            self._ledger_id = v1_claims.get("ledgerId", None)
+            self._application_name = v1_claims.get("applicationId", None)
+            self._token_version = 1
+        else:
+            self._token_version = 2
 
-        read_as = frozenset(claims.get("readAs", ()))
-        act_as = frozenset(claims.get("actAs", ()))
+    def _set(self, *, read_as: Collection[Party], act_as: Collection[Party], admin: bool):
+        """
+        Set the values of this :class:`TokenBasedAccessConfig`.
+
+        This is not a public API, and subject to change at any time.
+        """
+        read_as = frozenset(read_as)
+        act_as = frozenset(act_as)
 
         self._act_as = act_as
         self._read_only_as = read_as - act_as
         self._read_as = read_as.union(act_as)
-        self._admin = bool(claims.get("admin", False))
-        self._ledger_id = claims.get("ledgerId", None)
-        self._application_name = claims.get("applicationId", None)
+        self._admin = admin
 
     @property
     def read_as(self) -> AbstractSet[Party]:
@@ -266,8 +286,8 @@ def application_name(self) -> str:
         return self._application_name
 
     @property
-    def token_version(self) -> "Literal[1]":
-        return 1
+    def token_version(self) -> "Literal[1, 2]":
+        return self._token_version
 
 
 class TokenFileBasedAccessConfig(TokenBasedAccessConfig):
@@ -412,17 +432,27 @@ def parties(p: Union[None, Party, Collection[Party]]) -> Collection[Party]:
 
 
 def decode_token(token: str) -> Mapping[str, Any]:
+    warnings.warn("decode_token is deprecated; use decode_token_claims instead", DeprecationWarning)
+    claims = decode_token_claims(token)
+    claims_dict = claims.get(DamlLedgerApiNamespace)
+    if claims_dict is None:
+        raise ValueError(f"JWT is missing claim namespace: {DamlLedgerApiNamespace!r}")
+    return claims_dict
+
+
+def decode_token_claims(token: str) -> "Mapping[str, Any]":
+    """
+    Decode the claims section from a JSON Web Token (JWT).
+
+    Note that the signature is NOT verified; this is the responsibility of the caller!
+    """
     components = token.split(".", 3)
     if len(components) != 3:
         raise ValueError("not a JWT")
 
     pad_bytes = "=" * (-len(components[1]) % 4)
     claim_str = base64.urlsafe_b64decode(components[1] + pad_bytes)
-    claims = json.loads(claim_str)
-    claims_dict = claims.get(DamlLedgerApiNamespace)
-    if claims_dict is None:
-        raise ValueError(f"JWT is missing claim namespace: {DamlLedgerApiNamespace!r}")
-    return claims_dict
+    return json.loads(claim_str)
 
 
 def encode_unsigned_token(

python/dazl/ledger/config/access.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2017-2022 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import List, Tuple, Union, cast
+from typing import Any, AsyncIterable, Callable, Iterable, List, Tuple, TypeVar, Union, cast
 from urllib.parse import urlparse
 
 from grpc import (
@@ -13,12 +13,29 @@
     metadata_call_credentials,
     ssl_channel_credentials,
 )
-from grpc.aio import Channel, insecure_channel, secure_channel
+from grpc.aio import (
+    Channel,
+    ClientCallDetails,
+    StreamStreamCall,
+    StreamStreamClientInterceptor,
+    StreamUnaryCall,
+    StreamUnaryClientInterceptor,
+    UnaryStreamCall,
+    UnaryStreamClientInterceptor,
+    UnaryUnaryCall,
+    UnaryUnaryClientInterceptor,
+    insecure_channel,
+    secure_channel,
+)
 
 from ..config import Config
 
 __all__ = ["create_channel"]
 
+RequestType = TypeVar("RequestType")
+RequestIterableType = Union[Iterable[Any], AsyncIterable[Any]]
+ResponseIterableType = AsyncIterable[Any]
+
 
 def create_channel(config: "Config") -> "Channel":
     """
@@ -55,7 +72,15 @@ def create_channel(config: "Config") -> "Channel":
                 ),
             )
         return secure_channel(u.netloc, credentials, tuple(options))
+
+    elif config.access.token_version is not None:
+        # Python/C++ libraries refuse to allow "credentials" objects to be passed around on
+        # non-TLS channels, but they don't check interceptors; use an interceptor to inject
+        # an Authorization header instead
+        return insecure_channel(u.netloc, options, interceptors=[GrpcAuthInterceptor(config)])
+
     else:
+        # no TLS, no tokens--simply create an insecure channel with no adornments
         return insecure_channel(u.netloc, options)
 
 
@@ -74,3 +99,68 @@ def __call__(self, context: "AuthMetadataContext", callback: "AuthMetadataPlugin
             options.append(("authorization", "Bearer " + self._config.access.token))
 
         callback(tuple(options), None)
+
+
+class GrpcAuthInterceptor(
+    UnaryUnaryClientInterceptor,
+    UnaryStreamClientInterceptor,
+    StreamUnaryClientInterceptor,
+    StreamStreamClientInterceptor,
+):
+    """
+    An interceptor that injects "Authorization" metadata into a request.
+
+    This works around the fact that the C++ gRPC libraries (which Python is built on) highly
+    discourage sending authorization data over the wire unless the connection is protected with TLS.
+    """
+
+    # NOTE: There are a number of typing errors in the grpc.aio classes, so we're ignoring a handful
+    #  of lines until those problems are addressed.
+
+    def __init__(self, config: "Config"):
+        self._config = config
+
+    async def intercept_unary_unary(
+        self,
+        continuation: "Callable[[ClientCallDetails, RequestType], UnaryUnaryCall]",
+        client_call_details: ClientCallDetails,
+        request: RequestType,
+    ) -> "Union[UnaryUnaryCall, RequestType]":
+        return await continuation(self._modify_client_call_details(client_call_details), request)
+
+    async def intercept_unary_stream(
+        self,
+        continuation: "Callable[[ClientCallDetails, RequestType], UnaryStreamCall]",
+        client_call_details: ClientCallDetails,
+        request: RequestType,
+    ) -> "Union[ResponseIterableType, UnaryStreamCall]":
+        return await continuation(self._modify_client_call_details(client_call_details), request)
+
+    async def intercept_stream_unary(
+        self,
+        continuation: "Callable[[ClientCallDetails, RequestType], StreamUnaryCall]",
+        client_call_details: ClientCallDetails,
+        request_iterator: RequestIterableType,
+    ) -> StreamUnaryCall:
+        return await continuation(
+            self._modify_client_call_details(client_call_details), request_iterator  # type: ignore
+        )
+
+    async def intercept_stream_stream(
+        self,
+        continuation: Callable[[ClientCallDetails, RequestType], StreamStreamCall],
+        client_call_details: ClientCallDetails,
+        request_iterator: RequestIterableType,
+    ) -> "Union[ResponseIterableType, StreamStreamCall]":
+        return await continuation(
+            self._modify_client_call_details(client_call_details), request_iterator  # type: ignore
+        )
+
+    def _modify_client_call_details(self, client_call_details: ClientCallDetails):
+        if (
+            "authorization" not in client_call_details.metadata
+            and self._config.access.token_version is not None
+        ):
+            client_call_details.metadata.add("authorization", f"Bearer {self._config.access.token}")
+
+        return client_call_details