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