renamed onionr dir and bugfixes/linting progress

This commit is contained in:
Kevin Froman 2019-11-20 04:52:50 -06:00
parent 2b996da17f
commit 720efe4fca
226 changed files with 179 additions and 142 deletions

View file

@ -0,0 +1,15 @@
# Onionr Communicator
Onionr communicator is the Onionr client. It "connects" to remote Onionr peers and does things such as:
* Finding new peers
* Uploading blocks
* Downloading blocks
* Daemon maintenance/housekeeping
## Files
* \_\_init\_\_.py: Contains the main communicator code. Inits and launches the communicator and sets up the timers
* peeraction.py: contains a function to send commands to remote peers
* bootstrappers.py: adds peers from the bootstrap list to the communicator to try to connect to them
* onlinepeers: management of the online peer pool for the communicator

270
src/communicator/__init__.py Executable file
View file

@ -0,0 +1,270 @@
'''
Onionr - Private P2P Communication
This file contains both the OnionrCommunicate class for communcating with peers
and code to operate as a daemon, getting commands from the command queue database (see core.Core.daemonQueue)
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import sys, os, time
import config, logger
import onionrexceptions, onionrpeers
from onionrblocks import onionrblockapi as block
from onionrplugins import onionrevents as events
import onionrplugins as plugins
from . import onlinepeers, uploadqueue
from communicatorutils import servicecreator, onionrcommunicatortimers
from communicatorutils import downloadblocks, lookupblocks, lookupadders
from communicatorutils import servicecreator, connectnewpeers
from communicatorutils import uploadblocks
from communicatorutils import daemonqueuehandler, announcenode, deniableinserts
from communicatorutils import cooldownpeer, housekeeping, netcheck
from onionrutils import localcommand, epoch
from etc import humanreadabletime
import onionrservices, filepaths
from onionrblocks import storagecounter
from coredb import daemonqueue, dbfiles
from utils import gettransports
from netcontroller import NetController
OnionrCommunicatorTimers = onionrcommunicatortimers.OnionrCommunicatorTimers
config.reload()
class OnionrCommunicatorDaemon:
def __init__(self, shared_state, developmentMode=config.get('general.dev_mode', False)):
# configure logger and stuff
self.config = config
self.storage_counter = storagecounter.StorageCounter()
self.isOnline = True # Assume we're connected to the internet
self.shared_state = shared_state # TooManyObjects module
# list of timer instances
self.timers = []
# initialize core with Tor socks port being 3rd argument
self.proxyPort = shared_state.get(NetController).socksPort
# Upload information, list of blocks to upload
self.blocksToUpload = []
self.upload_session_manager = self.shared_state.get(uploadblocks.sessionmanager.BlockUploadSessionManager)
self.shared_state.share_object()
# loop time.sleep delay in seconds
self.delay = 1
# lists of connected peers and peers we know we can't reach currently
self.onlinePeers = []
self.offlinePeers = []
self.cooldownPeer = {}
self.connectTimes = {}
self.peerProfiles = [] # list of peer's profiles (onionrpeers.PeerProfile instances)
self.newPeers = [] # Peers merged to us. Don't add to db until we know they're reachable
self.announceProgress = {}
self.announceCache = {}
self.generating_blocks = []
# amount of threads running by name, used to prevent too many
self.threadCounts = {}
# set true when shutdown command received
self.shutdown = False
# list of new blocks to download, added to when new block lists are fetched from peers
self.blockQueue = {}
# list of blocks currently downloading, avoid s
self.currentDownloading = []
# timestamp when the last online node was seen
self.lastNodeSeen = None
# Dict of time stamps for peer's block list lookup times, to avoid downloading full lists all the time
self.dbTimestamps = {}
# Clear the daemon queue for any dead messages
if os.path.exists(dbfiles.daemon_queue_db):
daemonqueue.clear_daemon_queue()
# Loads in and starts the enabled plugins
plugins.reload()
# time app started running for info/statistics purposes
self.startTime = epoch.get_epoch()
uploadqueue.UploadQueue(self) # extends our upload list and saves our list when Onionr exits
if developmentMode:
OnionrCommunicatorTimers(self, self.heartbeat, 30)
# Set timers, function reference, seconds
# requires_peer True means the timer function won't fire if we have no connected peers
peerPoolTimer = OnionrCommunicatorTimers(self, onlinepeers.get_online_peers, 60, max_threads=1, my_args=[self])
OnionrCommunicatorTimers(self, self.runCheck, 2, max_threads=1)
# Timers to periodically lookup new blocks and download them
lookup_blocks_timer = OnionrCommunicatorTimers(self, lookupblocks.lookup_blocks_from_communicator, config.get('timers.lookupBlocks', 25), my_args=[self], requires_peer=True, max_threads=1)
# The block download timer is accessed by the block lookup function to trigger faster download starts
self.download_blocks_timer = OnionrCommunicatorTimers(self, self.getBlocks, config.get('timers.getBlocks', 10), requires_peer=True, max_threads=5)
# Timer to reset the longest offline peer so contact can be attempted again
OnionrCommunicatorTimers(self, onlinepeers.clear_offline_peer, 58, my_args=[self])
# Timer to cleanup old blocks
blockCleanupTimer = OnionrCommunicatorTimers(self, housekeeping.clean_old_blocks, 20, my_args=[self])
# Timer to discover new peers
OnionrCommunicatorTimers(self, lookupadders.lookup_new_peer_transports_with_communicator, 60, requires_peer=True, my_args=[self], max_threads=2)
# Timer for adjusting which peers we actively communicate to at any given time, to avoid over-using peers
OnionrCommunicatorTimers(self, cooldownpeer.cooldown_peer, 30, my_args=[self], requires_peer=True)
# Timer to read the upload queue and upload the entries to peers
OnionrCommunicatorTimers(self, uploadblocks.upload_blocks_from_communicator, 5, my_args=[self], requires_peer=True, max_threads=1)
# Timer to process the daemon command queue
OnionrCommunicatorTimers(self, daemonqueuehandler.handle_daemon_commands, 6, my_args=[self], max_threads=3)
# Setup direct connections
if config.get('general.socket_servers', False):
self.services = onionrservices.OnionrServices()
self.active_services = []
self.service_greenlets = []
OnionrCommunicatorTimers(self, servicecreator.service_creator, 5, max_threads=50, my_args=[self])
else:
self.services = None
# {peer_pubkey: ephemeral_address}, the address to reach them
self.direct_connection_clients = {}
# This timer creates deniable blocks, in an attempt to further obfuscate block insertion metadata
if config.get('general.insert_deniable_blocks', True):
deniableBlockTimer = OnionrCommunicatorTimers(self, deniableinserts.insert_deniable_block, 180, my_args=[self], requires_peer=True, max_threads=1)
deniableBlockTimer.count = (deniableBlockTimer.frequency - 175)
# Timer to check for connectivity, through Tor to various high-profile onion services
netCheckTimer = OnionrCommunicatorTimers(self, netcheck.net_check, 500, my_args=[self], max_threads=1)
# Announce the public API server transport address to other nodes if security level allows
if config.get('general.security_level', 1) == 0 and config.get('general.announce_node', True):
# Default to high security level incase config breaks
announceTimer = OnionrCommunicatorTimers(self, announcenode.announce_node, 3600, my_args=[self], requires_peer=True, max_threads=1)
announceTimer.count = (announceTimer.frequency - 120)
else:
logger.debug('Will not announce node.')
# Timer to delete malfunctioning or long-dead peers
cleanupTimer = OnionrCommunicatorTimers(self, self.peerCleanup, 300, requires_peer=True)
# Timer to cleanup dead ephemeral forward secrecy keys
forwardSecrecyTimer = OnionrCommunicatorTimers(self, housekeeping.clean_keys, 15, my_args=[self], max_threads=1)
# Adjust initial timer triggers
peerPoolTimer.count = (peerPoolTimer.frequency - 1)
cleanupTimer.count = (cleanupTimer.frequency - 60)
blockCleanupTimer.count = (blockCleanupTimer.frequency - 2)
lookup_blocks_timer = (lookup_blocks_timer.frequency - 2)
shared_state.add(self)
if config.get('general.use_bootstrap', True):
bootstrappeers.add_bootstrap_list_to_peer_list(self, [], db_only=True)
# Main daemon loop, mainly for calling timers, don't do any complex operations here to avoid locking
try:
while not self.shutdown:
for i in self.timers:
if self.shutdown:
break
i.processTimer()
time.sleep(self.delay)
# Debug to print out used FDs (regular and net)
#proc = psutil.Process()
#print(proc.open_files(), len(psutil.net_connections()))
except KeyboardInterrupt:
self.shutdown = True
pass
logger.info('Goodbye. (Onionr is cleaning up, and will exit)', terminal=True)
try:
self.service_greenlets
except AttributeError:
pass
else:
# Stop onionr direct connection services
for server in self.service_greenlets:
server.stop()
localcommand.local_command('shutdown') # shutdown the api
try:
time.sleep(0.5)
except KeyboardInterrupt:
pass
def getBlocks(self):
'''download new blocks in queue'''
downloadblocks.download_blocks_from_communicator(self)
def decrementThreadCount(self, threadName):
'''Decrement amount of a thread name if more than zero, called when a function meant to be run in a thread ends'''
try:
if self.threadCounts[threadName] > 0:
self.threadCounts[threadName] -= 1
except KeyError:
pass
def connectNewPeer(self, peer='', useBootstrap=False):
'''Adds a new random online peer to self.onlinePeers'''
connectnewpeers.connect_new_peer_to_communicator(self, peer, useBootstrap)
def peerCleanup(self):
'''This just calls onionrpeers.cleanupPeers, which removes dead or bad peers (offline too long, too slow)'''
onionrpeers.peer_cleanup()
self.decrementThreadCount('peerCleanup')
def getPeerProfileInstance(self, peer):
'''Gets a peer profile instance from the list of profiles, by address name'''
for i in self.peerProfiles:
# if the peer's profile is already loaded, return that
if i.address == peer:
retData = i
break
else:
# if the peer's profile is not loaded, return a new one. connectNewPeer also adds it to the list on connect
retData = onionrpeers.PeerProfiles(peer)
self.peerProfiles.append(retData)
return retData
def getUptime(self):
return epoch.get_epoch() - self.startTime
def heartbeat(self):
'''Show a heartbeat debug message'''
logger.debug('Heartbeat. Node running for %s.' % humanreadabletime.human_readable_time(self.getUptime()))
self.decrementThreadCount('heartbeat')
def runCheck(self):
if run_file_exists(self):
logger.debug('Status check; looks good.')
self.decrementThreadCount('runCheck')
def startCommunicator(shared_state):
OnionrCommunicatorDaemon(shared_state)
def run_file_exists(daemon):
if os.path.isfile(filepaths.run_check_file):
os.remove(filepaths.run_check_file)
return True
return False

View file

@ -0,0 +1,31 @@
'''
Onionr - Private P2P Communication
add bootstrap peers to the communicator peer list
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
from utils import readstatic, gettransports
from coredb import keydb
bootstrap_peers = readstatic.read_static('bootstrap-nodes.txt').split(',')
def add_bootstrap_list_to_peer_list(comm_inst, peerList, db_only=False):
'''
Add the bootstrap list to the peer list (no duplicates)
'''
for i in bootstrap_peers:
if i not in peerList and i not in comm_inst.offlinePeers and not i in gettransports.get() and len(str(i).strip()) > 0:
if not db_only: peerList.append(i)
keydb.addkeys.add_address(i)

