src/blockdb/__init__.py

@@ -14,6 +14,10 @@ def add_block_to_db(block: Block):
     # Raises db.DuplicateKey if dupe
     db.set_if_new(block_db_path, block.id, block.raw)
 
+    for func in block_storage_observers:
+        func(block)
+
+
 def get_blocks_by_type(block_type: str) -> "Generator[Block]":
     block_db = db.get_db_obj(block_db_path, 'u')
     for block_hash in db.list_keys(block_db_path):

src/gossip/client/dandelionstem/__init__.py

@@ -7,7 +7,6 @@ from typing import TYPE_CHECKING, Coroutine, List
 
 from ordered_set import OrderedSet
 
-import config
 from onionrthreads import add_delayed_thread
 from blockdb import add_block_to_db
 import logger
@@ -82,11 +81,10 @@ async def stem_out(d_phase: 'DandelionPhase'):
         sleep(1)
         return
     not_enough_edges = False
-    strict_dandelion = config.get('security.strict_dandelion', True)
 
     def blackhole_protection(q):
         for bl in q:
-            add_block_to_db(bl)
+            add_block_to_db(q)
 
 
     # Spawn threads with deep copied block queue to add to db after time
@@ -116,14 +114,7 @@ async def stem_out(d_phase: 'DandelionPhase'):
             #    "Making too few edges for stemout " +
             #    "this is bad for anonymity if frequent.",
             #    terminal=True)
-            if strict_dandelion:
             not_enough_edges = True
-            else:
-                if peer_sockets:
-                    # if we have at least 1 peer,
-                    # do dandelion anyway in non strict mode
-                    # Allow poorly connected networks to communicate faster
-                    break
             sleep(1)
         else:
             # Ran out of time for stem phase
@@ -131,7 +122,7 @@ async def stem_out(d_phase: 'DandelionPhase'):
                 logger.error(
                     "Did not stem out any blocks in time, " +
                     "if this happens regularly you may be under attack",
-                    terminal=False)
+                    terminal=True)
                 for s in peer_sockets:
                     if s:
                         s.close()

src/gossip/client/dandelionstem/stemstream.py

@@ -1,5 +1,3 @@
-from asyncio import sleep
-from queue import Empty
 from typing import TYPE_CHECKING
 
 import logger
@@ -24,14 +22,7 @@ async def do_stem_stream(
         while remaining_time > 5 and my_phase_id == d_phase.phase_id:
             # Primary client component that communicate's with gossip.server.acceptstem
             remaining_time = d_phase.remaining_time()
-            while remaining_time:
-                try:
-                    # queues can't block because we're in async
-                    bl = block_queue.get(block=False)
-                except Empty:
-                    await sleep(1)
-                else:
-                    break
+            bl: 'Block' = block_queue.get(block=True, timeout=remaining_time)
             logger.info("Sending block over dandelion++", terminal=True)
 
             block_size = str(len(bl.raw)).zfill(BLOCK_SIZE_LEN)

src/gossip/client/streamblocks.py

@@ -5,7 +5,6 @@ Download blocks that are being diffused
 doesn't apply for blocks in the gossip queue that are awaiting
 descision to fluff or stem
 """
-from ast import Index
 from threading import Thread, Semaphore
 from random import SystemRandom
 from time import sleep
@@ -121,12 +120,6 @@ def stream_from_peers():
     while True:
         need_socket_lock.acquire()
         available_set = gossip_peer_set - tried_peers
-        if not len(available_set) and len(tried_peers):
-            try:
-                tried_peers.pop()
-            except IndexError:
-                pass
-            available_set = gossip_peer_set - tried_peers
         peers = sys_rand.sample(
             available_set,
             min(MAX_STREAMS, len(available_set)))
@@ -144,4 +137,3 @@ def stream_from_peers():
             except IndexError:
                 need_socket_lock.release()
                 break
-            

src/gossip/constants.py

@@ -14,4 +14,4 @@ MAX_INBOUND_DANDELION_EDGE = 50  # Mainly picked to avoid slowloris
 OUTBOUND_DANDELION_EDGES = 2
 
 MAX_STEM_BLOCKS_PER_STREAM = 1000
-BLACKHOLE_EVADE_TIMER_SECS = 10
+BLACKHOLE_EVADE_TIMER_SECS = DANDELION_EPOCH_LENGTH * 3

src/gossip/server/diffuseblocks.py

@@ -20,7 +20,6 @@ from ..constants import BLOCK_MAX_SIZE, BLOCK_SIZE_LEN
 from ..constants import BLOCK_STREAM_OFFSET_DIGITS
 
 import logger
-import blockdb
 from blockdb import get_blocks_after_timestamp, block_storage_observers
 """
 This program is free software: you can redistribute it and/or modify
@@ -54,7 +53,13 @@ async def diffuse_blocks(reader: 'StreamReader', writer: 'StreamWriter'):
         raise ValueError(
             "Peer's specified time offset skewed too far into the future")
 
+    newly_stored_blocks = Queue()
 
+    def _add_to_queue(bl):
+        newly_stored_blocks.put_nowait(bl)
+    block_storage_observers.append(
+        _add_to_queue
+    )
 
     async def _send_block(block: 'Block'):
         writer.write(block.id)
@@ -85,20 +90,17 @@ async def diffuse_blocks(reader: 'StreamReader', writer: 'StreamWriter'):
                 )
             except IncompleteReadError:
                 keep_writing = False
