From cf64664a4461f88a7fbeaf7d969f4e2b15954dff Mon Sep 17 00:00:00 2001 From: Jonathan Zernik Date: Mon, 27 Dec 2021 01:15:05 -0800 Subject: [PATCH] Update squeaklib to support private messages (#1898) * Update squeaklib and update import from keys module * Rename signing key variable to private key in test util * Fix calling make squeak function * Update db data file to v3 * Add recipient public key column to squeak table * Got squeak entry db query working with alias for profiles table * Add test for make squeak with recipient * Add test case for get profile from squeak entry query * Got outer join with recipient profile working * Use outer join to get recipient profile for all squeak entry db queries * Got itest passing with refactored squeak_store * Remove unused config from squeak store class --- squeaknode/core/squeak_core.py | 2 + squeaknode/network/connection.py | 2 +- squeaknode/node/network_handler.py | 252 ++++++++++++------ .../node/peer_subscription_update_worker.py | 8 +- squeaknode/node/squeak_controller.py | 91 +++++-- squeaknode/node/squeak_node.py | 7 +- squeaknode/node/squeak_store.py | 193 +------------- squeaknode/node/update_follows_worker.py | 5 +- squeaknode/twitter/twitter_forwarder.py | 38 ++- tests/node/test_squeak_store.py | 35 +-- 10 files changed, 315 insertions(+), 318 deletions(-) diff --git a/squeaknode/core/squeak_core.py b/squeaknode/core/squeak_core.py index d38d7fff..eab67b30 100644 --- a/squeaknode/core/squeak_core.py +++ b/squeaknode/core/squeak_core.py @@ -65,6 +65,8 @@ class SqueakCore: def make_squeak(self, signing_profile: SqueakProfile, content_str: str, replyto_hash: Optional[bytes] = None) -> Tuple[CSqueak, bytes]: """Create a new squeak. + TODO: Include the block header in the result tuple. + Args: signing_profile: The profile of the author of the squeak. content_str: The content of the squeak as a string. diff --git a/squeaknode/network/connection.py b/squeaknode/network/connection.py index 8eee4cc3..9a7e8dee 100644 --- a/squeaknode/network/connection.py +++ b/squeaknode/network/connection.py @@ -233,7 +233,7 @@ class Connection(object): host=msg.offer.host.decode('utf-8'), port=msg.offer.port, ) - self.network_handler.save_received_offer( + self.network_handler.handle_received_offer( offer, self.peer.remote_address, ) diff --git a/squeaknode/node/network_handler.py b/squeaknode/node/network_handler.py index b4e03378..519b8e3f 100644 --- a/squeaknode/node/network_handler.py +++ b/squeaknode/node/network_handler.py @@ -20,11 +20,10 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import logging -from typing import List from typing import Optional +from bitcoin.core import CBlockHeader from squeak.core import CSqueak -from squeak.core.keys import SqueakPublicKey from squeak.messages import msg_getdata from squeak.messages import MSG_SECRET_KEY from squeak.messages import MSG_SQUEAK @@ -33,17 +32,23 @@ from squeak.net import CInterested from squeak.net import CInv from squeak.net import CSqueakLocator +from squeaknode.core.block_range import BlockRange from squeaknode.core.interests import squeak_matches_interest from squeaknode.core.lightning_address import LightningAddressHostPort from squeaknode.core.offer import Offer from squeaknode.core.peer_address import PeerAddress +from squeaknode.core.sent_offer import SentOffer +from squeaknode.core.squeak_core import SqueakCore from squeaknode.node.active_download_manager import ActiveDownload from squeaknode.node.downloaded_object import DownloadedOffer from squeaknode.node.downloaded_object import DownloadedSqueak from squeaknode.node.price_policy import PricePolicy +from squeaknode.node.secret_key_reply import FreeSecretKeyReply +from squeaknode.node.secret_key_reply import OfferReply from squeaknode.node.secret_key_reply import SecretKeyReply from squeaknode.node.squeak_store import SqueakStore + logger = logging.getLogger(__name__) @@ -55,20 +60,19 @@ class NetworkHandler: def __init__( self, squeak_store: SqueakStore, + squeak_core: SqueakCore, network_manager, download_manager, node_settings, config, ): self.squeak_store = squeak_store + self.squeak_core = squeak_core self.network_manager = network_manager self.active_download_manager = download_manager self.node_settings = node_settings self.config = config - def get_interested_locator(self) -> CSqueakLocator: - return self.squeak_store.get_interested_locator() - def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]: return self.squeak_store.get_squeak(squeak_hash) @@ -96,11 +100,11 @@ class NetworkHandler: ] def save_squeak(self, squeak: CSqueak) -> Optional[bytes]: - # return self.squeak_store.save_squeak(squeak) - return self.save_active_download_squeak(squeak) or \ - self.save_followed_squeak(squeak) + block_header = self.squeak_core.get_block_header(squeak) + return self.save_active_download_squeak(squeak, block_header) or \ + self.save_followed_squeak(squeak, block_header) - def save_active_download_squeak(self, squeak: CSqueak) -> Optional[bytes]: + def save_active_download_squeak(self, squeak: CSqueak, block_header: CBlockHeader) -> Optional[bytes]: """Save the given squeak as an active download. Returns: @@ -109,13 +113,13 @@ class NetworkHandler: counter = self.get_download_squeak_counter(squeak) if counter is None: return None - saved_squeak_hash = self.squeak_store.save_squeak(squeak) + saved_squeak_hash = self.squeak_store.save_squeak(squeak, block_header) if saved_squeak_hash is None: return None counter.increment() return saved_squeak_hash - def save_followed_squeak(self, squeak: CSqueak) -> Optional[bytes]: + def save_followed_squeak(self, squeak: CSqueak, block_header: CBlockHeader) -> Optional[bytes]: """Save the given squeak because it matches the followed interest criteria. @@ -125,7 +129,7 @@ class NetworkHandler: if not self.squeak_matches_interest(squeak): return None # TODO: catch exception if save_squeak fails (because of rate limit, for example). - return self.squeak_store.save_squeak(squeak) + return self.squeak_store.save_squeak(squeak, block_header) def squeak_matches_interest(self, squeak: CSqueak) -> bool: locator = self.get_interested_locator() @@ -135,67 +139,16 @@ class NetworkHandler: return False def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes): - return self.squeak_store.unlock_squeak(squeak_hash, secret_key) - - def get_reply_invs(self, interest): - squeak_hashes = self._get_local_squeaks(interest) - secret_key_hashes = self._get_local_secret_keys(interest) - squeak_invs = [ - CInv(type=MSG_SQUEAK, hash=squeak_hash) - for squeak_hash in squeak_hashes] - secret_key_invs = [ - CInv(type=MSG_SECRET_KEY, hash=squeak_hash) - for squeak_hash in secret_key_hashes] - return squeak_invs + secret_key_invs - - def _get_local_squeaks(self, interest: CInterested): - min_block = interest.nMinBlockHeight if interest.nMinBlockHeight != -1 else None - max_block = interest.nMaxBlockHeight if interest.nMaxBlockHeight != -1 else None - reply_to_hash = interest.hashReplySqk if interest.hashReplySqk != EMPTY_HASH else None - return self.lookup_squeaks( - public_keys=interest.pubkeys, - min_block=min_block, - max_block=max_block, - reply_to_hash=reply_to_hash, + # return self.squeak_store.unlock_squeak(squeak_hash, secret_key) + squeak = self.squeak_store.get_squeak(squeak_hash) + decrypted_content = self.squeak_core.get_decrypted_content( + squeak, + secret_key, ) - - def _get_local_secret_keys(self, interest: CInterested): - min_block = interest.nMinBlockHeight if interest.nMinBlockHeight != -1 else None - max_block = interest.nMaxBlockHeight if interest.nMaxBlockHeight != -1 else None - reply_to_hash = interest.hashReplySqk if interest.hashReplySqk != EMPTY_HASH else None - return self.lookup_secret_keys( - public_keys=interest.pubkeys, - min_block=min_block, - max_block=max_block, - reply_to_hash=reply_to_hash, - ) - - def lookup_squeaks( - self, - public_keys: List[SqueakPublicKey], - min_block: Optional[int], - max_block: Optional[int], - reply_to_hash: Optional[bytes], - ) -> List[bytes]: - return self.squeak_store.lookup_squeaks( - public_keys, - min_block, - max_block, - reply_to_hash, - ) - - def lookup_secret_keys( - self, - public_keys: List[SqueakPublicKey], - min_block: Optional[int], - max_block: Optional[int], - reply_to_hash: Optional[bytes], - ) -> List[bytes]: - return self.squeak_store.lookup_secret_keys( - public_keys, - min_block, - max_block, - reply_to_hash, + self.squeak_store.unlock_squeak( + squeak_hash, + secret_key, + decrypted_content, ) def get_secret_key_reply(self, squeak_hash: bytes, peer_address: PeerAddress) -> Optional[SecretKeyReply]: @@ -207,12 +160,132 @@ class NetworkHandler: host=self.config.lnd.external_host, port=self.config.lnd.port, ) - return self.squeak_store.get_secret_key_reply( + if price_msat == 0: + return self.get_free_squeak_secret_key_reply( + squeak_hash, + ) + else: + return self.get_offer_reply( + squeak_hash, + lnd_external_address, + peer_address, + price_msat, + ) + + def get_offer_reply( + self, + squeak_hash: bytes, + lnd_external_address: Optional[LightningAddressHostPort], + peer_address: PeerAddress, + price_msat: int, + ) -> Optional[OfferReply]: + sent_offer = self.get_sent_offer_for_peer( squeak_hash, - lnd_external_address, peer_address, price_msat, ) + if sent_offer is None: + return None + try: + offer = self.squeak_core.package_offer( + sent_offer, + lnd_external_address, + ) + return OfferReply( + squeak_hash=squeak_hash, + offer=offer, + ) + except Exception: + return None + + def get_free_squeak_secret_key_reply(self, squeak_hash: bytes) -> Optional[FreeSecretKeyReply]: + secret_key = self.squeak_store.get_squeak_secret_key(squeak_hash) + if secret_key is None: + return None + return FreeSecretKeyReply( + squeak_hash=squeak_hash, + secret_key=secret_key, + ) + + def get_sent_offer_for_peer( + self, + squeak_hash: bytes, + peer_address: PeerAddress, + price_msat: int, + ) -> Optional[SentOffer]: + # Check if there is an existing offer for the hash/peer_address combination + sent_offer = self.squeak_store.get_sent_offer_by_squeak_hash_and_peer( + squeak_hash, + peer_address, + ) + if sent_offer: + return sent_offer + squeak = self.squeak_store.get_squeak(squeak_hash) + secret_key = self.squeak_store.get_squeak_secret_key(squeak_hash) + if squeak is None or secret_key is None: + return None + try: + sent_offer = self.squeak_core.create_offer( + squeak, + secret_key, + peer_address, + price_msat, + ) + except Exception: + logger.exception("Failed to create offer.") + return None + self.squeak_store.save_sent_offer(sent_offer) + return sent_offer + + def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: + squeak = self.squeak_store.get_squeak(offer.squeak_hash) + secret_key = self.squeak_store.get_squeak_secret_key(offer.squeak_hash) + if squeak is None or secret_key is not None: + return None + try: + # TODO: Call unpack_offer with check_payment_point=True. + received_offer = self.squeak_core.unpack_offer( + squeak, + offer, + peer_address, + ) + except Exception: + logger.exception("Failed to save received offer.") + return None + return self.squeak_store.save_received_offer(received_offer) + + def get_reply_invs(self, interest): + squeak_hashes = self.get_local_squeaks(interest) + secret_key_hashes = self.get_local_secret_keys(interest) + squeak_invs = [ + CInv(type=MSG_SQUEAK, hash=squeak_hash) + for squeak_hash in squeak_hashes] + secret_key_invs = [ + CInv(type=MSG_SECRET_KEY, hash=squeak_hash) + for squeak_hash in secret_key_hashes] + return squeak_invs + secret_key_invs + + def get_local_squeaks(self, interest: CInterested): + min_block = interest.nMinBlockHeight if interest.nMinBlockHeight != -1 else None + max_block = interest.nMaxBlockHeight if interest.nMaxBlockHeight != -1 else None + reply_to_hash = interest.hashReplySqk if interest.hashReplySqk != EMPTY_HASH else None + return self.squeak_store.lookup_squeaks( + interest.pubkeys, + min_block, + max_block, + reply_to_hash, + ) + + def get_local_secret_keys(self, interest: CInterested): + min_block = interest.nMinBlockHeight if interest.nMinBlockHeight != -1 else None + max_block = interest.nMaxBlockHeight if interest.nMaxBlockHeight != -1 else None + reply_to_hash = interest.hashReplySqk if interest.hashReplySqk != EMPTY_HASH else None + return self.squeak_store.lookup_secret_keys( + interest.pubkeys, + min_block, + max_block, + reply_to_hash, + ) def request_offers(self, squeak_hash: bytes): logger.info("Requesting offers for squeak: {}".format( @@ -224,9 +297,9 @@ class NetworkHandler: getdata_msg = msg_getdata(inv=invs) self.broadcast_msg(getdata_msg) - def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: + def handle_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: # return self.squeak_store.save_received_offer(offer, peer_address) - received_offer_id = self.squeak_store.save_received_offer( + received_offer_id = self.save_received_offer( offer, peer_address, ) @@ -255,3 +328,30 @@ class NetworkHandler: def broadcast_msg(self, msg: MsgSerializable) -> int: return self.network_manager.broadcast_msg(msg) + + def get_interested_locator(self) -> CSqueakLocator: + block_range = self.get_interested_block_range() + followed_public_keys = self.squeak_store.get_followed_public_keys() + if len(followed_public_keys) == 0: + return CSqueakLocator( + vInterested=[], + ) + interests = [ + CInterested( + pubkeys=followed_public_keys, + nMinBlockHeight=block_range.min_block, + nMaxBlockHeight=block_range.max_block, + ) + ] + return CSqueakLocator( + vInterested=interests, + ) + + def get_interested_block_range(self) -> BlockRange: + max_block = self.squeak_core.get_best_block_height() + min_block = max( + 0, + # TODO: rename this. + max_block - self.config.node.interest_block_interval, + ) + return BlockRange(min_block, max_block) diff --git a/squeaknode/node/peer_subscription_update_worker.py b/squeaknode/node/peer_subscription_update_worker.py index b980b339..ee76ab1d 100644 --- a/squeaknode/node/peer_subscription_update_worker.py +++ b/squeaknode/node/peer_subscription_update_worker.py @@ -24,7 +24,7 @@ import threading from squeaknode.bitcoin.bitcoin_block_subscription_client import BitcoinBlockSubscriptionClient from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.squeak_store import SqueakStore +from squeaknode.node.network_handler import NetworkHandler logger = logging.getLogger(__name__) @@ -33,12 +33,12 @@ logger = logging.getLogger(__name__) class PeerSubscriptionUpdateWorker: def __init__( self, - squeak_store: SqueakStore, network_manager: NetworkManager, + network_handler: NetworkHandler, block_subscription_client: BitcoinBlockSubscriptionClient, ): - self.squeak_store = squeak_store self.network_manager = network_manager + self.network_handler = network_handler self.block_subscription_client = block_subscription_client def start_running(self): @@ -50,5 +50,5 @@ class PeerSubscriptionUpdateWorker: def subscribe_blocks(self): for block_hash in self.block_subscription_client.get_blocks(): logger.info("Got block from zeromq: {}".format(block_hash.hex())) - locator = self.squeak_store.get_interested_locator() + locator = self.network_handler.get_interested_locator() self.network_manager.update_local_subscriptions(locator) diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index b246452b..043b1fe6 100644 --- a/squeaknode/node/squeak_controller.py +++ b/squeaknode/node/squeak_controller.py @@ -36,13 +36,13 @@ from squeak.net import CInv from squeaknode.core.connected_peer import ConnectedPeer from squeaknode.core.download_result import DownloadResult -from squeaknode.core.offer import Offer from squeaknode.core.peer_address import PeerAddress from squeaknode.core.received_offer import ReceivedOffer from squeaknode.core.received_payment import ReceivedPayment from squeaknode.core.received_payment_summary import ReceivedPaymentSummary from squeaknode.core.sent_payment import SentPayment from squeaknode.core.sent_payment_summary import SentPaymentSummary +from squeaknode.core.squeak_core import SqueakCore from squeaknode.core.squeak_entry import SqueakEntry from squeaknode.core.squeak_peer import SqueakPeer from squeaknode.core.squeak_profile import SqueakProfile @@ -60,7 +60,7 @@ class SqueakController: def __init__( self, squeak_store: SqueakStore, - squeak_core, + squeak_core: SqueakCore, payment_processor, network_manager, download_manager, @@ -77,14 +77,72 @@ class SqueakController: self.node_settings = node_settings self.config = config - def save_squeak(self, squeak: CSqueak) -> Optional[bytes]: - return self.squeak_store.save_squeak(squeak) - - def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes): - return self.squeak_store.unlock_squeak(squeak_hash, secret_key) - def make_squeak(self, profile_id: int, content_str: str, replyto_hash: Optional[bytes]) -> Optional[bytes]: - return self.squeak_store.make_squeak(profile_id, content_str, replyto_hash) + # return self.squeak_store.make_squeak(profile_id, content_str, replyto_hash) + squeak_profile = self.squeak_store.get_squeak_profile(profile_id) + if squeak_profile is None: + raise Exception("Profile with id {} not found.".format( + profile_id, + )) + squeak, secret_key = self.squeak_core.make_squeak( + squeak_profile, + content_str, + replyto_hash, + ) + block_header = self.squeak_core.get_block_header(squeak) + inserted_squeak_hash = self.squeak_store.save_squeak( + squeak, + block_header, + ) + if inserted_squeak_hash is None: + return None + self.squeak_store.unlock_squeak( + inserted_squeak_hash, + secret_key, + content_str, + ) + return inserted_squeak_hash + + def pay_offer(self, received_offer_id: int) -> int: + # return self.squeak_store.pay_offer(received_offer_id) + # Get the offer from the database + received_offer = self.squeak_store.get_received_offer( + received_offer_id, + ) + if received_offer is None: + raise Exception("Received offer with id {} not found.".format( + received_offer_id, + )) + logger.info("Paying received offer: {}".format(received_offer)) + sent_payment = self.squeak_core.pay_offer(received_offer) + sent_payment_id = self.squeak_store.save_sent_payment(sent_payment) + # # Delete the received offer + # self.squeak_db.delete_offer(sent_payment.payment_hash) + # Mark the received offer as paid + self.squeak_store.mark_received_offer_paid( + sent_payment.payment_hash, + ) + # self.unlock_squeak( + # received_offer.squeak_hash, + # sent_payment.secret_key, + # ) + squeak = self.squeak_store.get_squeak(received_offer.squeak_hash) + decrypted_content = self.squeak_core.get_decrypted_content( + squeak, + sent_payment.secret_key, + ) + self.squeak_store.unlock_squeak( + received_offer.squeak_hash, + sent_payment.secret_key, + decrypted_content, + ) + return sent_payment_id + + # def save_squeak(self, squeak: CSqueak) -> Optional[bytes]: + # return self.squeak_store.save_squeak(squeak) + + # def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes): + # return self.squeak_store.unlock_squeak(squeak_hash, secret_key) def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]: return self.squeak_store.get_squeak(squeak_hash) @@ -173,9 +231,6 @@ class SqueakController: def get_received_offer(self, received_offer_id: int) -> Optional[ReceivedOffer]: return self.squeak_store.get_received_offer(received_offer_id) - def pay_offer(self, received_offer_id: int) -> int: - return self.squeak_store.pay_offer(received_offer_id) - def get_sent_payments( self, limit: int, @@ -272,12 +327,12 @@ class SqueakController: last_entry, ) - def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: - return self.squeak_store.save_received_offer(offer, peer_address) + # def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: + # return self.squeak_store.save_received_offer(offer, peer_address) - # TODO: remove from controller. - def get_followed_public_keys(self) -> List[SqueakPublicKey]: - return self.squeak_store.get_followed_public_keys() + # # TODO: remove from controller. + # def get_followed_public_keys(self) -> List[SqueakPublicKey]: + # return self.squeak_store.get_followed_public_keys() def get_received_payment_summary(self) -> ReceivedPaymentSummary: return self.squeak_store.get_received_payment_summary() @@ -461,7 +516,7 @@ class SqueakController: def subscribe_timeline_squeak_entries(self, stopped: threading.Event): for item in self.squeak_store.subscribe_new_squeaks(stopped): - followed_public_keys = self.get_followed_public_keys() + followed_public_keys = self.squeak_store.get_followed_public_keys() if item.GetPubKey() in set(followed_public_keys): squeak_hash = get_hash(item) yield self.get_squeak_entry(squeak_hash) diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index 3d9be3fc..b2211af6 100644 --- a/squeaknode/node/squeak_node.py +++ b/squeaknode/node/squeak_node.py @@ -171,13 +171,11 @@ class SqueakNode: def initialize_squeak_store(self): self.squeak_store = SqueakStore( self.squeak_db, - self.squeak_core, self.config.node.max_squeaks, self.config.node.max_squeaks_per_public_key_per_block, self.config.node.squeak_retention_s, self.config.node.received_offer_retention_s, self.config.node.sent_offer_retention_s, - self.config.node.interest_block_interval, ) def initialize_payment_processor(self): @@ -190,6 +188,7 @@ class SqueakNode: def initialize_twitter_forwarder(self): self.twitter_forwarder = TwitterForwarder( self.squeak_store, + self.squeak_core, self.config.twitter.forward_tweets_retry_s, ) @@ -211,6 +210,7 @@ class SqueakNode: def initialize_network_handler(self): self.network_handler = NetworkHandler( self.squeak_store, + self.squeak_core, self.network_manager, self.download_manager, self.node_settings, @@ -287,12 +287,13 @@ class SqueakNode: self.new_follow_worker = UpdateFollowsWorker( self.squeak_store, self.network_manager, + self.network_handler, ) def initialize_peer_subscription_update_worker(self): self.new_bitcoin_block_worker = PeerSubscriptionUpdateWorker( - self.squeak_store, self.network_manager, + self.network_handler, self.bitcoin_block_subscription_client, ) diff --git a/squeaknode/node/squeak_store.py b/squeaknode/node/squeak_store.py index ab570775..1c0e7cd5 100644 --- a/squeaknode/node/squeak_store.py +++ b/squeaknode/node/squeak_store.py @@ -25,15 +25,12 @@ from typing import Iterator from typing import List from typing import Optional +from bitcoin.core import CBlockHeader +from squeak.core import CheckSqueak from squeak.core import CSqueak from squeak.core.keys import SqueakPrivateKey from squeak.core.keys import SqueakPublicKey -from squeak.net import CInterested -from squeak.net import CSqueakLocator -from squeaknode.core.block_range import BlockRange -from squeaknode.core.lightning_address import LightningAddressHostPort -from squeaknode.core.offer import Offer from squeaknode.core.peer_address import PeerAddress from squeaknode.core.peers import create_saved_peer from squeaknode.core.profiles import create_contact_profile @@ -45,7 +42,6 @@ from squeaknode.core.received_payment_summary import ReceivedPaymentSummary from squeaknode.core.sent_offer import SentOffer from squeaknode.core.sent_payment import SentPayment from squeaknode.core.sent_payment_summary import SentPaymentSummary -from squeaknode.core.squeak_core import SqueakCore from squeaknode.core.squeak_entry import SqueakEntry from squeaknode.core.squeak_peer import SqueakPeer from squeaknode.core.squeak_profile import SqueakProfile @@ -53,9 +49,6 @@ from squeaknode.core.twitter_account import TwitterAccount from squeaknode.core.twitter_account_entry import TwitterAccountEntry from squeaknode.core.update_subscriptions_event import UpdateSubscriptionsEvent from squeaknode.node.listener_subscription_client import EventListener -from squeaknode.node.secret_key_reply import FreeSecretKeyReply -from squeaknode.node.secret_key_reply import OfferReply -from squeaknode.node.secret_key_reply import SecretKeyReply logger = logging.getLogger(__name__) @@ -66,33 +59,27 @@ class SqueakStore: def __init__( self, squeak_db, - squeak_core: SqueakCore, max_squeaks, max_squeaks_per_public_key_per_block, squeak_retention_s, received_offer_retention_s, sent_offer_retention_s, - interested_block_range_size, ): self.squeak_db = squeak_db - self.squeak_core = squeak_core self.max_squeaks = max_squeaks self.max_squeaks_per_public_key_per_block = max_squeaks_per_public_key_per_block self.squeak_retention_s = squeak_retention_s self.received_offer_retention_s = received_offer_retention_s self.sent_offer_retention_s = sent_offer_retention_s - self.interested_block_range_size = interested_block_range_size self.new_squeak_listener = EventListener() self.new_received_offer_listener = EventListener() self.new_secret_key_listener = EventListener() self.new_follow_listener = EventListener() self.twitter_stream_change_listener = EventListener() - def save_squeak(self, squeak: CSqueak) -> Optional[bytes]: + def save_squeak(self, squeak: CSqueak, block_header: CBlockHeader) -> Optional[bytes]: # Check if the squeak is valid - self.squeak_core.check_squeak(squeak) - # Get the block header for the squeak. - block_header = self.squeak_core.get_block_header(squeak) + CheckSqueak(squeak) # Check if limit exceeded. if self.squeak_db.get_number_of_squeaks() >= self.max_squeaks: raise Exception("Exceeded max number of squeaks.") @@ -113,16 +100,11 @@ class SqueakStore: logger.info("Saved squeak: {}".format( inserted_squeak_hash.hex(), )) - # Notify the listener self.new_squeak_listener.handle_new_item(squeak) return inserted_squeak_hash - def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes): + def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes, decrypted_content: str): squeak = self.squeak_db.get_squeak(squeak_hash) - decrypted_content = self.squeak_core.get_decrypted_content( - squeak, - secret_key, - ) self.squeak_db.set_squeak_decryption_key( squeak_hash, secret_key, @@ -131,126 +113,8 @@ class SqueakStore: logger.info("Unlocked squeak: {}".format( squeak_hash.hex(), )) - # Notify the listener self.new_secret_key_listener.handle_new_item(squeak) - def make_squeak(self, profile_id: int, content_str: str, replyto_hash: Optional[bytes]) -> Optional[bytes]: - squeak_profile = self.squeak_db.get_profile(profile_id) - squeak, decryption_key = self.squeak_core.make_squeak( - squeak_profile, - content_str, - replyto_hash, - ) - inserted_squeak_hash = self.save_squeak(squeak) - if inserted_squeak_hash is None: - return None - self.unlock_squeak( - inserted_squeak_hash, - decryption_key, - ) - return inserted_squeak_hash - - def get_secret_key_reply( - self, - squeak_hash: bytes, - lnd_external_address: Optional[LightningAddressHostPort], - peer_address: PeerAddress, - price_msat: int, - ) -> Optional[SecretKeyReply]: - if price_msat == 0: - return self.get_free_squeak_secret_key_reply( - squeak_hash, - ) - else: - return self.get_offer_reply( - squeak_hash, - lnd_external_address, - peer_address, - price_msat, - ) - - def get_offer_reply( - self, - squeak_hash: bytes, - lnd_external_address: Optional[LightningAddressHostPort], - peer_address: PeerAddress, - price_msat: int, - ) -> Optional[OfferReply]: - sent_offer = self.get_sent_offer_for_peer( - squeak_hash, - peer_address, - price_msat, - ) - if sent_offer is None: - return None - try: - offer = self.squeak_core.package_offer( - sent_offer, - lnd_external_address, - ) - return OfferReply( - squeak_hash=squeak_hash, - offer=offer, - ) - except Exception: - return None - - def get_free_squeak_secret_key_reply(self, squeak_hash: bytes) -> Optional[FreeSecretKeyReply]: - secret_key = self.get_squeak_secret_key(squeak_hash) - if secret_key is None: - return None - return FreeSecretKeyReply( - squeak_hash=squeak_hash, - secret_key=secret_key, - ) - - def pay_offer(self, received_offer_id: int) -> int: - # Get the offer from the database - received_offer = self.squeak_db.get_received_offer( - received_offer_id) - if received_offer is None: - raise Exception("Received offer with id {} not found.".format( - received_offer_id, - )) - logger.info("Paying received offer: {}".format(received_offer)) - sent_payment = self.squeak_core.pay_offer(received_offer) - sent_payment_id = self.squeak_db.insert_sent_payment(sent_payment) - # # Delete the received offer - # self.squeak_db.delete_offer(sent_payment.payment_hash) - # Mark the received offer as paid - self.squeak_db.set_received_offer_paid( - sent_payment.payment_hash, - True, - ) - self.unlock_squeak( - received_offer.squeak_hash, - sent_payment.secret_key, - ) - return sent_payment_id - - def get_interested_locator(self) -> CSqueakLocator: - block_range = self.get_interested_block_range() - followed_public_keys = self.get_followed_public_keys() - if len(followed_public_keys) == 0: - return CSqueakLocator( - vInterested=[], - ) - interests = [ - CInterested( - pubkeys=followed_public_keys, - nMinBlockHeight=block_range.min_block, - nMaxBlockHeight=block_range.max_block, - ) - ] - return CSqueakLocator( - vInterested=interests, - ) - - def get_interested_block_range(self) -> BlockRange: - max_block = self.squeak_core.get_best_block_height() - min_block = max(0, max_block - self.interested_block_range_size) - return BlockRange(min_block, max_block) - def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]: return self.squeak_db.get_squeak(squeak_hash) @@ -260,35 +124,14 @@ class SqueakStore: def delete_squeak(self, squeak_hash: bytes) -> None: self.squeak_db.delete_squeak(squeak_hash) - def get_sent_offer_for_peer( - self, - squeak_hash: bytes, - peer_address: PeerAddress, - price_msat: int, - ) -> Optional[SentOffer]: - # Check if there is an existing offer for the hash/peer_address combination - sent_offer = self.squeak_db.get_sent_offer_by_squeak_hash_and_peer( + def get_sent_offer_by_squeak_hash_and_peer(self, squeak_hash: bytes, peer_address: PeerAddress) -> Optional[SentOffer]: + return self.squeak_db.get_sent_offer_by_squeak_hash_and_peer( squeak_hash, peer_address, ) - if sent_offer: - return sent_offer - squeak = self.get_squeak(squeak_hash) - secret_key = self.get_squeak_secret_key(squeak_hash) - if squeak is None or secret_key is None: - return None - try: - sent_offer = self.squeak_core.create_offer( - squeak, - secret_key, - peer_address, - price_msat, - ) - except Exception: - logger.exception("Failed to create offer.") - return None - self.squeak_db.insert_sent_offer(sent_offer) - return sent_offer + + def save_sent_offer(self, sent_offer: SentOffer) -> int: + return self.squeak_db.insert_sent_offer(sent_offer) def create_signing_profile(self, profile_name: str) -> int: squeak_profile = create_signing_profile( @@ -517,21 +360,7 @@ class SqueakStore: last_entry, ) - def save_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: - squeak = self.get_squeak(offer.squeak_hash) - secret_key = self.get_squeak_secret_key(offer.squeak_hash) - if squeak is None or secret_key is not None: - return None - try: - # TODO: Call unpack_offer with check_payment_point=True. - received_offer = self.squeak_core.unpack_offer( - squeak, - offer, - peer_address, - ) - except Exception: - logger.exception("Failed to save received offer.") - return None + def save_received_offer(self, received_offer: ReceivedOffer) -> Optional[int]: received_offer_id = self.squeak_db.insert_received_offer( received_offer) if received_offer_id is None: diff --git a/squeaknode/node/update_follows_worker.py b/squeaknode/node/update_follows_worker.py index afa7af55..35142f1c 100644 --- a/squeaknode/node/update_follows_worker.py +++ b/squeaknode/node/update_follows_worker.py @@ -23,6 +23,7 @@ import logging import threading from squeaknode.network.network_manager import NetworkManager +from squeaknode.node.network_handler import NetworkHandler from squeaknode.node.squeak_store import SqueakStore @@ -35,9 +36,11 @@ class UpdateFollowsWorker: self, squeak_store: SqueakStore, network_manager: NetworkManager, + network_handler: NetworkHandler, ): self.squeak_store = squeak_store self.network_manager = network_manager + self.network_handler = network_handler self.stopped = threading.Event() def start_running(self): @@ -56,5 +59,5 @@ class UpdateFollowsWorker: self.stopped, ): logger.debug("Handling update subscriptions event") - locator = self.squeak_store.get_interested_locator() + locator = self.network_handler.get_interested_locator() self.network_manager.update_local_subscriptions(locator) diff --git a/squeaknode/twitter/twitter_forwarder.py b/squeaknode/twitter/twitter_forwarder.py index a0d99a2d..87079bbb 100644 --- a/squeaknode/twitter/twitter_forwarder.py +++ b/squeaknode/twitter/twitter_forwarder.py @@ -22,7 +22,9 @@ import logging import threading from typing import Dict +from typing import Optional +from squeaknode.core.squeak_core import SqueakCore from squeaknode.core.twitter_account_entry import TwitterAccountEntry from squeaknode.node.squeak_store import SqueakStore from squeaknode.twitter.twitter_stream import TwitterStream @@ -36,9 +38,11 @@ class TwitterForwarder: def __init__( self, squeak_store: SqueakStore, + squeak_core: SqueakCore, retry_s: int, ): self.squeak_store = squeak_store + self.squeak_core = squeak_core self.retry_s = retry_s self.lock = threading.Lock() self.current_tasks: Dict[str, TwitterForwarderTask] = {} @@ -54,6 +58,7 @@ class TwitterForwarder: for account in self.squeak_store.get_twitter_accounts(): task = TwitterForwarderTask( self.squeak_store, + self.squeak_core, account, self.retry_s, ) @@ -80,10 +85,12 @@ class TwitterForwarderTask: def __init__( self, squeak_store: SqueakStore, + squeak_core: SqueakCore, twitter_account: TwitterAccountEntry, retry_s: int, ): self.squeak_store = squeak_store + self.squeak_core = squeak_core self.twitter_account = twitter_account self.retry_s = retry_s self.stopped = threading.Event() @@ -148,10 +155,9 @@ class TwitterForwarderTask: return False def forward_tweet(self, tweet: dict) -> None: - self.squeak_store.make_squeak( + self.make_squeak( profile_id=self.twitter_account.profile_id, content_str=tweet['data']['text'], - replyto_hash=None, ) def handle_tweet(self, tweet: dict): @@ -159,3 +165,31 @@ class TwitterForwarderTask: "Got tweet: {}".format(tweet)) if self.is_tweet_a_match(tweet): self.forward_tweet(tweet) + + def make_squeak( + self, + profile_id: int, + content_str: str, + ) -> Optional[bytes]: + squeak_profile = self.squeak_store.get_squeak_profile(profile_id) + if squeak_profile is None: + raise Exception("Profile with id {} not found.".format( + profile_id, + )) + squeak, secret_key = self.squeak_core.make_squeak( + squeak_profile, + content_str, + ) + block_header = self.squeak_core.get_block_header(squeak) + inserted_squeak_hash = self.squeak_store.save_squeak( + squeak, + block_header, + ) + if inserted_squeak_hash is None: + return None + self.squeak_store.unlock_squeak( + inserted_squeak_hash, + secret_key, + content_str, + ) + return inserted_squeak_hash diff --git a/tests/node/test_squeak_store.py b/tests/node/test_squeak_store.py index b50d6ce2..725c12f2 100644 --- a/tests/node/test_squeak_store.py +++ b/tests/node/test_squeak_store.py @@ -24,7 +24,6 @@ import pytest from sqlalchemy import create_engine from squeaknode.core.lightning_address import LightningAddressHostPort -from squeaknode.core.squeak_core import SqueakCore from squeaknode.db.squeak_db import SqueakDb from squeaknode.node.squeak_store import SqueakStore @@ -42,11 +41,6 @@ def squeak_db(db_engine): yield db -@pytest.fixture -def squeak_core(): - return mock.Mock(spec=SqueakCore) - - @pytest.fixture def lightning_host_port(): return LightningAddressHostPort(host="my_lightning_host", port=8765) @@ -82,53 +76,33 @@ def sent_offer_retention_s(): return 7200 -@pytest.fixture -def interested_block_range_size(): - return 2016 - - @pytest.fixture def inserted_signing_profile_id(squeak_db, signing_profile): yield squeak_db.insert_profile(signing_profile) -# @pytest.fixture -# def inserted_received_offer_id(squeak_db, received_offer, creation_date): -# with mock.patch.object(SqueakDb, 'timestamp_now_ms', new_callable=mock.PropertyMock) as mock_timestamp_ms: -# mock_timestamp_ms.return_value = creation_date / 1000 -# yield squeak_db.insert_received_offer(received_offer) - - @pytest.fixture def squeak_store( squeak_db, - squeak_core, max_squeaks, max_squeaks_per_public_key_per_block, squeak_retention_s, received_offer_retention_s, sent_offer_retention_s, - interested_block_range_size, ): return SqueakStore( squeak_db, - squeak_core, max_squeaks, max_squeaks_per_public_key_per_block, squeak_retention_s, received_offer_retention_s, sent_offer_retention_s, - interested_block_range_size, ) -def test_save_squeak(squeak_store, squeak_db, squeak_core, block_header, squeak): - with mock.patch.object(squeak_core, 'check_squeak', autospec=True) as mock_check_squeak, \ - mock.patch.object(squeak_core, 'get_block_header', autospec=True) as mock_get_block_header, \ - mock.patch.object(squeak_db, 'insert_squeak', autospec=True) as mock_insert_squeak: - mock_check_squeak.return_value = None - mock_get_block_header.return_value = block_header - squeak_store.save_squeak(squeak) +def test_save_squeak(squeak_store, squeak_db, block_header, squeak): + with mock.patch.object(squeak_db, 'insert_squeak', autospec=True) as mock_insert_squeak: + squeak_store.save_squeak(squeak, block_header) mock_insert_squeak.assert_called_once_with(squeak, block_header) @@ -179,10 +153,9 @@ def test_get_sent_offer_already_exists(squeak_store, squeak_db, sent_offer): with mock.patch.object(squeak_db, 'get_sent_offer_by_squeak_hash_and_peer', autospec=True) as mock_get_sent_offer_by_squeak_hash_and_peer: mock_get_sent_offer_by_squeak_hash_and_peer.return_value = sent_offer - retrieved_sent_offer = squeak_store.get_sent_offer_for_peer( + retrieved_sent_offer = squeak_store.get_sent_offer_by_squeak_hash_and_peer( sent_offer.squeak_hash, sent_offer.peer_address, - sent_offer.price_msat, ) assert retrieved_sent_offer == sent_offer