mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 14:14:22 +01:00
272e61922d
When an appointment was triggered a flag was set in the Watcher, and removed later on in the Responder if the transaction ended up being rejected. That's pretty annoying. Since we have information about whether a transaction has made it to the mempool or not via the Carrier's receipt, this can be all done in the Watcher, which makes more sense and reduces the interaction with the db (1 write if succeeds, 0 otherwise instead of 1 write if succeeds, 2 otherwise).
290 lines
13 KiB
Python
290 lines
13 KiB
Python
from uuid import uuid4
|
|
from queue import Queue
|
|
from threading import Thread
|
|
|
|
from common.cryptographer import Cryptographer
|
|
from common.appointment import Appointment
|
|
from common.tools import compute_locator
|
|
|
|
from common.logger import Logger
|
|
|
|
from pisa.cleaner import Cleaner
|
|
from pisa.responder import Responder
|
|
from pisa.block_processor import BlockProcessor
|
|
|
|
logger = Logger("Watcher")
|
|
|
|
|
|
class Watcher:
|
|
"""
|
|
The :class:`Watcher` is the class in charge to watch for channel breaches for the appointments accepted by the
|
|
tower.
|
|
|
|
The :class:`Watcher` keeps track of the accepted appointments in ``appointments`` and, for new received block,
|
|
checks if any breach has happened by comparing the txids with the appointment locators. If a breach is seen, the
|
|
:obj:`EncryptedBlob <pisa.encrypted_blob.EncryptedBlob>` of the corresponding appointment is decrypted and the data
|
|
is passed to the :obj:`Responder <pisa.responder.Responder>`.
|
|
|
|
If an appointment reaches its end with no breach, the data is simply deleted.
|
|
|
|
The :class:`Watcher` receives information about new received blocks via the ``block_queue`` that is populated by the
|
|
:obj:`ChainMonitor <pisa.chain_monitor.ChainMonitor>`.
|
|
|
|
Args:
|
|
db_manager (:obj:`DBManager <pisa.db_manager>`): a ``DBManager`` instance to interact with the database.
|
|
chain_monitor (:obj:`ChainMonitor <pisa.chain_monitor.ChainMonitor>`): a ``ChainMonitor`` instance used to track
|
|
new blocks received by ``bitcoind``.
|
|
sk_der (:obj:`bytes`): a DER encoded private key used to sign appointment receipts (signaling acceptance).
|
|
config (:obj:`dict`): a dictionary containing all the configuration parameters. Used locally to retrieve
|
|
``MAX_APPOINTMENTS`` and ``EXPIRY_DELTA``.
|
|
responder (:obj:`Responder <pisa.responder.Responder>`): a ``Responder`` instance. If ``None`` is passed, a new
|
|
instance is created. Populated instances are useful when bootstrapping the system from backed-up data.
|
|
|
|
|
|
Attributes:
|
|
appointments (:obj:`dict`): a dictionary containing a simplification of the appointments (:obj:`Appointment
|
|
<pisa.appointment.Appointment>` instances) accepted by the tower (``locator`` and ``end_time``).
|
|
It's populated trough ``add_appointment``.
|
|
locator_uuid_map (:obj:`dict`): a ``locator:uuid`` map used to allow the :obj:`Watcher` to deal with several
|
|
appointments with the same ``locator``.
|
|
asleep (:obj:`bool`): A flag that signals whether the :obj:`Watcher` is asleep or awake.
|
|
block_queue (:obj:`Queue`): A queue used by the :obj:`Watcher` to receive block hashes from ``bitcoind``. It is
|
|
populated by the :obj:`ChainMonitor <pisa.chain_monitor.ChainMonitor>`.
|
|
chain_monitor (:obj:`ChainMonitor <pisa.chain_monitor.ChainMonitor>`): a ``ChainMonitor`` instance used to track
|
|
new blocks received by ``bitcoind``.
|
|
config (:obj:`dict`): a dictionary containing all the configuration parameters. Used locally to retrieve
|
|
``MAX_APPOINTMENTS`` and ``EXPIRY_DELTA``.
|
|
db_manager (:obj:`DBManager <pisa.db_manager>`): A db manager instance to interact with the database.
|
|
signing_key (:mod:`EllipticCurvePrivateKey`): a private key used to sign accepted appointments.
|
|
|
|
Raises:
|
|
ValueError: if `pisa_sk_file` is not found.
|
|
|
|
"""
|
|
|
|
def __init__(self, db_manager, chain_monitor, sk_der, config, responder=None):
|
|
self.appointments = dict()
|
|
self.locator_uuid_map = dict()
|
|
self.asleep = True
|
|
self.block_queue = Queue()
|
|
self.chain_monitor = chain_monitor
|
|
self.config = config
|
|
self.db_manager = db_manager
|
|
self.signing_key = Cryptographer.load_private_key_der(sk_der)
|
|
|
|
if not isinstance(responder, Responder):
|
|
self.responder = Responder(db_manager, chain_monitor)
|
|
|
|
def add_appointment(self, appointment):
|
|
"""
|
|
Adds a new appointment to the ``appointments`` dictionary if ``max_appointments`` has not been reached.
|
|
|
|
``add_appointment`` is the entry point of the Watcher. Upon receiving a new appointment, if the :obj:`Watcher`
|
|
is asleep, it will be awaken and start monitoring the blockchain (``do_watch``) until ``appointments`` is empty.
|
|
It will go back to sleep once there are no more pending appointments.
|
|
|
|
Once a breach is seen on the blockchain, the :obj:`Watcher` will decrypt the corresponding
|
|
:obj:`EncryptedBlob <pisa.encrypted_blob.EncryptedBlob>` and pass the information to the
|
|
:obj:`Responder <pisa.responder.Responder>`.
|
|
|
|
The tower may store multiple appointments with the same ``locator`` to avoid DoS attacks based on data
|
|
rewriting. `locators`` should be derived from the ``dispute_txid``, but that task is performed by the user, and
|
|
the tower has no way of verifying whether or not they have been properly derived. Therefore, appointments are
|
|
identified by ``uuid`` and stored in ``appointments`` and ``locator_uuid_map``.
|
|
|
|
Args:
|
|
appointment (:obj:`Appointment <pisa.appointment.Appointment>`): the appointment to be added to the
|
|
:obj:`Watcher`.
|
|
|
|
Returns:
|
|
:obj:`tuple`: A tuple signaling if the appointment has been added or not (based on ``max_appointments``).
|
|
The structure looks as follows:
|
|
|
|
- ``(True, signature)`` if the appointment has been accepted.
|
|
- ``(False, None)`` otherwise.
|
|
|
|
"""
|
|
|
|
if len(self.appointments) < self.config.get("MAX_APPOINTMENTS"):
|
|
|
|
uuid = uuid4().hex
|
|
self.appointments[uuid] = {"locator": appointment.locator, "end_time": appointment.end_time}
|
|
|
|
if appointment.locator in self.locator_uuid_map:
|
|
self.locator_uuid_map[appointment.locator].append(uuid)
|
|
|
|
else:
|
|
self.locator_uuid_map[appointment.locator] = [uuid]
|
|
|
|
if self.asleep:
|
|
self.asleep = False
|
|
self.chain_monitor.watcher_asleep = False
|
|
Thread(target=self.do_watch).start()
|
|
|
|
logger.info("Waking up")
|
|
|
|
self.db_manager.store_watcher_appointment(uuid, appointment.to_json())
|
|
self.db_manager.create_append_locator_map(appointment.locator, uuid)
|
|
|
|
appointment_added = True
|
|
signature = Cryptographer.sign(appointment.serialize(), self.signing_key)
|
|
|
|
logger.info("New appointment accepted", locator=appointment.locator)
|
|
|
|
else:
|
|
appointment_added = False
|
|
signature = None
|
|
|
|
logger.info("Maximum appointments reached, appointment rejected", locator=appointment.locator)
|
|
|
|
return appointment_added, signature
|
|
|
|
def do_watch(self):
|
|
"""
|
|
Monitors the blockchain whilst there are pending appointments.
|
|
|
|
This is the main method of the :obj:`Watcher` and the one in charge to pass appointments to the
|
|
:obj:`Responder <pisa.responder.Responder>` upon detecting a breach.
|
|
"""
|
|
|
|
while len(self.appointments) > 0:
|
|
block_hash = self.block_queue.get()
|
|
logger.info("New block received", block_hash=block_hash)
|
|
|
|
block = BlockProcessor.get_block(block_hash)
|
|
|
|
if block is not None:
|
|
txids = block.get("tx")
|
|
|
|
logger.info("List of transactions", txids=txids)
|
|
|
|
expired_appointments = [
|
|
uuid
|
|
for uuid, appointment_data in self.appointments.items()
|
|
if block["height"] > appointment_data.get("end_time") + self.config.get("EXPIRY_DELTA")
|
|
]
|
|
|
|
Cleaner.delete_expired_appointments(
|
|
expired_appointments, self.appointments, self.locator_uuid_map, self.db_manager
|
|
)
|
|
|
|
valid_breaches, invalid_breaches = self.filter_valid_breaches(self.get_breaches(txids))
|
|
|
|
for uuid, breach in valid_breaches.items():
|
|
logger.info(
|
|
"Notifying responder and deleting appointment",
|
|
penalty_txid=breach["penalty_txid"],
|
|
locator=breach["locator"],
|
|
uuid=uuid,
|
|
)
|
|
|
|
receipt = self.responder.handle_breach(
|
|
uuid,
|
|
breach["locator"],
|
|
breach["dispute_txid"],
|
|
breach["penalty_txid"],
|
|
breach["penalty_rawtx"],
|
|
self.appointments[uuid].get("end_time"),
|
|
block_hash,
|
|
)
|
|
|
|
Cleaner.delete_appointment_from_memory(uuid, self.appointments, self.locator_uuid_map)
|
|
|
|
# Appointments are only flagged as triggered if they are delivered, otherwise they are just deleted.
|
|
# FIXME: This is only necessary because of the triggered appointment approach. Fix if it changes.
|
|
if receipt.delivered:
|
|
self.db_manager.create_triggered_appointment_flag(uuid)
|
|
|
|
else:
|
|
self.db_manager.delete_watcher_appointment(uuid)
|
|
Cleaner.update_delete_db_locator_map(uuid, breach["locator"], self.db_manager)
|
|
|
|
Cleaner.delete_completed_appointments(
|
|
invalid_breaches, self.appointments, self.locator_uuid_map, self.db_manager
|
|
)
|
|
|
|
# Register the last processed block for the watcher
|
|
self.db_manager.store_last_block_hash_watcher(block_hash)
|
|
|
|
# Go back to sleep if there are no more appointments
|
|
self.asleep = True
|
|
self.chain_monitor.watcher_asleep = True
|
|
|
|
logger.info("No more pending appointments, going back to sleep")
|
|
|
|
def get_breaches(self, txids):
|
|
"""
|
|
Gets a list of channel breaches given the list of transaction ids.
|
|
|
|
Args:
|
|
txids (:obj:`list`): the list of transaction ids included in the last received block.
|
|
|
|
Returns:
|
|
:obj:`dict`: A dictionary (``locator:txid``) with all the breaches found. An empty dictionary if none are
|
|
found.
|
|
"""
|
|
|
|
potential_locators = {compute_locator(txid): txid for txid in txids}
|
|
|
|
# Check is any of the tx_ids in the received block is an actual match
|
|
intersection = set(self.locator_uuid_map.keys()).intersection(potential_locators.keys())
|
|
breaches = {locator: potential_locators[locator] for locator in intersection}
|
|
|
|
if len(breaches) > 0:
|
|
logger.info("List of breaches", breaches=breaches)
|
|
|
|
else:
|
|
logger.info("No breaches found")
|
|
|
|
return breaches
|
|
|
|
def filter_valid_breaches(self, breaches):
|
|
"""
|
|
Filters what of the found breaches contain valid transaction data.
|
|
|
|
The :obj:`Watcher` cannot if a given :obj:`EncryptedBlob <pisa.encrypted_blob.EncryptedBlob>` contains a valid
|
|
transaction until a breach if seen. Blobs that contain arbitrary data are dropped and not sent to the
|
|
:obj:`Responder <pisa.responder.Responder>`.
|
|
|
|
Args:
|
|
breaches (:obj:`dict`): a dictionary containing channel breaches (``locator:txid``).
|
|
|
|
Returns:
|
|
:obj:`dict`: A dictionary containing all the breaches flagged either as valid or invalid.
|
|
The structure is as follows:
|
|
|
|
``{locator, dispute_txid, penalty_txid, penalty_rawtx, valid_breach}``
|
|
"""
|
|
|
|
valid_breaches = {}
|
|
invalid_breaches = []
|
|
|
|
for locator, dispute_txid in breaches.items():
|
|
for uuid in self.locator_uuid_map[locator]:
|
|
appointment = Appointment.from_dict(self.db_manager.load_watcher_appointment(uuid))
|
|
|
|
try:
|
|
penalty_rawtx = Cryptographer.decrypt(appointment.encrypted_blob, dispute_txid)
|
|
|
|
except ValueError:
|
|
penalty_rawtx = None
|
|
|
|
penalty_tx = BlockProcessor.decode_raw_transaction(penalty_rawtx)
|
|
|
|
if penalty_tx is not None:
|
|
valid_breaches[uuid] = {
|
|
"locator": locator,
|
|
"dispute_txid": dispute_txid,
|
|
"penalty_txid": penalty_tx.get("txid"),
|
|
"penalty_rawtx": penalty_rawtx,
|
|
}
|
|
|
|
logger.info(
|
|
"Breach found for locator", locator=locator, uuid=uuid, penalty_txid=penalty_tx.get("txid")
|
|
)
|
|
|
|
else:
|
|
invalid_breaches.append(uuid)
|
|
|
|
return valid_breaches, invalid_breaches
|