Merge pull request #802 from vesellov/master

populate details about service_proxy_server() inf the api.network_status

Author: vesellov

CHANGELOG.txt

@@ -1,6 +1,30 @@
 Change Log
 ==========
 
+
+2024-09-04 Veselin Penev [[email protected]](mailto:[email protected])
+
+* bump pycryptodomex==3.20.0
+* solved another issue with removing old versions of the same file
+* calculate original file size before start the backup
+* able to populate web-socket stream data for specific files/locations
+* fix in bitdust/userid/id_url.py
+* fix bug in logging
+* fixed bug related to auto-removing oldest versions of the files
+
+
+
+2024-07-29 Veselin Penev [[email protected]](mailto:[email protected])
+
+* downgrade coincurve version
+* built supplier.storage_contract.scan_recent_storage_transactions() method
+* added caching of identities unique names
+* fix in supplier.storage_contract.verify_accept_storage_payment()
+* fix in bitdust/supplier/customer_space.py
+* fix in bitdust/blockchain/bismuth_wallet.py
+
+
+
 2024-07-12 Veselin Penev [[email protected]](mailto:[email protected])
 
 * do not reject unpaid customers for now, storage contracts are still in progress

bitdust/access/key_ring.py

@@ -462,8 +462,11 @@ def on_key_received(newpacket, info, status, error_message):
                 # but we already have a key with that ID
                 if my_keys.is_key_private(key_id):
                     # we should not overwrite existing private key
-                    # TODO: check other scenarios
-                    raise Exception('private key already registered with %r' % key_id)
+                    if my_keys.get_public_key_raw(key_id) != key_object.toPublicString():
+                        raise Exception('private key already registered : %r' % key_id)
+                    lg.warn('received public key, but matching private key is already registered %r' % key_id)
+                    p2p_service.SendAck(newpacket)
+                    return True
                 if my_keys.get_public_key_raw(key_id) != key_object.toPublicString():
                     my_keys.erase_key(key_id)
                     if not my_keys.register_key(key_id, key_object, label=key_label):

bitdust/access/shared_access_coordinator.py

@@ -669,7 +669,7 @@ def doConnectCustomerSuppliers(self, *args, **kwargs):
         self.suppliers_in_progress.clear()
         self.suppliers_succeed.clear()
         for supplier_idurl in self.known_suppliers_list:
-            self.suppliers_in_progress.append(supplier_idurl)
+            self.suppliers_in_progress.append(id_url.field(supplier_idurl))
             if id_url.is_cached(supplier_idurl):
                 self._do_connect_with_supplier(supplier_idurl)
             else:
@@ -689,7 +689,7 @@ def doSupplierRequestListFiles(self, event, *args, **kwargs):
         """
         Action method.
         """
-        supplier_idurl = kwargs['supplier_idurl']
+        supplier_idurl = id_url.field(kwargs['supplier_idurl'])
         pkt_out = None
         if event == 'supplier-file-modified':
             remote_path = kwargs['remote_path']
@@ -797,8 +797,8 @@ def doRemember(self, event, *args, **kwargs):
         """
         Action method.
         """
-        supplier_idurl = kwargs['supplier_idurl']
-        if supplier_idurl in self.suppliers_in_progress:
+        supplier_idurl = id_url.field(kwargs['supplier_idurl'])
+        if id_url.is_in(supplier_idurl, self.suppliers_in_progress):
             self.suppliers_in_progress.remove(supplier_idurl)
             if event in ['index-sent', 'index-up-to-date']:
                 if supplier_idurl not in self.suppliers_succeed:

bitdust/crypt/encrypted.py

@@ -294,7 +294,7 @@ def Unserialize(data, decrypt_key=None):
             exc,
             list(dct.keys()),
             (dct.get('c'), dct.get('b'), dct.get('i'), dct.get('r')),
-            traceback.format_tb(),
+            '\n'.join(traceback.format_stack()),
         ))
         if _Debug:
             lg.out(_DebugLevel, repr(dct))

