Onionr/src/communicator/__init__.py

331 lines
12 KiB
Python
Raw Normal View History

2020-01-27 08:09:28 +00:00
"""Onionr - Private P2P Communication.
2018-06-10 08:00:01 +00:00
2020-01-27 08:09:28 +00:00
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
"""
import time
import config
import logger
import onionrpeers
import onionrplugins as plugins
from . import onlinepeers, uploadqueue
from communicatorutils import servicecreator
from communicatorutils import onionrcommunicatortimers
from communicatorutils import downloadblocks
from communicatorutils import lookupblocks
from communicatorutils import lookupadders
from communicatorutils import connectnewpeers
from communicatorutils import uploadblocks
from communicatorutils import announcenode, deniableinserts
from communicatorutils import cooldownpeer
from communicatorutils import housekeeping
from communicatorutils import netcheck
from onionrutils import epoch
from onionrcommands.openwebinterface import get_url
from etc import humanreadabletime
import onionrservices
from netcontroller import NetController
from . import bootstrappeers
from . import daemoneventhooks
"""
2018-06-10 08:00:01 +00:00
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/>.
"""
2019-11-16 04:18:38 +00:00
2019-02-28 03:02:44 +00:00
OnionrCommunicatorTimers = onionrcommunicatortimers.OnionrCommunicatorTimers
config.reload()
class OnionrCommunicatorDaemon:
def __init__(self, shared_state, developmentMode=None):
if developmentMode is None:
developmentMode = config.get('general.dev_mode', False)
# configure logger and stuff
2019-07-20 06:02:30 +00:00
self.config = config
self.isOnline = True # Assume we're connected to the internet
self.shared_state = shared_state # TooManyObjects module
2018-08-21 20:01:50 +00:00
# populate kv values
self.kv = self.shared_state.get_by_string('DeadSimpleKV')
self.kv.put('blockQueue', {})
self.kv.put('shutdown', False)
self.kv.put('onlinePeers', [])
self.kv.put('currentDownloading', [])
self.kv.put('announceCache', {})
2019-12-18 10:05:37 +00:00
if config.get('general.offline_mode', False):
self.isOnline = False
# list of timer instances
self.timers = []
2019-02-12 19:18:08 +00:00
# initialize core with Tor socks port being 3rd argument
2019-08-04 04:52:57 +00:00
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.offlinePeers = []
2018-08-31 22:53:48 +00:00
self.cooldownPeer = {}
self.connectTimes = {}
# list of peer's profiles (onionrpeers.PeerProfile instances)
self.peerProfiles = []
# Peers merged to us. Don't add to db until we know they're reachable
self.newPeers = []
self.announceProgress = {}
self.generating_blocks = []
# amount of threads running by name, used to prevent too many
self.threadCounts = {}
2018-07-06 04:27:12 +00:00
2018-12-09 17:29:39 +00:00
# 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 = {}
# Loads in and starts the enabled plugins
plugins.reload()
# time app started running for info/statistics purposes
self.startTime = epoch.get_epoch()
2018-09-15 01:05:25 +00:00
# extends our upload list and saves our list when Onionr exits
uploadqueue.UploadQueue(self)
if developmentMode:
OnionrCommunicatorTimers(self, self.heartbeat, 30)
2018-07-06 04:27:12 +00:00
2018-07-02 04:04:14 +00:00
# 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])
# 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],
max_threads=1)
# Timer to cleanup old blocks
blockCleanupTimer = OnionrCommunicatorTimers(
self, housekeeping.clean_old_blocks, 20, my_args=[self],
max_threads=1)
# 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)
# Setup direct connections
if config.get('general.ephemeral_tunnels', False):
2019-07-18 17:40:48 +00:00
self.services = onionrservices.OnionrServices()
2019-03-24 02:56:46 +00:00
self.active_services = []
2019-03-26 04:25:46 +00:00
self.service_greenlets = []
OnionrCommunicatorTimers(
self, servicecreator.service_creator, 5,
max_threads=50, my_args=[self])
2019-03-24 02:56:46 +00:00
else:
self.services = None
2019-08-16 20:41:56 +00:00
# {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)
2018-12-09 17:29:39 +00:00
# Timer to check for connectivity,
# through Tor to various high-profile onion services
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 - 60)
2018-12-09 17:29:39 +00:00
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
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)
2019-08-04 04:52:57 +00:00
shared_state.add(self)
if config.get('general.use_bootstrap_list', True):
bootstrappeers.add_bootstrap_list_to_peer_list(
self, [], db_only=True)
2019-08-04 04:52:57 +00:00
daemoneventhooks.daemon_event_handlers(shared_state)
if not config.get('onboarding.done', True):
logger.info(
'First run detected. Run openhome to get setup.',
terminal=True)
get_url()
while not config.get('onboarding.done', True) and \
not self.shared_state.get_by_string(
'DeadSimpleKV').get('shutdown'):
try:
time.sleep(2)
except KeyboardInterrupt:
self.shared_state.get_by_string(
'DeadSimpleKV').put('shutdown', True)
# Main daemon loop, mainly for calling timers,
# don't do any complex operations here to avoid locking
try:
while not self.shared_state.get_by_string(
'DeadSimpleKV').get('shutdown'):
for i in self.timers:
if self.shared_state.get_by_string(
'DeadSimpleKV').get('shutdown'):
break
i.processTimer()
time.sleep(self.delay)
except KeyboardInterrupt:
self.shared_state.get_by_string(
'DeadSimpleKV').put('shutdown', True)
2018-07-06 04:27:12 +00:00
logger.info(
'Goodbye. (Onionr is cleaning up, and will exit)', terminal=True)
2019-03-27 18:55:43 +00:00
try:
self.service_greenlets
except AttributeError:
pass
else:
# Stop onionr direct connection services
2019-03-27 18:55:43 +00:00
for server in self.service_greenlets:
server.stop()
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)
2018-07-06 04:27:12 +00:00
2018-08-02 07:28:26 +00:00
def peerCleanup(self):
"""This just calls onionrpeers.cleanupPeers.
Remove dead or bad peers (offline too long, too slow)"""
2019-08-02 23:00:04 +00:00
onionrpeers.peer_cleanup()
self.decrementThreadCount('peerCleanup')
2018-08-02 07:28:26 +00:00
def getPeerProfileInstance(self, peer):
"""Gets a peer profile instance from the list of profiles"""
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
2019-07-18 17:40:48 +00:00
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')
2019-08-04 04:52:57 +00:00
def startCommunicator(shared_state):
OnionrCommunicatorDaemon(shared_state)