Merge branch 'online_backup' of https://github.com/RCasatta/opentimestamps-server into online_backup

Author: RCasatta

otsserver/backup.py

@@ -18,8 +18,8 @@
 import time
 from urllib.parse import urlparse, urljoin
 
-PAGING = 1000
-SLEEP_SECS = 600
+PAGING = 1000  # Number of commitments per chunk
+SLEEP_SECS = 60  # Once the backup is synced this is the polling interval to check for new chunks
 
 
 class Backup:
@@ -29,27 +29,36 @@ def __init__(self, journal, calendar, cache_path):
         self.cache_path = cache_path
         os.makedirs(cache_path, exist_ok=True)
 
+    # Return the bytes of the chunk
     def __getitem__(self, chunk):
+
+        # We use a disk cache, creating a chunk of 1000 commitments is a quite expensive operation of about 10s.
+        # The server isn't blocked in the meantime but this could be used by an attacker to degrade calendar performance
+        # Moreover is not recommended to set up an HTTP cache more than one year (see RFC 2616), thus, a disk cache is
+        #  mandatory.
         cached_kv_bytes = self.read_disk_cache(chunk)
         if cached_kv_bytes is not None:
             return cached_kv_bytes
 
         backup_map = {}
         start = chunk*PAGING
         end = start+PAGING
-        for i in range(start, end)[::-1]:  # iterate in reverse to fail fast
+
+        # Iterate in reverse to fail fast if this chunk is not complete, a chunk is considered complete if all relative
+        # 1000 commitments are complete. Which means a tx with more of 6 confirmations timestamp them
+        for i in range(start, end)[::-1]:
             try:
                 current = self.journal[i]
-                # print(str(i) +":"+b2x(journal[i]))
                 current_el = self.calendar[current]
-                # print("\t"+str(current_el))
                 self.__create_kv_map(current_el, current_el.msg, backup_map)
             except KeyError:
+                # according to https://docs.python.org/3/library/exceptions.html#IndexError IndexError is the more
+                # appropriate exception for this case
                 raise IndexError
             if i % 100 == 0:
-                logging.info(str(i) + ":" + b2x(self.journal[i]))
+                logging.debug("Got commitment " + str(i) + ":" + b2x(self.journal[i]))
 
-        logging.info("map len " + str(len(backup_map)) + " start:" + str(start) + " end:" + str(end))
+        logging.debug("map len " + str(len(backup_map)) + " start:" + str(start) + " end:" + str(end))
         kv_bytes = self.__kv_map_to_bytes(backup_map)
         self.write_disk_cache(chunk, kv_bytes)
 
@@ -92,6 +101,7 @@ def __create_kv_map(ts, msg, kv_map):
     @staticmethod
     def __kv_map_to_bytes(kv_map):
         ctx = BytesSerializationContext()
+        # Sorting the map elements to create chunks deterministically, but this is not mandatory for importing the chunk
         for key, value in sorted(kv_map.items()):
             ctx.write_varuint(len(key))
             ctx.write_bytes(key)
@@ -101,8 +111,11 @@ def __kv_map_to_bytes(kv_map):
         return ctx.getbytes()
 
     def read_disk_cache(self, chunk):
+        # For the disk cache we are using 6 digits file name which will support a total of 1 billion commitments,
+        # because every chunk contain 1000 commitments. Supposing 1 commitment per second this could last for 32 years
+        # which appear to be ok for this version
         chunk_str = "{0:0>6}".format(chunk)
-        chunk_path = chunk_str[0:3]
+        chunk_path = chunk_str[0:3]  # we create a path to avoid creating more than 1000 files per directory
 
         try:
             cache_file = self.cache_path + '/' + chunk_path + '/' + chunk_str
@@ -120,7 +133,8 @@ def write_disk_cache(self, chunk, bytes):
         with open(cache_file, 'wb') as fd:
             fd.write(bytes)
 
-
+# The following is a shrinked version of the standard calendar http server, it only support the '/timestamp' endpoint
+# This way the backup server could serve request in place of the calendar serve which is backupping
 class RPCRequestHandler(http.server.BaseHTTPRequestHandler):
 
     def do_GET(self):
@@ -197,6 +211,9 @@ def serve_forever(self):
         super().serve_forever()
 
 
+# This is the thread responsible for asking the chunks to the running calendar and import them in the db.
+# The main script allow to launch 1 thread of this for every calendar to backup, thus a backup server could
+# theoretically serve timestamp in place of every calendar server which supports this incremental live backup mechanism
 class AskBackup(threading.Thread):
 
     def __init__(self, db, calendar_url, base_path):
@@ -208,36 +225,33 @@ def __init__(self, db, calendar_url, base_path):
         super().__init__(target=self.loop)
 
     def loop(self):
-        print("Starting loop for %s" % self.calendar_url)
+        logging.info("Starting loop for %s" % self.calendar_url)
 
         try:
             with open(self.up_to_path, 'r') as up_to_fd:
                 last_known = int(up_to_fd.read().strip())
         except FileNotFoundError as exp:
             last_known = -1
