Python zeebe-grpc and OAuth

Hi there,

I’m trying to test my Python gRPC Library (https://gitlab.com/stephane.ludwig/zeebe_python_grpc) with Camunda Cloud but I’m a little bit lost in space with OAuth 2.0.
I would like to integrate oauthlib (https://oauthlib.readthedocs.io/) and gRPC. Does anyone have know-how on this topic?

Yo @gizmo84 , “OAuth” is a red herring. See the Node client implementation here:

You just send the env vars over REST to the specified validator service (also in the env vars), and it will give you back a token.

You then put that token in the headers for the gRPC calls. See here: https://github.com/creditsenseau/zeebe-client-node-js/blob/master/src/lib/GrpcClient.ts#L257

So if you are googling for how to implement this in Python, you would never touch an OAuth lib. You would be googling for:

  • How to read an env var in Python
  • How to do a REST POST in Python
  • How to set a metadata header in gRPC in Python

That’s pretty much it!

HTH,
Josh

Thanks Josh! You helped me a lot. The two first points are easy. I will see how to manage the third one.

1 Like

@gizmo84 How did it go? Were you able to use zeebe_python_grpc with Camunda Cloud (with OAuth) and would you be able to share your solution?

Hi @datakurre,

I didn’t had time to work on it lately but last time I tried it following code worked for me:

import requests
import grpc
from zeebe_grpc import gateway_pb2, gateway_pb2_grpc

ZEEBE_ADDRESS = "<CLUSTER_ID>.zeebe.camunda.io:443"
ZEEBE_CLIENT_ID = "<CLIENT_ID>"
ZEEBE_CLIENT_SECRET = <SECRET>
ZEEBE_AUTHORIZATION_SERVER_URL = "https://login.cloud.camunda.io/oauth/token"

audience = ZEEBE_ADDRESS.split(":")[0]
p = {
    "audience": audience,
    "client_id": ZEEBE_CLIENT_ID,
    "client_secret": ZEEBE_CLIENT_SECRET,
}

r = requests.post(ZEEBE_AUTHORIZATION_SERVER_URL, json=p)
access_token = r.json().get("access_token")

class AuthGateway(grpc.AuthMetadataPlugin):
    def __call__(self, context, callback):
        callback((("authorization", f"Bearer {access_token}"),), None)

call_credentials = grpc.metadata_call_credentials(AuthGateway())

creds = grpc.ssl_channel_credentials()

composite_credentials = grpc.composite_channel_credentials(creds, call_credentials)

with grpc.secure_channel(ZEEBE_ADDRESS, credentials=composite_credentials) as channel:
    stub = gateway_pb2_grpc.GatewayStub(channel)

    topology = stub.Topology(gateway_pb2.TopologyRequest())
    print(topology)

Let me know if it’s working for you! :slight_smile:

@gizmo84 Thanks!

For the record, I finally ended up with something like

        headers = {"Accept": "application/json",
                   "Content-Type": "application/json"}
        data = {
            "audience": ZEEBE_ADDRESS.rsplit(":", 1)[0],
            "client_id": ZEEBE_CLIENT_ID,
            "client_secret": ZEEBE_CLIENT_SECRET,
            "grant_type": "client_credentials",            
        }
        response = requests.post(
            ZEEBE_AUTHORIZATION_SERVER_URL,
            data=json.dumps(data),
            headers=headers,
        )
        token = response.json()["access_token"]  
        auth_creds = grpc.access_token_call_credentials(token)
        ssl_creds = grpc.ssl_channel_credentials()
        channel_creds = grpc.composite_channel_credentials(ssl_creds, auth_creds)
        with grpc.secure_channel(ZEEBE_ADDRESS, channel_creds) as channel:
            ...
1 Like