implemented client jwt for hercules sdk

py/hercules/internal/connection.py

@@ -12,6 +12,8 @@
 from tucana.generated.aquila import action_pb2, action_pb2_grpc
 from tucana.generated.shared import module_pb2
 
+from hercules.internal.jwt import build_logon_token
+
 
 async def create_connection(
     module: module_pb2.Module,
@@ -22,8 +24,10 @@ async def create_connection(
     channel = grpc.aio.insecure_channel(aquila_url, options=grpc_options)
     stub = action_pb2_grpc.ActionTransferServiceStub(channel)
 
+    logon_token = build_logon_token(auth_token, module.identifier)
+
     # Bi-directional stream. Authorization is passed as call metadata.
-    stream = stub.Transfer(metadata=(("authorization", auth_token),))
+    stream = stub.Transfer(metadata=(("authorization", logon_token),))
 
     await stream.write(
         action_pb2.ActionTransferRequest(

py/hercules/internal/jwt.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import time
+
+TOKEN_LIFETIME_SECS = 300
+
+_HEADER = base64.urlsafe_b64encode(
+    json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode()
+).rstrip(b"=")
+
+
+def build_logon_token(secret: str, identifier: str) -> str:
+    payload = base64.urlsafe_b64encode(
+        json.dumps(
+            {"sub": identifier, "exp": int(time.time()) + TOKEN_LIFETIME_SECS},
+            separators=(",", ":"),
+        ).encode()
+    ).rstrip(b"=")
+
+    signing_input = _HEADER + b"." + payload
+    signature = hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()
+    encoded_signature = base64.urlsafe_b64encode(signature).rstrip(b"=")
+
+    return (signing_input + b"." + encoded_signature).decode()