diff --git a/itests/test.sh b/itests/test.sh index 52dffb05..c8e170c6 100644 --- a/itests/test.sh +++ b/itests/test.sh @@ -1,6 +1,7 @@ #!/bin/bash pytest -s tests +#pytest -s tests -k "test_buy_squeak" #pytest -s tests -k "test_sell_squeak" #pytest -s tests -k "test_download_single_squeak" #pytest -s tests -k "test_connect_peer" @@ -8,5 +9,4 @@ pytest -s tests #pytest -s tests -k "test_download_single_squeak" #pytest -s tests -k "test_share_single_squeak" #pytest -s tests -k "test_delete_squeak" -#pytest -s tests -k "test_get_squeak_by_lookup" #pytest -s tests -k "test_subscribe_squeaks" diff --git a/requirements.txt b/requirements.txt index 75aa9607..58e8b40d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ alembic argparse +expiringdict Flask flask-cors flask-login diff --git a/squeaknode/config/config.py b/squeaknode/config/config.py index 1f5b10dc..ce5883c9 100644 --- a/squeaknode/config/config.py +++ b/squeaknode/config/config.py @@ -38,7 +38,7 @@ DEFAULT_NETWORK = "testnet" DEFAULT_PRICE_MSAT = 10000 DEFAULT_LOG_LEVEL = "INFO" DEFAULT_MAX_SQUEAKS = 10000 -DEFAULT_MAX_SQUEAKS_PER_ADDRESS_PER_BLOCK = 100 +DEFAULT_MAX_SQUEAKS_PER_ADDRESS_IN_BLOCK_RANGE = 100 DEFAULT_SERVER_RPC_HOST = "0.0.0.0" DEFAULT_SERVER_RPC_PORT = None DEFAULT_ADMIN_RPC_HOST = "0.0.0.0" @@ -121,8 +121,8 @@ class NodeConfig(Config): cast=int, required=False, default=DEFAULT_PRICE_MSAT) max_squeaks = key( cast=int, required=False, default=DEFAULT_MAX_SQUEAKS) - max_squeaks_per_address_per_block = key( - cast=int, required=False, default=DEFAULT_MAX_SQUEAKS_PER_ADDRESS_PER_BLOCK) + max_squeaks_per_address_in_block_range = key( + cast=int, required=False, default=DEFAULT_MAX_SQUEAKS_PER_ADDRESS_IN_BLOCK_RANGE) sqk_dir_path = key( cast=str, required=False, default=DEFAULT_SQK_DIR_PATH) log_level = key( diff --git a/squeaknode/core/util.py b/squeaknode/core/util.py index 7ab5c5ef..88a3443e 100644 --- a/squeaknode/core/util.py +++ b/squeaknode/core/util.py @@ -24,16 +24,21 @@ import random from bitcoin.base58 import Base58ChecksumError from bitcoin.wallet import CBitcoinAddressError +from squeak.core import CSqueak from squeak.core.elliptic import generate_random_scalar from squeak.core.elliptic import scalar_difference from squeak.core.elliptic import scalar_from_bytes from squeak.core.elliptic import scalar_sum from squeak.core.elliptic import scalar_to_bytes from squeak.core.signing import CSqueakAddress +from squeak.net import CInterested DATA_KEY_LENGTH = 32 VERSION_NONCE_LENGTH = 8 +HASH_LENGTH = 32 +EMPTY_HASH = b'\x00' * HASH_LENGTH + def get_hash(squeak): return squeak.GetHash()[::-1] @@ -89,3 +94,19 @@ def is_address_valid(address: str) -> bool: except (Base58ChecksumError, CBitcoinAddressError): return False return True + + +def squeak_matches_interest(squeak: CSqueak, interest: CInterested) -> bool: + if len(interest.addresses) > 0 \ + and squeak.GetAddress() not in interest.addresses: + return False + # if interest.nMinBlockHeight != -1 \ + # and squeak.nBlockHeight < interest.nMinBlockHeight: + # return False + # if interest.nMaxBlockHeight != -1 \ + # and squeak.nBlockHeight > interest.nMaxBlockHeight: + # return False + if interest.hashReplySqk != EMPTY_HASH \ + and squeak.hashReplySqk != interest.hashReplySqk: + return False + return True diff --git a/squeaknode/db/squeak_db.py b/squeaknode/db/squeak_db.py index b0d39197..7b9dc409 100644 --- a/squeaknode/db/squeak_db.py +++ b/squeaknode/db/squeak_db.py @@ -504,6 +504,28 @@ class SqueakDb: num_squeaks = row["num_squeaks"] return num_squeaks + def number_of_squeaks_with_address_in_block_range( + self, + address: str, + min_block: int, + max_block: int, + ) -> int: + """ Get number of squeaks with address in block range. """ + s = ( + select([ + func.count().label("num_squeaks"), + ]) + .select_from(self.squeaks) + .where(self.squeaks.c.author_address == address) + .where(self.squeaks.c.n_block_height >= min_block) + .where(self.squeaks.c.n_block_height <= max_block) + ) + with self.get_connection() as connection: + result = connection.execute(s) + row = result.fetchone() + num_squeaks = row["num_squeaks"] + return num_squeaks + # def lookup_squeaks_needing_offer( # self, # addresses: List[str], diff --git a/squeaknode/network/peer_message_handler.py b/squeaknode/network/peer_message_handler.py index 629bb133..e67535de 100644 --- a/squeaknode/network/peer_message_handler.py +++ b/squeaknode/network/peer_message_handler.py @@ -169,7 +169,7 @@ class PeerMessageHandler: def handle_squeak(self, msg): squeak = msg.squeak # TODO: check if interested before saving. - self.squeak_controller.save_squeak(squeak) + self.squeak_controller.save_received_squeak(squeak) def handle_offer(self, msg): # TODO: check if interested before saving. diff --git a/squeaknode/node/new_squeak_worker.py b/squeaknode/node/new_squeak_worker.py index ad7aeaa8..be36cb8a 100644 --- a/squeaknode/node/new_squeak_worker.py +++ b/squeaknode/node/new_squeak_worker.py @@ -24,10 +24,10 @@ import threading from squeak.core import CSqueak from squeak.messages import msg_inv -from squeak.net import CInterested from squeak.net import CInv from squeaknode.core.util import get_hash +from squeaknode.core.util import squeak_matches_interest from squeaknode.network.network_manager import NetworkManager from squeaknode.network.peer import Peer from squeaknode.node.squeak_controller import SqueakController @@ -102,22 +102,22 @@ class NewSqueakWorker: return False locator = peer.subscription.locator for interest in locator.vInterested: - if self.squeak_matches_interest(squeak, interest): + if squeak_matches_interest(squeak, interest): logger.debug("Found a match!") return True return False - def squeak_matches_interest(self, squeak: CSqueak, interest: CInterested) -> bool: - if len(interest.addresses) > 0 \ - and squeak.GetAddress() not in interest.addresses: - return False - # if interest.nMinBlockHeight != -1 \ - # and squeak.nBlockHeight < interest.nMinBlockHeight: - # return False - # if interest.nMaxBlockHeight != -1 \ - # and squeak.nBlockHeight > interest.nMaxBlockHeight: - # return False - if interest.hashReplySqk != EMPTY_HASH \ - and squeak.hashReplySqk != interest.hashReplySqk: - return False - return True + # def squeak_matches_interest(self, squeak: CSqueak, interest: CInterested) -> bool: + # if len(interest.addresses) > 0 \ + # and squeak.GetAddress() not in interest.addresses: + # return False + # # if interest.nMinBlockHeight != -1 \ + # # and squeak.nBlockHeight < interest.nMinBlockHeight: + # # return False + # # if interest.nMaxBlockHeight != -1 \ + # # and squeak.nBlockHeight > interest.nMaxBlockHeight: + # # return False + # if interest.hashReplySqk != EMPTY_HASH \ + # and squeak.hashReplySqk != interest.hashReplySqk: + # return False + # return True diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index a05e5650..d43a1a7b 100644 --- a/squeaknode/node/squeak_controller.py +++ b/squeaknode/node/squeak_controller.py @@ -49,9 +49,12 @@ from squeaknode.core.squeak_entry import SqueakEntry from squeaknode.core.squeak_peer import SqueakPeer from squeaknode.core.squeak_profile import SqueakProfile from squeaknode.core.util import is_address_valid +from squeaknode.core.util import squeak_matches_interest from squeaknode.network.peer import Peer from squeaknode.node.listener_subscription_client import EventListener from squeaknode.node.received_payments_subscription_client import ReceivedPaymentsSubscriptionClient +from squeaknode.node.temporary_interest_manager import TemporaryInterest +from squeaknode.node.temporary_interest_manager import TemporaryInterestManager logger = logging.getLogger(__name__) @@ -63,18 +66,17 @@ class SqueakController: self, squeak_db, squeak_core, - squeak_rate_limiter, payment_processor, network_manager, config, ): self.squeak_db = squeak_db self.squeak_core = squeak_core - self.squeak_rate_limiter = squeak_rate_limiter self.payment_processor = payment_processor self.network_manager = network_manager self.new_squeak_listener = EventListener() self.new_received_offer_listener = EventListener() + self.temporary_interest_manager = TemporaryInterestManager() self.config = config def save_squeak(self, squeak: CSqueak) -> bytes: @@ -131,6 +133,31 @@ class SqueakController: logger.info("Deleted number of offers : {}".format(num_deleted_offers)) self.squeak_db.delete_squeak(squeak_hash) + def save_received_squeak(self, squeak: CSqueak) -> None: + if self.get_temporary_interest_counter(squeak): + logger.debug("Saving squeak based on temporary interest.") + self.save_squeak(squeak) + elif self.squeak_matches_interest(squeak): + self.save_squeak(squeak) + + def squeak_matches_interest(self, squeak: CSqueak) -> bool: + locator = self.get_interested_locator() + for interest in locator.vInterested: + if squeak_matches_interest(squeak, interest) \ + and self.squeak_in_limit_of_interest(squeak, interest): + return True + return False + + def squeak_in_limit_of_interest(self, squeak: CSqueak, interest: CInterested) -> bool: + return self.squeak_db.number_of_squeaks_with_address_in_block_range( + str(squeak.GetAddress), + interest.nMinBlockHeight, + interest.nMaxBlockHeight, + ) < self.config.node.max_squeaks_per_address_in_block_range + + def get_temporary_interest_counter(self, squeak: CSqueak) -> Optional[TemporaryInterest]: + return self.temporary_interest_manager.lookup_counter(squeak) + def get_buy_offer(self, squeak_hash: bytes, peer_address: PeerAddress) -> Offer: # Check if there is an existing offer for the hash/peer_address combination sent_offer = self.get_saved_sent_offer(squeak_hash, peer_address) @@ -577,6 +604,8 @@ class SqueakController: logger.info("Downloading single squeak: {}".format( squeak_hash.hex(), )) + # Add the temporary interest in this hash. + self.temporary_interest_manager.add_hash_interest(1, squeak_hash) invs = [ CInv(type=1, hash=squeak_hash) ] diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index 007662b4..4a726178 100644 --- a/squeaknode/node/squeak_node.py +++ b/squeaknode/node/squeak_node.py @@ -41,7 +41,6 @@ from squeaknode.node.process_received_payments_worker import ProcessReceivedPaym from squeaknode.node.squeak_controller import SqueakController from squeaknode.node.squeak_deletion_worker import SqueakDeletionWorker from squeaknode.node.squeak_offer_expiry_worker import SqueakOfferExpiryWorker -from squeaknode.node.squeak_rate_limiter import SqueakRateLimiter logger = logging.getLogger(__name__) @@ -62,7 +61,6 @@ class SqueakNode: self.initialize_lightning_client() self.initialize_bitcoin_client() self.initialize_squeak_core() - self.initialize_rate_limiter() self.initialize_payment_processor() self.initialize_network_manager() self.initialize_squeak_controller() @@ -138,12 +136,6 @@ class SqueakNode: self.lightning_client, ) - def initialize_rate_limiter(self): - self.squeak_rate_limiter = SqueakRateLimiter( - self.squeak_db, - self.config.node.max_squeaks_per_address_per_block, - ) - def initialize_payment_processor(self): self.payment_processor = PaymentProcessor( self.squeak_db, @@ -158,7 +150,6 @@ class SqueakNode: self.squeak_controller = SqueakController( self.squeak_db, self.squeak_core, - self.squeak_rate_limiter, self.payment_processor, self.network_manager, self.config, diff --git a/squeaknode/node/squeak_rate_limiter.py b/squeaknode/node/squeak_rate_limiter.py deleted file mode 100644 index ffeedf2d..00000000 --- a/squeaknode/node/squeak_rate_limiter.py +++ /dev/null @@ -1,66 +0,0 @@ -# MIT License -# -# Copyright (c) 2020 Jonathan Zernik -# -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files (the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions: -# -# The above copyright notice and this permission notice shall be included in all -# copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -# SOFTWARE. -import logging - -from squeaknode.core.util import get_hash - -logger = logging.getLogger(__name__) - - -HOUR_IN_SECONDS = 3600 - - -class SqueakRateLimiter: - def __init__( - self, - squeak_db, - max_squeaks_per_address_per_block, - ): - self.squeak_db = squeak_db - self.max_squeaks_per_address_per_block = max_squeaks_per_address_per_block - - def should_rate_limit_allow(self, squeak): - logger.info("Checking rate limit for squeak: {!r}".format( - get_hash(squeak).hex(), - )) - current_squeak_count = self._get_num_squeaks_with_address_with_block( - squeak) - logger.info( - "Current squeak count: {}, limit: {}".format( - current_squeak_count, self.max_squeaks_per_address_per_block - ) - ) - return current_squeak_count < self.max_squeaks_per_address_per_block - - def _get_num_squeaks_with_address_with_block(self, squeak): - address = str(squeak.GetAddress()) - block_height = squeak.nBlockHeight - logger.info( - "Getting squeak count for address: {} with height: {}".format( - address, - block_height, - ) - ) - return self.squeak_db.number_of_squeaks_with_address_with_block( - address, - block_height, - ) diff --git a/squeaknode/node/temporary_interest_manager.py b/squeaknode/node/temporary_interest_manager.py new file mode 100644 index 00000000..1df9f3e3 --- /dev/null +++ b/squeaknode/node/temporary_interest_manager.py @@ -0,0 +1,101 @@ +# MIT License +# +# Copyright (c) 2020 Jonathan Zernik +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +import logging +import threading +import uuid +from abc import ABC +from abc import abstractmethod +from typing import Dict +from typing import Optional + +from expiringdict import ExpiringDict +from squeak.core import CSqueak +from squeak.net import CInterested + +from squeaknode.core.util import get_hash +from squeaknode.core.util import squeak_matches_interest + +logger = logging.getLogger(__name__) + + +class TemporaryInterest(ABC): + + def __init__(self, limit: int): + self.limit = limit + self.count = 0 + self._lock = threading.Lock() + + @abstractmethod + def is_interested(self, squeak: CSqueak) -> bool: + pass + + def increment(self) -> None: + with self._lock: + self.count += 1 + + def is_under_limit(self) -> bool: + with self._lock: + return self.count < self.limit + + +class TemporaryRangeInterest(TemporaryInterest): + + def __init__(self, limit: int, interest: CInterested): + self.interest = interest + super().__init__(limit) + + def is_interested(self, squeak: CSqueak) -> bool: + return squeak_matches_interest(squeak, self.interest) + + +class TemporaryHashInterest(TemporaryInterest): + + def __init__(self, limit: int, squeak_hash: bytes): + self.squeak_hash = squeak_hash + super().__init__(limit) + + def is_interested(self, squeak: CSqueak) -> bool: + return self.squeak_hash == get_hash(squeak) + + +class TemporaryInterestManager: + + def __init__(self): + self.interests: Dict[str, TemporaryInterest] = ExpiringDict( + max_len=100, max_age_seconds=10) + + def lookup_counter(self, squeak: CSqueak) -> Optional[TemporaryInterest]: + for name, interest in self.interests.items(): + if interest.is_interested(squeak) \ + and interest.is_under_limit(): + return interest + return None + + def add_interest(self, interest: TemporaryInterest) -> None: + name_key = "interest_key_{}".format(uuid.uuid1()) + self.interests[name_key] = interest + + def add_range_interest(self, limit: int, interest: CInterested) -> None: + self.add_interest(TemporaryRangeInterest(limit, interest)) + + def add_hash_interest(self, limit: int, squeak_hash: bytes) -> None: + self.add_interest(TemporaryHashInterest(limit, squeak_hash)) diff --git a/tests/node/test_squeak_controller.py b/tests/node/test_squeak_controller.py index 717b7e50..443046fc 100644 --- a/tests/node/test_squeak_controller.py +++ b/tests/node/test_squeak_controller.py @@ -32,7 +32,6 @@ from squeaknode.db.squeak_db import SqueakDb from squeaknode.network.network_manager import NetworkManager from squeaknode.node.payment_processor import PaymentProcessor from squeaknode.node.squeak_controller import SqueakController -from squeaknode.node.squeak_rate_limiter import SqueakRateLimiter @pytest.fixture @@ -87,11 +86,6 @@ def price_msat(): return 777 -@pytest.fixture -def squeak_rate_limiter(): - return mock.Mock(spec=SqueakRateLimiter) - - @pytest.fixture def payment_processor(): return mock.Mock(spec=PaymentProcessor) @@ -101,7 +95,6 @@ def payment_processor(): def squeak_controller( squeak_db, squeak_core, - squeak_rate_limiter, payment_processor, network_manager, config, @@ -109,7 +102,6 @@ def squeak_controller( return SqueakController( squeak_db, squeak_core, - squeak_rate_limiter, payment_processor, network_manager, config, @@ -120,7 +112,6 @@ def squeak_controller( def regtest_squeak_controller( squeak_db, squeak_core, - squeak_rate_limiter, payment_processor, network_manager, regtest_config, @@ -128,7 +119,6 @@ def regtest_squeak_controller( return SqueakController( squeak_db, squeak_core, - squeak_rate_limiter, payment_processor, network_manager, regtest_config,