-        time_offset = time()
 
         # Diffuse blocks stored since we started this stream
         while keep_writing:
-            bls = blockdb.get_blocks_after_timestamp(time_offset)
-            for bl in bls:
-                await _send_block(bl)
+            await _send_block(await newly_stored_blocks.get())
             try:
                 keep_writing = bool(
                     int.from_bytes(await reader.readexactly(1), 'big')
                 )
             except IncompleteReadError:
                 keep_writing = False
-                    break
     except Exception:
         logger.warn(traceback.format_exc(), terminal=True)
 
+    block_storage_observers.remove(_add_to_queue)

src/httpapi/addblock/__init__.py

@@ -8,7 +8,6 @@ import secrets
 from flask import Blueprint, Response, request
 
 from onionrblocks import Block
-import blockdb
 
 import logger
 from gossip import blockqueues
@@ -39,7 +38,6 @@ def block_serialized():
     req_data = request.data
     block_id = req_data[:BLOCK_ID_SIZE]
     block_data = req_data[BLOCK_ID_SIZE:]
-    blockdb.add_block_to_db(Block(block_id, block_data, auto_verify=False))
-    #blockqueues.gossip_block_queues[stream_to_use].put(
-    #Block(block_id, block_data, auto_verify=False), block=False)
+    blockqueues.gossip_block_queues[stream_to_use].put(
+        Block(block_id, block_data, auto_verify=False))
     return "ok"

src/onionrcommands/daemonlaunch/__init__.py

@@ -112,6 +112,7 @@ def daemon():
         f"Onionr daemon is running under pid {os.getpid()}", terminal=True)
     events.event('init', threaded=False)
     events.event('daemon_start')
+
     Thread(target=gossip.start_gossip_threads, daemon=True).start()
 
     try:

src/onionrplugins/__init__.py

@@ -1,9 +1,9 @@
-"""
-Onionr - Private P2P Communication.
+'''
+    Onionr - Private P2P Communication
 
-<anagement of modules/plugins.
-"""
-"""
+    This file deals with management of modules/plugins.
+'''
+'''
     This program is free software: you can redistribute it and/or modify
     it under the terms of the GNU General Public License as published by
     the Free Software Foundation, either version 3 of the License, or
@@ -16,7 +16,7 @@ GNU General Public License for more details.
 
     You should have received a copy of the GNU General Public License
     along with this program.  If not, see <https://www.gnu.org/licenses/>.
-"""
+'''
 import os, re, importlib
 import traceback
 
@@ -32,9 +32,9 @@ _instances = dict()
 config.reload()
 
 def reload(stop_event = True):
-    """
+    '''
         Reloads all the plugins
-    """
+    '''
 
     check()
 
@@ -93,9 +93,9 @@ def enable(name, start_event = True):
 
 
 def disable(name, stop_event = True):
-    """
+    '''
         Disables a plugin
-    """
+    '''
 
     check()
 
@@ -111,9 +111,9 @@ def disable(name, stop_event = True):
             stop(name)
 
 def start(name):
-    """
+    '''
         Starts the plugin
-    """
+    '''
 
     check()
 
@@ -135,9 +135,9 @@ def start(name):
     return None
 
 def stop(name):
-    """
+    '''
         Stops the plugin