-        print("Checking calendar " + str(self.calendar_url) + ", last_known commitment:" + str(last_known))
+        logging.info("Checking calendar " + str(self.calendar_url) + ", last_known commitment:" + str(last_known))
 
         while True:
             start_time = time.time()
             backup_url = urljoin(self.calendar_url, "/experimental/backup/%d" % (last_known + 1))
-            print(str(backup_url))
+            logging.debug("Asking " + str(backup_url))
             try:
                 r = requests.get(backup_url)
             except Exception as err:
-                print("Exception asking " + str(backup_url) + " message " + str(err))
+                logging.error("Exception asking " + str(backup_url) + " message " + str(err))
                 break
 
-            if r.status_code == 404:
-                print("%s not found, sleeping for %s seconds" % (backup_url, SLEEP_SECS) )
+            if r.status_code != 200:
+                logging.info("%s not found, sleeping for %s seconds" % (backup_url, SLEEP_SECS) )
                 time.sleep(SLEEP_SECS)
                 continue
 
-            # print(r.raw.read(10))
             kv_map = Backup.bytes_to_kv_map(r.content)
-            # print(str(map))
             attestations = {}
             ops = {}
-            print("kv_maps elements " + str(len(kv_map)))
             for key, value in kv_map.items():
                 # print("--- key=" + b2x(key) + " value=" + b2x(value))
                 ctx = BytesDeserializationContext(value)
@@ -252,31 +266,28 @@ def loop(self):
 
             proxy = bitcoin.rpc.Proxy()
 
-            # verify all bitcoin attestation are valid
-            print("total attestations: " + str(len(attestations)))
+            # Verify all bitcoin attestation are valid
+            logging.debug("Total attestations: " + str(len(attestations)))
             for key, attestation in attestations.items():
                 if attestation.__class__ == BitcoinBlockHeaderAttestation:
                     blockhash = proxy.getblockhash(attestation.height)
                     block_header = proxy.getblockheader(blockhash)
+                    # the following raise an exception and block computation if the attestation does not verify
                     attested_time = attestation.verify_against_blockheader(key, block_header)
-                    print("verifying " + b2x(key) + " result " + str(attested_time))
+                    logging.debug("Verifying " + b2x(key) + " result " + str(attested_time))
 
             # verify all ops connects to an attestation
-            print("total ops: " + str(len(ops)))
+            logging.debug("Total ops: " + str(len(ops)))
             for key, op in ops.items():
-
-                # print("key " + b2x(key) + " op " + str(op))
                 current_key = key
                 current_op = op
                 while True:
                     next_key = current_op(current_key)
-                    # print("next_key " + b2x(next_key))
                     if next_key in ops:
                         current_key = next_key
                         current_op = ops[next_key]
                     else:
                         break
-                # print("maps to " + b2x(next_key))
                 assert next_key in attestations
 
             batch = leveldb.WriteBatch()
@@ -289,11 +300,11 @@ def loop(self):
                 with open(self.up_to_path, 'w') as up_to_fd:
                     up_to_fd.write('%d\n' % last_known)
             except FileNotFoundError as exp:
-                print(str(exp))
+                logging.error(str(exp))
                 break
 
             elapsed_time = time.time() - start_time
-            print("Took %ds" % elapsed_time)
+            logging.info("Took %ds for %s" % (elapsed_time, str(backup_url)))
 
 
 

otsserver/rpc.py

@@ -15,6 +15,9 @@
 import socketserver
 import threading
 import time
+import pystache
+import datetime 
+from functools import reduce
 
 import bitcoin.core
 from bitcoin.core import b2lx, b2x
@@ -24,7 +27,7 @@
 from opentimestamps.core.serialize import StreamSerializationContext
 
 from otsserver.calendar import Journal
-
+renderer = pystache.Renderer()
 
 class RPCRequestHandler(http.server.BaseHTTPRequestHandler):
     MAX_DIGEST_LENGTH = 64
@@ -181,42 +184,62 @@ def do_GET(self):
             # need to investigate further, but this seems to work.
             str_wallet_balance = str(proxy._call("getbalance"))
 
-            welcome_page = """\
-<html>
+            transactions = proxy._call("listtransactions", "", 50)
+            # We want only the confirmed txs containing an OP_RETURN, from most to least recent
+            transactions = list(filter(lambda x: x["confirmations"] > 0 and x["amount"] == 0, transactions))
+            a_week_ago = (datetime.date.today() - datetime.timedelta(days=7)).timetuple()
+            a_week_ago_posix = time.mktime(a_week_ago)
+            transactions_in_last_week = list(filter(lambda x: x["time"] > a_week_ago_posix, transactions))
+            fees_in_last_week = reduce(lambda a,b: a-b["fee"], transactions_in_last_week, 0)
+            time_between_transactions = round(168 / len(transactions_in_last_week)) # in hours based on 168 hours in a week
+            transactions.sort(key=lambda x: x["confirmations"])
+            homepage_template = """<html>
 <head>
     <title>OpenTimestamps Calendar Server</title>
 </head>
 <body>