bitdust/interface/api.py

@@ -515,13 +515,14 @@ def process_info():
             }
     if driver.is_on('service_backup_db'):
         from bitdust.storage import backup_fs
+        v = backup_fs.total_stats()
         result['file'] = {
-            'items': backup_fs.counter(),
-            'files': backup_fs.numberfiles(),
-            'files_size': backup_fs.sizefiles(),
-            'folders': backup_fs.numberfolders(),
-            'folders_size': backup_fs.sizefolders(),
-            'backups_size': backup_fs.sizebackups(),
+            'items': v['items'],
+            'files': v['files'],
+            'folders': v['folders'],
+            'files_size': v['size_files'],
+            'folders_size': v['size_folders'],
+            'backups_size': v['size_backups'],
             'customers': len(backup_fs.known_customers()),
         }
     if driver.is_on('service_shared_data'):
@@ -693,6 +694,7 @@ def network_select(name):
 
     def _on_network_disconnected(x):
         cur_base_dir = deploy.current_base_dir()
+        # TODO: must wait shutdown and init to complete with defered
         shutdowner.shutdown_services()
         shutdowner.shutdown_local()
         shutdowner.shutdown_automats()
@@ -969,6 +971,20 @@ def network_status(suppliers=False, customers=False, cache=False, tcp=False, udp
                         i['queue'] = len(s.pending_packets)
                     sessions.append(i)
                 r['proxy']['sessions'] = sessions
+            if driver.is_on('service_proxy_server'):
+                from bitdust.transport.proxy import proxy_router
+                if proxy_router.A():
+                    r['proxy']['routes'] = []
+                    for v in proxy_router.A().routes.values():
+                        _r = v['connection_info'].copy()
+                        _r['contacts'] = ', '.join(['{}:{}'.format(c[0], c[1]) for c in v['contacts']])
+                        _r['address'] = ', '.join(['{}:{}'.format(a[0], a[1]) for a in v['address']])
+                        _r.pop('id', None)
+                        _r.pop('index', None)
+                        r['proxy']['routes'].append(_r)
+                    r['proxy']['closed_routes'] = [(strng.to_text(k), strng.to_text(v)) for k, v in proxy_router.A().closed_routes.items()]
+                    r['proxy']['acks'] = len(proxy_router.A().acks)
+                    r['proxy']['hosts'] = ', '.join([('{}://{}:{}'.format(strng.to_text(k), strng.to_text(v[0]), strng.to_text(v[1]))) for k, v in proxy_router.A().my_hosts.items()])
     if dht:
         from bitdust.dht import dht_service
         r['dht'] = {}
@@ -2292,7 +2308,7 @@ def file_delete(remote_path):
     backup_fs.DeleteLocalDir(settings.getLocalBackupsDir(), pathIDfull)
     backup_fs.DeleteByID(pathID, iter=backup_fs.fs(customer_idurl, key_alias), iterID=backup_fs.fsID(customer_idurl, key_alias))
     backup_fs.Scan(customer_idurl=customer_idurl, key_alias=key_alias)
-    backup_fs.Calculate(iterID=backup_fs.fsID(customer_idurl, key_alias))
+    backup_fs.Calculate(customer_idurl=customer_idurl, key_alias=key_alias)
     if key_alias != 'master':
         if driver.is_on('service_shared_data'):
             from bitdust.access import shared_access_coordinator
@@ -2428,33 +2444,17 @@ def file_upload_start(local_path, remote_path, wait_result=False, publish_events
     if not pathID:
         return ERROR('path %s was not registered yet' % remote_path)
     keyID = my_keys.make_key_id(alias=key_alias, creator_glob_id=parts['customer'])
-    # customerID = global_id.MakeGlobalID(customer=parts['customer'], key_alias=key_alias)
     pathIDfull = packetid.MakeBackupID(keyID, pathID)
     if key_alias != 'master':
         if not driver.is_on('service_shared_data'):
             return ERROR('service_shared_data() is not started')
-
-
-#     def _restart_active_share(result):
-#         if _Debug:
-#             lg.args(_DebugLevel, result=result, key_id=keyID, path=path, pathID=pathID)
-#         if key_alias != 'master':
-#             from bitdust.access import shared_access_coordinator
-#             active_share = shared_access_coordinator.get_active_share(keyID)
-#             if not active_share:
-#                 active_share = shared_access_coordinator.SharedAccessCoordinator(key_id=keyID, publish_events=publish_events)
-#             active_share.automat('restart')
-#         return result
-
     if wait_result:
         task_created_defer = Deferred()
         tsk = backup_control.StartSingle(
             pathID=pathIDfull,
             localPath=local_path,
             keyID=keyID,
         )
-        # if key_alias != 'master':
-        #     tsk.result_defer.addCallback(_restart_active_share)
         tsk.result_defer.addCallback(
             lambda result: task_created_defer.callback(
                 OK(
@@ -2470,15 +2470,16 @@ def file_upload_start(local_path, remote_path, wait_result=False, publish_events
                 )
             )
         )
-        tsk.result_defer.addErrback(lambda result: task_created_defer.callback(ERROR(
-            'uploading task %d for %s failed: %s' % (
-                tsk.number,
-                tsk.pathID,
-                result[1],
-            ),
-            api_method='file_upload_start',
-        ), ), )
-        backup_fs.Calculate(iterID=backup_fs.fsID(customer_idurl, key_alias))
+        tsk.result_defer.addErrback(lambda result: task_created_defer.callback(ERROR(result, api_method='file_upload_start')))
+        # tsk.result_defer.addErrback(lambda result: task_created_defer.callback(ERROR(
+        #     'uploading task %d for %s failed: %s' % (
+        #         tsk.number,
+        #         tsk.pathID,
+        #         result,
+        #     ),
+        #     api_method='file_upload_start',
+        # ), ), )
+        backup_fs.Calculate(customer_idurl=customer_idurl, key_alias=key_alias)
         backup_control.SaveFSIndex(customer_idurl, key_alias)
         if _Debug:
             lg.out(_DebugLevel, 'api.file_upload_start %s with %s, wait_result=True' % (remote_path, pathIDfull))
@@ -2489,10 +2490,8 @@ def file_upload_start(local_path, remote_path, wait_result=False, publish_events
         localPath=local_path,
         keyID=keyID,
     )
-    # if key_alias != 'master':
-    #     tsk.result_defer.addCallback(_restart_active_share)
-    tsk.result_defer.addErrback(lambda result: lg.err('errback from api.file_upload_start.task(%s) failed with %s' % (result[0], result[1])))
-    backup_fs.Calculate(iterID=backup_fs.fsID(customer_idurl, key_alias))
+    tsk.result_defer.addErrback(lambda result: lg.err('errback from api.file_upload_start.task() failed with %r' % result))
+    backup_fs.Calculate(customer_idurl=customer_idurl, key_alias=key_alias)
     backup_control.SaveFSIndex(customer_idurl, key_alias)
     if _Debug:
         lg.out(_DebugLevel, 'api.file_upload_start %s with %s' % (remote_path, pathIDfull))

bitdust/p2p/p2p_service.py

@@ -51,6 +51,10 @@
 
 #------------------------------------------------------------------------------
 
+from twisted.internet.defer import Deferred
+
+#------------------------------------------------------------------------------
+
 from bitdust.logs import lg
 
 from bitdust.contacts import contactsdb
@@ -66,8 +70,11 @@
 from bitdust.crypt import signed
 from bitdust.crypt import my_keys
 
+from bitdust.services import driver
+
 from bitdust.transport import gateway
 from bitdust.transport import callback
+from bitdust.transport import packet_out
 
 from bitdust.userid import my_id
 
@@ -195,6 +202,91 @@ def outbox(outpacket, wide, callbacks, target=None, route=None, response_timeout
 #------------------------------------------------------------------------------
 
 
+def on_request_service_received(newpacket, info):
+    if len(newpacket.Payload) > 1024*10:
+        lg.warn('too long payload')
+        SendFail(newpacket, 'too long payload')
+        return True
+    try:
+        json_payload = serialization.BytesToDict(newpacket.Payload, keys_to_text=True, values_to_text=True)
+        json_payload['name']
+        json_payload['payload']
+    except:
+        lg.warn('json payload invalid')
+        SendFail(newpacket, 'json payload invalid')
+        return True
+    service_name = str(json_payload['name'])
+    if _Debug:
+        lg.out(_Debug, 'service_p2p_hookups.RequestService {%s} from %s' % (service_name, newpacket.OwnerID))
+    if not driver.is_exist(service_name):
+        lg.warn('got wrong payload in %s' % service_name)
+        SendFail(newpacket, 'service %s not exist' % service_name)
+        return True
+    if not driver.is_on(service_name):
+        SendFail(newpacket, 'service %s is off' % service_name)
+        return True
+    try:
+        result = driver.request(service_name, json_payload['payload'], newpacket, info)
+    except:
+        lg.exc()
+        SendFail(newpacket, 'request processing failed with exception')
+        return True
+    if not result:
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_request_service SKIP request %s' % service_name)
+        return False
+    if isinstance(result, Deferred):
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_request_service fired delayed execution')
+    elif isinstance(result, packet_out.PacketOut):
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_request_service outbox packet sent')
+    return True
+
+
+def on_cancel_service_received(newpacket, info):
+    if len(newpacket.Payload) > 1024*10:
+        SendFail(newpacket, 'too long payload')
+        return True
+    try:
+        json_payload = serialization.BytesToDict(newpacket.Payload, keys_to_text=True, values_to_text=True)
+        json_payload['name']
+        json_payload['payload']
+    except:
+        SendFail(newpacket, 'json payload invalid')
+        return True
+    service_name = json_payload['name']
+    if _Debug:
+        lg.out(_Debug, 'service_p2p_hookups.CancelService {%s} from %s' % (service_name, newpacket.OwnerID))
+    if not driver.is_exist(service_name):
+        lg.warn('got wrong payload in %s' % newpacket)
+        SendFail(newpacket, 'service %s not exist' % service_name)
+        return True
+    if not driver.is_on(service_name):
+        SendFail(newpacket, 'service %s is off' % service_name)
+        return True
+    try:
+        result = driver.cancel(service_name, json_payload['payload'], newpacket, info)
+    except:
+        lg.exc()
+        SendFail(newpacket, 'request processing failed with exception')
+        return True
+    if not result:
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_cancel_service SKIP request %s' % service_name)
+        return False
+    if isinstance(result, Deferred):
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_cancel_service fired delayed execution')
+    elif isinstance(result, packet_out.PacketOut):
+        if _Debug:
+            lg.out(_Debug, 'service_p2p_hookups._send_cancel_service outbox packet sent')
+    return True
+
+
+#------------------------------------------------------------------------------
+
+
 def Ack(newpacket, info):
     if _Debug:
         lg.out(_DebugLevel, 'p2p_service.Ack %s from [%s] at %s://%s with %d bytes payload' % (newpacket.PacketID, nameurl.GetName(newpacket.CreatorID), info.proto, info.host, len(newpacket.Payload)))

bitdust/services/driver.py

@@ -474,7 +474,7 @@ def _on_started(start_result, stop_result, dependencies_results):
 
     def _on_failed(err):
         lg.warn('failed service %s in driver.restart() : %r' % (service_name, err))
-        restart_result.errback(str(err))
+        restart_result.errback(Exception(str(err)))
         return None
 
     def _do_start(stop_result=None, dependencies_results=None):