-    """
+    '''
 
     check()
 
@@ -183,11 +183,12 @@ def import_module_from_file(full_path_to_module):
     return module
 
 def get_plugin(name):
-    """
+    '''
         Returns the instance of a module
-    """
+    '''
 
     check()
+
     if str(name).lower() in _instances:
         return _instances[str(name).lower()]
     else:
@@ -195,23 +196,23 @@ def get_plugin(name):
         return get_plugin(name)
 
 def get_plugins():
-    """
+    '''
         Returns a list of plugins (deprecated)
-    """
+    '''
 
     return _instances
 
 def exists(name):
-    """
+    '''
         Return value indicates whether or not the plugin exists
-    """
+    '''
 
     return os.path.isdir(get_plugins_folder(str(name).lower()))
 
 def get_enabled_plugins():
-    """
+    '''
         Returns a list of the enabled plugins
-    """
+    '''
 
     check()
     config.reload()
@@ -219,16 +220,16 @@ def get_enabled_plugins():
     return list(config.get('plugins.enabled', list()))
 
 def is_enabled(name):
-    """
+    '''
         Return value indicates whether or not the plugin is enabled
-    """
+    '''
 
     return name in get_enabled_plugins()
 
 def get_plugins_folder(name = None, absolute = True):
-    """
+    '''
         Returns the path to the plugins folder
-    """
+    '''
 
     path = ''
 
@@ -245,16 +246,16 @@ def get_plugins_folder(name = None, absolute = True):
     return path + '/'
 
 def get_plugin_data_folder(name, absolute = True):
-    """
+    '''
         Returns the location of a plugin's data folder
-    """
+    '''
 
     return get_plugins_folder(name, absolute)
 
 def check():
-    """
+    '''
         Checks to make sure files exist
-    """
+    '''
 
     if not config.is_set('plugins'):
         logger.debug('Generating plugin configuration data...')

src/onionrplugins/onionrevents.py

@@ -40,10 +40,11 @@ def __event_caller(event_name, data = {}):
         if plugin in disabled: continue
         try:
             call(plugins.get_plugin(plugin), event_name, data, get_pluginapi(data))
-        except ModuleNotFoundError as _:
+        except ModuleNotFoundError as e:
             logger.warn('Disabling nonexistant plugin "%s"...' % plugin, terminal=True)
             plugins.disable(plugin, stop_event = False)
-        except Exception as _:
+        except Exception as e:
+
             logger.error('Event "%s" failed for plugin "%s".' % (event_name, plugin), terminal=True)
             logger.error('\n' + traceback.format_exc(), terminal=True)
 

static-data/default-plugins/unixtransport/bootstrap.py

@@ -34,7 +34,8 @@ def load_existing_peers(callback: Callable):
                 daemon=True).start()
 
 
-def on_bootstrap(api, data=None):
+def on_bootstrap(api, data):
+
     callback_func = data['callback']
 
     try:

static-data/default-plugins/unixtransport/bootstrap.txt

@@ -1 +1 @@
-/dev/shm/onionr3177415330/gossip-server.sock
+/dev/shm/onionr442130151/gossip-server.sock

static-data/default-plugins/unixtransport/main.py

@@ -25,8 +25,8 @@ sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
 # import after path insert
 from unixpeer import UnixPeer
 
-from unixbootstrap import on_bootstrap
-from unixannounce import on_announce_rec
+from bootstrap import on_bootstrap
+from announce import on_announce_rec
 #from shutdown import on_shutdown_event
 
 """
@@ -57,6 +57,7 @@ def on_init(api, data=None):
     logger.info(
         f"Peers can connect to {gossip_server_socket_file}", terminal=True)
 
+
 def on_get_our_transport(api, data=None):
     callback_func = data['callback']
     for_peer = data['peer']
@@ -64,3 +65,9 @@ def on_get_our_transport(api, data=None):
         return
     if data['peer'].__class__ == UnixPeer:
         callback_func(for_peer, gossip_server_socket_file)
+
+
+def on_gossip_start(api, data: Set[Peer] = None):
+    # We don't do gossip logic
+    return
+

static-data/default-plugins/unixtransport/unixpeer.py

@@ -12,6 +12,7 @@ class UnixPeer:
 
 
     def get_socket(self, connect_timeout) -> socket:
+
         s = socket(AF_UNIX, SOCK_STREAM)
         #s.settimeout(connect_timeout)
         s.connect(self.transport_address)