View file

@ -0,0 +1,12 @@
# Online Peers
Manages a pool of peers to perform actions with. Since Onionr does not maintain socket connections, it holds a list of peers.
## Files
* \_\_init\_\_.py: exposes some functions to interact with the pool
* clearofflinepeer.py: Pop the oldest peer in the offline list
* onlinepeers.py: communicator timer to add new peers to the pool randomly
* pickonlinepeers.py: returns a random peer from the online pool
* removeonlinepeer.py: removes a specified peer from the online pool

View file

@ -0,0 +1,6 @@
from . import clearofflinepeer, onlinepeers, pickonlinepeers, removeonlinepeer
clear_offline_peer = clearofflinepeer.clear_offline_peer
get_online_peers = onlinepeers.get_online_peers
pick_online_peer = pickonlinepeers.pick_online_peer
remove_online_peer = removeonlinepeer.remove_online_peer

View file

@ -0,0 +1,29 @@
'''
Onionr - Private P2P Communication
clear offline peer in a communicator instance
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import logger
def clear_offline_peer(comm_inst):
'''Removes the longest offline peer to retry later'''
try:
removed = comm_inst.offlinePeers.pop(0)
except IndexError:
pass
else:
logger.debug('Removed ' + removed + ' from offline list, will try them again.')
comm_inst.decrementThreadCount('clear_offline_peer')

View file

@ -0,0 +1,45 @@
'''
Onionr - Private P2P Communication
get online peers in a communicator instance
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import time
from etc import humanreadabletime
import logger
def get_online_peers(comm_inst):
'''
Manages the comm_inst.onlinePeers attribute list, connects to more peers if we have none connected
'''
config = comm_inst.config
logger.debug('Refreshing peer pool...')
maxPeers = int(config.get('peers.max_connect', 10))
needed = maxPeers - len(comm_inst.onlinePeers)
for i in range(needed):
if len(comm_inst.onlinePeers) == 0:
comm_inst.connectNewPeer(useBootstrap=True)
else:
comm_inst.connectNewPeer()
if comm_inst.shutdown:
break
else:
if len(comm_inst.onlinePeers) == 0:
logger.debug('Couldn\'t connect to any peers.' + (' Last node seen %s ago.' % humanreadabletime.human_readable_time(time.time() - comm_inst.lastNodeSeen) if not comm_inst.lastNodeSeen is None else ''), terminal=True)
else:
comm_inst.lastNodeSeen = time.time()
comm_inst.decrementThreadCount('get_online_peers')

