Python: How to pass header and data into gRPC API code

Viewed 31

Hi folks I am new to gRPCs. I am currently building a client that can send requests to the server. I have hit a bit of a roadblock and cant figure out how to continue. I need to code into python the following gRPCurl call:

grpcurl -H "Authorization: Bearer aAbcdEefgHh123poIjuYHtgbNmCxxx" -H "x-team-mesh: TEST" -d '{"context":{"project_id":"this-is-my-project-id"}}' api.app.co.com:443 team.app.infrastructure.v3.App.envs_pb2_grpc.ListEnvs ../protobuf/app-api/team/app/infrastructure/v3/envs.proto

I have looked at examples such as these but they are not quite the same structure as my code. I cannot figure out how to pass in the header and data body when instantiating a channel on the client side.

How and where do I place the -H (header) and -d (body?) parts of the gRPCurl call?

import grpc
import team.app.infrastructure.v3.branches_pb2 as pb2
import team.app.infrastructure.v3.branches_pb2_grpc as pb2_grpc
import google.api
import sys
from okta.token_auth import OktaToken


class AppAuth(grpc.AuthMetadataPlugin):
    def __init__(self, key):
        self._key = key
    def __call__(self, context, callback):
        callback((('rpc-auth-header', self._key),), None)


class AppClient(object):

    def __init__(self):
        self.server = 'api.app.co.com:443'

        with open('auth/certs/DigiCertClientRSA4096RootG5.crt.pem', 'rb') as fh:
            rootCert = fh.read()

        new_token = OktaToken()

        metadata = []
        metadata.append(('context', 'project_id', 'this-is-my-project-id'))
        metadata.append(('x-team-mesh', 'TEST'))
        metadata.append(('authorization', 'Bearer ' + new_token.okta_token_request()))


        self.channel = grpc.secure_channel('{}'.format(self.server),
                                    grpc.composite_channel_credentials(
                                        grpc.ssl_channel_credentials(),
                                        grpc.ssl_channel_credentials(rootCert),
                                        grpc.metadata_call_credentials(
                                            BitcAuth(new_token.okta_token_request())
                                            )
                                        )
                                    )

        try:
            grpc.channel_ready_future(self.channel).result(timeout=3)
        except grpc.FutureTimeoutError:
            sys.exit('Error connecting to server')
        else:
            self.stub = pb2_grpc.EnvsStub(self.channel)

    def get_envs(self):

        request = pb2.ListEnvsRequest()
        response = self.stub.ListEnvs(request)
        print(f'{response}')

        return response


if __name__ == '__main__':

    client = AppClient()
    result = client.get_envs()

I suspect that this line may be where to make the necessary changes based on what I have read so far but all my attempts to add metadata and payload(?) have been unsuccessful: response = self.stub.ListEnvs(request).

Since the curl call works fine, and my python code does not Im pretty sure I am not passing on the right pieces in the gRPC API call:

custom error:

debug_error_string = "UNKNOWN:Error received from peer ipv4:127.0.0.1:443 {grpc_message:"RBAC: access denied", grpc_status:7, created_time:"2022-09-12T17:59:22.654641-07:00"}"

Help is greatly appreciated

0 Answers
Related