Insecure credentials

example.py

from typing import Any, Callable

import grpc
from grpc_interceptor import ClientCallDetails, ClientInterceptor

from authzed.api.v1 import Client, ReadSchemaRequest
from authzed.api.v1.permission_service_pb2_grpc import PermissionsServiceStub
from authzed.api.v1.schema_service_pb2_grpc import SchemaServiceStub


class TokenAuthorization(ClientInterceptor):
    def __init__(self, token: str):
        self._token = token

    def intercept(
        self,
        method: Callable,
        request_or_iterator: Any,
        call_details: grpc.ClientCallDetails,
    ):
        metadata = [("authorization", f"Bearer {self._token}")]
        if call_details.metadata is not None:
            metadata = metadata.extend(call_details.metadata)

        print("final metadata: ", metadata)

        new_details = ClientCallDetails(
            call_details.method,
            call_details.timeout,
            metadata,
            call_details.credentials,
            call_details.wait_for_ready,
            call_details.compression,
        )

        return method(request_or_iterator, new_details)


class CustomClient(Client):
    """
    v1 Authzed gRPC API client.
    """

    def __init__(
        self,
        target: str,
        token: str,
        options=None,
        compression=None,
    ):
        fake_credentials = grpc.local_channel_credentials()
        super().__init__(target, fake_credentials, options, compression)
        auth_interceptor = TokenAuthorization(token)

        insecure_channel = grpc.insecure_channel(target, options, compression)
        channel = grpc.intercept_channel(insecure_channel, auth_interceptor)
        SchemaServiceStub.__init__(self, channel)
        PermissionsServiceStub.__init__(self, channel)


client = CustomClient(
    "10.0.4.80:50051",
    "testtesttesttest",
)


resp = client.ReadSchema(ReadSchemaRequest())
print(resp.schema_text)