View file

@ -0,0 +1,35 @@
'''
Onionr - Private P2P Communication
pick online peers in a communicator instance
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import secrets
def pick_online_peer(comm_inst):
'''randomly picks peer from pool without bias (using secrets module)'''
ret_data = ''
while True:
peer_length = len(comm_inst.onlinePeers)
if peer_length <= 0:
break
try:
# get a random online peer, securely. May get stuck in loop if network is lost or if all peers in pool magically disconnect at once
ret_data = comm_inst.onlinePeers[secrets.randbelow(peer_length)]
except IndexError:
pass
else:
break
return ret_data

View file

@ -0,0 +1,33 @@
'''
Onionr - Private P2P Communication
remove an online peer from the pool in a communicator instance
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
def remove_online_peer(comm_inst, peer):
'''Remove an online peer'''
try:
del comm_inst.connectTimes[peer]
except KeyError:
pass
try:
del comm_inst.dbTimestamps[peer]
except KeyError:
pass
try:
comm_inst.onlinePeers.remove(peer)
except ValueError:
pass

View file

@ -0,0 +1,55 @@
'''
Onionr - Private P2P Communication
This file implements logic for performing requests to Onionr peers
'''
'''
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
import streamedrequests
import logger
from onionrutils import epoch, basicrequests
from coredb import keydb
from . import onlinepeers
def peer_action(comm_inst, peer, action, returnHeaders=False, max_resp_size=5242880):
'''Perform a get request to a peer'''
penalty_score = -10
if len(peer) == 0:
return False
url = 'http://%s/%s' % (peer, action)
try:
ret_data = basicrequests.do_get_request(url, port=comm_inst.proxyPort,
max_size=max_resp_size)
except streamedrequests.exceptions.ResponseLimitReached:
logger.warn('Request failed due to max response size being overflowed', terminal=True)
ret_data = False
penalty_score = -100
# if request failed, (error), mark peer offline
if ret_data == False: # For some reason "if not" breaks this. Prob has to do with empty string.
try:
comm_inst.getPeerProfileInstance(peer).addScore(penalty_score)
onlinepeers.remove_online_peer(comm_inst, peer)
keydb.transportinfo.set_address_info(peer, 'lastConnectAttempt', epoch.get_epoch())
if action != 'ping' and not comm_inst.shutdown:
logger.warn(f'Lost connection to {peer}', terminal=True)
onlinepeers.get_online_peers(comm_inst) # Will only add a new peer to pool if needed
except ValueError:
pass
else:
peer_profile = comm_inst.getPeerProfileInstance(peer)
peer_profile.update_connect_time()
peer_profile.addScore(1)
return ret_data # If returnHeaders, returns tuple of data, headers. if not, just data string

View file

@ -0,0 +1,60 @@
"""
Onionr - Private P2P Communication
Class to remember blocks that need to be uploaded and not shared on startup/shutdown
"""
"""
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
import atexit
import json
import deadsimplekv
import filepaths
from onionrutils import localcommand
UPLOAD_MEMORY_FILE = filepaths.upload_list
def _add_to_hidden_blocks(cache):
for bl in cache:
localcommand.local_command('waitforshare/' + bl, post=True)
class UploadQueue:
"""
Saves and loads block upload info from json file
"""
def __init__(self, communicator: 'OnionrCommunicatorDaemon'):
"""Start the UploadQueue object, loading left over uploads into queue
and registering save shutdown function
"""
self.communicator = communicator
cache = deadsimplekv.DeadSimpleKV(UPLOAD_MEMORY_FILE)
self.store_obj = cache
cache: list = cache.get('uploads')
if cache == None:
cache = []
_add_to_hidden_blocks(cache)
self.communicator.blocksToUpload.extend(cache)
atexit.register(self.save)
def save(self):
"""Saves to disk on shutdown or if called manually"""
bl: list = self.communicator.blocksToUpload
self.store_obj.put('uploads', bl)
self.store_obj.flush()