GlcoinCore/squeaknode
1
0
Fork 0
mirror of https://github.com/yzernik/squeaknode.git synced 2026-08-13 12:33:25 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use http for peer connection (#2002)

Browse source 
* Got basic web server running for peer

* Got squeak route and lookup route working

* Got secret key route working

* Got offer endpoint working

* Remove old comment

* Move peer server code to server subdir

* Create client subdir for peer client

* Return empty list for lookup when pubkeys is empty

* Add peer client and network controller class

* Remove old network code

* Remove old block subscription worker

* Remove network handler class

* Remove more old classes

* Add back twitter forwarder worker

* Remove download methods from controller

* Add websocket endpoint

* Add websocket subscribe method

* Add websocket client library

* Fetch squeaks from peer (slowly) on UI timeline load

* Fetch squeaks from peer (slowly) on UI pubkey squeaks load

* Got download squeaks working with HTTP requests

* Fixed external address method in controller

* Remove download pubkey squeak button and action

* Remove unused connection displays from peer ana peers page

* Fix 404 not found for offer endpoint

* Better handling of null cases in peer client downloader

* Simplify peer downloader

* Create subclass for range download

* Got single squeak download working

* Fix itest for get external address
This commit is contained in:
Jonathan Zernik 2022-03-17 14:38:46 -07:00 • committed by GitHub
parent 51f39427a9
commit 68e69e55c8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
50 changed files with 790 additions and 3638 deletions
Download patch file Download diff file Expand all files Collapse all files

12
frontend/src/api/client.js
View file

@ -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,
});
}

13
frontend/src/components/Nav/index.js
View file

@ -38,7 +38,6 @@ import {
selectMakeSqueakStatus,
selectBuySqueakStatus,
selectDownloadSqueakStatus,
selectDownloadPubkeySqueakStatus,
} from '../../features/squeaks/squeaksSlice'
const Nav = ({history}) => {
@ -303,18 +302,6 @@ const Nav = ({history}) => {
</div>
</div>
{/* Block screen with modal when download pubkey squeaks is waiting. */}
<div style={{display: useSelector(selectDownloadPubkeySqueakStatus) === 'loading' ? 'block' : 'none'}} className="modal-edit">
<div style={{minHeight: '270px', height: 'initial'}} className="modal-content">
<div className="modal-header">
<p className="modal-title">Downloading pubkey squeaks...</p>
</div>
<div style={{marginTop:'5px'}} className="modal-body">
<Loader />
</div>
</div>
</div>
</div>
)
}

66
frontend/src/components/Peer/index.js
View file

@ -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(
}
</div>
<div className="profile-options">
<div onClick={(e)=>
peerConnection ?
disconnectPeer(e) :
connectPeer(e)
}
className={peerConnection ? 'disconnect peer-connect-button' : 'peer-connect-button'}>
<span><span>{ peerConnection ? 'Connected' : 'Connect'}</span></span>
</div>
</div>
<div className="profile-options">
{peer &&
<div onClick={(e)=>
@ -215,34 +175,20 @@ return(
</div>
<div className="feed-wrapper">
{peerConnection &&
<div className="feed-trending-card">
<div className="feed-card-trend">
<div>Connection Time</div>
<div>{moment(peerConnection.getConnectTimeS() * 1000).fromNow(true)}</div>
<div>Number of downloads</div>
<div>TODO</div>
</div>
<div className="feed-card-trend">
<div>Bytes received</div>
<div>{peerConnection.getNumberBytesReceived()}</div>
<div>Number of purchases</div>
<div>TODO</div>
</div>
<div className="feed-card-trend">
<div>Messages received</div>
<div>{peerConnection.getNumberMessagesReceived()}</div>
</div>
<div className="feed-card-trend">
<div>Last Message Received Time</div>
<div>{moment(peerConnection.getLastMessageReceivedTimeS() * 1000).fromNow(true)}</div>
</div>
<div className="feed-card-trend">
<div>Bytes sent</div>
<div>{peerConnection.getNumberBytesSent()}</div>
</div>
<div className="feed-card-trend">
<div>Messages sent</div>
<div>{peerConnection.getNumberMessagesSent()}</div>
<div>Last connection time</div>
<div>TODO</div>
</div>
</div>
}
</div>

46
frontend/src/components/Peers/index.js
View file

@ -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) => {
<div>
<div className="explore-nav-menu">
<div onClick={()=>setTab('Saved Peers')} className={tab === 'Saved Peers' ? `explore-nav-item activeTab` : `explore-nav-item`}>
Saved Peers
</div>
<div onClick={()=>setTab('Connected Peers')} className={tab === 'Connected Peers' ? `explore-nav-item activeTab` : `explore-nav-item`}>
Connected Peers
Peers
</div>
</div>
{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 <div onClick={()=>goToPeer(peerAddress)} key={peerId} className="search-result-wapper">
{isConnected ?
<ICON_LAPTOPFILL styles={{fill:"rgb(0,128,0)", width:'48px', height:"48px"}} /> :
<ICON_LAPTOPFILL styles={{fill:"rgb(255,0,0)", width:'48px', height:"48px"}} />
}
<ICON_LAPTOPFILL styles={{fill:"rgb(0,128,0)", width:'48px', height:"48px"}} /> :
<div className="search-user-details">
<div className="search-user-warp">
<div className="search-user-info">
@ -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 <div onClick={()=>goToPeer(peerAddress)} key={addrStr} className="search-result-wapper">
<ICON_LAPTOPFILL styles={{fill:"rgb(0,128,0)", width:'48px', height:"48px"}} />
<div className="search-user-details">
<div className="search-user-warp">
<div className="search-user-info">
<div className="search-user-name">{savedPeerName}</div>
<div className="search-user-username">{addrStr}</div>
</div>
</div>
<div className="search-user-bio">
&nbsp;
</div>
</div>
</div>
})
<></>
: <div className="try-searching">
Nothing to see here ..
<div/>

16
frontend/src/components/Profile/index.js
View file

@ -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) => {
</div>
}
<div onClick={(e)=>
downloadUserSqueaks()
}
className={'profile-edit-button'}>
<span><span>{'Download Squeaks'}</span></span>
</div>
{user &&
<div onClick={(e)=>
user.getFollowing() ?

23
frontend/src/features/squeaks/squeaksSlice.js
View file

@ -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

4
itests/config.ini
View file

@ -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

6
itests/tests/test_squeak_node.py
View file

@ -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):

2
requirements.txt
View file

@ -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

25
squeaknode/admin/messages.py
View file

@ -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)

59
squeaknode/admin/squeak_admin_server_handler.py
View file

@ -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)

