Reproduce attack and implement fix for liveness issue with ABA/commoncoin

honeybadgerbft/core/binaryagreement.py

@@ -1,10 +1,61 @@
 import gevent
 from gevent.event import Event
+
 from collections import defaultdict
+import logging
 
 from honeybadgerbft.exceptions import RedundantMessageError, AbandonedNodeError
 
 
+logger = logging.getLogger(__name__)
+
+
+def handle_conf_messages(*, sender, message, conf_values, pid, bv_signal):
+    _, r, v = message
+    assert v in ((0,), (1,), (0, 1))
+    if sender in conf_values[r][v]:
+        logger.warn(f'Redundant CONF received {message} by {sender}',
+                    extra={'nodeid': pid, 'epoch': r})
+        # FIXME: Raise for now to simplify things & be consistent
+        # with how other TAGs are handled. Will replace the raise
+        # with a continue statement as part of
+        # https://github.com/initc3/HoneyBadgerBFT-Python/issues/10
+        raise RedundantMessageError(
+            'Redundant CONF received {}'.format(message))
+
+    conf_values[r][v].add(sender)
+    logger.debug(
+        f'add v = {v} to conf_value[{r}] = {conf_values[r]}',
+        extra={'nodeid': pid, 'epoch': r},
+    )
+
+    bv_signal.set()
+
+
+def wait_for_conf_values(*, pid, N, f, epoch, conf_sent, bin_values,
+                         values, conf_values, bv_signal, broadcast):
+    conf_sent[epoch][tuple(values)] = True
+    logger.debug(f"broadcast {('CONF', epoch, tuple(values))}",
+                 extra={'nodeid': pid, 'epoch': epoch})
+    broadcast(('CONF', epoch, tuple(bin_values[epoch])))
+    while True:
+        logger.debug(
+            f'looping ... conf_values[epoch] is: {conf_values[epoch]}',
+            extra={'nodeid': pid, 'epoch': epoch},
+        )
+        if 1 in bin_values[epoch] and len(conf_values[epoch][(1,)]) >= N - f:
+            return set((1,))
+        if 0 in bin_values[epoch] and len(conf_values[epoch][(0,)]) >= N - f:
+            return set((0,))
+        if (sum(len(senders) for conf_value, senders in
+                conf_values[epoch].items() if senders and
+                set(conf_value).issubset(bin_values[epoch])) >= N - f):
+            return set((0, 1))
+
+        bv_signal.clear()
+        bv_signal.wait()
+
+
 def binaryagreement(sid, pid, N, f, coin, input, decide, broadcast, receive):
     """Binary consensus from [MMR14]. It takes an input ``vi`` and will
     finally write the decided value into ``decide`` channel.
@@ -23,7 +74,9 @@ def binaryagreement(sid, pid, N, f, coin, input, decide, broadcast, receive):
     # Messages received are routed to either a shared coin, the broadcast, or AUX
     est_values = defaultdict(lambda: [set(), set()])
     aux_values = defaultdict(lambda: [set(), set()])
+    conf_values = defaultdict(lambda: {(0,): set(), (1,): set(), (0, 1): set()})
     est_sent = defaultdict(lambda: [False, False])
+    conf_sent = defaultdict(lambda: {(0,): False, (1,): False, (0, 1): False})
     bin_values = defaultdict(set)
 
     # This event is triggered whenever bin_values or aux_values changes
@@ -32,6 +85,8 @@ def binaryagreement(sid, pid, N, f, coin, input, decide, broadcast, receive):
     def _recv():
         while True:  # not finished[pid]:
             (sender, msg) = receive()
+            logger.debug(f'receive {msg} from node {sender}',
+                         extra={'nodeid': pid, 'epoch': msg[1]})
             assert sender in range(N)
             if msg[0] == 'EST':
                 # BV_Broadcast message
@@ -41,7 +96,11 @@ def _recv():
                     # FIXME: raise or continue? For now will raise just
                     # because it appeared first, but maybe the protocol simply
                     # needs to continue.
-                    print('Redundant EST received', msg)
+                    print(f'Redundant EST received by {sender}', msg)
+                    logger.warn(
+                        f'Redundant EST message received by {sender}: {msg}',
+                        extra={'nodeid': pid, 'epoch': msg[1]}
+                    )
                     raise RedundantMessageError(
                         'Redundant EST received {}'.format(msg))
                     # continue
@@ -51,10 +110,18 @@ def _recv():
                 if len(est_values[r][v]) >= f + 1 and not est_sent[r][v]:
                     est_sent[r][v] = True
                     broadcast(('EST', r, v))
+                    logger.debug(f"broadcast {('EST', r, v)}",
+                                 extra={'nodeid': pid, 'epoch': r})
 
                 # Output after reaching second threshold
                 if len(est_values[r][v]) >= 2 * f + 1:
+                    logger.debug(
+                        f'add v = {v} to bin_value[{r}] = {bin_values[r]}',
+                        extra={'nodeid': pid, 'epoch': r},
+                    )
                     bin_values[r].add(v)
+                    logger.debug(f'bin_values[{r}] is now: {bin_values[r]}',
+                                 extra={'nodeid': pid, 'epoch': r})
                     bv_signal.set()
 
             elif msg[0] == 'AUX':
@@ -68,11 +135,28 @@ def _recv():
                     print('Redundant AUX received', msg)
                     raise RedundantMessageError(
                         'Redundant AUX received {}'.format(msg))
-                    # continue
 
+                logger.debug(
+                    f'add sender = {sender} to aux_value[{r}][{v}] = {aux_values[r][v]}',
+                    extra={'nodeid': pid, 'epoch': r},
+                )
                 aux_values[r][v].add(sender)
+                logger.debug(
+                    f'aux_value[{r}][{v}] is now: {aux_values[r][v]}',
+                    extra={'nodeid': pid, 'epoch': r},
+                )
+
                 bv_signal.set()
 
+            elif msg[0] == 'CONF':
+                handle_conf_messages(
+                    sender=sender,
+                    message=msg,
+                    conf_values=conf_values,
+                    pid=pid,
+                    bv_signal=bv_signal,
+                )
+
     # Translate mmr14 broadcast into coin.broadcast
     # _coin_broadcast = lambda (r, sig): broadcast(('COIN', r, sig))
     # _coin_recv = Queue()
@@ -88,6 +172,9 @@ def _recv():
     r = 0
     already_decided = None
     while True:  # Unbounded number of rounds
+        logger.info(f'Starting with est = {est}',
+                    extra={'nodeid': pid, 'epoch': r})
+
         if not est_sent[r][est]:
             est_sent[r][est] = True
             broadcast(('EST', r, est))
@@ -98,10 +185,19 @@ def _recv():
             bv_signal.wait()
 
         w = next(iter(bin_values[r]))  # take an element
+        logger.debug(f"broadcast {('AUX', r, w)}",
+                     extra={'nodeid': pid, 'epoch': r})
         broadcast(('AUX', r, w))
 
         values = None
+        logger.debug(
+            f'block until at least N-f ({N-f}) AUX values are received',
+            extra={'nodeid': pid, 'epoch': r})
         while True:
+            logger.debug(f'bin_values[{r}]: {bin_values[r]}',
+                         extra={'nodeid': pid, 'epoch': r})
+            logger.debug(f'aux_values[{r}]: {aux_values[r]}',
+                         extra={'nodeid': pid, 'epoch': r})
             # Block until at least N-f AUX values are received
             if 1 in bin_values[r] and len(aux_values[r][1]) >= N - f:
                 values = set((1,))
@@ -118,8 +214,37 @@ def _recv():
             bv_signal.clear()
             bv_signal.wait()
 
+        logger.debug(f'Completed AUX phase with values = {values}',
+                     extra={'nodeid': pid, 'epoch': r})
+
+        # CONF phase
+        logger.debug(
+            f'block until at least N-f ({N-f}) CONF values are received',
+            extra={'nodeid': pid, 'epoch': r})
+        if not conf_sent[r][tuple(values)]:
+            values = wait_for_conf_values(
+                pid=pid,
+                N=N,
+                f=f,
+                epoch=r,
+                conf_sent=conf_sent,
+                bin_values=bin_values,
+                values=values,
+                conf_values=conf_values,
+                bv_signal=bv_signal,
+                broadcast=broadcast,
+            )
+        logger.debug(f'Completed CONF phase with values = {values}',
+                     extra={'nodeid': pid, 'epoch': r})
+
+        logger.debug(
+            f'Block until receiving the common coin value',
+            extra={'nodeid': pid, 'epoch': r},
+        )
         # Block until receiving the common coin value
         s = coin(r)
+        logger.info(f'Received coin with value = {s}',
+                    extra={'nodeid': pid, 'epoch': r})
 
         try:
             est, already_decided = set_new_estimate(
@@ -130,6 +255,8 @@ def _recv():
             )
         except AbandonedNodeError:
             # print('[sid:%s] [pid:%d] QUITTING in round %d' % (sid,pid,r)))
+            logger.debug(f'QUIT!',
+                         extra={'nodeid': pid, 'epoch': r})
             _thread_recv.kill()
             return
 

honeybadgerbft/core/commoncoin.py

@@ -1,9 +1,13 @@
+import logging
+
 from honeybadgerbft.crypto.threshsig.boldyreva import serialize
 from collections import defaultdict
 from gevent import Greenlet
 from gevent.queue import Queue
 import hashlib
 
+logger = logging.getLogger(__name__)
+
 
 class CommonCoinFailureException(Exception):
     """Raised for common coin failures."""
@@ -34,8 +38,12 @@ def shared_coin(sid, pid, N, f, PK, SK, broadcast, receive):
 
     def _recv():
         while True:     # main receive loop
+            logger.debug(f'entering loop',
+                         extra={'nodeid': pid, 'epoch': '?'})
             # New shares for some round r, from sender i
             (i, (_, r, sig)) = receive()
+            logger.debug(f'received i, _, r, sig: {i, _, r, sig}',
+                         extra={'nodeid': pid, 'epoch': r})
             assert i in range(N)
             assert r >= 0
             if i in received[r]:
@@ -56,6 +64,10 @@ def _recv():
 
             # After reaching the threshold, compute the output and
             # make it available locally
+            logger.debug(
+                f'if len(received[r]) == f + 1: {len(received[r]) == f + 1}',
+                extra={'nodeid': pid, 'epoch': r},
+            )
             if len(received[r]) == f + 1:
 
                 # Verify and get the combined signature
@@ -65,6 +77,8 @@ def _recv():
 
                 # Compute the bit from the least bit of the hash
                 bit = hash(serialize(sig))[0] % 2
+                logger.debug(f'put bit {bit} in output queue',
+                             extra={'nodeid': pid, 'epoch': r})
                 outputQueue[r].put_nowait(bit)
 
     # greenletPacker(Greenlet(_recv), 'shared_coin', (pid, N, f, broadcast, receive)).start()
@@ -79,6 +93,8 @@ def getCoin(round):
         """
         # I have to do mapping to 1..l
         h = PK.hash_message(str((sid, round)))
+        logger.debug(f"broadcast {('COIN', round, SK.sign(h))}",
+                     extra={'nodeid': pid, 'epoch': round})
         broadcast(('COIN', round, SK.sign(h)))
         return outputQueue[round].get()
 

setup.py

@@ -34,6 +34,7 @@
 tests_require = [
     'coverage',
     'flake8',
+    'logutils',
     'pytest',
     'pytest-cov',
     'pytest-mock',