open-traffic-generator / openapiart

OpenAPI artifact generator
MIT License
6 stars 4 forks source link

Add ability to Connect to secured server using SSL/TLS in python SDK #446

Closed Vibaswan closed 11 months ago

Vibaswan commented 11 months ago

OTG-HW is using snappi python to run all tests. We are planning to support secure gRPC. But snappi is creating insecure channel (Please check code snippet )

def _get_stub(self):
    if self._stub is None:
        CHANNEL_OPTIONS = [
            ("grpc.enable_retries", 0),
            ("grpc.keepalive_timeout_ms", self._keep_alive_timeout),
        ]
        self._channel = grpc.insecure_channel(
            self._location, options=CHANNEL_OPTIONS
        )
        self._stub = pb2_grpc.OpenapiStub(self._channel)
    return self._stub

I think gosnappi has some hook to pass any custom channel clientConnApi := openapiart.NewApi() clientConnApi.NewGrpcTransport().SetClientConnection(conn)

It will better to provision some provision for python snappi as well.

Note: Currently this hack can use in client pyhton snappi to run secure channel (I have checked it using our certificate):

def _get_stub(self):
    if self._stub is None:
        CHANNEL_OPTIONS = [
            ("grpc.enable_retries", 0),
            ("grpc.keepalive_timeout_ms", self._keep_alive_timeout),
            ('grpc.ssl_target_name_override', [www.keysight.com](http://www.keysight.com/)),
        ]

        cafile = 'D:/OTG/Athena/certificates/server/server.crt'
        ca_chain = open(cafile,'rb').read()
        creds = grpc.ssl_channel_credentials(root_certificates=ca_chain,
                                                private_key=None,
                                                certificate_chain=None)
        self._channel = grpc.secure_channel(self._location, creds,
            options=CHANNEL_OPTIONS)
        try:
            grpc.channel_ready_future(self._channel).result(timeout=5)
            print(f"Successfully established secure channel to {self._location}")
        except:
            print(f"Timeout. Failed to establish secure channel to {self._location}")

        #self._channel = grpc.insecure_channel(
        #    self._location, options=CHANNEL_OPTIONS
        #)

        self._stub = pb2_grpc.OpenapiStub(self._channel)
    return self._stub