Merge pull request #32 from aperture-data/release-0.1.0

Release 0.1.0

Author: luisremis

aperturedb/Connector.py

@@ -1,37 +1,220 @@
-import os
+#
+# The MIT License
+#
+# @copyright Copyright (c) 2017 Intel Corporation
+# @copyright Copyright (c) 2021 ApertureData Inc
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"),
+# to deal in the Software without restriction,
+# including without limitation the rights to use, copy, modify,
+# merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+# ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+#
+
 import sys
+import os
+import socket
+import struct
 import time
+import json
+import ssl
 
-import vdms
+# VDMS Protobuf import (autogenerated)
+from . import queryMessage_pb2
 
 BACKOFF_TIME = 0.2 # seconds
 BACKOFF_MULTIPLIER = 1.2
 
+PROTOCOL_VERSION = 1
+
 class Connector(object):
 
-    def __init__(self, ip="localhost", port=55555):
+    def __init__(self, host="localhost", port=55555,
+                 user="", password="", token="",
+                 session=None,
+                 use_ssl=True):
 
-        self.ip   = ip
-        self.port = port
+        self.use_ssl = use_ssl
 
-        self.connector = vdms.vdms()
-        self.connector.connect(ip, port)
+        self.host = host
+        self.port = port
 
+        self.connected = False
+        self.last_response   = ''
         self.last_query_time = 0
 
+        self.session_token = ""
+
+        self._connect()
+
+        if session:
+            self.session = session
+        else:
+            try:
+                self._authenticate(user, password, token)
+            except Exception as e:
+                raise Exception("Authentication failed:", str(e))
+
+        self.session_token = self.session["session_token"]
+
+    def _send_msg(self, data):
+
+        sent_len = struct.pack('@I', len(data)) # send size first
+        self.conn.send(sent_len)
+        self.conn.send(data)
+
+    def _recv_msg(self):
+
+        recv_len = self.conn.recv(4) # get message size
+
+        recv_len = struct.unpack('@I', recv_len)[0]
+        response = b''
+        while len(response) < recv_len:
+            packet = self.conn.recv(recv_len - len(response))
+            if not packet:
+                print("Error receiving")
+            response += packet
+
+        return response
+
+    def _authenticate(self, user, password="", token=""):
+
+        query = [{
+            "Authenticate": {
+                "username": user,
+            }
+        }]
+
+        if password:
+            query[0]["Authenticate"]["password"] = password
+        elif token:
+            query[0]["Authenticate"]["token"] = token
+        else:
+            raise Exception("Either password or token must be specified for authentication")
+
+        response, _ = self.query(query)
+
+        self.session = response[0]["Authenticate"]
+        if self.session["status"] != 0:
+            raise Exception(self.session["info"])
+
+        self.session_token = self.session["session_token"]
+
+    def _connect(self):
+
+        self.conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.conn.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
+
+        # TCP_QUICKACK only supported in Linux 2.4.4+.
+        # We use startswith for checking the platform following Python's
+        # documentation:
+        # https://docs.python.org/dev/library/sys.html#sys.platform
+        if sys.platform.startswith('linux'):
+            self.conn.setsockopt(socket.SOL_TCP, socket.TCP_QUICKACK, 1)
+
+        self.conn.connect((self.host, self.port))
+
+        # Handshake with server to negotiate protocol
+
+        protocol = 2 if self.use_ssl else 1
+
+        hello_msg = struct.pack('@II', PROTOCOL_VERSION, protocol)
+
+        # Send desire protocol
+        self._send_msg(hello_msg)
+
+        # Receive response from server
+        response = self._recv_msg()
+
+        version, server_protocol = struct.unpack('@II', response)
+
+        if version != PROTOCOL_VERSION:
+            print("WARNING: Protocol version differ from server")
+
+        if server_protocol != protocol:
+            self.conn.close()
+            self.connected = False
+            raise Exception("Server did not accept protocol. Aborting Connection.")
+
+        if self.use_ssl:
+
+            # Server is ok with SSL, we switch over SSL.
+            self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+            self.context.check_hostname = False
+            # TODO, we need to add support for local certificates
+            # For now, we let the server send us the certificate
+            self.context.verify_mode = ssl.VerifyMode.CERT_NONE
+            self.conn = self.context.wrap_socket(self.conn)
+
+        self.connected = True
+
     def __del__(self):
 
-        self.connector.disconnect()
+        self.conn.close()
+        self.connected = False
 
     def create_new_connection(self):
 
-        return Connector(self.ip, self.port)
+        return Connector(self.host, self.port, session=self.session)
+
+    def _query(self, query, blob_array = []):
+
+        # Check the query type
+        if not isinstance(query, str): # assumes json
+            query_str = json.dumps(query)
+        else:
+            query_str = query
 
+        if not self.connected:
+            return "NOT CONNECTED"
+
+        query_msg = queryMessage_pb2.queryMessage()
+        # query has .json and .blobs
+        query_msg.json = query_str
+
+        # Set Auth token, only when not authenticated before
+        query_msg.token = self.session_token
+
+        for blob in blob_array:
+            query_msg.blobs.append(blob)
+
+        # Serialize with protobuf and send
+        data = query_msg.SerializeToString();
+        self._send_msg(data)
+
+        response = self._recv_msg()
+
+        querRes = queryMessage_pb2.queryMessage()
+        querRes.ParseFromString(response)
+
+        response_blob_array = []
+        for b in querRes.blobs:
+            response_blob_array.append(b)
+
+        self.last_response = json.loads(querRes.json)
+
+        return (self.last_response, response_blob_array)
+
+    # This is the API method, that has a retry mechanism
     def query(self, q, blobs=[], n_retries=0):
 
         if n_retries == 0:
             start = time.time()
-            self.response, self.blobs = self.connector.query(q, blobs)
+            self.response, self.blobs = self._query(q, blobs)
             self.last_query_time = time.time() - start
             return self.response, self.blobs
 
@@ -49,15 +232,15 @@ def query(self, q, blobs=[], n_retries=0):
                 error_msg += " retries\n"
                 sys.stderr.write(error_msg)
                 sys.stderr.write("Response: \n")
-                sys.stderr.write(self.connector.get_last_response_str())
+                sys.stderr.write(self.get_last_response_str())
                 sys.stderr.write("\n")
                 sys.stderr.write("Query: \n")
                 sys.stderr.write(str(q))
                 sys.stderr.write("\n")
                 break
 
             start = time.time()
-            self.response, self.blobs = self.connector.query(q, blobs)
+            self.response, self.blobs = self._query(q, blobs)
             self.last_query_time = time.time() - start
             status = self.check_status(self.response)
 
@@ -67,7 +250,11 @@ def query(self, q, blobs=[], n_retries=0):
 
     def get_last_response_str(self):
 
-        return self.connector.get_last_response_str()
+        return json.dumps(self.last_response, indent=4, sort_keys=False)
+
+    def print_last_response(self):
+
+        print(self.get_last_response_str())
 
     def get_last_query_time(self):
 

aperturedb/queryMessage_pb2.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="aperturedb",
-    version="0.0.13",
+    version="0.1.0",
     description="ApertureDB Client Module",
     install_requires=['vdms', 'scikit-image', 'image',
                       'opencv-python', 'numpy', 'matplotlib', 'pandas'],