-<p>This is an <a href="https://opentimestamps.org">OpenTimestamps</a> <a href="https://github.com/opentimestamps/opentimestamps-server">Calendar Server</a> (v%s)</p>
-
+<p>This is an <a href="https://opentimestamps.org">OpenTimestamps</a> <a href="https://github.com/opentimestamps/opentimestamps-server">Calendar Server</a> (v{{ version }})</p>
 <p>
-Pending commitments: %d</br>
-Transactions waiting for confirmation: %d</br>
-Most recent timestamp tx: %s (%d prior versions)</br>
-Most recent merkle tree tip: %s</br>
-Best-block: %s, height %d</br>
+Pending commitments: {{ pending_commitments }}</br>
+Transactions waiting for confirmation: {{ txs_waiting_for_confirmation }}</br>
+Most recent timestamp tx: {{ most_recent_tx }} ({{ prior_versions }} prior versions)</br>
+Most recent merkle tree tip: {{ tip }}</br>
+Best-block: {{ best_block }}, height {{ block_height }}</br>
 </br>
-Wallet balance: %s BTC</br>
+Wallet balance: {{ balance }} BTC</br>
 </p>
-
 <p>
-You can donate to the wallet by sending funds to: %s</br>
+You can donate to the wallet by sending funds to: {{ address }}</br>
 This address changes after every donation.
 </p>
-
+<p>
+Average time between transactions in the last week: {{ time_between_transactions }} hour(s)</br>
+Fees used in the last week: {{ fees_in_last_week }} BTC</br>
+Latest transactions: </br>
+{{#transactions}}
+    {{txid}} </br>
+{{/transactions}}
+</p>
 </body>
-</html>
-""" % (otsserver.__version__,
-       len(self.calendar.stamper.pending_commitments),
-       len(self.calendar.stamper.txs_waiting_for_confirmation),
-       b2lx(self.calendar.stamper.unconfirmed_txs[-1].tx.GetTxid()) if self.calendar.stamper.unconfirmed_txs else 'None',
-       max(0, len(self.calendar.stamper.unconfirmed_txs) - 1),
-       b2x(self.calendar.stamper.unconfirmed_txs[-1].tip_timestamp.msg) if self.calendar.stamper.unconfirmed_txs else 'None',
-       bitcoin.core.b2lx(proxy.getbestblockhash()), proxy.getblockcount(),
-       str_wallet_balance,
-       str(proxy.getaccountaddress('')))
-
-            self.wfile.write(welcome_page.encode())
+</html>"""
+
+            stats = { 'version': otsserver.__version__,
+              'pending_commitments': len(self.calendar.stamper.pending_commitments),
+              'txs_waiting_for_confirmation':len(self.calendar.stamper.txs_waiting_for_confirmation),
+              'most_recent_tx': b2lx(self.calendar.stamper.unconfirmed_txs[-1].tx.GetTxid()) if self.calendar.stamper.unconfirmed_txs else 'None',
+              'prior_versions': max(0, len(self.calendar.stamper.unconfirmed_txs) - 1),
+              'tip': b2x(self.calendar.stamper.unconfirmed_txs[-1].tip_timestamp.msg) if self.calendar.stamper.unconfirmed_txs else 'None',
+              'best_block': bitcoin.core.b2lx(proxy.getbestblockhash()),
+              'block_height': proxy.getblockcount(),
+              'balance': str_wallet_balance,
+              'address': str(proxy.getaccountaddress('')),
+              'transactions': transactions[:5],
+              'time_between_transactions': time_between_transactions,
+              'fees_in_last_week': fees_in_last_week,
+            }
+            welcome_page = renderer.render(homepage_template, stats)
+            self.wfile.write(str.encode(welcome_page))
+
 
         elif self.path.startswith('/timestamp/'):
             self.get_timestamp()

otsserver/stamper.py

@@ -300,13 +300,18 @@ def __do_bitcoin(self):
                     self.pending_commitments.add(reorged_commitment_timestamp.msg)
 
             # Check if this block contains any of the pending transactions
-
-            try:
-                block = proxy.getblock(block_hash)
-            except KeyError:
-                # Must have been a reorg or something, return
-                logging.error("Failed to get block")
-                return
+            block = None
+            while block is None:
+                try:
+                    block = proxy.getblock(block_hash)
+                except KeyError:
+                    # Must have been a reorg or something, return
+                    logging.error("Failed to get block")
+                    return
+                except BrokenPipeError:
+                    logging.error("BrokenPipeError to get block")
+                    time.sleep(5)
+                    proxy = bitcoin.rpc.Proxy()
 
             # the following is an optimization, by pre computing the tx_id we rapidly check if our unconfirmed tx
             # is in the block

requirements.txt

@@ -2,3 +2,4 @@ opentimestamps>=0.2.1
 GitPython>=2.0.8
 leveldb>=0.20
 pysha3>=1.0.2
+pystache>=0.5