12
squeaknode/admin/webapp/app.py
View file

@ -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())

57
squeaknode/bitcoin/bitcoin_block_subscription_client.py
View file

@ -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('<I', seq)[-1])
if topic == b"hashblock":
logger.debug('- HASH BLOCK (' + sequence + ') -')
logger.debug(binascii.hexlify(body))
logger.debug('body type: {}'.format(type(body)))
logger.debug('body: {}'.format(body))
yield body

0
squeaknode/network/__init__.py → squeaknode/client/__init__.py
View file

79
squeaknode/client/network_controller.py Normal file
View file

@ -0,0 +1,79 @@
# 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 squeak.core.keys import SqueakPublicKey
from squeaknode.client.peer_downloader import RangeDownloader
from squeaknode.client.peer_downloader import SingleDownloader
from squeaknode.node.squeak_store import SqueakStore
logger = logging.getLogger(__name__)
DOWNLOAD_TIMEOUT_S = 10
class NetworkController:
def __init__(self, squeak_store: SqueakStore, config):
self.squeak_store = squeak_store
self.config = config
def download_timeline(self) -> 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()

100
squeaknode/client/peer_client.py Normal file
View file

@ -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

146
squeaknode/client/peer_downloader.py Normal file
View file

@ -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

32
squeaknode/core/connected_peer.py
View file

@ -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]

365
squeaknode/network/connection.py
View file

@ -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()

188
squeaknode/network/connection_manager.py
View file

@ -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

156
squeaknode/network/network_manager.py
View file

@ -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,
)

403
squeaknode/network/peer.py
View file

@ -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

147
squeaknode/network/peer_client.py
View file

@ -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,
)

87
squeaknode/network/peer_handler.py
View file

@ -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()

91
squeaknode/network/peer_server.py
View file

@ -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()

207
squeaknode/node/active_download_manager.py
View file

@ -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)

72
squeaknode/node/downloaded_object.py
View file

@ -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

347
squeaknode/node/network_handler.py
View file

@ -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)

54
squeaknode/node/peer_connection_worker.py
View file

@ -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"

54
squeaknode/node/peer_subscription_update_worker.py
View file

@ -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)

8
squeaknode/node/price_policy.py
View file

@ -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()

66
squeaknode/node/secret_key_reply.py
View file

@ -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,
)

152
squeaknode/node/squeak_controller.py
View file

@ -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

122
squeaknode/node/squeak_node.py
View file

@ -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,
)

59
squeaknode/node/squeak_store.py
View file

@ -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]

63
squeaknode/node/update_follows_worker.py
View file

@ -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)

77
squeaknode/node/update_subscribed_secret_key_worker.py
View file

@ -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.")

77
squeaknode/node/update_subscribed_squeak_worker.py
View file

@ -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.")

5
squeaknode/network/util.py → squeaknode/server/__init__.py
View file

@ -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())

199
squeaknode/server/app.py Normal file
View file

@ -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/<hash>')
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/<hash>')
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/<hash>')
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.")

136
squeaknode/server/squeak_peer_server_handler.py Normal file
View file

@ -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()

15
tests/admin/conftest.py
View file

@ -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,

20
tests/admin/test_messages.py
View file

@ -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

21
tests/conftest.py
View file

@ -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(

203
tests/network/test_connection_manager.py
View file

@ -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

87
tests/network/util.py
View file

@ -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

136
tests/node/test_active_download.py
View file

@ -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,
)

26
tests/node/test_price_policy.py
View file

@ -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

67
tests/node/test_secret_key_reply.py
View file

@ -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,
)
)

20
tests/node/test_squeak_controller.py
View file

@ -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,