xos/coreapi/grpc_client.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import print_function
import base64
import grpc
import orm
from protos import xos_pb2_grpc, modeldefs_pb2_grpc, utility_pb2_grpc
from google.protobuf.empty_pb2 import Empty
from grpc import (
    metadata_call_credentials,
    composite_channel_credentials,
    ssl_channel_credentials,
)

SERVER_CA = "/usr/local/share/ca-certificates/local_certs.crt"


class UsernamePasswordCallCredentials(grpc.AuthMetadataPlugin):
    """Metadata wrapper for raw access token credentials."""

    def __init__(self, username, password):
        self._username = username
        self._password = password

    def __call__(self, context, callback):
        basic_auth = "Basic %s" % base64.b64encode(
            "%s:%s" % (self._username, self._password)
        )
        metadata = (("Authorization", basic_auth),)
        callback(metadata, None)


class SessionIdCallCredentials(grpc.AuthMetadataPlugin):
    """Metadata wrapper for raw access token credentials."""

    def __init__(self, sessionid):
        self._sessionid = sessionid

    def __call__(self, context, callback):
        metadata = (("x-xossession", self._sessionid),)
        callback(metadata, None)


class XOSClient(object):
    def __init__(self, hostname, port):
        self.hostname = hostname
        self.port = port


class InsecureClient(XOSClient):
    def __init__(self, hostname, port=50055):
        super(InsecureClient, self).__init__(hostname, port)
        self.channel = grpc.insecure_channel("%s:%d" % (self.hostname, self.port))
        self.stub = xos_pb2_grpc.xosStub(self.channel)
        self.modeldefs = modeldefs_pb2_grpc.modeldefsStub(self.channel)
        self.utility = utility_pb2_grpc.utilityStub(self.channel)

        self.xos_orm = orm.ORMStub(self.stub, "xos")


class SecureClient(XOSClient):
    def __init__(
        self,
        hostname,
        port=50051,
        cacert=SERVER_CA,
        username=None,
        password=None,
        sessionid=None,
    ):
        super(SecureClient, self).__init__(hostname, port)

        server_ca = open(cacert, "r").read()
        if sessionid:
            call_creds = metadata_call_credentials(SessionIdCallCredentials(sessionid))
        else:
            call_creds = metadata_call_credentials(
                UsernamePasswordCallCredentials(username, password)
            )
        chan_creds = ssl_channel_credentials(server_ca)
        chan_creds = composite_channel_credentials(chan_creds, call_creds)

        self.channel = grpc.secure_channel(
            "%s:%d" % (self.hostname, self.port), chan_creds
        )
        self.stub = xos_pb2_grpc.xosStub(self.channel)
        self.modeldefs = modeldefs_pb2_grpc.modeldefsStub(self.channel)
        self.utility = utility_pb2_grpc.utilityStub(self.channel)

        self.xos_orm = orm.ORMStub(self.stub, "xos")


def main():  # self-test
    client = InsecureClient("xos-core.cord.lab")
    print(client.stub.ListUser(Empty()))

    client = SecureClient(
        "xos-core.cord.lab", username="padmin@vicci.org", password="letmein"
    )
    print(client.stub.ListUser(Empty()))


if __name__ == "__main__":
    main()