feat: Notification monitoring system

at_client/atclient.py

@@ -1,4 +1,11 @@
+import base64
 import json
+from queue import Empty
+import time
+import traceback
+
+from at_client.connections.notification.atevents import AtEvent, AtEventType
+
 
 from .common.atsign import AtSign
 from .util.verbbuilder import *
@@ -8,20 +15,26 @@
 from .exception.atexception import *
 from .connections.atrootconnection import AtRootConnection
 from .connections.atsecondaryconnection import AtSecondaryConnection
+from .connections.atmonitorconnection import AtMonitorConnection
+from .util.atconstants import *
 from .connections.address import Address
 from .common.keys import Keys, SharedKey, PrivateHiddenKey, PublicKey, SelfKey
 from .util.authutil import AuthUtil
 
 class AtClient(ABC):
     def __init__(self, atsign:AtSign, root_address:Address=Address("root.atsign.org", 64), secondary_address:Address=None, verbose:bool = False):
+        global shared_queue
         self.atsign = atsign
+        self.queue = shared_queue
+        self.monitor_connection = None
         self.keys = KeysUtil.load_keys(atsign)
         self.verbose = verbose
         if secondary_address is None:
             self.root_connection = AtRootConnection.get_instance(host=root_address.host, 
                                                             port=root_address.port, 
                                                             verbose=verbose)
             secondary_address = self.root_connection.find_secondary(atsign)
+        self.secondary_address = secondary_address
         self.secondary_connection = AtSecondaryConnection(secondary_address, verbose=verbose)
         self.secondary_connection.connect()
         AuthUtil.authenticate_with_pkam(self.secondary_connection, self.atsign, self.keys)
@@ -320,7 +333,77 @@ def delete(self, key):
                 raise AtSecondaryConnectException(f"Failed to execute {command} - {e}")
         else:
             raise NotImplementedError(f"No implementation found for key type: {type(key)}")
-
+    
     def __del__(self):
         if self.secondary_connection:
             self.secondary_connection.disconnect()
+
+    def start_monitor(self):
+        global should_be_running_lock
+        what = ""
+        try:
+            if self.monitor_connection == None:
+                what = "construct an AtMonitorConnection"
+                self.monitor_connection = AtMonitorConnection(queue=self.queue, atsign=self.atsign, address=self.secondary_address, verbose=True)
+                self.monitor_connection.connect()
+                AuthUtil.authenticate_with_pkam(self.monitor_connection, self.atsign, self.keys)
+            should_be_running_lock.acquire(blocking=1)
+            if not self.monitor_connection.running:
+                should_be_running_lock.release()
+                what = "call monitor_connection.start_monitor()"
+                self.monitor_connection.start_monitor()
+            else:
+                should_be_running_lock.release()
+        except Exception as e:
+            print("SEVERE: failed to " + what + " : " + str(e))
+            traceback.print_exc()
+
+    def stop_monitor(self):
+        global should_be_running_lock
+        what = ""
+        try:
+            if self.monitor_connection == None:
+                return
+            should_be_running_lock.acquire(blocking=1)
+            if not self.monitor_connection.running:
+                should_be_running_lock.release()
+                what = "call monitor_connection.stop_monitor()"
+                self.monitor_connection.stop_monitor()
+            else:
+                should_be_running_lock.release()
+        except Exception as e:
+            print("SEVERE: failed to " + what + " : " + str(e))
+            traceback.print_exc()
+
+    def handle_event(self, queue, at_event):
+        try:
+            event_type = at_event.event_type
+            event_data = at_event.event_data
+
+            if event_type == AtEventType.SHARED_KEY_NOTIFICATION:
+                if event_data["value"] != None:
+                    shared_shared_key_name = event_data["key"]
+                    shared_shared_key_encrypted_value = event_data["value"]
+                    try:
+                        shared_key_decrypted_value = EncryptionUtil.rsa_decrypt_from_base64(shared_shared_key_encrypted_value, self.keys[KeysUtil.encryption_private_key_name])
+                        self.keys[shared_shared_key_name] = shared_key_decrypted_value
+                    except Exception as e:
+                        print(str(time.time()) + ": caught exception " + str(e) + " while decrypting received shared key " + shared_shared_key_name)
+            elif event_type == AtEventType.UPDATE_NOTIFICATION:
+                if event_data["value"] != None:
+                    key = event_data["key"]
+                    encrypted_value = event_data["value"]
+                    ivNonce = event_data["metadata"]["ivNonce"]
+                    try:
+                        encryption_key_shared_by_other = self.get_encryption_key_shared_by_other(SharedKey.from_string(key=key))
+                        decrypted_value = EncryptionUtil.aes_decrypt_from_base64(encrypted_text=encrypted_value.encode(), self_encryption_key=encryption_key_shared_by_other, iv=base64.b64decode(ivNonce))
+                        new_event_data = dict(event_data)
+                        new_event_data["decryptedValue"] = decrypted_value
+                        new_at_event = AtEvent(AtEventType.DECRYPTED_UPDATE_NOTIFICATION, new_event_data)
+                        queue.put(new_at_event)
+                    except Exception as e:
+                        print(str(time.time()) + ": caught exception " + str(e) + " while decrypting received data with key name [" + key + "]")
+        except Empty:
+            pass
+
+

at_client/common/keys.py

@@ -134,7 +134,7 @@ def from_string(key: str) -> 'SharedKey':
             raise AtException("SharedKey.from_string('" + key + "'): key must have structure @bob:foo.bar@alice")
         key_name = split_by_at_sign[0]
         shared_by = split_by_at_sign[1]
