diff --git a/frontend/src/api/client.js b/frontend/src/api/client.js index 89719c2a..38950158 100644 --- a/frontend/src/api/client.js +++ b/frontend/src/api/client.js @@ -744,15 +744,3 @@ export const downloadSqueak = (squeakHash) => { deser: deser, }); } - -export const downloadPubkeySqueaks = (pubkey) => { - console.log('Calling downloadPubkeySqueaks'); - const request = new DownloadPubKeySqueaksRequest(); - request.setPubkey(pubkey); - const deser = DownloadPubKeySqueaksReply.deserializeBinary; - return baseRequest({ - url: '/downloadaddresssqueaks', - req: request, - deser: deser, - }); -} diff --git a/frontend/src/components/Nav/index.js b/frontend/src/components/Nav/index.js index 8bb64298..aba179c4 100644 --- a/frontend/src/components/Nav/index.js +++ b/frontend/src/components/Nav/index.js @@ -38,7 +38,6 @@ import { selectMakeSqueakStatus, selectBuySqueakStatus, selectDownloadSqueakStatus, - selectDownloadPubkeySqueakStatus, } from '../../features/squeaks/squeaksSlice' const Nav = ({history}) => { @@ -303,18 +302,6 @@ const Nav = ({history}) => { - {/* Block screen with modal when download pubkey squeaks is waiting. */} -
-
-
-

Downloading pubkey squeaks...

-
-
- -
-
-
- ) } diff --git a/frontend/src/components/Peer/index.js b/frontend/src/components/Peer/index.js index 8c1e393c..da4fd902 100644 --- a/frontend/src/components/Peer/index.js +++ b/frontend/src/components/Peer/index.js @@ -14,13 +14,8 @@ import { useDispatch } from 'react-redux' import { fetchPeer, - fetchConnectedPeers, selectCurrentPeer, selectCurrentPeerStatus, - selectConnectedPeers, - selectPeerConnectionByAddress, - setConnectPeer, - setDisconnectPeer, setSavePeer, setDeletePeer, setPeerAutoconnectEnabled, @@ -43,12 +38,6 @@ const [bannerLoading, setBannerLoading] = useState(false) const [styleBody, setStyleBody] = useState(false) const peer = useSelector(selectCurrentPeer); -const connectedPeers = useSelector(selectConnectedPeers); -const peerConnection = useSelector(state => selectPeerConnectionByAddress(state, { - network: props.match.params.network, - host: props.match.params.host, - port: props.match.params.port, -})); const dispatch = useDispatch(); @@ -59,7 +48,6 @@ useEffect(() => { host: props.match.params.host, port: props.match.params.port, })); - dispatch(fetchConnectedPeers()); }, []) const isInitialMount = useRef(true); @@ -73,22 +61,6 @@ useEffect(() => { useEffect( () => () => document.getElementsByTagName("body")[0].style.cssText = "", [] ) -const connectPeer = (e) => { - dispatch(setConnectPeer({ - host: props.match.params.host, - port: props.match.params.port, - network: props.match.params.network, - })); -} - -const disconnectPeer = (e) => { - dispatch(setDisconnectPeer({ - host: props.match.params.host, - port: props.match.params.port, - network: props.match.params.network, - })); -} - const deletePeer = () => { let values = { peerId: peer.getPeerId(), @@ -187,18 +159,6 @@ return( } - -
-
- peerConnection ? - disconnectPeer(e) : - connectPeer(e) - } - className={peerConnection ? 'disconnect peer-connect-button' : 'peer-connect-button'}> - { peerConnection ? 'Connected' : 'Connect'} -
-
-
{peer &&
@@ -215,34 +175,20 @@ return(
- {peerConnection &&
-
Connection Time
-
{moment(peerConnection.getConnectTimeS() * 1000).fromNow(true)}
+
Number of downloads
+
TODO
-
Bytes received
-
{peerConnection.getNumberBytesReceived()}
+
Number of purchases
+
TODO
-
Messages received
-
{peerConnection.getNumberMessagesReceived()}
-
-
-
Last Message Received Time
-
{moment(peerConnection.getLastMessageReceivedTimeS() * 1000).fromNow(true)}
-
-
-
Bytes sent
-
{peerConnection.getNumberBytesSent()}
-
-
-
Messages sent
-
{peerConnection.getNumberMessagesSent()}
+
Last connection time
+
TODO
- }
diff --git a/frontend/src/components/Peers/index.js b/frontend/src/components/Peers/index.js index 0b721568..5da11041 100644 --- a/frontend/src/components/Peers/index.js +++ b/frontend/src/components/Peers/index.js @@ -11,12 +11,8 @@ import { useSelector } from 'react-redux' import { useDispatch } from 'react-redux' import { - fetchConnectedPeers, - selectConnectedPeers, - selectConnectedPeersStatus, selectSavedPeers, fetchSavedPeers, - setConnectPeer, setSavePeer, } from '../../features/peers/peersSlice' import { @@ -38,14 +34,12 @@ const Peers = (props) => { const externalAddress = useSelector(selectExternalAddress); const peers = useSelector(selectSavedPeers); - const connectedPeers = useSelector(selectConnectedPeers); const dispatch = useDispatch(); useEffect(() => { window.scrollTo(0, 0) dispatch(fetchExternalAddress()); - dispatch(fetchConnectedPeers()); dispatch(fetchSavedPeers()); }, []) @@ -61,12 +55,6 @@ const Peers = (props) => { return `${peerAddress.getNetwork()}/${peerAddress.getHost()}:${peerAddress.getPort()}`; } - const isPeerConnected = (peerAddress) => { - const peerAddressStr = peerAddressToStr(peerAddress); - const connectedPeerAddresses = connectedPeers.map((p) => peerAddressToStr(p.getPeerAddress())); - return connectedPeerAddresses.includes(peerAddressStr); - }; - const toggleSavePeerModal = (param, type) => { setStyleBody(!styleBody) setTimeout(()=>{ setSavePeerModalOpen(!savePeerModalOpen) },20) @@ -135,10 +123,7 @@ const Peers = (props) => {
setTab('Saved Peers')} className={tab === 'Saved Peers' ? `explore-nav-item activeTab` : `explore-nav-item`}> - Saved Peers -
-
setTab('Connected Peers')} className={tab === 'Connected Peers' ? `explore-nav-item activeTab` : `explore-nav-item`}> - Connected Peers + Peers
{tab === 'Saved Peers' ? @@ -149,13 +134,9 @@ const Peers = (props) => { const host = peerAddress.getHost(); const port = peerAddress.getPort(); const addrStr = host + ':' + port; - const isConnected = isPeerConnected(peerAddress); return
goToPeer(peerAddress)} key={peerId} className="search-result-wapper"> - {isConnected ? - : - - } + :
@@ -171,28 +152,7 @@ const Peers = (props) => { }) : tab === 'Connected Peers' ? - connectedPeers.map(p=>{ - const peerAddress = p.getPeerAddress(); - const host = peerAddress.getHost(); - const port = peerAddress.getPort(); - const addrStr = host + ':' + port; - const savedPeer = p.getSavedPeer(); - const savedPeerName = savedPeer && savedPeer.getPeerName(); - return
goToPeer(peerAddress)} key={addrStr} className="search-result-wapper"> - -
-
-
-
{savedPeerName}
-
{addrStr}
-
-
-
-   -
-
-
- }) + <> :
Nothing to see here ..
diff --git a/frontend/src/components/Profile/index.js b/frontend/src/components/Profile/index.js index 9b222de3..db024ed0 100644 --- a/frontend/src/components/Profile/index.js +++ b/frontend/src/components/Profile/index.js @@ -30,7 +30,6 @@ import { selectLastProfileSqueak, selectProfileSqueaksStatus, clearProfileSqueaks, - setDownloadPubkeySqueaks, } from '../../features/squeaks/squeaksSlice' @@ -118,14 +117,6 @@ const Profile = (props) => { }); } - const downloadUserSqueaks = () => { - dispatch(setDownloadPubkeySqueaks(props.match.params.username)) - .then(() => { - console.log('Finished downloading pubkey squeaks.'); - }); - } - - const createContactProfile = () => { dispatch(setCreateContactProfile({ pubkey: userParam, @@ -311,13 +302,6 @@ const Profile = (props) => {
} -
- downloadUserSqueaks() - } - className={'profile-edit-button'}> - {'Download Squeaks'} -
- {user &&
user.getFollowing() ? diff --git a/frontend/src/features/squeaks/squeaksSlice.js b/frontend/src/features/squeaks/squeaksSlice.js index 9b1eb426..620c5ae2 100644 --- a/frontend/src/features/squeaks/squeaksSlice.js +++ b/frontend/src/features/squeaks/squeaksSlice.js @@ -19,7 +19,6 @@ import { getSqueakOffers, buySqueak, downloadSqueak, - downloadPubkeySqueaks, } from '../../api/client' const initialState = { @@ -40,7 +39,6 @@ const initialState = { squeakOffers: [], buySqueakStatus: 'idle', downloadSqueakStatus: 'idle', - downloadPubkeySqueakStatus: 'idle', } // Thunk functions @@ -199,15 +197,6 @@ export const setDownloadSqueak = createAsyncThunk( } ) -export const setDownloadPubkeySqueaks = createAsyncThunk( - 'squeaks/setDownloadPubkeySqueaks', - async (pubkey) => { - console.log('Downloading pubkey squeaks'); - const response = await downloadPubkeySqueaks(pubkey); - return response; - } -) - const updatedSqueakInArray = (squeakArr, newSqueak) => { const currentIndex = squeakArr.findIndex(squeak => squeak.getSqueakHash() === newSqueak.getSqueakHash()); if (currentIndex != -1) { @@ -394,16 +383,6 @@ const squeaksSlice = createSlice({ // TODO: check if current squeak is the one that got downloaded. state.currentSqueak = newSqueak; }) - .addCase(setDownloadPubkeySqueaks.pending, (state, action) => { - state.downloadPubkeySqueakStatus = 'loading' - }) - .addCase(setDownloadPubkeySqueaks.rejected, (state, action) => { - state.downloadPubkeySqueakStatus = 'idle' - }) - .addCase(setDownloadPubkeySqueaks.fulfilled, (state, action) => { - console.log(action); - state.downloadPubkeySqueakStatus = 'idle'; - }) }, }) @@ -466,5 +445,3 @@ export const selectSqueakOffersStatus = state => state.squeaks.squeakOffersStatu export const selectBuySqueakStatus = state => state.squeaks.buySqueakStatus export const selectDownloadSqueakStatus = state => state.squeaks.downloadSqueakStatus - -export const selectDownloadPubkeySqueakStatus = state => state.squeaks.downloadPubkeySqueakStatus diff --git a/itests/config.ini b/itests/config.ini index 3ab842b6..74a60325 100644 --- a/itests/config.ini +++ b/itests/config.ini @@ -16,6 +16,10 @@ rpc_pass=devpass rpc_use_ssl=true rpc_ssl_cert=/rpc/rpc.cert +[server] +external_address=myexternaladdress.com +port=8765 + [postgresql] host=db user=postgres diff --git a/itests/tests/test_squeak_node.py b/itests/tests/test_squeak_node.py index 92b88016..87590089 100644 --- a/itests/tests/test_squeak_node.py +++ b/itests/tests/test_squeak_node.py @@ -104,8 +104,10 @@ def test_get_external_address(admin_stub): external_address = get_external_address(admin_stub) print(external_address) - assert external_address.host is not None and len(external_address.host) > 0 - assert external_address.port > 0 + # assert external_address.host is not None and len(external_address.host) > 0 + assert external_address.host == 'myexternaladdress.com' + # assert external_address.port > 0 + assert external_address.port == 8765 def test_get_default_peer_port(admin_stub): diff --git a/requirements.txt b/requirements.txt index 4f59ec04..01e9d692 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,6 +2,7 @@ alembic==1.7.1 Flask==2.0.1 flask-cors==3.0.10 flask-login==0.5.0 +flask-sock==0.5.1 Flask-WTF==0.15.1 googleapis-common-protos==1.53.0 importlib_resources==1.4.0 @@ -14,3 +15,4 @@ requests==2.26.0 SQLAlchemy==1.4.25 squeaklib==0.11.0 typed-config==0.2.5 +websocket-client==1.3.1 diff --git a/squeaknode/admin/messages.py b/squeaknode/admin/messages.py index f6a7e001..df80cadd 100644 --- a/squeaknode/admin/messages.py +++ b/squeaknode/admin/messages.py @@ -27,7 +27,6 @@ from squeak.core.keys import SqueakPublicKey from proto import squeak_admin_pb2 from squeaknode.admin.profile_image_util import bytes_to_base64_string from squeaknode.admin.profile_image_util import load_default_profile_image -from squeaknode.core.connected_peer import ConnectedPeer from squeaknode.core.download_result import DownloadResult from squeaknode.core.peer_address import Network from squeaknode.core.peer_address import PeerAddress @@ -160,24 +159,6 @@ def payment_summary_to_message( ) -def connected_peer_to_message(connected_peer: ConnectedPeer) -> squeak_admin_pb2.ConnectedPeer: - return squeak_admin_pb2.ConnectedPeer( - peer_address=peer_address_to_message( - connected_peer.peer.remote_address), - connect_time_s=connected_peer.peer.connect_time, - last_message_received_time_s=connected_peer.peer.last_msg_revc_time, - number_messages_received=connected_peer.peer.num_msgs_received, - number_bytes_received=connected_peer.peer.num_bytes_received, - number_messages_sent=connected_peer.peer.num_msgs_sent, - number_bytes_sent=connected_peer.peer.num_bytes_sent, - is_peer_saved=(connected_peer.saved_peer is not None), - saved_peer=( - squeak_peer_to_message(connected_peer.saved_peer) - if connected_peer.saved_peer else None - ), - ) - - def peer_address_to_message(peer_address: PeerAddress) -> squeak_admin_pb2.PeerAddress: return squeak_admin_pb2.PeerAddress( network=peer_address.network.name, @@ -300,9 +281,3 @@ def optional_sent_payment_to_message(sent_payment: Optional[SentPayment]) -> Opt if sent_payment is None: return None return sent_payment_to_message(sent_payment) - - -def optional_connected_peer_to_message(connected_peer: Optional[ConnectedPeer]) -> Optional[squeak_admin_pb2.ConnectedPeer]: - if connected_peer is None: - return None - return connected_peer_to_message(connected_peer) diff --git a/squeaknode/admin/squeak_admin_server_handler.py b/squeaknode/admin/squeak_admin_server_handler.py index 96a330ba..1aed1a9b 100644 --- a/squeaknode/admin/squeak_admin_server_handler.py +++ b/squeaknode/admin/squeak_admin_server_handler.py @@ -25,13 +25,11 @@ from squeak.core.keys import SqueakPrivateKey from squeak.core.keys import SqueakPublicKey from proto import squeak_admin_pb2 -from squeaknode.admin.messages import connected_peer_to_message from squeaknode.admin.messages import download_result_to_message from squeaknode.admin.messages import message_to_peer_address from squeaknode.admin.messages import message_to_received_payment from squeaknode.admin.messages import message_to_sent_payment from squeaknode.admin.messages import message_to_squeak_entry -from squeaknode.admin.messages import optional_connected_peer_to_message from squeaknode.admin.messages import optional_received_offer_to_message from squeaknode.admin.messages import optional_sent_payment_to_message from squeaknode.admin.messages import optional_squeak_entry_to_message @@ -884,35 +882,6 @@ class SqueakAdminServerHandler(object): self.squeak_controller.connect_peer(peer_address) return squeak_admin_pb2.ConnectPeerReply() - def handle_get_connected_peers(self, request): - logger.info("Handle get connected peers.") - connected_peers = self.squeak_controller.get_connected_peers() - logger.info("Connected peers: {}".format( - connected_peers, - )) - connected_peers_display_msgs = [ - connected_peer_to_message(peer) for peer in connected_peers - ] - return squeak_admin_pb2.GetConnectedPeersReply( - connected_peers=connected_peers_display_msgs - ) - - def handle_get_connected_peer(self, request): - peer_address = message_to_peer_address(request.peer_address) - logger.info("Handle get connected peer for address: {}".format( - peer_address, - )) - connected_peer = self.squeak_controller.get_connected_peer( - peer_address) - logger.info("Connected peer: {}".format( - connected_peer, - )) - connected_peers_display_msg = optional_connected_peer_to_message( - connected_peer) - return squeak_admin_pb2.GetConnectedPeerReply( - connected_peer=connected_peers_display_msg - ) - def handle_disconnect_peer(self, request): peer_address = message_to_peer_address(request.peer_address) logger.info( @@ -920,34 +889,6 @@ class SqueakAdminServerHandler(object): self.squeak_controller.disconnect_peer(peer_address) return squeak_admin_pb2.DisconnectPeerReply() - def handle_subscribe_connected_peers(self, request, stopped): - logger.info("Handle subscribe connected peers") - connected_peers_stream = self.squeak_controller.subscribe_connected_peers( - stopped, - ) - for connected_peers in connected_peers_stream: - connected_peers_display_msgs = [ - connected_peer_to_message(peer) for peer in connected_peers - ] - yield squeak_admin_pb2.GetConnectedPeersReply( - connected_peers=connected_peers_display_msgs - ) - - def handle_subscribe_connected_peer(self, request, stopped): - peer_address = message_to_peer_address(request.peer_address) - logger.info( - "Handle subscribe connected peer with peer address: {}".format(peer_address)) - connected_peer_stream = self.squeak_controller.subscribe_connected_peer( - peer_address, - stopped, - ) - for connected_peer in connected_peer_stream: - connected_peer_display_msg = optional_connected_peer_to_message( - connected_peer) - yield squeak_admin_pb2.GetConnectedPeerReply( - connected_peer=connected_peer_display_msg, - ) - def handle_subscribe_buy_offers(self, request, stopped): squeak_hash_str = request.squeak_hash squeak_hash = bytes.fromhex(squeak_hash_str) diff --git a/squeaknode/admin/webapp/app.py b/squeaknode/admin/webapp/app.py index c95d4a69..fc539bd6 100644 --- a/squeaknode/admin/webapp/app.py +++ b/squeaknode/admin/webapp/app.py @@ -481,18 +481,6 @@ def create_app(handler, username, password): def getlikedsqueakdisplays(msg): return handler.handle_get_liked_squeak_display_entries(msg) - @app.route("/getconnectedpeers", methods=["POST"]) - @login_required - @protobuf_serialized(squeak_admin_pb2.GetConnectedPeersRequest()) - def getconnectedpeers(msg): - return handler.handle_get_connected_peers(msg) - - @app.route("/getconnectedpeer", methods=["POST"]) - @login_required - @protobuf_serialized(squeak_admin_pb2.GetConnectedPeerRequest()) - def getconnectedpeer(msg): - return handler.handle_get_connected_peer(msg) - @app.route("/connectpeer", methods=["POST"]) @login_required @protobuf_serialized(squeak_admin_pb2.ConnectPeerRequest()) diff --git a/squeaknode/bitcoin/bitcoin_block_subscription_client.py b/squeaknode/bitcoin/bitcoin_block_subscription_client.py deleted file mode 100644 index e0f3c5c0..00000000 --- a/squeaknode/bitcoin/bitcoin_block_subscription_client.py +++ /dev/null @@ -1,57 +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 binascii -import logging -import struct -from typing import Iterable - -import zmq.asyncio - -logger = logging.getLogger(__name__) - - -class BitcoinBlockSubscriptionClient: - """Get an iterator of new bitcoin blocks.""" - - def __init__( - self, - host: str, - port: int, - ) -> None: - self.zmqContext = zmq.Context() - self.zmqSubSocket = self.zmqContext.socket(zmq.SUB) - self.zmqSubSocket.setsockopt(zmq.RCVHWM, 0) - self.zmqSubSocket.setsockopt_string(zmq.SUBSCRIBE, "hashblock") - self.zmqSubSocket.connect("tcp://{}:{}".format(host, port)) - - def get_blocks(self) -> Iterable[bytes]: - while True: - topic, body, seq = self.zmqSubSocket.recv_multipart() - sequence = "Unknown" - if len(seq) == 4: - sequence = str(struct.unpack(' None: + min_block = 0 # TODO + max_block = 999999999999 # TODO + followed_public_keys = self.squeak_store.get_followed_public_keys() + peers = self.squeak_store.get_autoconnect_peers() + for peer in peers: + downloader = RangeDownloader( + peer, + self.squeak_store, + min_block, + max_block, + followed_public_keys, + ) + downloader.download() + + def download_pubkey_squeaks(self, pubkey: SqueakPublicKey) -> None: + min_block = 0 # TODO + max_block = 999999999999 # TODO + peers = self.squeak_store.get_autoconnect_peers() + for peer in peers: + downloader = RangeDownloader( + peer, + self.squeak_store, + min_block, + max_block, + [pubkey], + ) + downloader.download() + + def download_single_squeak(self, squeak_hash: bytes) -> None: + peers = self.squeak_store.get_autoconnect_peers() + for peer in peers: + downloader = SingleDownloader( + peer, + self.squeak_store, + squeak_hash, + ) + downloader.download() diff --git a/squeaknode/client/peer_client.py b/squeaknode/client/peer_client.py new file mode 100644 index 00000000..8f01d9e1 --- /dev/null +++ b/squeaknode/client/peer_client.py @@ -0,0 +1,100 @@ +# 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 typing import List +from typing import Optional + +import requests +from squeak.core import CSqueak +from squeak.core.keys import SqueakPublicKey + +from squeaknode.core.offer import Offer +from squeaknode.core.squeak_peer import SqueakPeer + +logger = logging.getLogger(__name__) + + +DOWNLOAD_TIMEOUT_S = 10 + + +class PeerClient: + + def __init__(self, peer: SqueakPeer): + self.peer = peer + self.base_url = f"http://{peer.address.host}:{peer.address.port}" + + def lookup( + self, + min_block: int, + max_block: int, + pubkeys: List[SqueakPublicKey], + ) -> List[bytes]: + pubkeys_str = [ + pubkey.to_bytes().hex() + for pubkey in pubkeys + ] + payload = { + 'minblock': min_block, + 'maxblock': max_block, + 'pubkeys': pubkeys_str, + } + url = f"{self.base_url}/lookup" + r = requests.get(url, params=payload) # type: ignore + squeak_hashes_str = r.json() + return [ + bytes.fromhex(squeak_hash_str) + for squeak_hash_str in squeak_hashes_str + ] + + def get_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]: + squeak_hash_str = squeak_hash.hex() + url = f"{self.base_url}/squeak/{squeak_hash_str}" + r = requests.get(url) + if r.status_code != requests.codes.ok: + return None + squeak_bytes = r.content + return CSqueak.deserialize(squeak_bytes) + + def get_secret_key(self, squeak_hash: bytes) -> Optional[bytes]: + squeak_hash_str = squeak_hash.hex() + url = f"{self.base_url}/secretkey/{squeak_hash_str}" + r = requests.get(url) + if r.status_code != requests.codes.ok: + return None + secret_key = r.content + return secret_key + + def get_offer(self, squeak_hash: bytes) -> Optional[Offer]: + squeak_hash_str = squeak_hash.hex() + url = f"{self.base_url}/offer/{squeak_hash_str}" + r = requests.get(url) + if r.status_code != requests.codes.ok: + return None + offer_json = r.json() + offer = Offer( + squeak_hash=bytes.fromhex(offer_json['squeak_hash']), + nonce=bytes.fromhex(offer_json['nonce']), + payment_request=offer_json['payment_request'], + host=offer_json['host'], + port=int(offer_json['port']), + ) + return offer diff --git a/squeaknode/client/peer_downloader.py b/squeaknode/client/peer_downloader.py new file mode 100644 index 00000000..a84071a2 --- /dev/null +++ b/squeaknode/client/peer_downloader.py @@ -0,0 +1,146 @@ +# 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 abc import ABC +from abc import abstractmethod +from typing import List + +from squeak.core import CSqueak +from squeak.core.keys import SqueakPublicKey + +from squeaknode.client.peer_client import PeerClient +from squeaknode.core.squeak_peer import SqueakPeer +from squeaknode.core.squeaks import get_hash +from squeaknode.node.squeak_store import SqueakStore + +logger = logging.getLogger(__name__) + + +DOWNLOAD_TIMEOUT_S = 10 + + +class PeerDownloader(ABC): + + def __init__(self, peer: SqueakPeer, squeak_store: SqueakStore): + self.peer = peer + self.client = PeerClient(peer) + self.squeak_store = squeak_store + + @abstractmethod + def get_hashes(self) -> List[bytes]: + """Get list of squeak hashes to download. + """ + + @abstractmethod + def is_squeak_wanted(self, squeak: CSqueak) -> bool: + """Return true if squeak is supposed to be downloaded. + """ + + def download(self) -> None: + squeak_hashes = self.get_hashes() + for squeak_hash in squeak_hashes: + # Download the squeak if not already owned. + self.download_squeak(squeak_hash) + # Download the secret key if not already unlocked. + self.download_secret_key(squeak_hash) + # Download the offer if not already unlocked. + self.download_offer(squeak_hash) + + def download_squeak(self, squeak_hash: bytes) -> None: + # Download the squeak if not already owned. + if self.squeak_store.get_squeak(squeak_hash): + return + squeak = self.client.get_squeak(squeak_hash) + if squeak and self.is_squeak_wanted(squeak): + self.squeak_store.save_squeak(squeak) + + def download_secret_key(self, squeak_hash: bytes) -> None: + # Get the squeak from the database. + squeak = self.squeak_store.get_squeak(squeak_hash) + if squeak and self.is_squeak_wanted(squeak): + # Download the secret key is not already unlocked. + if self.squeak_store.get_squeak_secret_key(squeak_hash): + return + secret_key = self.client.get_secret_key(squeak_hash) + if secret_key: + self.squeak_store.save_secret_key(squeak_hash, secret_key) + + def download_offer(self, squeak_hash: bytes) -> None: + # Get the squeak from the database. + squeak = self.squeak_store.get_squeak(squeak_hash) + if squeak and self.is_squeak_wanted(squeak): + # Download the secret key is not already unlocked. + if self.squeak_store.get_squeak_secret_key(squeak_hash): + return + offer = self.client.get_offer(squeak_hash) + if offer: + self.squeak_store.handle_offer( + squeak, + offer, + self.peer.address, + ) + + +class RangeDownloader(PeerDownloader): + + def __init__( + self, + peer: SqueakPeer, + squeak_store: SqueakStore, + min_block: int, + max_block: int, + pubkeys: List[SqueakPublicKey], + ): + super().__init__(peer, squeak_store) + self.min_block = min_block + self.max_block = max_block + self.pubkeys = pubkeys + + def get_hashes(self) -> List[bytes]: + return self.client.lookup( + self.min_block, + self.max_block, + self.pubkeys, + ) + + def is_squeak_wanted(self, squeak: CSqueak) -> bool: + return squeak.nBlockHeight >= self.min_block and \ + squeak.nBlockHeight <= self.max_block and \ + squeak.GetPubKey() in self.pubkeys + + +class SingleDownloader(PeerDownloader): + + def __init__( + self, + peer: SqueakPeer, + squeak_store: SqueakStore, + squeak_hash: bytes, + ): + super().__init__(peer, squeak_store) + self.squeak_hash = squeak_hash + + def get_hashes(self) -> List[bytes]: + return [self.squeak_hash] + + def is_squeak_wanted(self, squeak: CSqueak) -> bool: + return get_hash(squeak) == self.squeak_hash diff --git a/squeaknode/core/connected_peer.py b/squeaknode/core/connected_peer.py deleted file mode 100644 index 9c17ef12..00000000 --- a/squeaknode/core/connected_peer.py +++ /dev/null @@ -1,32 +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. -from typing import NamedTuple -from typing import Optional - -from squeaknode.core.squeak_peer import SqueakPeer -from squeaknode.network.peer import Peer - - -class ConnectedPeer(NamedTuple): - """Represents another node in the network.""" - peer: Peer - saved_peer: Optional[SqueakPeer] diff --git a/squeaknode/network/connection.py b/squeaknode/network/connection.py deleted file mode 100644 index 9a7e8dee..00000000 --- a/squeaknode/network/connection.py +++ /dev/null @@ -1,365 +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 -import threading - -from squeak.messages import msg_addr -from squeak.messages import msg_getaddr -from squeak.messages import msg_getdata -from squeak.messages import msg_inv -from squeak.messages import msg_notfound -from squeak.messages import msg_ping -from squeak.messages import msg_pong -from squeak.messages import MSG_SECRET_KEY -from squeak.messages import MSG_SQUEAK -from squeak.messages import msg_squeak - -from squeaknode.core.crypto import generate_ping_nonce -from squeaknode.core.offer import Offer -from squeaknode.network.peer import Peer -from squeaknode.node.network_handler import NetworkHandler - - -logger = logging.getLogger(__name__) - - -HANDSHAKE_TIMEOUT = 30 -PING_TIMEOUT = 60 -PONG_TIMEOUT = 30 - - -class Connection(object): - """Handles lifecycle of a connection to a peer. - """ - - def __init__(self, peer: Peer, network_handler: NetworkHandler): - self.peer = peer - self.network_handler = network_handler - self.handshake_timer = HandshakeTimer(self) - self.ping_timer = PingTimer(self) - self.pong_timer = PongTimer(self) - - def handshake(self): - """Do a handshake with a peer. - """ - self.handshake_timer.start_timer() - - if self.peer.outgoing: - self.peer.send_version() - self.peer.receive_version() - if not self.peer.outgoing: - self.peer.send_version() - - self.peer.set_connected() - self.handshake_timer.cancel() - logger.debug("HANDSHAKE COMPLETE-----------") - - def shutdown(self): - logger.info("Peer shutting down...") - self.peer.stop() - self.handshake_timer.cancel() - self.ping_timer.cancel() - self.pong_timer.cancel() - - def handle_connection(self): - try: - self.initial_sync() - self.handle_msgs() - except Exception: - logger.exception("Error in handle_connection") - finally: - self.shutdown() - # self._stopped.set() - - def initial_sync(self): - self.send_ping() - self.update_addrs() - self.update_subscription() - - def send_ping(self): - ping_msg = msg_ping() - ping_msg.nonce = generate_ping_nonce() - self.pong_timer.start_timer(ping_msg.nonce) - self.peer.send_msg(ping_msg) - - def start_ping_timer(self): - self.ping_timer.start_timer() - - def update_subscription(self): - locator = self.network_handler.get_interested_locator() - self.peer.update_local_subscription(locator) - - def update_addrs(self): - getaddr_msg = msg_getaddr() - self.peer.send_msg(getaddr_msg) - - def handle_msgs(self): - """Handles messages from the peer if there are any available. - - This method blocks when the peer has not sent any messages. - """ - msg = self.peer.recv_msg() - while msg is not None: - self.handle_peer_message(msg) - msg = self.peer.recv_msg() - logger.info("Finished handle_msgs") - - def handle_peer_message(self, msg): - """Handle messages from a peer with completed handshake.""" - if msg.command == b'ping': - self.handle_ping(msg) - elif msg.command == b'pong': - self.handle_pong(msg) - elif msg.command == b'addr': - self.handle_addr(msg) - elif msg.command == b'getaddr': - self.handle_getaddr(msg) - elif msg.command == b'inv': - self.handle_inv(msg) - elif msg.command == b'getsqueaks': - self.handle_getsqueaks(msg) - elif msg.command == b'squeak': - self.handle_squeak(msg) - elif msg.command == b'getdata': - self.handle_getdata(msg) - elif msg.command == b'notfound': - self.handle_notfound(msg) - elif msg.command == b'secretkey': - self.handle_secret_key(msg) - elif msg.command == b'subscribe': - self.handle_subscribe(msg) - else: - raise Exception("Unrecognized message: {}".format( - msg.command - )) - - def handle_ping(self, msg): - nonce = msg.nonce - pong = msg_pong() - pong.nonce = nonce - self.peer.set_last_recv_ping() - self.peer.send_msg(pong) - - def handle_pong(self, msg): - nonce = msg.nonce - self.pong_timer.stop_timer(nonce) - - def handle_addr(self, msg): - # TODO: Save new address in table. - for addr in msg.addrs: - pass - - def handle_getaddr(self, msg): - # TODO: Get known peers from table in database. - addr_msg = msg_addr(addrs=[]) - self.peer.send_msg(addr_msg) - - def handle_inv(self, msg): - invs = msg.inv - unknown_invs = self.network_handler.get_unknown_invs(invs) - if unknown_invs: - getdata_msg = msg_getdata(inv=unknown_invs) - self.peer.send_msg(getdata_msg) - - def handle_getdata(self, msg): - invs = msg.inv - for inv in invs: - if inv.type == MSG_SQUEAK: - squeak = self.network_handler.get_squeak(inv.hash) - if squeak is None: - reply_msg = msg_notfound(inv=[inv]) - self.peer.send_msg(reply_msg) - else: - reply_msg = msg_squeak(squeak=squeak) - self.peer.send_msg(reply_msg) - if inv.type == MSG_SECRET_KEY: - reply = self.network_handler.get_secret_key_reply( - inv.hash, - self.peer.remote_address, - ) - if reply is None: - reply_msg = msg_notfound(inv=[inv]) - self.peer.send_msg(reply_msg) - else: - reply_msg = reply.get_msg() - self.peer.send_msg(reply_msg) - - def handle_notfound(self, msg): - pass - - def handle_getsqueaks(self, msg): - locator = msg.locator - for interest in locator.vInterested: - reply_invs = self.network_handler.get_reply_invs(interest) - inv_msg = msg_inv(inv=reply_invs) - self.peer.send_msg(inv_msg) - - def handle_squeak(self, msg): - squeak = msg.squeak - saved_squeak_hash = self.network_handler.save_squeak(squeak) - if saved_squeak_hash is not None: - self.network_handler.request_offers(saved_squeak_hash) - - def handle_secret_key(self, msg): - if msg.has_secret_key(): - self.network_handler.unlock_squeak( - msg.hashSqk, - msg.secretKey, - ) - elif msg.has_offer(): - offer = Offer( - squeak_hash=msg.hashSqk, - nonce=msg.offer.nonce, - payment_request=msg.offer.strPaymentInfo.decode('utf-8'), - host=msg.offer.host.decode('utf-8'), - port=msg.offer.port, - ) - self.network_handler.handle_received_offer( - offer, - self.peer.remote_address, - ) - - def handle_subscribe(self, msg): - self.peer.set_remote_subscription(msg.locator) - - -class HandshakeTimer: - """Stop the peer if handshake is not complete before timeout. - """ - - def __init__(self, connection: Connection): - self.connection = connection - self.timer = None - self._lock = threading.Lock() - - def start_timer(self): - with self._lock: - self.timer = threading.Timer( - HANDSHAKE_TIMEOUT, - self.shutdown, - ) - self.timer.name = "handshake_timer_thread_{}".format( - self.connection.peer) - self.timer.start() - - def cancel(self): - logger.debug("Cancelling handshake timer.") - with self._lock: - if self.timer: - self.timer.cancel() - - def shutdown(self): - logger.info("Shutdown connection triggered by handshake timer.") - self.connection.shutdown() - - -class PingTimer: - """Send a ping message when the timer expires. - """ - - def __init__(self, connection: Connection): - self.connection = connection - self.timer = None - self._lock = threading.Lock() - - def start_timer(self): - logger.debug("Starting ping timer.") - with self._lock: - # Cancel the existing timer. - if self.timer: - self.timer.cancel() - - # Start a new timer. - self.timer = threading.Timer( - PING_TIMEOUT, - self.send_ping, - ) - self.timer.name = "ping_timer_thread_{}".format( - self.connection.peer) - self.timer.start() - - def cancel(self): - logger.debug("Cancelling ping timer.") - with self._lock: - if self.timer: - self.timer.cancel() - - def send_ping(self): - logger.debug("Sending ping triggered by timer.") - self.connection.send_ping() - - -class PongTimer: - """Shut down the connection when the timer expires. - """ - - def __init__(self, connection: Connection): - self.connection = connection - self.timer = None - self.expected_nonce = None - self._lock = threading.Lock() - - def start_timer(self, nonce): - logger.debug("Starting pong timer.") - with self._lock: - # Cancel the existing timer. - if self.timer: - return - - # Start a new timer. - self.expected_nonce = nonce - self.timer = threading.Timer( - PONG_TIMEOUT, - self.shutdown, - ) - self.timer.name = "pong_timer_thread_{}".format( - self.connection.peer) - self.timer.start() - - def stop_timer(self, nonce): - logger.debug("Stopping pong timer.") - with self._lock: - if nonce != self.expected_nonce: - self.shutdown() - - # Cancel the existing timer. - if self.timer: - self.timer.cancel() - self.timer = None - self.expected_nonce = None - - # Start a new ping timer. - self.start_ping_timer() - - def cancel(self): - logger.debug("Cancelling pong timer.") - with self._lock: - if self.timer: - self.timer.cancel() - - def shutdown(self): - logger.info("Shutdown connection triggered by pong timer.") - self.connection.shutdown() - - def start_ping_timer(self): - logger.debug("Starting ping timer triggered by pong response.") - self.connection.start_ping_timer() diff --git a/squeaknode/network/connection_manager.py b/squeaknode/network/connection_manager.py deleted file mode 100644 index 87d71775..00000000 --- a/squeaknode/network/connection_manager.py +++ /dev/null @@ -1,188 +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 -import queue -import socket -import threading -from contextlib import contextmanager -from typing import Dict -from typing import List -from typing import Optional - -from squeaknode.core.peer_address import PeerAddress -from squeaknode.network.connection import Connection -from squeaknode.network.peer import Peer -from squeaknode.network.peer_client import ConnectPeerResult -from squeaknode.node.listener_subscription_client import EventListener -from squeaknode.node.network_handler import NetworkHandler - - -MIN_PEERS = 5 -MAX_PEERS = 10 -UPDATE_THREAD_SLEEP_TIME = 10 - - -logger = logging.getLogger(__name__) - - -class ConnectionManager(object): - """Maintains connections to other peers in the network. - """ - - def __init__(self, local_address): - self._peers: Dict[PeerAddress, Peer] = {} - self.peers_lock = threading.Lock() - self.peer_changed_listener = EventListener() - self.single_peer_changed_listener = EventListener() - self.accept_connections = True - self.local_address = local_address - - @contextmanager - def connect( - self, - peer_socket: socket.socket, - address: PeerAddress, - outgoing: bool, - network_handler: NetworkHandler, - result_queue: queue.Queue, - ): - try: - peer = Peer( - peer_socket, - self.local_address, - address, - outgoing, - self.single_peer_changed_listener, - ) - connection = Connection(peer, network_handler) - logger.debug("Doing handshake.") - connection.handshake() - logger.debug("Adding peer.") - self._add_peer(peer) - result_queue.put( - ConnectPeerResult.from_success(peer.remote_address)) - logger.debug("Yielding connection.") - yield connection - logger.debug("Removing peer.") - self._remove_peer(peer) - except Exception as e: - logger.exception("Error in connection.") - result_queue.put(ConnectPeerResult.from_failure(e)) - raise - finally: - logger.info("Disconnected peer.") - peer.stop() - - @property - def peers(self) -> List[Peer]: - return list(self._peers.values()) - - def has_connection(self, address): - """Return True if the address is already connected.""" - return address in self._peers - - def _on_peers_changed(self): - peers = self.peers - logger.info('Current number of peers {}'.format(len(peers))) - logger.info('Current peers:--------') - for peer in peers: - logger.info(peer) - logger.info('--------------') - self.peer_changed_listener.handle_new_item(peers) - - def _is_duplicate_nonce(self, peer): - for other_peer in self.peers: - if other_peer.local_version: - if peer.remote_version == other_peer.local_version.nNonce: - return True - return False - - def _add_peer(self, peer: Peer): - """Add a peer. - """ - with self.peers_lock: - if not self.accept_connections: - raise NotAcceptingConnectionsError() - if self._is_duplicate_nonce(peer): - logger.debug('Failed to add peer {}'.format(peer)) - raise DuplicateNonceError() - if self.has_connection(peer.remote_address): - logger.debug('Failed to add peer {}'.format(peer)) - raise DuplicatePeerError() - self._peers[peer.remote_address] = peer - logger.debug('Added peer {}'.format(peer)) - self._on_peers_changed() - - def _remove_peer(self, peer: Peer): - """Remove a peer. - """ - with self.peers_lock: - if not self.has_connection(peer.remote_address): - return - del self._peers[peer.remote_address] - logger.debug('Removed peer {}'.format(peer)) - self._on_peers_changed() - - def get_peer(self, address) -> Optional[Peer]: - """Get a peer info by address. - """ - return self._peers.get(address) - - def stop_connection(self, address): - """Stop peer connections for address. - """ - with self.peers_lock: - peer = self.get_peer(address) - if peer is not None: - peer.stop() - - def stop_all_connections(self): - """Stop all peer connections. - """ - self.accept_connections = False - with self.peers_lock: - for peer in self.peers: - peer.stop() - - def yield_peers_changed(self, stopped: threading.Event): - yield from self.peer_changed_listener.yield_items(stopped) - - def yield_single_peer_changed(self, peer_address: PeerAddress, stopped: threading.Event): - for peer in self.single_peer_changed_listener.yield_items(stopped): - logger.debug('yield_single_peer_changed: {}'.format(peer)) - if peer.remote_address == peer_address: - if peer.connect_time is None: - yield None - else: - yield peer - - -class DuplicatePeerError(Exception): - pass - - -class DuplicateNonceError(Exception): - pass - - -class NotAcceptingConnectionsError(Exception): - pass diff --git a/squeaknode/network/network_manager.py b/squeaknode/network/network_manager.py deleted file mode 100644 index 0d17a71f..00000000 --- a/squeaknode/network/network_manager.py +++ /dev/null @@ -1,156 +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 -import socket -from typing import Iterable -from typing import List -from typing import Optional - -from squeak.messages import MsgSerializable -from squeak.net import CSqueakLocator - -from squeaknode.core.peer_address import Network -from squeaknode.core.peer_address import PeerAddress -from squeaknode.network.connection_manager import ConnectionManager -from squeaknode.network.peer import Peer -from squeaknode.network.peer_client import PeerClient -from squeaknode.network.peer_handler import PeerHandler -from squeaknode.network.peer_server import PeerServer - - -MIN_PEERS = 5 -MAX_PEERS = 10 -UPDATE_THREAD_SLEEP_TIME = 10 - - -logger = logging.getLogger(__name__) - - -class NetworkManager(object): - """Interface for doing things involving the network. - """ - - def __init__(self, config, default_port): - self.config = config - self.external_host = self.config.server.external_address - self.local_ip = socket.gethostbyname('localhost') - self.local_port = self.config.server.port or default_port - self.peer_server = None - self.peer_client = None - self.tor_proxy_ip = self.config.tor.proxy_ip - self.tor_proxy_port = self.config.tor.proxy_port - self.connection_manager = ConnectionManager(self.local_address) - - def start(self, network_handler): - peer_handler = PeerHandler( - self.connection_manager, - network_handler, - ) - self.peer_server = PeerServer( - peer_handler, - self.local_port, - ) - self.peer_client = PeerClient( - peer_handler, - self.tor_proxy_ip, - self.tor_proxy_port, - ) - self.peer_server.start() - - def stop(self): - self.peer_server.stop() - self.connection_manager.stop_all_connections() - - def connect_peer_sync(self, peer_address: PeerAddress) -> None: - if self.connection_manager.has_connection(peer_address): - raise Exception("Already connected to: {}".format(peer_address)) - self.peer_client.connect_address(peer_address) - - def connect_peer_async(self, peer_address: PeerAddress) -> None: - if self.connection_manager.has_connection(peer_address): - return - self.peer_client.connect_address_async(peer_address) - - def disconnect_peer(self, peer_address: PeerAddress) -> None: - self.connection_manager.stop_connection(peer_address) - - def get_connected_peer(self, peer_address: PeerAddress) -> Optional[Peer]: - return self.connection_manager.get_peer(peer_address) - - def get_connected_peers(self) -> List[Peer]: - return self.connection_manager.peers - - def broadcast_msg(self, msg: MsgSerializable) -> int: - """Send a message to all connected peers. - - Returns: - int: the number of peers message was sent to. - """ - count = 0 - for peer in self.connection_manager.peers: - try: - peer.send_msg(msg) - count += 1 - except Exception: - logger.exception("Failed to send msg to peer: {}".format( - peer, - )) - return count - - def update_local_subscriptions(self, locator: CSqueakLocator) -> None: - for peer in self.connection_manager.peers: - try: - peer.update_local_subscription(locator) - except Exception: - logger.exception("Failed to update local subcription with peer: {}".format( - peer, - )) - - @property - def local_address(self) -> PeerAddress: - return PeerAddress( - network=Network.IPV4, - host=self.local_ip, - port=self.local_port, - ) - - @property - def external_address(self) -> PeerAddress: - return PeerAddress( - network=Network.IPV4, - host=self.external_host or self.local_ip, - port=self.local_port, - ) - - def subscribe_connected_peers(self, stopped) -> Iterable[List[Peer]]: - # yield from self.connection_manager.yield_peers_changed(stopped) - for item in self.connection_manager.yield_peers_changed(stopped): - logger.info("subscribe_connected_peers yielding item: {}".format( - item, - )) - yield item - - def subscribe_connected_peer(self, peer_address: PeerAddress, stopped) -> Iterable[Peer]: - yield from self.connection_manager.yield_single_peer_changed( - peer_address, - stopped, - ) diff --git a/squeaknode/network/peer.py b/squeaknode/network/peer.py deleted file mode 100644 index 675e406a..00000000 --- a/squeaknode/network/peer.py +++ /dev/null @@ -1,403 +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 -import socket -import threading -import time -from io import BytesIO -from typing import Optional - -from bitcoin.core.serialize import SerializationTruncationError -from bitcoin.net import CAddress -from squeak.core import CSqueak -from squeak.messages import msg_getsqueaks -from squeak.messages import msg_subscribe -from squeak.messages import msg_verack -from squeak.messages import msg_version -from squeak.messages import MsgSerializable -from squeak.net import CSqueakLocator - -from squeaknode.core.crypto import generate_version_nonce -from squeaknode.core.interests import get_differential_squeaks -from squeaknode.core.interests import squeak_matches_interest -from squeaknode.core.peer_address import PeerAddress -from squeaknode.network.util import time_now -from squeaknode.node.listener_subscription_client import EventListener - - -MAX_MESSAGE_LEN = 1048576 -SOCKET_READ_LEN = 1024 -LAST_MESSAGE_TIMEOUT = 600 -PING_TIMEOUT = 10 -PING_INTERVAL = 60 - -UPDATE_TIME_INTERVAL = 10 -HANDSHAKE_VERSION = 70002 - - -logger = logging.getLogger(__name__) - - -class Peer(object): - """Maintains the internal state of a peer connection. - """ - - def __init__( - self, - peer_socket: socket.socket, - local_address: PeerAddress, - remote_address: PeerAddress, - outgoing: bool, - peer_changed_listener: EventListener, - ): - self._peer_socket = peer_socket - self._peer_socket_lock = threading.Lock() - self._local_address = local_address - self._remote_address = remote_address - self._outgoing = outgoing - self._connect_time = 0 - self._local_version = None - self._remote_version = None - self._last_msg_revc_time = None - self._last_sent_ping_nonce = None - self._last_sent_ping_time = None - self._last_recv_ping_time = None - self._num_msgs_received = 0 - self._num_bytes_received = 0 - self._num_msgs_sent = 0 - self._num_bytes_sent = 0 - - self._remote_subscription = None - self._local_subscription = None - - self.msg_receiver = MessageReceiver( - self._peer_socket, - ) - self.received_msgs_iter = self.msg_receiver.recv_msgs() - - self.peer_changed_listener = peer_changed_listener - - @property - def nVersion(self): - remote_version = self._remote_version - if remote_version: - return remote_version.nVersion - - @property - def local_address(self): - return self._local_address - - @property - def remote_address(self): - return self._remote_address - - @property - def remote_subscription(self): - return self._remote_subscription - - @property - def local_subscription(self): - return self._local_subscription - - @property - def local_caddress(self): - caddress = CAddress() - caddress.nTime = self.connect_time - caddress.ip = self.local_address.host - caddress.port = self.local_address.port - return caddress - - @property - def remote_caddress(self): - caddress = CAddress() - caddress.nTime = self.connect_time - # TODO: Set the remote address ip - # caddress.ip = - caddress.port = self.remote_address.port - return caddress - - @property - def outgoing(self): - return self._outgoing - - @property - def connect_time(self): - return self._connect_time - - @property - def num_msgs_received(self): - return self._num_msgs_received - - @property - def num_bytes_received(self): - return self._num_bytes_received - - @property - def num_msgs_sent(self): - return self._num_msgs_sent - - @property - def num_bytes_sent(self): - return self._num_bytes_sent - - @property - def local_version(self): - return self._local_version - - @local_version.setter - def local_version(self, local_version): - self._local_version = local_version - - @property - def remote_version(self): - return self._remote_version - - @remote_version.setter - def remote_version(self, remote_version): - self._remote_version = remote_version - - @property - def last_msg_revc_time(self): - return self._last_msg_revc_time - - @property - def last_sent_ping_time(self): - return self._last_sent_ping_time - - def set_last_sent_ping(self, nonce, timestamp=None): - timestamp = timestamp or time.time() - self._last_sent_ping_nonce = nonce - self._last_sent_ping_time = time.time() - - @property - def last_recv_ping_time(self): - return self._last_recv_ping_time - - def set_last_recv_ping(self, timestamp=None): - timestamp = timestamp or time.time() - self._last_recv_ping_time = timestamp - - # @property - # def peer_state(self): - # return ConnectedPeer( - # peer_address=self.remote_address, - # connect_time_s=self.connect_time, - # outgoing=self.outgoing, - # sent_bytes=0, - # sent_messages=0, - # received_bytes=0, - # received_messages=0, - # ) - - def recv_msg(self): - """Read data from the peer socket, and yield messages as they are decoded. - - This method blocks when the socket has no data to read. - """ - try: - msg = next(self.received_msgs_iter) - except Exception: - return None - self.record_msg_received(msg) - self.on_peer_updated() - logger.info('Received msg {} from {}'.format(msg, self)) - return msg - - def stop(self): - logger.info("Stopping peer socket: {}".format(self._peer_socket)) - try: - self._peer_socket.shutdown(socket.SHUT_RDWR) - self._peer_socket.stop() - except Exception: - pass - finally: - self.set_disconnected() - self.on_peer_updated() - - def send_msg(self, msg): - logger.info('Sending msg {} to {}'.format(msg, self)) - data = msg.to_bytes() - try: - with self._peer_socket_lock: - self._peer_socket.send(data) - self.record_msg_sent(msg) - self.on_peer_updated() - except Exception: - logger.info('Failed to send msg to {}'.format(self)) - self.stop() - - def send_version(self): - local_version = self.version_pkt() - self.local_version = local_version - self.send_msg(local_version) - verack = self.recv_msg() - if not isinstance(verack, msg_verack): - raise Exception('Expected verack response: {}'.format( - verack, - )) - - def receive_version(self): - remote_version = self.recv_msg() - if not isinstance(remote_version, msg_version): - raise Exception('Expected version message. Received: {}'.format( - remote_version, - )) - self.remote_version = remote_version - verack = msg_verack() - self.send_msg(verack) - - def version_pkt(self): - """Get the version message for this peer.""" - msg = msg_version() - msg.nVersion = HANDSHAKE_VERSION - msg.addrTo = self.remote_caddress - msg.addrFrom = self.local_caddress - msg.nNonce = generate_version_nonce() - return msg - - def set_connected(self): - self._connect_time = time_now() - - def set_disconnected(self): - self._connect_time = None - - def set_remote_subscription(self, locator: Optional[CSqueakLocator]): - self._remote_subscription = locator - - def set_local_subscription(self, locator: Optional[CSqueakLocator]): - self._local_subscription = locator - - def is_remote_subscribed(self, squeak: CSqueak): - if self.remote_subscription is None: - return False - for interest in self.remote_subscription.vInterested: - if squeak_matches_interest(squeak, interest): - return True - return False - - def update_local_subscription(self, locator: CSqueakLocator): - assert len(locator.vInterested) <= 1 - if len(locator.vInterested) == 0: - locator = None - - # Send the subscribe message - subscribe_msg = msg_subscribe( - locator=locator, - ) - self.send_msg(subscribe_msg) - - # Send the getsqueaks messages for differential interests - if locator is not None: - for interest in self.get_differential_interests(locator): - diff_locator = CSqueakLocator( - vInterested=[interest] - ) - getsqueaks_msg = msg_getsqueaks( - locator=diff_locator, - ) - self.send_msg(getsqueaks_msg) - - # Set the local subscription with the new value - self.set_local_subscription(locator) - - def get_differential_interests(self, locator: CSqueakLocator): - if self._local_subscription is None: - yield locator.vInterested[0] - else: - new_interest = locator.vInterested[0] - old_interest = self._local_subscription.vInterested[0] - for interest in get_differential_squeaks(new_interest, old_interest): - yield interest - - def on_peer_updated(self): - logger.debug('on_peer_updated: {}'.format(self)) - self.peer_changed_listener.handle_new_item(self) - - def record_msg_received(self, msg): - self._num_msgs_received += 1 - self._num_bytes_received += len(msg.to_bytes()) - self._last_msg_revc_time = time_now() - - def record_msg_sent(self, msg): - if msg: - self._num_msgs_sent += 1 - self._num_bytes_sent += len(msg.to_bytes()) - - # def subscribe_peer_state(self, stopped): - # for result in self.peer_changed_listener.yield_items(stopped): - # yield result - - def __repr__(self): - return "Peer(%s)" % (str(self.remote_address)) - - -class MessageDecoder: - """Handles the incoming binary data from a peer and buffers and decodes. - """ - - def __init__(self): - self.recv_data_buffer = BytesIO() - - def process_recv_data(self, recv_data): - data = self.read_data_buffer() + recv_data - try: - while data: - self.set_data_buffer(data) - msg = MsgSerializable.stream_deserialize(self.recv_data_buffer) - if msg is None: - raise Exception('Invalid data') - else: - yield msg - data = self.read_data_buffer() - except SerializationTruncationError: - self.set_data_buffer(data) - - def read_data_buffer(self): - return self.recv_data_buffer.read() - - def set_data_buffer(self, data): - if len(data) > MAX_MESSAGE_LEN: - raise Exception('Message size too large') - self.recv_data_buffer = BytesIO(data) - - -class MessageReceiver: - """Reads bytes from the socket and return decoded messages in an iterator. - """ - - def __init__(self, socket): - self.socket = socket - self.decoder = MessageDecoder() - - def recv_msgs(self): - while True: - try: - recv_data = self.socket.recv(SOCKET_READ_LEN) - except Exception: - logger.error("Error in recv") - return - if not recv_data: - logger.error("revc_data is None") - return - - for msg in self.decoder.process_recv_data(recv_data): - yield msg diff --git a/squeaknode/network/peer_client.py b/squeaknode/network/peer_client.py deleted file mode 100644 index f1d03642..00000000 --- a/squeaknode/network/peer_client.py +++ /dev/null @@ -1,147 +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 -import queue -import socket -import threading - -import socks - -from squeaknode.core.peer_address import Network -from squeaknode.core.peer_address import PeerAddress - - -SOCKET_CONNECT_TIMEOUT = 30 - - -logger = logging.getLogger(__name__) - - -class PeerClient(object): - """Creates outgoing connections to other peers in the network. - """ - - def __init__(self, peer_handler, tor_proxy_ip, tor_proxy_port): - self.peer_handler = peer_handler - self.tor_proxy_ip = tor_proxy_ip - self.tor_proxy_port = tor_proxy_port - - def connect_address(self, address: PeerAddress): - logger.info('Making connection to {}'.format(address)) - result_queue: queue.Queue = queue.Queue() - threading.Thread( - target=self.make_connection, - args=(address, result_queue,), - ).start() - - # Wait for connect result from the queue. - connect_result = result_queue.get() - logger.debug("connect_result: {}".format(connect_result)) - if connect_result.failure is not None: - raise connect_result.failure - - def connect_address_async(self, address: PeerAddress): - logger.info('Making connection async to {}'.format(address)) - result_queue: queue.Queue = queue.Queue() - threading.Thread( - target=self.make_connection, - args=(address, result_queue,), - ).start() - - def make_connection(self, address: PeerAddress, result_queue: queue.Queue): - logger.info('Conecting to address: {}'.format(address)) - try: - peer_socket = self.get_socket(address) - peer_socket.settimeout(SOCKET_CONNECT_TIMEOUT) - connect_address = (address.host, address.port) - peer_socket.connect(connect_address) - peer_socket.setblocking(True) - self.handle_connection( - peer_socket, - address, - result_queue, - ) - except Exception as e: - logger.exception('Failed to connect to {}'.format(address)) - result_queue.put(ConnectPeerResult.from_failure(e)) - - def handle_connection( - self, - peer_socket: socket.socket, - peer_address: PeerAddress, - result_queue: queue.Queue, - ): - """Handle a newly connected peer socket.""" - self.peer_handler.handle_connection( - peer_socket, - peer_address, - outgoing=True, - result_queue=result_queue, - ) - - def get_socket(self, address: PeerAddress): - if address.network not in [ - Network.IPV4, - Network.IPV6, - Network.TORV3, - ]: - raise Exception("Unsupported network: {}".format(address.network)) - if address.network == Network.TORV3 and self.tor_proxy_ip is None: - raise Exception( - "Unable to connect to tor address without tor proxy ip configured.") - if address.network == Network.TORV3 and self.tor_proxy_port is None: - raise Exception( - "Unable to connect to tor address without tor proxy port configured.") - if address.network == Network.TORV3: - s = socks.socksocket() # Same API as socket.socket in the standard lib - s.set_proxy(socks.SOCKS5, self.tor_proxy_ip, self.tor_proxy_port) - return s - return socket.socket() - - -class ConnectPeerResult(object): - """Result of a connect peer attempt. - """ - - def __init__( - self, - success: PeerAddress = None, - failure: Exception = None, - ): - self.success = success - self.failure = failure - - @classmethod - def from_success(cls, success): - return cls(success=success) - - @classmethod - def from_failure(cls, failure): - return cls(failure=failure) - - def __repr__(self): - return "ConnectPeerResult: \ - success: {} \ - failure: {}".format( - self.success, - self.failure, - ) diff --git a/squeaknode/network/peer_handler.py b/squeaknode/network/peer_handler.py deleted file mode 100644 index 198c1fe4..00000000 --- a/squeaknode/network/peer_handler.py +++ /dev/null @@ -1,87 +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 -import queue -import socket -import threading -from typing import Optional - -from squeaknode.core.peer_address import PeerAddress - - -logger = logging.getLogger(__name__) - - -class PeerHandler(): - """Handles new peer connection. - """ - - def __init__( - self, - connection_manager, - network_handler, - ): - self.connection_manager = connection_manager - self.network_handler = network_handler - - def handle_connection( - self, - peer_socket: socket.socket, - address: PeerAddress, - outgoing: bool, - result_queue: Optional[queue.Queue] = None, - ): - """Handle a new socket connection. - - This method blocks until the peer connection is established. - """ - # Create a dummy queue if not needed. - if result_queue is None: - result_queue = queue.Queue() - - threading.Thread( - target=self.start_connection, - args=( - peer_socket, - address, - outgoing, - result_queue, - ), - ).start() - - def start_connection( - self, - peer_socket: socket.socket, - address: PeerAddress, - outgoing: bool, - result_queue: queue.Queue, - ): - """Start a connection - """ - with self.connection_manager.connect( - peer_socket, - address, - outgoing, - self.network_handler, - result_queue, - ) as connection: - connection.handle_connection() diff --git a/squeaknode/network/peer_server.py b/squeaknode/network/peer_server.py deleted file mode 100644 index 46ab5dcf..00000000 --- a/squeaknode/network/peer_server.py +++ /dev/null @@ -1,91 +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 -import socket -import threading - -from squeaknode.core.peer_address import Network -from squeaknode.core.peer_address import PeerAddress - - -MIN_PEERS = 5 -MAX_PEERS = 10 -UPDATE_THREAD_SLEEP_TIME = 10 - - -logger = logging.getLogger(__name__) - - -class PeerServer(object): - """Accepts incoming connections from other peers in the network. - """ - - def __init__(self, peer_handler, port): - self.peer_handler = peer_handler - self.port = port - self.listen_socket = socket.socket() - - def start(self): - logger.info("Starting peer server with port: {}".format( - self.port, - )) - # Start Listen thread - threading.Thread( - target=self.accept_connections, - name="peer_server_listen_thread", - ).start() - - def stop(self): - logger.info("Stopping peer server listener thread...") - self.listen_socket.shutdown(socket.SHUT_RDWR) - self.listen_socket.close() - - def accept_connections(self): - try: - self.listen_socket.bind(('', self.port)) - self.listen_socket.listen() - while True: - peer_socket, address = self.listen_socket.accept() - host, port = address - peer_address = PeerAddress( - network=Network.IPV4, - host=host, - port=port, - ) - peer_socket.setblocking(True) - self.handle_connection( - peer_socket, - peer_address, - ) - except Exception: - logger.exception("Accept peer connections failed.") - - def handle_connection( - self, - peer_socket: socket.socket, - peer_address: PeerAddress, - ): - """Handle a newly connected peer socket.""" - threading.Thread( - target=self.peer_handler.handle_connection, - args=(peer_socket, peer_address, False), - ).start() diff --git a/squeaknode/node/active_download_manager.py b/squeaknode/node/active_download_manager.py deleted file mode 100644 index 35d0ee4c..00000000 --- a/squeaknode/node/active_download_manager.py +++ /dev/null @@ -1,207 +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 -import threading -import time -import uuid -from abc import ABC -from abc import abstractmethod -from concurrent.futures import ThreadPoolExecutor -from concurrent.futures import TimeoutError -from typing import Dict -from typing import Optional - -from squeak.messages import msg_getdata -from squeak.messages import msg_getsqueaks -from squeak.messages import MsgSerializable -from squeak.net import CInterested -from squeak.net import CInv -from squeak.net import CSqueakLocator - -from squeaknode.core.download_result import DownloadResult -from squeaknode.node.downloaded_object import DownloadedObject - -logger = logging.getLogger(__name__) - - -DOWNLOAD_TIMEOUT_S = 10 - - -class ActiveDownload(ABC): - - def __init__(self, limit: int): - self.limit = limit - self.count = 0 - self._lock = threading.Lock() - self.stopped = threading.Event() - self.num_peers = 0 - self.start_time_ms: Optional[int] = None - - @abstractmethod - def is_interested(self, downloaded_object: DownloadedObject) -> bool: - """Return True if the given squeak matches the download interest.""" - - @abstractmethod - def get_download_msg(self) -> MsgSerializable: - """Get the message to send to peers to get download response.""" - - def initiate_download(self, broadcast_fn) -> None: - self.start_time_ms = int(time.time() * 1000) - msg = self.get_download_msg() - self.num_peers = broadcast_fn(msg) - if self.num_peers == 0: - self.mark_complete() - - def increment(self) -> None: - with self._lock: - self.count += 1 - if self.count >= self.limit: - self.mark_complete() - - def mark_complete(self): - self.stopped.set() - - def cancel(self): - self.stopped.set() - - def get_elapsed_time_ms(self): - if self.start_time_ms is None: - return 0 - end_time_ms = int(time.time() * 1000) - return end_time_ms - self.start_time_ms - - def wait_for_complete(self, timeout_s: int) -> None: - self.stopped.wait(timeout=timeout_s) - - def get_result(self) -> DownloadResult: - return DownloadResult( - number_downloaded=self.count, - number_requested=self.limit, - elapsed_time_ms=self.get_elapsed_time_ms(), - number_peers=self.num_peers, - ) - - -class InterestDownload(ActiveDownload): - - def __init__(self, limit: int, interest: CInterested): - self.interest = interest - super().__init__(limit) - - def is_interested(self, downloaded_object: DownloadedObject) -> bool: - return downloaded_object.matches_requested_squeak_range(self.interest) - - def get_download_msg(self) -> MsgSerializable: - locator = CSqueakLocator( - vInterested=[self.interest], - ) - return msg_getsqueaks( - locator=locator, - ) - - -class HashDownload(ActiveDownload): - - def __init__(self, squeak_hash: bytes): - self.squeak_hash = squeak_hash - super().__init__(1) - - def is_interested(self, downloaded_object: DownloadedObject) -> bool: - return downloaded_object.matches_requested_squeak_hash(self.squeak_hash) - - def get_download_msg(self) -> MsgSerializable: - invs = [ - CInv(type=1, hash=self.squeak_hash) - ] - return msg_getdata( - inv=invs, - ) - - -class OffersDownload(ActiveDownload): - - def __init__(self, limit: int, squeak_hash: bytes): - self.squeak_hash = squeak_hash - super().__init__(limit) - - def is_interested(self, downloaded_object: DownloadedObject) -> bool: - return downloaded_object.matches_requested_offer_hash(self.squeak_hash) - - def get_download_msg(self) -> MsgSerializable: - invs = [ - CInv(type=2, hash=self.squeak_hash) - ] - return msg_getdata(inv=invs) - - -class ActiveDownloadManager: - - def __init__(self, network_manager): - self.network_manager = network_manager - self.downloads: Dict[str, ActiveDownload] = dict() - self.executor = None - - def start(self): - logger.info("Starting Download Manager...") - self.executor = ThreadPoolExecutor(max_workers=10) - - def stop(self): - for download in self.downloads.values(): - download.cancel() - logger.info("Stopping Download Manager...") - self.executor.shutdown(wait=True) - logger.info("Stopped Download Manager.") - - def lookup_counter(self, downloaded_object: DownloadedObject) -> Optional[ActiveDownload]: - for name, interest in self.downloads.items(): - if interest.is_interested(downloaded_object): - return interest - return None - - def run_download(self, download: ActiveDownload) -> DownloadResult: - name_key = "download_key_{}".format(uuid.uuid1()) - self.downloads[name_key] = download - future = self.executor.submit(self.download_task, download) - try: - return future.result() - except TimeoutError: - return download.get_result() - finally: - del self.downloads[name_key] - - def download_task(self, download: ActiveDownload) -> DownloadResult: - broadcast_fn = self.network_manager.broadcast_msg - download.initiate_download(broadcast_fn) - download.wait_for_complete(DOWNLOAD_TIMEOUT_S) - return download.get_result() - - def download_interest(self, limit: int, interest: CInterested) -> DownloadResult: - download = InterestDownload(limit, interest) - return self.run_download(download) - - def download_hash(self, squeak_hash: bytes) -> DownloadResult: - download = HashDownload(squeak_hash) - return self.run_download(download) - - def download_offers(self, limit: int, squeak_hash: bytes) -> DownloadResult: - download = OffersDownload(limit, squeak_hash) - return self.run_download(download) diff --git a/squeaknode/node/downloaded_object.py b/squeaknode/node/downloaded_object.py deleted file mode 100644 index fa0ebc6e..00000000 --- a/squeaknode/node/downloaded_object.py +++ /dev/null @@ -1,72 +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. -from squeak.core import CSqueak -from squeak.net import CInterested - -from squeaknode.core.interests import squeak_matches_interest -from squeaknode.core.offer import Offer -from squeaknode.core.squeaks import get_hash - - -class DownloadedObject: - - def matches_requested_squeak_range(self, interest: CInterested) -> bool: - """Return True if the object matches the requested interest. - """ - - def matches_requested_squeak_hash(self, squeak_hash: bytes) -> bool: - """Return True if the object matches the requested offer hash. - """ - - def matches_requested_offer_hash(self, squeak_hash: bytes) -> bool: - """Return True if the object matches the requested squeak hash. - """ - - -class DownloadedSqueak(DownloadedObject): - - def __init__(self, squeak: CSqueak): - self.squeak = squeak - - def matches_requested_squeak_range(self, interest: CInterested) -> bool: - return squeak_matches_interest(self.squeak, interest) - - def matches_requested_squeak_hash(self, squeak_hash: bytes) -> bool: - return squeak_hash == get_hash(self.squeak) - - def matches_requested_offer_hash(self, squeak_hash: bytes) -> bool: - return False - - -class DownloadedOffer(DownloadedObject): - - def __init__(self, offer: Offer): - self.offer = offer - - def matches_requested_squeak_range(self, interest: CInterested) -> bool: - return False - - def matches_requested_squeak_hash(self, squeak_hash: bytes) -> bool: - return False - - def matches_requested_offer_hash(self, squeak_hash: bytes) -> bool: - return self.offer.squeak_hash == squeak_hash diff --git a/squeaknode/node/network_handler.py b/squeaknode/node/network_handler.py deleted file mode 100644 index f3bfbf69..00000000 --- a/squeaknode/node/network_handler.py +++ /dev/null @@ -1,347 +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 typing import Optional - -from squeak.core import CSqueak -from squeak.messages import msg_getdata -from squeak.messages import MSG_SECRET_KEY -from squeak.messages import MSG_SQUEAK -from squeak.messages import MsgSerializable -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__) - - -EMPTY_HASH = b'\x00' * 32 - - -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_squeak(self, squeak_hash: bytes) -> Optional[CSqueak]: - return self.squeak_store.get_squeak(squeak_hash) - - def get_squeak_secret_key(self, squeak_hash: bytes) -> Optional[bytes]: - return self.squeak_store.get_squeak_secret_key(squeak_hash) - - def get_unknown_invs(self, invs): - unknown_squeak_invs = self.get_unknown_squeaks(invs) - unknown_secret_key_invs = self.get_unknown_secret_keys(invs) - return unknown_squeak_invs + unknown_secret_key_invs - - def get_unknown_squeaks(self, invs): - return [ - inv for inv in invs - if inv.type == MSG_SQUEAK - and self.get_squeak(inv.hash) is None - ] - - def get_unknown_secret_keys(self, invs): - return [ - inv for inv in invs - if inv.type == MSG_SECRET_KEY - and self.get_squeak(inv.hash) is not None - and self.get_squeak_secret_key(inv.hash) is None - ] - - def save_squeak(self, squeak: CSqueak) -> Optional[bytes]: - return self.save_active_download_squeak(squeak) or \ - self.save_followed_squeak(squeak) - - def save_active_download_squeak(self, squeak: CSqueak) -> Optional[bytes]: - """Save the given squeak as an active download. - - Returns: - bytes: the hash of the saved squeak. - """ - counter = self.get_download_squeak_counter(squeak) - if counter is None: - return None - saved_squeak_hash = self.squeak_store.save_squeak(squeak) - if saved_squeak_hash is None: - return None - counter.increment() - return saved_squeak_hash - - def save_followed_squeak(self, squeak: CSqueak) -> Optional[bytes]: - """Save the given squeak because it matches the followed - interest criteria. - - Returns: - bytes: the hash of the saved squeak. - """ - 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) - - def squeak_matches_interest(self, squeak: CSqueak) -> bool: - locator = self.get_interested_locator() - for interest in locator.vInterested: - if squeak_matches_interest(squeak, interest): - return True - return False - - def unlock_squeak(self, squeak_hash: bytes, secret_key: bytes): - self.squeak_store.save_secret_key( - squeak_hash, - secret_key, - ) - - def get_secret_key_reply(self, squeak_hash: bytes, peer_address: PeerAddress) -> Optional[SecretKeyReply]: - squeak = self.get_squeak(squeak_hash) - price_msat = self.get_price_for_squeak(squeak, peer_address) - lnd_external_address: Optional[LightningAddressHostPort] = None - if self.config.lnd.external_host: - lnd_external_address = LightningAddressHostPort( - host=self.config.lnd.external_host, - port=self.config.lnd.port, - ) - 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.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( - squeak_hash.hex(), - )) - invs = [ - CInv(type=2, hash=squeak_hash) - ] - getdata_msg = msg_getdata(inv=invs) - self.broadcast_msg(getdata_msg) - - def handle_received_offer(self, offer: Offer, peer_address: PeerAddress) -> Optional[int]: - received_offer_id = self.save_received_offer( - offer, - peer_address, - ) - if received_offer_id is None: - return None - counter = self.get_download_offer_counter(offer) - if counter is not None: - counter.increment() - return received_offer_id - - def get_download_offer_counter(self, offer: Offer) -> Optional[ActiveDownload]: - downloaded_offer = DownloadedOffer(offer) - return self.active_download_manager.lookup_counter(downloaded_offer) - - def get_download_squeak_counter(self, squeak: CSqueak) -> Optional[ActiveDownload]: - downloaded_squeak = DownloadedSqueak(squeak) - return self.active_download_manager.lookup_counter(downloaded_squeak) - - def get_price_for_squeak(self, squeak: CSqueak, peer_address: PeerAddress) -> int: - price_policy = PricePolicy( - self.squeak_store, - self.config, - self.node_settings, - ) - return price_policy.get_price(squeak, peer_address) - - 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_connection_worker.py b/squeaknode/node/peer_connection_worker.py deleted file mode 100644 index 007b4c29..00000000 --- a/squeaknode/node/peer_connection_worker.py +++ /dev/null @@ -1,54 +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.network.network_manager import NetworkManager -from squeaknode.node.periodic_worker import PeriodicWorker -from squeaknode.node.squeak_store import SqueakStore - - -logger = logging.getLogger(__name__) - - -class PeerConnectionWorker(PeriodicWorker): - def __init__( - self, - squeak_store: SqueakStore, - network_manager: NetworkManager, - connect_interval_s: int, - ): - self.squeak_store = squeak_store - self.network_manager = network_manager - self.connect_interval_s = connect_interval_s - - def work_fn(self): - peers = self.squeak_store.get_autoconnect_peers() - for peer in peers: - self.network_manager.connect_peer_async( - peer.address, - ) - - def get_interval_s(self): - return self.connect_interval_s - - def get_name(self): - return "peer_connection_worker" diff --git a/squeaknode/node/peer_subscription_update_worker.py b/squeaknode/node/peer_subscription_update_worker.py deleted file mode 100644 index ee76ab1d..00000000 --- a/squeaknode/node/peer_subscription_update_worker.py +++ /dev/null @@ -1,54 +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 -import threading - -from squeaknode.bitcoin.bitcoin_block_subscription_client import BitcoinBlockSubscriptionClient -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.network_handler import NetworkHandler - - -logger = logging.getLogger(__name__) - - -class PeerSubscriptionUpdateWorker: - def __init__( - self, - network_manager: NetworkManager, - network_handler: NetworkHandler, - block_subscription_client: BitcoinBlockSubscriptionClient, - ): - self.network_manager = network_manager - self.network_handler = network_handler - self.block_subscription_client = block_subscription_client - - def start_running(self): - threading.Thread( - target=self.subscribe_blocks, - daemon=True, - ).start() - - 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.network_handler.get_interested_locator() - self.network_manager.update_local_subscriptions(locator) diff --git a/squeaknode/node/price_policy.py b/squeaknode/node/price_policy.py index b3e9f867..bcbc6824 100644 --- a/squeaknode/node/price_policy.py +++ b/squeaknode/node/price_policy.py @@ -22,8 +22,6 @@ import logging from typing import Optional -from squeak.core import CSqueak - from squeaknode.config.config import SqueaknodeConfig from squeaknode.core.peer_address import PeerAddress from squeaknode.core.squeak_peer import SqueakPeer @@ -42,14 +40,10 @@ class PricePolicy: self.config = config self.node_settings = node_settings - def get_price(self, squeak: CSqueak, peer_address: PeerAddress) -> int: + def get_price(self) -> int: """Get the price to sell this squeak to this peer. """ - # Return zero for price if peer is configured to be share for free. - peer = self.get_peer(peer_address) - if peer is not None and peer.share_for_free: - return 0 sell_price_msat = self.get_sell_price_msat() if sell_price_msat is None: return self.get_default_price() diff --git a/squeaknode/node/secret_key_reply.py b/squeaknode/node/secret_key_reply.py deleted file mode 100644 index 6a64d230..00000000 --- a/squeaknode/node/secret_key_reply.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. -from squeak.messages import msg_secretkey -from squeak.messages import MsgSerializable -from squeak.net import COffer - -from squeaknode.core.offer import Offer - - -class SecretKeyReply: - - def get_msg(self) -> MsgSerializable: - """Return the message to reply to the other peer. - """ - - -class FreeSecretKeyReply(SecretKeyReply): - - def __init__(self, squeak_hash: bytes, secret_key: bytes): - self.squeak_hash = squeak_hash - self.secret_key = secret_key - - def get_msg(self) -> MsgSerializable: - return msg_secretkey( - hashSqk=self.squeak_hash, - secretKey=self.secret_key, - ) - - -class OfferReply(SecretKeyReply): - - def __init__(self, squeak_hash: bytes, offer: Offer): - self.squeak_hash = squeak_hash - self.offer = offer - - def get_msg(self) -> MsgSerializable: - offer = COffer( - nonce=self.offer.nonce, - strPaymentInfo=self.offer.payment_request.encode( - 'utf-8'), - host=self.offer.host.encode('utf-8'), - port=self.offer.port, - ) - return msg_secretkey( - hashSqk=self.squeak_hash, - offer=offer, - ) diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index 64b11fce..5160263c 100644 --- a/squeaknode/node/squeak_controller.py +++ b/squeaknode/node/squeak_controller.py @@ -21,20 +21,16 @@ # SOFTWARE. import logging import threading -from typing import Iterable from typing import List from typing import Optional from squeak.core import CSqueak from squeak.core.keys import SqueakPrivateKey from squeak.core.keys import SqueakPublicKey -from squeak.messages import msg_getdata -from squeak.messages import MsgSerializable -from squeak.net import CInterested -from squeak.net import CInv -from squeaknode.core.connected_peer import ConnectedPeer +from squeaknode.client.network_controller import NetworkController from squeaknode.core.download_result import DownloadResult +from squeaknode.core.peer_address import Network from squeaknode.core.peer_address import PeerAddress from squeaknode.core.received_offer import ReceivedOffer from squeaknode.core.received_payment import ReceivedPayment @@ -64,8 +60,6 @@ class SqueakController: squeak_store: SqueakStore, squeak_core: SqueakCore, payment_processor, - network_manager, - download_manager, tweet_forwarder, node_settings, config, @@ -74,8 +68,6 @@ class SqueakController: self.squeak_store = squeak_store self.squeak_core = squeak_core self.payment_processor = payment_processor - self.network_manager = network_manager - self.active_download_manager = download_manager self.tweet_forwarder = tweet_forwarder self.node_settings = node_settings self.config = config @@ -279,11 +271,21 @@ class SqueakController: def get_squeak_entry(self, squeak_hash: bytes) -> Optional[SqueakEntry]: return self.squeak_store.get_squeak_entry(squeak_hash) + def download_single_squeak(self, squeak_hash: bytes) -> DownloadResult: + network_controller = NetworkController(self.squeak_store, self.config) + network_controller.download_single_squeak(squeak_hash) + return DownloadResult(1, 1, 0, 9999) + def get_timeline_squeak_entries( self, limit: int, last_entry: Optional[SqueakEntry], ) -> List[SqueakEntry]: + # TODO: remove this temporary hack, after converting this to websockets. + logger.info('Start downloading timeline...') + network_controller = NetworkController(self.squeak_store, self.config) + network_controller.download_timeline() + logger.info('Finished downloading timeline.') return self.squeak_store.get_timeline_squeak_entries(limit, last_entry) def get_liked_squeak_entries( @@ -299,6 +301,11 @@ class SqueakController: limit: int, last_entry: Optional[SqueakEntry], ) -> List[SqueakEntry]: + # TODO: remove this temporary hack, after converting this to websockets. + logger.info('Start downloading pubkey squeaks...') + network_controller = NetworkController(self.squeak_store, self.config) + network_controller.download_pubkey_squeaks(public_key) + logger.info('Finished downloading pubkey squeaks.') return self.squeak_store.get_squeak_entries_for_public_key( public_key, limit, @@ -351,125 +358,6 @@ class SqueakController: def unlike_squeak(self, squeak_hash: bytes): return self.squeak_store.unlike_squeak(squeak_hash) - def connect_peer(self, peer_address: PeerAddress) -> None: - logger.info("Connect to peer: {}".format( - peer_address, - )) - self.network_manager.connect_peer_sync(peer_address) - - def get_connected_peer(self, peer_address: PeerAddress) -> Optional[ConnectedPeer]: - peer = self.network_manager.get_connected_peer(peer_address) - if peer is None: - return None - return ConnectedPeer( - peer=peer, - saved_peer=self.squeak_store.get_peer_by_address( - peer_address, - ), - ) - - def get_connected_peers(self) -> List[ConnectedPeer]: - peers = self.network_manager.get_connected_peers() - return [ - ConnectedPeer( - peer=peer, - saved_peer=self.squeak_store.get_peer_by_address( - peer.remote_address, - ), - ) for peer in peers - ] - - def download_squeaks( - self, - public_keys: List[SqueakPublicKey], - min_block: int, - max_block: int, - replyto_hash: Optional[bytes], - ) -> DownloadResult: - interest = CInterested( - pubkeys=public_keys, - nMinBlockHeight=min_block, - nMaxBlockHeight=max_block, - replyto_squeak_hash=replyto_hash, - ) if replyto_hash else CInterested( - pubkeys=public_keys, - nMinBlockHeight=min_block, - nMaxBlockHeight=max_block, - ) - return self.active_download_manager.download_interest(10, interest) - - def download_single_squeak(self, squeak_hash: bytes) -> DownloadResult: - logger.info("Downloading single squeak: {}".format( - squeak_hash.hex(), - )) - return self.active_download_manager.download_hash(squeak_hash) - - def download_offers(self, squeak_hash: bytes) -> DownloadResult: - logger.info("Downloading offers for squeak: {}".format( - squeak_hash.hex(), - )) - return self.active_download_manager.download_offers(10, squeak_hash) - - def request_offers(self, squeak_hash: bytes): - logger.info("Requesting offers for squeak: {}".format( - squeak_hash.hex(), - )) - invs = [ - CInv(type=2, hash=squeak_hash) - ] - getdata_msg = msg_getdata(inv=invs) - self.broadcast_msg(getdata_msg) - - def download_replies(self, squeak_hash: bytes) -> DownloadResult: - logger.info("Downloading replies for squeak: {}".format( - squeak_hash.hex(), - )) - interest = CInterested( - hashReplySqk=squeak_hash, - ) - return self.active_download_manager.download_interest(10, interest) - - def download_public_key_squeaks(self, public_key: SqueakPublicKey) -> DownloadResult: - logger.info("Downloading squeaks for public key: {}".format( - public_key, - )) - interest = CInterested( - pubkeys=[public_key], - ) - return self.active_download_manager.download_interest(10, interest) - - def broadcast_msg(self, msg: MsgSerializable) -> int: - return self.network_manager.broadcast_msg(msg) - - def disconnect_peer(self, peer_address: PeerAddress) -> None: - logger.info("Disconnect to peer: {}".format( - peer_address, - )) - self.network_manager.disconnect_peer(peer_address) - - def subscribe_connected_peers(self, stopped: threading.Event) -> Iterable[List[ConnectedPeer]]: - for peers in self.network_manager.subscribe_connected_peers(stopped): - yield [ - ConnectedPeer( - peer=peer, - saved_peer=self.squeak_store.get_peer_by_address( - peer.remote_address, - ) - ) for peer in peers - ] - - def subscribe_connected_peer(self, peer_address: PeerAddress, stopped: threading.Event) -> Iterable[Optional[ConnectedPeer]]: - for peer in self.network_manager.subscribe_connected_peer(peer_address, stopped): - if peer is None: - yield None - else: - yield ConnectedPeer( - peer=peer, - saved_peer=self.squeak_store.get_peer_by_address( - peer.remote_address, - ) - ) - def subscribe_new_squeaks(self, stopped: threading.Event): yield from self.squeak_store.subscribe_new_squeaks(stopped) @@ -520,7 +408,11 @@ class SqueakController: yield self.get_squeak_entry(squeak_hash) def get_external_address(self) -> PeerAddress: - return self.network_manager.external_address + return PeerAddress( + network=Network.IPV4, + host=self.config.server.external_address or '', + port=self.config.server.port or 0, + ) def get_default_peer_port(self) -> int: return self.default_port diff --git a/squeaknode/node/squeak_node.py b/squeaknode/node/squeak_node.py index 657701b5..5a16a5ce 100644 --- a/squeaknode/node/squeak_node.py +++ b/squeaknode/node/squeak_node.py @@ -27,7 +27,6 @@ from squeak.params import SelectParams from squeaknode.admin.squeak_admin_server_handler import SqueakAdminServerHandler from squeaknode.admin.squeak_admin_server_servicer import SqueakAdminServerServicer from squeaknode.admin.webapp.app import SqueakAdminWebServer -from squeaknode.bitcoin.bitcoin_block_subscription_client import BitcoinBlockSubscriptionClient from squeaknode.bitcoin.bitcoin_core_client import BitcoinCoreClient from squeaknode.config.config import SqueaknodeConfig from squeaknode.core.squeak_core import SqueakCore @@ -35,22 +34,16 @@ from squeaknode.db.db_engine import get_connection_string from squeaknode.db.db_engine import get_engine from squeaknode.db.squeak_db import SqueakDb from squeaknode.lightning.lnd_lightning_client import LNDLightningClient -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.active_download_manager import ActiveDownloadManager -from squeaknode.node.network_handler import NetworkHandler from squeaknode.node.node_settings import NodeSettings from squeaknode.node.payment_processor import PaymentProcessor -from squeaknode.node.peer_connection_worker import PeerConnectionWorker -from squeaknode.node.peer_subscription_update_worker import PeerSubscriptionUpdateWorker from squeaknode.node.process_forward_tweets_worker import ProcessForwardTweetsWorker from squeaknode.node.process_received_payments_worker import ProcessReceivedPaymentsWorker 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_store import SqueakStore -from squeaknode.node.update_follows_worker import UpdateFollowsWorker -from squeaknode.node.update_subscribed_secret_key_worker import UpdateSubscribedSecretKeysWorker -from squeaknode.node.update_subscribed_squeak_worker import UpdateSubscribedSqueaksWorker +from squeaknode.server.app import SqueakPeerWebServer +from squeaknode.server.squeak_peer_server_handler import SqueakPeerServerHandler from squeaknode.twitter.twitter_forwarder import TwitterForwarder logger = logging.getLogger(__name__) @@ -65,56 +58,43 @@ class SqueakNode: self.create_node_settings() self.create_lightning_client() self.create_bitcoin_client() - self.create_bitcoin_block_subscription_client() self.create_squeak_core() self.create_squeak_store() self.create_payment_processor() self.create_twitter_forwarder() - self.create_network_manager() - self.create_download_manager() self.create_squeak_controller() - self.create_network_handler() + + self.create_peer_handler() + self.create_peer_web_server() + self.create_admin_handler() self.create_admin_rpc_server() self.create_admin_web_server() self.create_received_payment_processor_worker() - self.create_forward_tweets_processor_worker() - self.create_peer_connection_worker() self.create_squeak_deletion_worker() self.create_offer_expiry_worker() - self.create_new_squeak_worker() - self.create_new_secret_key_worker() - self.create_new_follow_worker() - self.create_peer_subscription_update_worker() + self.create_forward_tweets_processor_worker() def start_running(self): self.squeak_db.init_with_retries() self.lightning_client.init() - self.network_manager.start(self.network_handler) if self.config.rpc.enabled: self.admin_rpc_server.start() if self.config.webadmin.enabled: self.admin_web_server.start() + self.peer_web_server.start() self.received_payment_processor_worker.start_running() - self.forward_tweets_processor_worker.start_running() - self.peer_connection_worker.start() self.squeak_deletion_worker.start() self.offer_expiry_worker.start() - self.new_squeak_worker.start_running() - self.new_secret_key_worker.start_running() - self.new_follow_worker.start_running() - self.new_bitcoin_block_worker.start_running() - self.download_manager.start() + self.forward_tweets_processor_worker.start_running() def stop_running(self): - self.download_manager.stop() self.admin_web_server.stop() self.admin_rpc_server.stop() - self.network_manager.stop() + self.peer_web_server.stop() self.received_payment_processor_worker.stop_running() self.forward_tweets_processor_worker.stop_running() - self.new_squeak_worker.stop_running() def set_network_params(self): SelectParams(self.config.node.network) @@ -150,12 +130,6 @@ class SqueakNode: self.config.bitcoin.rpc_ssl_cert, ) - def create_bitcoin_block_subscription_client(self): - self.bitcoin_block_subscription_client = BitcoinBlockSubscriptionClient( - self.config.bitcoin.rpc_host, - self.config.bitcoin.zeromq_hashblock_port, - ) - def create_squeak_core(self): self.squeak_core = SqueakCore( self.bitcoin_client, @@ -187,41 +161,30 @@ class SqueakNode: self.config.twitter.forward_tweets_retry_s, ) - def create_network_manager(self): - self.network_manager = NetworkManager( - self.config, - squeak.params.params.DEFAULT_PORT, - ) - def create_squeak_controller(self): self.squeak_controller = SqueakController( self.squeak_store, self.squeak_core, self.payment_processor, - self.network_manager, - self.download_manager, self.twitter_forwarder, self.node_settings, self.config, squeak.params.params.DEFAULT_PORT, ) - def create_network_handler(self): - self.network_handler = NetworkHandler( - self.squeak_store, - self.squeak_core, - self.network_manager, - self.download_manager, - self.node_settings, - self.config, - ) - def create_admin_handler(self): self.admin_handler = SqueakAdminServerHandler( self.lightning_client, self.squeak_controller, ) + def create_peer_handler(self): + self.peer_handler = SqueakPeerServerHandler( + self.squeak_store, + self.node_settings, + self.config, + ) + def create_admin_rpc_server(self): self.admin_rpc_server = SqueakAdminServerServicer( self.config.rpc.host, @@ -241,23 +204,18 @@ class SqueakNode: self.admin_handler, ) + def create_peer_web_server(self): + self.peer_web_server = SqueakPeerWebServer( + self.config.server.host, + squeak.params.params.DEFAULT_PORT, + self.peer_handler, + ) + def create_received_payment_processor_worker(self): self.received_payment_processor_worker = ProcessReceivedPaymentsWorker( self.payment_processor, ) - def create_forward_tweets_processor_worker(self): - self.forward_tweets_processor_worker = ProcessForwardTweetsWorker( - self.twitter_forwarder, - ) - - def create_peer_connection_worker(self): - self.peer_connection_worker = PeerConnectionWorker( - self.squeak_store, - self.network_manager, - self.config.node.peer_autoconnect_interval_s, - ) - def create_squeak_deletion_worker(self): self.squeak_deletion_worker = SqueakDeletionWorker( self.squeak_store, @@ -270,33 +228,7 @@ class SqueakNode: self.config.node.offer_deletion_interval_s, ) - def create_new_squeak_worker(self): - self.new_squeak_worker = UpdateSubscribedSqueaksWorker( - self.squeak_store, - self.network_manager, - ) - - def create_new_secret_key_worker(self): - self.new_secret_key_worker = UpdateSubscribedSecretKeysWorker( - self.squeak_store, - self.network_manager, - ) - - def create_new_follow_worker(self): - self.new_follow_worker = UpdateFollowsWorker( - self.squeak_store, - self.network_manager, - self.network_handler, - ) - - def create_peer_subscription_update_worker(self): - self.new_bitcoin_block_worker = PeerSubscriptionUpdateWorker( - self.network_manager, - self.network_handler, - self.bitcoin_block_subscription_client, - ) - - def create_download_manager(self): - self.download_manager = ActiveDownloadManager( - self.network_manager, + def create_forward_tweets_processor_worker(self): + self.forward_tweets_processor_worker = ProcessForwardTweetsWorker( + self.twitter_forwarder, ) diff --git a/squeaknode/node/squeak_store.py b/squeaknode/node/squeak_store.py index 3b103fbb..c4f2bc40 100644 --- a/squeaknode/node/squeak_store.py +++ b/squeaknode/node/squeak_store.py @@ -31,6 +31,8 @@ from squeak.core import CSqueak from squeak.core.keys import SqueakPrivateKey from squeak.core.keys import SqueakPublicKey +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 @@ -189,6 +191,55 @@ class SqueakStore: def save_sent_offer(self, sent_offer: SentOffer) -> int: return self.squeak_db.insert_sent_offer(sent_offer) + 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.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.save_sent_offer(sent_offer) + return sent_offer + + def get_packaged_offer( + self, + squeak_hash: bytes, + peer_address: PeerAddress, + price_msat: int, + lnd_external_address: Optional[LightningAddressHostPort], + ) -> Optional[Offer]: + sent_offer = self.get_sent_offer_for_peer( + squeak_hash, + peer_address, + price_msat, + ) + if sent_offer is None: + return None + return self.squeak_core.package_offer( + sent_offer, + lnd_external_address, + ) + def create_signing_profile(self, profile_name: str) -> int: squeak_profile = create_signing_profile( profile_name, @@ -427,6 +478,14 @@ class SqueakStore: self.new_received_offer_listener.handle_new_item(received_offer) return received_offer_id + def handle_offer(self, squeak: CSqueak, offer: Offer, peer_address: PeerAddress): + received_offer = self.squeak_core.unpack_offer( + squeak, + offer, + peer_address, + ) + self.save_received_offer(received_offer) + def get_followed_public_keys(self) -> List[SqueakPublicKey]: followed_profiles = self.squeak_db.get_following_profiles() return [profile.public_key for profile in followed_profiles] diff --git a/squeaknode/node/update_follows_worker.py b/squeaknode/node/update_follows_worker.py deleted file mode 100644 index 35142f1c..00000000 --- a/squeaknode/node/update_follows_worker.py +++ /dev/null @@ -1,63 +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 -import threading - -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.network_handler import NetworkHandler -from squeaknode.node.squeak_store import SqueakStore - - -logger = logging.getLogger(__name__) - - -class UpdateFollowsWorker: - - def __init__( - 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): - threading.Thread( - target=self.handle_new_follow, - name="new_follows_thread", - daemon=True, - ).start() - - def stop_running(self): - self.stopped.set() - - def handle_new_follow(self): - logger.debug("Starting UpdateFollowsWorker...") - for _ in self.squeak_store.subscribe_follows( - self.stopped, - ): - logger.debug("Handling update subscriptions event") - locator = self.network_handler.get_interested_locator() - self.network_manager.update_local_subscriptions(locator) diff --git a/squeaknode/node/update_subscribed_secret_key_worker.py b/squeaknode/node/update_subscribed_secret_key_worker.py deleted file mode 100644 index 1b79862f..00000000 --- a/squeaknode/node/update_subscribed_secret_key_worker.py +++ /dev/null @@ -1,77 +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 -import threading - -from squeak.messages import msg_inv -from squeak.messages import MSG_SECRET_KEY -from squeak.net import CInv - -from squeaknode.core.squeaks import get_hash -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.squeak_store import SqueakStore - - -logger = logging.getLogger(__name__) - - -class UpdateSubscribedSecretKeysWorker: - - def __init__(self, squeak_store: SqueakStore, network_manager: NetworkManager): - self.squeak_store = squeak_store - self.network_manager = network_manager - self.stopped = threading.Event() - - def start_running(self): - threading.Thread( - target=self.handle_new_secret_keys, - name="new_secret_keys_worker_thread", - daemon=True, - ).start() - - def stop_running(self): - self.stopped.set() - - def handle_new_secret_keys(self): - logger.debug("Starting UpdateSubscribedSecretKeysWorker...") - for squeak in self.squeak_store.subscribe_new_secret_keys( - self.stopped, - ): - logger.debug("Handling new secret key for squeak hash: {!r}".format( - get_hash(squeak).hex(), - )) - self.forward_secret_key(squeak) - - def forward_secret_key(self, squeak): - logger.debug("Forward new secret key for hash: {!r}".format( - get_hash(squeak).hex(), - )) - for peer in self.network_manager.get_connected_peers(): - if peer.is_remote_subscribed(squeak): - logger.debug("Forwarding to peer: {}".format( - peer, - )) - squeak_hash = get_hash(squeak) - inv = CInv(type=MSG_SECRET_KEY, hash=squeak_hash) - inv_msg = msg_inv(inv=[inv]) - peer.send_msg(inv_msg) - logger.debug("Finished checking peers to forward.") diff --git a/squeaknode/node/update_subscribed_squeak_worker.py b/squeaknode/node/update_subscribed_squeak_worker.py deleted file mode 100644 index 81c1619c..00000000 --- a/squeaknode/node/update_subscribed_squeak_worker.py +++ /dev/null @@ -1,77 +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 -import threading - -from squeak.messages import msg_inv -from squeak.messages import MSG_SQUEAK -from squeak.net import CInv - -from squeaknode.core.squeaks import get_hash -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.squeak_store import SqueakStore - - -logger = logging.getLogger(__name__) - - -class UpdateSubscribedSqueaksWorker: - - def __init__(self, squeak_store: SqueakStore, network_manager: NetworkManager): - self.squeak_store = squeak_store - self.network_manager = network_manager - self.stopped = threading.Event() - - def start_running(self): - threading.Thread( - target=self.handle_new_squeaks, - name="new_squeaks_worker_thread", - daemon=True, - ).start() - - def stop_running(self): - self.stopped.set() - - def handle_new_squeaks(self): - logger.debug("Starting UpdateSubscribedSqueaksWorker...") - for squeak in self.squeak_store.subscribe_new_squeaks( - self.stopped, - ): - logger.debug("Handling new squeak: {!r}".format( - get_hash(squeak).hex(), - )) - self.forward_squeak(squeak) - - def forward_squeak(self, squeak): - logger.debug("Forward new squeak: {!r}".format( - get_hash(squeak).hex(), - )) - for peer in self.network_manager.get_connected_peers(): - if peer.is_remote_subscribed(squeak): - logger.debug("Forwarding to peer: {}".format( - peer, - )) - squeak_hash = get_hash(squeak) - inv = CInv(type=MSG_SQUEAK, hash=squeak_hash) - inv_msg = msg_inv(inv=[inv]) - peer.send_msg(inv_msg) - logger.debug("Finished checking peers to forward.") diff --git a/squeaknode/network/util.py b/squeaknode/server/__init__.py similarity index 95% rename from squeaknode/network/util.py rename to squeaknode/server/__init__.py index 575353a3..f0ab91ef 100644 --- a/squeaknode/network/util.py +++ b/squeaknode/server/__init__.py @@ -19,8 +19,3 @@ # 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 time - - -def time_now(): - return int(time.time()) diff --git a/squeaknode/server/app.py b/squeaknode/server/app.py new file mode 100644 index 00000000..0263002a --- /dev/null +++ b/squeaknode/server/app.py @@ -0,0 +1,199 @@ +# 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 os +import threading + +from flask import Flask +from flask import jsonify +from flask import request +from flask_sock import Sock +from werkzeug.serving import make_server + +from squeaknode.server.squeak_peer_server_handler import NotFoundError +from squeaknode.server.squeak_peer_server_handler import PaymentRequiredError + +logger = logging.getLogger(__name__) + + +def create_app(handler): + # create and configure the app + logger.debug("Starting flask app from directory: {}".format(os.getcwd())) + app = Flask( + __name__, + static_url_path="/", + static_folder="static/build", + ) + app.config.from_mapping( + SECRET_KEY="dev", + ) + logger.debug("Starting flask with app.root_path: {}".format(app.root_path)) + logger.debug("Files in root path: {}".format(os.listdir(app.root_path))) + sock = Sock(app) + + @app.route("/") + def index(): + logger.info("Getting index route.") + return "Hello, Index!" + + @app.route("/hello") + def hello_world(): + logger.info("Getting hello route.") + return "Hello, World!" + + @app.route('/squeak/') + def squeak(hash): + logger.info("Getting squeak route.") + logger.info(hash) + squeak_bytes = handler.handle_get_squeak_bytes(hash) + logger.info(squeak_bytes) + return squeak_bytes + + @app.route('/secretkey/') + def secret_key(hash): + logger.info("Getting secretkey route.") + logger.info(hash) + try: + secret_key_bytes = handler.handle_get_secret_key(hash) + except NotFoundError: + return "Not found", 404 + except PaymentRequiredError: + return "Payment required", 402 + logger.info(secret_key_bytes) + return secret_key_bytes + + @app.route('/offer/') + def offer(hash): + logger.info("Getting offer route.") + logger.info(hash) + client_host = request.remote_addr + logger.info(client_host) + try: + offer = handler.handle_get_offer(hash, client_host) + except NotFoundError: + return "Not found", 404 + return jsonify({ + 'squeak_hash': offer.squeak_hash.hex(), + 'nonce': offer.nonce.hex(), + 'payment_request': offer.payment_request, + 'host': offer.host, + 'port': offer.port, + }) + + @app.route("/lookup") + def lookup(): + logger.info("Getting lookup route.") + min_block = request.args.get('minblock') + max_block = request.args.get('maxblock') + pubkeys = request.args.getlist('pubkeys') + logger.info("Hello, lookup! Min block {}, Max block {}, pubkeys: {}".format( + min_block, + max_block, + pubkeys, + )) + if len(pubkeys) == 0: + squeak_hashes = [] + else: + squeak_hashes = handler.handle_lookup_squeaks( + pubkeys, + min_block, + max_block, + ) + squeak_hashes_str = [ + squeak_hash.hex() + for squeak_hash in squeak_hashes + ] + logger.info(squeak_hashes_str) + return jsonify(squeak_hashes_str) + + @sock.route('/echo') + def echo(ws): + count = 0 + while True: + data = f'hello_{count}' + logger.info(data) + ws.send(data) + count += 1 + import time + time.sleep(5) + + @sock.route('/subscribetimeline') + def subscribe_timeline(ws): + logger.info("Getting lookup route.") + min_block = request.args.get('minblock') + max_block = request.args.get('maxblock') + pubkeys = request.args.getlist('pubkeys') + logger.info("Hello, lookup! Min block {}, Max block {}, pubkeys: {}".format( + min_block, + max_block, + pubkeys, + )) + # TODO: This special case should not need to be handled here. + if len(pubkeys) == 0: + squeak_hashes = [] + else: + squeak_hashes = handler.handle_lookup_squeaks( + pubkeys, + min_block, + max_block, + ) + for squeak_hash in squeak_hashes: + ws.send(squeak_hash.hex()) + + return app + + +class SqueakPeerWebServer: + def __init__( + self, + host, + port, + handler, + ): + self.host = host + self.port = port + self.app = create_app(handler) + self.server = None + + def get_app(self): + return self.app + + def start(self): + self.server = make_server( + self.host, + self.port, + self.get_app(), + threaded=True, + ssl_context=None, + ) + + logger.info("Starting SqueakPeerWebServer...") + threading.Thread( + target=self.server.serve_forever, + ).start() + + def stop(self): + if self.server is None: + return + logger.info("Stopping SqueakPeerWebServer....") + self.server.shutdown() + logger.info("Stopped SqueakPeerWebServer.") diff --git a/squeaknode/server/squeak_peer_server_handler.py b/squeaknode/server/squeak_peer_server_handler.py new file mode 100644 index 00000000..b59f03bc --- /dev/null +++ b/squeaknode/server/squeak_peer_server_handler.py @@ -0,0 +1,136 @@ +# 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 typing import List +from typing import Optional + +from squeak.core.keys import SqueakPublicKey + +from squeaknode.core.lightning_address import LightningAddressHostPort +from squeaknode.core.offer import Offer +from squeaknode.core.peer_address import Network +from squeaknode.core.peer_address import PeerAddress +from squeaknode.node.price_policy import PricePolicy +from squeaknode.node.squeak_store import SqueakStore + +logger = logging.getLogger(__name__) + + +class PaymentRequiredError(Exception): + pass + + +class NotFoundError(Exception): + pass + + +class SqueakPeerServerHandler(object): + """Handles peer server commands.""" + + def __init__( + self, + squeak_store: SqueakStore, + node_settings, + config, + ): + self.squeak_store = squeak_store + self.node_settings = node_settings + self.config = config + + def handle_get_squeak_bytes(self, squeak_hash_str) -> Optional[bytes]: + squeak_hash = bytes.fromhex(squeak_hash_str) + logger.info("Handle get squeak for hash: {}".format(squeak_hash_str)) + squeak = self.squeak_store.get_squeak(squeak_hash) + if not squeak: + raise NotFoundError() + return squeak.serialize() + + def handle_get_secret_key(self, squeak_hash_str) -> bytes: + squeak_hash = bytes.fromhex(squeak_hash_str) + logger.info( + "Handle get secret key for hash: {}".format(squeak_hash_str)) + price_msat = self.get_price_for_squeak() + if price_msat > 0: + raise PaymentRequiredError() + secret_key = self.squeak_store.get_squeak_secret_key(squeak_hash) + if not secret_key: + raise NotFoundError() + return secret_key + + def handle_get_offer(self, squeak_hash_str, client_host) -> Offer: + squeak_hash = bytes.fromhex(squeak_hash_str) + logger.info("Handle get offer for hash: {}, client_host: {}".format( + squeak_hash_str, client_host)) + client_addr = PeerAddress( + network=Network.IPV4, + host=client_host, + port=0, + ) + logger.info("client_addr: {}".format(client_addr)) + price_msat = self.get_price_for_squeak() + if price_msat == 0: + raise NotFoundError() + # TODO: lnd_external_address should be configured inside SqueakStore. + lnd_external_address: Optional[LightningAddressHostPort] = None + if self.config.lnd.external_host: + lnd_external_address = LightningAddressHostPort( + host=self.config.lnd.external_host, + port=self.config.lnd.port, + ) + logger.info(lnd_external_address) + offer = self.squeak_store.get_packaged_offer( + squeak_hash, + client_addr, + price_msat, + lnd_external_address, + ) + if not offer: + raise NotFoundError() + return offer + + def handle_lookup_squeaks( + self, + pubkey_strs: List[str], + min_block: Optional[int], + max_block: Optional[int], + ) -> List[bytes]: + pubkeys = [ + SqueakPublicKey.from_bytes(bytes.fromhex(pubkey_str)) + for pubkey_str in pubkey_strs + ] + + # Add separate endpoint for replies. + # reply_to_hash = interest.hashReplySqk if interest.hashReplySqk != EMPTY_HASH else None + return self.squeak_store.lookup_squeaks( + pubkeys, + min_block, + max_block, + None, + ) + + def get_price_for_squeak(self) -> int: + price_policy = PricePolicy( + self.squeak_store, + self.config, + self.node_settings, + ) + return price_policy.get_price() diff --git a/tests/admin/conftest.py b/tests/admin/conftest.py index 86f5fa0b..e6d18081 100644 --- a/tests/admin/conftest.py +++ b/tests/admin/conftest.py @@ -218,21 +218,6 @@ def payment_summary_msg( ) -@pytest.fixture -def connected_peer_msg(peer_address_message, peer_msg): - yield squeak_admin_pb2.ConnectedPeer( - peer_address=peer_address_message, - connect_time_s=0, - last_message_received_time_s=0, - number_messages_received=0, - number_bytes_received=0, - number_messages_sent=0, - number_bytes_sent=0, - is_peer_saved=True, - saved_peer=peer_msg, - ) - - @pytest.fixture def download_result_msg( download_result, diff --git a/tests/admin/test_messages.py b/tests/admin/test_messages.py index 7f7417aa..f7130e05 100644 --- a/tests/admin/test_messages.py +++ b/tests/admin/test_messages.py @@ -19,13 +19,11 @@ # 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. -from squeaknode.admin.messages import connected_peer_to_message from squeaknode.admin.messages import download_result_to_message from squeaknode.admin.messages import message_to_peer_address from squeaknode.admin.messages import message_to_received_payment from squeaknode.admin.messages import message_to_sent_payment from squeaknode.admin.messages import message_to_squeak_entry -from squeaknode.admin.messages import optional_connected_peer_to_message from squeaknode.admin.messages import optional_received_offer_to_message from squeaknode.admin.messages import optional_sent_payment_to_message from squeaknode.admin.messages import optional_squeak_entry_to_message @@ -145,12 +143,6 @@ def test_payment_summary_to_message( assert msg == payment_summary_msg -def test_connected_peer_to_message(connected_peer, connected_peer_msg): - msg = connected_peer_to_message(connected_peer) - - assert msg == connected_peer_msg - - def test_optional_profile_to_message_none(): msg = optional_squeak_profile_to_message(None) @@ -221,15 +213,3 @@ def test_optional_sent_payment_to_message(sent_payment, sent_payment_msg): msg = optional_sent_payment_to_message(sent_payment) assert msg == sent_payment_msg - - -def test_optional_connected_peer_to_message_none(): - msg = optional_connected_peer_to_message(None) - - assert msg is None - - -def test_optional_connected_peer_to_message(connected_peer, connected_peer_msg): - msg = optional_connected_peer_to_message(connected_peer) - - assert msg == connected_peer_msg diff --git a/tests/conftest.py b/tests/conftest.py index ae269309..e1e364ee 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,7 +25,6 @@ from squeak.core.elliptic import payment_point_bytes_from_scalar_bytes from squeak.core.keys import SqueakPrivateKey from squeaknode.bitcoin.block_info import BlockInfo -from squeaknode.core.connected_peer import ConnectedPeer from squeaknode.core.download_result import DownloadResult from squeaknode.core.lightning_address import LightningAddressHostPort from squeaknode.core.offer import Offer @@ -44,7 +43,6 @@ from squeaknode.core.squeak_peer import SqueakPeer from squeaknode.core.squeaks import get_hash from squeaknode.core.squeaks import make_squeak_with_block from squeaknode.core.user_config import UserConfig -from squeaknode.network.peer import Peer from tests.utils import gen_contact_profile from tests.utils import gen_signing_profile from tests.utils import sha256 @@ -557,25 +555,6 @@ def sent_payment_summary( ) -@pytest.fixture -def peer_object(peer_address): - yield Peer( - peer_socket=None, - local_address=peer_address, - remote_address=peer_address, - outgoing=True, - peer_changed_listener=None, - ) - - -@pytest.fixture -def connected_peer(peer, peer_object): - yield ConnectedPeer( - peer=peer_object, - saved_peer=peer, - ) - - @pytest.fixture def download_result(): yield DownloadResult( diff --git a/tests/network/test_connection_manager.py b/tests/network/test_connection_manager.py deleted file mode 100644 index d1be7c01..00000000 --- a/tests/network/test_connection_manager.py +++ /dev/null @@ -1,203 +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 queue -import socket -import threading - -import mock -import pytest -from util import yield_inbound_socket_and_outbound_socket - -from squeaknode.core.peer_address import Network -from squeaknode.core.peer_address import PeerAddress -from squeaknode.network.connection_manager import ConnectionManager -from squeaknode.node.squeak_controller import SqueakController - - -@pytest.fixture -def local_address(): - local_ip = socket.gethostbyname('localhost') - local_port = 56789 - yield PeerAddress( - network=Network.IPV4, - host=local_ip, - port=local_port, - ) - - -@pytest.fixture -def inbound_socket_and_outbound_socket(): - yield from yield_inbound_socket_and_outbound_socket() - - -@pytest.fixture -def inbound_socket(inbound_socket_and_outbound_socket): - inbound_socket, _ = inbound_socket_and_outbound_socket - yield inbound_socket - - -@pytest.fixture -def outbound_socket(inbound_socket_and_outbound_socket): - _, outbound_socket = inbound_socket_and_outbound_socket - yield outbound_socket - - -@pytest.fixture -def inbound_local_address(): - yield PeerAddress( - network=Network.IPV4, - host='inbound.com', - port=56789, - ) - - -@pytest.fixture -def outbound_local_address(): - yield PeerAddress( - network=Network.IPV4, - host='outbound.com', - port=4321, - ) - - -@pytest.fixture -def inbound_connection_manager(local_address): - yield ConnectionManager(local_address) - - -@pytest.fixture -def outbound_connection_manager(local_address): - yield ConnectionManager(local_address) - - -@pytest.fixture -def squeak_controller(): - return mock.Mock(spec=SqueakController) - - -def start_connection( - connection_manager, - connected_event, - disconnect_event, - disconnected_event, - socket, - remote_address, - is_outbound, - squeak_controller, - result_q, -): - if is_outbound: - print('------Starting outbound connection------') - else: - print('------Starting inbound connection------') - with connection_manager.connect( - socket, - remote_address, - is_outbound, - squeak_controller, - result_q, - ): - connected_event.set() - disconnect_event.wait() - disconnected_event.set() - - -def test_connect_peers( - inbound_socket, - outbound_socket, - inbound_connection_manager, - outbound_connection_manager, - inbound_local_address, - outbound_local_address, - squeak_controller, - caplog, -): - import logging - caplog.set_level(logging.INFO) - inbound_q = queue.Queue() - outbound_q = queue.Queue() - - assert inbound_socket is not None - assert outbound_socket is not None - - assert len(inbound_connection_manager.peers) == 0 - assert len(outbound_connection_manager.peers) == 0 - - inbound_connected_event = threading.Event() - inbound_disconnect_event = threading.Event() - inbound_disconnected_event = threading.Event() - # Start the inbound connection in a thread. - inbound_connection_thread = threading.Thread( - target=start_connection, - args=( - inbound_connection_manager, - inbound_connected_event, - inbound_disconnect_event, - inbound_disconnected_event, - inbound_socket, - outbound_local_address, - False, - squeak_controller, - inbound_q, - )) - inbound_connection_thread.start() - - outbound_connected_event = threading.Event() - outbound_disconnect_event = threading.Event() - outbound_disconnected_event = threading.Event() - # Start the outbound connection in a thread. - outbound_connection_thread = threading.Thread( - target=start_connection, - args=( - outbound_connection_manager, - outbound_connected_event, - outbound_disconnect_event, - outbound_disconnected_event, - outbound_socket, - inbound_local_address, - True, - squeak_controller, - outbound_q, - )) - outbound_connection_thread.start() - - # import time - # time.sleep(5) - # assert False - - # Wait for both sides of connection to be connected - inbound_connected_event.wait() - outbound_connected_event.wait() - - print('Opened connection.') - assert len(inbound_connection_manager.peers) == 1 - assert len(outbound_connection_manager.peers) == 1 - - inbound_disconnect_event.set() - outbound_disconnect_event.set() - - inbound_disconnected_event.wait() - outbound_disconnected_event.wait() - - print('Closed connection.') - assert len(inbound_connection_manager.peers) == 0 - assert len(outbound_connection_manager.peers) == 0 diff --git a/tests/network/util.py b/tests/network/util.py deleted file mode 100644 index 85c761fd..00000000 --- a/tests/network/util.py +++ /dev/null @@ -1,87 +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 socket -import threading -from concurrent.futures import ThreadPoolExecutor -from contextlib import closing - - -TEST_SOCKET_PORT = 19999 -SOCKET_CONNECT_TIMEOUT = 10 - - -def accept_connections(listen_socket, started_event): - try: - print('Trying to bind and listen on port: {}'.format( - TEST_SOCKET_PORT), flush=True) - listen_socket.bind(('', TEST_SOCKET_PORT)) - listen_socket.listen() - started_event.set() - print('Started event set.', flush=True) - peer_socket, address = listen_socket.accept() - # host, port = address - # peer_address = PeerAddress( - # host=host, - # port=port, - # ) - peer_socket.setblocking(True) - return peer_socket - except Exception as e: - print(e) - started_event.set() - return e - - -def make_connection(peer_socket, started): - address = ('localhost', TEST_SOCKET_PORT) - started.wait() - print('Conecting to address: {}'.format(address)) - try: - peer_socket.settimeout(SOCKET_CONNECT_TIMEOUT) - peer_socket.connect(address) - peer_socket.setblocking(True) - return peer_socket - except Exception: - print('Failed to connect to {}'.format(address)) - - -def yield_inbound_socket_and_outbound_socket(): - # TODO: set up inbound and outbound sockets - started = threading.Event() - - # Use futures - with closing(socket.socket()) as listen_socket, \ - closing(socket.socket()) as peer_socket, \ - ThreadPoolExecutor(max_workers=2) as executor: - inbound_future = executor.submit( - accept_connections, listen_socket, started) - outbound_future = executor.submit( - make_connection, peer_socket, started) - - inbound_socket = inbound_future.result() - print(inbound_socket) - if isinstance(inbound_socket, Exception): - raise inbound_socket - outbound_socket = outbound_future.result() - print(outbound_socket) - - yield inbound_socket, outbound_socket diff --git a/tests/node/test_active_download.py b/tests/node/test_active_download.py deleted file mode 100644 index 032fbf85..00000000 --- a/tests/node/test_active_download.py +++ /dev/null @@ -1,136 +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 mock -import pytest -from squeak.messages import msg_getdata -from squeak.messages import msg_getsqueaks -from squeak.net import CInterested -from squeak.net import CInv -from squeak.net import CSqueakLocator - -from squeaknode.core.download_result import DownloadResult -from squeaknode.node.active_download_manager import HashDownload -from squeaknode.node.active_download_manager import InterestDownload -from squeaknode.node.downloaded_object import DownloadedSqueak -from tests.utils import gen_squeak - - -@pytest.fixture() -def download_hash(squeak_hash): - yield HashDownload(squeak_hash=squeak_hash) - - -@pytest.fixture() -def interest(public_key, block_count): - yield CInterested( - pubkeys=(public_key,), - nMinBlockHeight=block_count - 100, - nMaxBlockHeight=block_count + 100, - ) - - -@pytest.fixture() -def download_interest(interest): - yield InterestDownload(limit=10, interest=interest) - - -def test_download_hash_is_interested(download_hash, squeak): - downloaded_squeak = DownloadedSqueak(squeak) - - assert download_hash.is_interested(downloaded_squeak) - - -def test_download_hash_is_not_interested(download_hash, private_key, block_count): - other_squeak = gen_squeak(private_key, block_count) - downloaded_squeak = DownloadedSqueak(other_squeak) - - assert not download_hash.is_interested(downloaded_squeak) - - -def test_download_interest_is_interested(download_interest, squeak): - downloaded_squeak = DownloadedSqueak(squeak) - - assert download_interest.is_interested(downloaded_squeak) - - -def test_download_interest_is_not_interested(download_interest, private_key, block_count): - other_squeak = gen_squeak(private_key, block_count + 200) - downloaded_squeak = DownloadedSqueak(other_squeak) - - assert not download_interest.is_interested(downloaded_squeak) - - -def test_download_hash_initiate(download_hash, squeak_hash): - broadcast_fn = mock.Mock() - download_hash.initiate_download(broadcast_fn) - - expected_msg = msg_getdata( - inv=[CInv(type=1, hash=squeak_hash)] - ) - - broadcast_fn.assert_called_once_with(expected_msg) - - -def test_download_interest_initiate(download_interest, interest): - broadcast_fn = mock.Mock() - download_interest.initiate_download(broadcast_fn) - - expected_msg = msg_getsqueaks( - locator=CSqueakLocator( - vInterested=[interest], - ) - ) - - broadcast_fn.assert_called_once_with(expected_msg) - - -def test_download_hash_mark_complete(download_hash, squeak): - with mock.patch.object(download_hash, 'mark_complete', autospec=True) as mock_mark_complete: - mock_mark_complete.return_value = None - download_hash.increment() - - mock_mark_complete.assert_called_once_with() - - -def test_download_hash_mark_complete_not_called(download_hash, squeak): - with mock.patch.object(download_hash, 'mark_complete', autospec=True) as mock_mark_complete: - mock_mark_complete.return_value = None - - mock_mark_complete.assert_not_called() - - -def test_download_hash_wait_for_complete(download_hash): - download_hash.mark_complete() - - download_hash.wait_for_complete(50) - - -def test_download_hash_get_result(download_hash, squeak): - download_hash.increment() - download_result = download_hash.get_result() - - assert download_result == DownloadResult( - number_downloaded=1, - number_requested=1, - elapsed_time_ms=0, - number_peers=0, - ) diff --git a/tests/node/test_price_policy.py b/tests/node/test_price_policy.py index 90e059b1..a1de941b 100644 --- a/tests/node/test_price_policy.py +++ b/tests/node/test_price_policy.py @@ -30,32 +30,10 @@ def price_policy(): yield PricePolicy(None, None, None) -def test_get_price(price_policy, squeak, peer_address, user_config): +def test_get_price(price_policy, user_config): with mock.patch.object(price_policy, 'get_peer', autospec=True) as mock_get_peer, \ mock.patch.object(price_policy, 'get_sell_price_msat', autospec=True) as mock_get_sell_price_msat: mock_get_peer.return_value = None mock_get_sell_price_msat.return_value = 555 - assert price_policy.get_price(squeak, peer_address) == 555 - - -def test_get_price_profile_share_free_peer(price_policy, squeak, peer_address, peer, user_config): - with mock.patch.object(price_policy, 'get_peer', autospec=True) as mock_get_peer, \ - mock.patch.object(price_policy, 'get_sell_price_msat', autospec=True) as mock_get_sell_price_msat: - mock_get_peer.return_value = peer._replace( - share_for_free=True, - ) - mock_get_sell_price_msat.return_value = 555 - - assert price_policy.get_price(squeak, peer_address) == 0 - - -def test_get_price_profile_no_share_free_peer(price_policy, squeak, peer_address, peer, user_config): - with mock.patch.object(price_policy, 'get_peer', autospec=True) as mock_get_peer, \ - mock.patch.object(price_policy, 'get_sell_price_msat', autospec=True) as mock_get_sell_price_msat: - mock_get_peer.return_value = peer._replace( - share_for_free=False, - ) - mock_get_sell_price_msat.return_value = 555 - - assert price_policy.get_price(squeak, peer_address) == 555 + assert price_policy.get_price() == 555 diff --git a/tests/node/test_secret_key_reply.py b/tests/node/test_secret_key_reply.py deleted file mode 100644 index 6eb9fd53..00000000 --- a/tests/node/test_secret_key_reply.py +++ /dev/null @@ -1,67 +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 pytest -from squeak.messages import msg_secretkey -from squeak.net import COffer - -from squeaknode.node.secret_key_reply import FreeSecretKeyReply -from squeaknode.node.secret_key_reply import OfferReply - - -@pytest.fixture() -def free_secret_key_reply(squeak_hash, secret_key): - yield FreeSecretKeyReply( - squeak_hash=squeak_hash, - secret_key=secret_key, - ) - - -@pytest.fixture() -def offer_reply(squeak_hash, offer): - yield OfferReply( - squeak_hash=squeak_hash, - offer=offer, - ) - - -def test_free_secret_key_reply_msg(squeak_hash, secret_key, free_secret_key_reply): - reply_msg = free_secret_key_reply.get_msg() - - assert reply_msg == msg_secretkey( - hashSqk=squeak_hash, - secretKey=secret_key, - ) - - -def test_offer_reply_msg(squeak_hash, offer, offer_reply): - reply_msg = offer_reply.get_msg() - - assert reply_msg == msg_secretkey( - hashSqk=squeak_hash, - offer=COffer( - nonce=offer.nonce, - strPaymentInfo=offer.payment_request.encode( - 'utf-8'), - host=offer.host.encode('utf-8'), - port=offer.port, - ) - ) diff --git a/tests/node/test_squeak_controller.py b/tests/node/test_squeak_controller.py index e4ff2aa9..2698aeb8 100644 --- a/tests/node/test_squeak_controller.py +++ b/tests/node/test_squeak_controller.py @@ -27,8 +27,6 @@ from squeaknode.core.lightning_address import LightningAddressHostPort from squeaknode.core.peer_address import Network from squeaknode.core.peer_address import PeerAddress from squeaknode.core.squeak_core import SqueakCore -from squeaknode.network.network_manager import NetworkManager -from squeaknode.node.active_download_manager import ActiveDownloadManager from squeaknode.node.node_settings import NodeSettings from squeaknode.node.payment_processor import PaymentProcessor from squeaknode.node.squeak_controller import SqueakController @@ -62,11 +60,6 @@ def node_settings(): return mock.Mock(spec=NodeSettings) -@pytest.fixture -def network_manager(): - return mock.Mock(spec=NetworkManager) - - @pytest.fixture def squeak_core(): return mock.Mock(spec=SqueakCore) @@ -110,11 +103,6 @@ def payment_processor(): return mock.Mock(spec=PaymentProcessor) -@pytest.fixture -def download_manager(): - return mock.Mock(spec=ActiveDownloadManager) - - @pytest.fixture def twitter_forwarder(): return mock.Mock(spec=TwitterForwarder) @@ -125,8 +113,6 @@ def squeak_controller( squeak_store, squeak_core, payment_processor, - network_manager, - download_manager, twitter_forwarder, node_settings, config, @@ -136,8 +122,6 @@ def squeak_controller( squeak_store, squeak_core, payment_processor, - network_manager, - download_manager, twitter_forwarder, node_settings, config, @@ -150,8 +134,6 @@ def regtest_squeak_controller( squeak_store, squeak_core, payment_processor, - network_manager, - download_manager, twitter_forwarder, node_settings, regtest_config, @@ -161,8 +143,6 @@ def regtest_squeak_controller( squeak_store, squeak_core, payment_processor, - network_manager, - download_manager, twitter_forwarder, node_settings, regtest_config,