-        shared_key = SharedKey(AtSign(shared_by), AtSign(shared_with))
+        shared_key = SharedKey(key_name, AtSign(shared_by), AtSign(shared_with))
         shared_key.name = key_name
         return shared_key
 

at_client/common/metadata.py

@@ -3,6 +3,8 @@
 from dateutil.parser import parse
 from dataclasses import dataclass
 
+# from ..util.atconstants import *
+
 
 @dataclass
 class Metadata:
@@ -30,6 +32,12 @@ class Metadata:
     shared_key_enc: str = None
     pub_key_cs: str = None
     encoding: str = None
+
+    # enc_key_name: str = None
+    # enc_algo: str = None
+    # iv_nonce: str = None
+    # ske_enc_key_name: str = None
+    # ske_enc_algo: str = None
 
     def parse_datetime(datetime_str):
         if datetime_str is not None:
@@ -64,6 +72,13 @@ def from_json(json_str):
         metadata.shared_key_enc = data.get('sharedKeyEnc')
         metadata.pub_key_cs = data.get('pubKeyCS')
         metadata.encoding = data.get('encoding')
+
+        # metadata.encKeyName = data.get(ENCRYPTING_KEY_NAME)
+        # metadata.encAlgo = data.get(ENCRYPTING_ALGO)
+        # metadata.ivNonce = data.get(IV_OR_NONCE)
+        # metadata.skeEncKeyName = data.get(SHARED_KEY_ENCRYPTED_ENCRYPTING_KEY_NAME)
+        # metadata.skeEncAlgo = data.get(SHARED_KEY_ENCRYPTED_ENCRYPTING_ALGO)
+
         return metadata
 
     @staticmethod
@@ -93,6 +108,13 @@ def from_dict(data_dict):
         metadata.shared_key_enc = data_dict.get('sharedKeyEnc')
         metadata.pub_key_cs = data_dict.get('pubKeyCS')
         metadata.encoding = data_dict.get('encoding')
+
+        # metadata.enc_key_name = data_dict.get(ENCRYPTING_KEY_NAME)
+        # metadata.enc_algo = data_dict.get(ENCRYPTING_ALGO)
+        # metadata.iv_nonce = data_dict.get(IV_OR_NONCE)
+        # metadata.ske_enc_key_name = data_dict.get(SHARED_KEY_ENCRYPTED_ENCRYPTING_KEY_NAME)
+        # metadata.ske_enc_algo = data_dict.get(SHARED_KEY_ENCRYPTED_ENCRYPTING_ALGO)
+
         return metadata
 
 
@@ -120,6 +142,9 @@ def __str__(self):
             s += f":isEncrypted:{'true' if self.is_encrypted else 'false'}"
         if self.encoding:
             s += f":encoding:{self.encoding}"
+
+        # TO?DO: Add new parameters    
+
         return s
 
     @staticmethod
@@ -146,4 +171,11 @@ def squash(first_metadata, second_metadata):
         metadata.shared_key_enc = first_metadata.shared_key_enc if first_metadata.shared_key_enc is not None else second_metadata.shared_key_enc
         metadata.pub_key_cs = first_metadata.pub_key_cs if first_metadata.pub_key_cs is not None else second_metadata.pub_key_cs
         metadata.encoding = first_metadata.encoding if first_metadata.encoding is not None else second_metadata.encoding
+
+        # metadata.enc_key_name = first_metadata.enc_key_name if first_metadata.enc_key_name is not None else second_metadata.enc_key_name
+        # metadata.enc_algo = first_metadata.enc_algo if first_metadata.enc_algo is not None else second_metadata.enc_algo
+        # metadata.iv_nonce = first_metadata.iv_nonce if first_metadata.iv_nonce is not None else second_metadata.iv_nonce
+        # metadata.ske_enc_key_name = first_metadata.ske_enc_key_name if first_metadata.ske_enc_key_name is not None else second_metadata.ske_enc_key_name
+        # metadata.ske_enc_algo = first_metadata.ske_enc_algo if first_metadata.ske_enc_algo is not None else second_metadata.ske_enc_algo
+
         return metadata

at_client/connections/__init__.py

@@ -1,4 +1,7 @@
 from .address import Address
 from .atconnection import AtConnection
 from .atrootconnection import AtRootConnection
-from .atsecondaryconnection import AtSecondaryConnection
+from .atsecondaryconnection import AtSecondaryConnection
+from .atmonitorconnection import AtMonitorConnection
+from .notification.atevents import AtEventType
+from .notification.atnotification import AtNotification

at_client/connections/atconnection.py

@@ -1,6 +1,10 @@
+import asyncio
 import socket
 import ssl
 from abc import ABC, abstractmethod
+import traceback
+
+from ..util.socketutil import SocketUtil
 
 from ..exception.atexception import AtException
 from .response import Response
@@ -29,6 +33,7 @@ def __init__(self, host:str, port:int, context:ssl.SSLContext, verbose:bool=Fals
         self._secure_root_socket = None
         self._verbose = verbose
         self._connected = False
+        self.monitor_connection = None
 
     def __str__(self):
         """
@@ -77,10 +82,15 @@ def connect(self):
         Establish a connection to the server. Throws IOException
         """
         if not self._connected:
+            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self._socket.settimeout(None) 
             self._socket.connect(self._addr_info)
             self._secure_root_socket = self._context.wrap_socket(
                 self._socket, server_hostname=self._host, do_handshake_on_connect=True
             )
+
+            self._stream_reader = SocketUtil(self._secure_root_socket)
+
             self._connected = True
             self.read()