kev/Onionr
1
0
Fork 0
Code Issues 24 Pull Requests Projects Releases 2 Wiki Activity

* Refactored onionrcommands py to be linting compliant (started using mypy)

Browse Source 
+ git ignore mypy cache
This commit is contained in:
Kevin Froman 2019-12-19 04:34:19 -06:00
parent 6d123b93fc
commit 906219fe30
20 changed files with 617 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -21,6 +21,8 @@ testdata/*
*.dll
*.exe
.mypy_cache/
dist/*
# log files

35
src/onionrcommands/banblocks.py
View File

@ -1,9 +1,15 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This file contains the command for banning blocks from the node
'''
'''
"""
import sys
import logger
from onionrutils import stringvalidators
from onionrstorage import removeblock
from onionrstorage import deleteBlock
from onionrblocks import onionrblacklist
from utils import reconstructhash
"""
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
@ -16,17 +22,11 @@
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
import logger
from onionrutils import stringvalidators
from onionrstorage import removeblock
from onionrstorage import deleteBlock
from onionrblocks import onionrblacklist
from utils import reconstructhash
"""
def ban_block():
"""Deletes a block, permanently blacklisting it"""
"""Delete a block, permanently blacklisting it."""
blacklist = onionrblacklist.OnionrBlackList()
try:
ban = sys.argv[2]
@ -41,8 +41,9 @@ def ban_block():
blacklist.addToDB(ban)
removeblock.remove_block(ban)
deleteBlock(ban)
except Exception as error:
logger.error('Could not blacklist block', error=error, terminal=True)
except Exception as error: # pylint: disable=W0703
logger.error('Could not blacklist block',
error=error, terminal=True)
else:
logger.info('Block blacklisted', terminal=True)
else:
@ -50,4 +51,6 @@ def ban_block():
else:
logger.error('Invalid block hash', terminal=True)
ban_block.onionr_help = "<block hash>: deletes and blacklists a block"
ban_block.onionr_help = "<block hash>: " # type: ignore
ban_block.onionr_help += "deletes and blacklists a block" # type: ignore

51
src/onionrcommands/daemonlaunch.py
View File

@ -1,4 +1,5 @@
""" Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
launch the api servers and communicator
"""
import os
@ -47,10 +48,7 @@ def _proper_shutdown():
def daemon():
"""
Starts the Onionr communication daemon
"""
"""Start the Onionr communication daemon."""
offline_mode = config.get('general.offline_mode', False)
if not hastor.has_tor():
@ -143,23 +141,23 @@ def daemon():
cleanup.delete_run_files()
def _ignore_sigint(sig, frame):
"""This space intentionally left blank"""
def _ignore_sigint(sig, frame): # pylint: disable=W0612,W0613
"""Space intentionally left blank."""
return
def kill_daemon():
"""
Shutdown the Onionr daemon (communicator)
"""
"""Shutdown the Onionr daemon (communicator)."""
logger.warn('Stopping the running daemon...', timestamp=False,
terminal=True)
try:
# On platforms where we can, fork out to prevent locking
try:
pid = os.fork()
if pid != 0: return
except (AttributeError, OSError): pass
if pid != 0:
return
except (AttributeError, OSError):
pass
events.event('daemon_stop')
net = NetController(config.get('client.port', 59496))
@ -169,18 +167,17 @@ def kill_daemon():
pass
net.killTor()
except Exception as e:
logger.error('Failed to shutdown daemon: ' + str(e),
error=e, timestamp=False, terminal=True)
return
kill_daemon.onionr_help = "Gracefully stops the Onionr API servers"
kill_daemon.onionr_help = "Gracefully stops the " # type: ignore
kill_daemon.onionr_help += "Onionr API servers" # type: ignore
def start(input: bool = False, override: bool = False):
"""If no lock file, make one and start onionr,
error if there is and its not overridden"""
def start(override: bool = False):
"""If no lock file, make one and start onionr.
Error exit if there is and its not overridden
"""
if os.path.exists(filepaths.lock_file) and not override:
logger.fatal('Cannot start. Daemon is already running,'
+ ' or it did not exit cleanly.\n'
@ -188,14 +185,18 @@ def start(input: bool = False, override: bool = False):
+ ' delete onionr.lock & try again).', terminal=True)
else:
if not onionrvalues.DEVELOPMENT_MODE:
lockFile = open(filepaths.lock_file, 'w')
lockFile.write('delete at your own risk')
lockFile.close()
lock_file = open(filepaths.lock_file, 'w')
lock_file.write('delete at your own risk')
lock_file.close()
# Start Onionr daemon
daemon()
try:
os.remove(filepaths.lock_file)
except FileNotFoundError:
pass
start.onionr_help = "Start Onionr node (public and clients API servers)"
start.onionr_help = "Start Onionr node " # type: ignore
start.onionr_help += "(public and clients API servers)" # type: ignore

47
src/onionrcommands/exportblocks.py
View File

@ -1,9 +1,15 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This file handles the command for exporting blocks to disk
'''
'''
"""
import sys
import logger
import onionrstorage
from utils import createdirs
from onionrutils import stringvalidators
import filepaths
"""
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
@ -16,28 +22,31 @@
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
import logger, onionrstorage
from utils import createdirs
from onionrutils import stringvalidators
import filepaths
def doExport(bHash):
"""
def _do_export(b_hash):
createdirs.create_dirs()
data = onionrstorage.getData(bHash)
with open('%s/%s.dat' % (filepaths.export_location, bHash), 'wb') as exportFile:
exportFile.write(data)
data = onionrstorage.getData(b_hash)
with open('%s/%s.dat' % (filepaths.export_location,
b_hash), 'wb') as export:
export.write(data)
logger.info('Block exported as file', terminal=True)
def export_block():
exportDir = filepaths.export_location
"""Export block based on hash from stdin or argv."""
try:
if not stringvalidators.validate_hash(sys.argv[2]): raise ValueError
if not stringvalidators.validate_hash(sys.argv[2]):
raise ValueError
except (IndexError, ValueError):
logger.error('No valid block hash specified.', terminal=True)
sys.exit(1)
else:
bHash = sys.argv[2]
doExport(bHash)
b_hash = sys.argv[2]
_do_export(b_hash)
export_block.onionr_help = "<block hash>: Export an Onionr block to a file. Export directory is in the Onionr data directory under block-export/"
export_block.onionr_help = "<block hash>: Export block to " # type: ignore
export_block.onionr_help += "a file. Export directory is in " # type: ignore
export_block.onionr_help += "Onionr home under block-export" # type: ignore

107
src/onionrcommands/filecommands.py
View File

@ -1,8 +1,16 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This file handles the commands for adding and getting files from the Onionr network
This file handles the commands for adding
and getting files from the Onionr network
"""
import sys
import os
import logger
from onionrblocks.onionrblockapi import Block
import onionrexceptions
from onionrutils import stringvalidators
from etc import onionrvalues
from onionrblocks import insert
"""
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
@ -18,62 +26,67 @@
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
import base64, sys, os
import logger
from onionrblocks.onionrblockapi import Block
import onionrexceptions
from onionrutils import stringvalidators
from etc import onionrvalues
from onionrblocks import insert
_ORIG_DIR = onionrvalues.ORIG_RUN_DIR_ENV_VAR
def _get_dir(path: str) -> str:
if not os.getenv(_ORIG_DIR) is None: return os.getenv(_ORIG_DIR) + '/' + path
else: return path
if not os.getenv(_ORIG_DIR) is None:
return os.getenv(_ORIG_DIR) + '/' + path # type: ignore
else:
return path
def add_html(singleBlock=True, blockType='html'):
add_file(singleBlock, blockType)
"""Create one-off web page from HTML file, no ext resources."""
add_file(blockType=blockType)
add_html.onionr_help = "Adds an HTML file into Onionr. Does not currently include dependant resources"
def add_file(singleBlock=False, blockType='bin'):
"""
Adds a file to the onionr network
"""
add_html.onionr_help = "Adds an HTML file into Onionr. Does " # type: ignore
add_html.onionr_help += "not include dependant resources" # type: ignore
def add_file(blockType='bin'):
"""Add a file to the onionr network."""
if len(sys.argv) >= 3:
filename = sys.argv[2]
contents = None
if not os.path.exists(_get_dir(filename)):
logger.error('That file does not exist. Improper path (specify full path)?', terminal=True)
return
logger.info('Adding file... this might take a long time.', terminal=True)
try:
with open(_get_dir(filename), 'rb') as singleFile:
blockhash = insert(singleFile.read(), header=blockType)
if len(blockhash) > 0:
logger.info('File %s saved in block %s' % (filename, blockhash), terminal=True)
except Exception as e:
logger.error('Failed to save file in block ' + str(e), timestamp = False, terminal=True)
else:
logger.error('%s add-file <filename>' % sys.argv[0], timestamp = False, terminal=True)
add_file.onionr_help = "<file path> Add a file into the Onionr network"
if not os.path.exists(_get_dir(filename)):
logger.error(
'That file does not exist. Improper path (specify full path)?',
terminal=True)
return
logger.info('Adding file, this might take a long time.',
terminal=True)
try:
with open(_get_dir(filename), 'rb') as single_file:
blockhash = insert(single_file.read(), header=blockType)
if len(blockhash) > 0:
logger.info('File %s saved in block %s' %
(filename, blockhash), terminal=True)
except Exception as err: # pylint: disable=W0703
logger.error('Failed to save file in block ' +
str(err), timestamp=False, terminal=True)
else:
logger.error('%s add-file <filename>' %
sys.argv[0], timestamp=False, terminal=True)
add_file.onionr_help = "<file path> Add a file into " # type: ignore
add_file.onionr_help += "the Onionr network" # type: ignore
def get_file():
"""
Get a file from onionr blocks
"""
"""Get a file from onionr blocks."""
try:
fileName = _get_dir(sys.argv[2])
file_name = _get_dir(sys.argv[2])
bHash = sys.argv[3]
except IndexError:
logger.error("Syntax %s %s" % (sys.argv[0], '/path/to/filename <blockhash>'), terminal=True)
logger.error("Syntax %s %s" % (
sys.argv[0], '/path/to/filename <blockhash>'), terminal=True)
else:
logger.info(fileName, terminal=True)
logger.info(file_name, terminal=True)
contents = None
if os.path.exists(fileName):
if os.path.exists(file_name):
logger.error("File already exists", terminal=True)
return
if not stringvalidators.validate_hash(bHash):
@ -81,9 +94,13 @@ def get_file():
return
try:
with open(fileName, 'wb') as myFile:
myFile.write(Block(bHash).bcontent)
with open(file_name, 'wb') as my_file:
my_file.write(Block(bHash).bcontent)
except onionrexceptions.NoDataAvailable:
logger.error('That block is not available. Trying again later may work.', terminal=True)
logger.error(
'That block is not available. Trying again later may work.',
terminal=True)
get_file.onionr_help = "<file path> <block hash>: Download a file from the onionr network."
get_file.onionr_help = "<file path> <block hash>: Download " # type: ignore
get_file.onionr_help += "a file from the onionr network." # type: ignore

7
src/onionrcommands/keyadders.py
View File

@ -1,5 +1,4 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
add keys (transport and pubkey)
"""
@ -23,7 +22,7 @@ from coredb import keydb
def add_address():
"""Command to add a peer address from either an arg or stdin"""
"""Command to add a peer address from either an arg or stdin."""
try:
newAddress = sys.argv[2]
newAddress = newAddress.replace('http:', '').replace('/', '')
@ -38,4 +37,4 @@ def add_address():
logger.warn("Unable to add address.", terminal=True)
add_address.onionr_help = "Adds a node transport address to the node list"
add_address.onionr_help = "Adds a node transport address" # type: ignore

27
src/onionrcommands/motdcreator.py
View File

@ -1,15 +1,36 @@
"""Onionr - Private P2P Communication.
Command to make new network-wide MOTD message. Only network admin can do this
The key is set in onionrvalues
"""
import onionrblocks
"""
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 motd_creator():
"""Create a new MOTD message for the Onionr network"""
"""Create a new MOTD message for the Onionr network."""
motd = ''
new = ''
print('Enter a new MOTD, quit on a new line:')
while new != 'quit':
new = input()
new = input() # nosec B323
if new != 'quit':
motd += new
bl = onionrblocks.insert(motd, header='motd', sign=True)
print(f"inserted in {bl}")
motd_creator.onionr_help = "Create a new MOTD message for the onionr network"
motd_creator.onionr_help = "Create a new MOTD for the network" # type: ignore

109
src/onionrcommands/onionrstatistics.py
View File

@ -1,9 +1,21 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This module defines commands to show stats/details about the local node
'''
'''
"""
import os
import logger
from onionrblocks import onionrblockapi
from onionrblocks import onionrblacklist
from onionrutils import checkcommunicator, mnemonickeys
from utils import sizeutils, gethostname, getconsolewidth, identifyhome
from coredb import blockmetadb, keydb
import onionrcrypto
import config
from etc import onionrvalues
check_communicator = checkcommunicator.is_communicator_running
"""
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
@ -16,17 +28,11 @@
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 os, uuid, time
import logger
from onionrblocks import onionrblockapi
from onionrblocks import onionrblacklist
from onionrutils import checkcommunicator, mnemonickeys
from utils import sizeutils, gethostname, getconsolewidth, identifyhome
from coredb import blockmetadb, daemonqueue, keydb
import onionrcrypto, config
from etc import onionrvalues
"""
def show_stats():
"""Print/log statistic info about our Onionr install."""
try:
# define stats messages here
totalBlocks = len(blockmetadb.get_block_list())
@ -36,21 +42,31 @@ def show_stats():
messages = {
# info about local client
'Onionr Daemon Status' : ((logger.colors.fg.green + 'Online') if checkcommunicator.is_communicator_running(timeout = 9) else logger.colors.fg.red + 'Offline'),
'Onionr Daemon Status':
((logger.colors.fg.green + 'Online')
if check_communicator(timeout=9)
else logger.colors.fg.red + 'Offline'),
# file and folder size stats
'div1': True, # this creates a solid line across the screen, a div
'Total Block Size' : sizeutils.human_size(sizeutils.size(home + 'blocks/')),
'Total Plugin Size' : sizeutils.human_size(sizeutils.size(home + 'plugins/')),
'Log File Size' : sizeutils.human_size(sizeutils.size(home + 'output.log')),
'Total Block Size':
sizeutils.human_size(sizeutils.size(home + 'blocks/')),
'Total Plugin Size':
sizeutils.human_size(sizeutils.size(home + 'plugins/')),
'Log File Size':
sizeutils.human_size(sizeutils.size(home + 'output.log')),
# count stats
'div2': True,
'Known Peers (nodes)' : str(max(len(keydb.listkeys.list_adders()) - 1, 0)),
'Enabled Plugins' : str(len(config.get('plugins.enabled', list()))) + ' / ' + str(len(os.listdir(home + 'plugins/'))),
'Known Peers (nodes)':
str(max(len(keydb.listkeys.list_adders()) - 1, 0)),
'Enabled Plugins':
str(len(config.get('plugins.enabled', list()))) + ' / ' +
str(len(os.listdir(home + 'plugins/'))),
'Stored Blocks': str(totalBlocks),
'Deleted Blocks': str(totalBanned),
'Percent Blocks Signed' : str(round(100 * signedBlocks / max(totalBlocks, 1), 2)) + '%'
'Percent Blocks Signed':
str(round(100 * signedBlocks / max(totalBlocks, 1), 2)) + '%'
}
# color configuration
@ -73,23 +89,43 @@ def show_stats():
groupsize = width - prewidth - len('[+] ')
# generate stats table
logger.info(colors['title'] + 'Onionr v%s Statistics' % onionrvalues.ONIONR_VERSION + colors['reset'], terminal=True)
logger.info(colors['border'] + '-' * (maxlength + 1) + '+' + colors['reset'], terminal=True)
logger.info(colors['title'] + 'Onionr v%s Statistics' %
onionrvalues.ONIONR_VERSION + colors['reset'],
terminal=True)
logger.info(colors['border'] + '-' * (maxlength + 1) +
'+' + colors['reset'], terminal=True)
for key, val in messages.items():
if not (type(val) is bool and val is True):
val = [str(val)[i:i + groupsize] for i in range(0, len(str(val)), groupsize)]
val = [str(val)[i:i + groupsize]
for i in range(0, len(str(val)), groupsize)]
logger.info(colors['key'] + str(key).rjust(maxlength) + colors['reset'] + colors['border'] + ' | ' + colors['reset'] + colors['val'] + str(val.pop(0)) + colors['reset'], terminal=True)
logger.info(colors['key'] + str(key).rjust(maxlength) +
colors['reset'] + colors['border'] +
' | ' + colors['reset'] + colors['val'] +
str(val.pop(0)) + colors['reset'], terminal=True)
for value in val:
logger.info(' ' * maxlength + colors['border'] + ' | ' + colors['reset'] + colors['val'] + str(value) + colors['reset'], terminal=True)
logger.info(' ' * maxlength + colors['border'] + ' | ' +
colors['reset'] + colors['val'] + str(
value) + colors['reset'], terminal=True)
else:
logger.info(colors['border'] + '-' * (maxlength + 1) + '+' + colors['reset'], terminal=True)
logger.info(colors['border'] + '-' * (maxlength + 1) + '+' + colors['reset'], terminal=True)
except Exception as e:
logger.error('Failed to generate statistics table. ' + str(e), error = e, timestamp = False, terminal=True)
logger.info(colors['border'] + '-' * (maxlength +
1) + '+' +
colors['reset'], terminal=True)
logger.info(colors['border'] + '-' * (maxlength + 1) +
'+' + colors['reset'], terminal=True)
except Exception as e: # pylint: disable=W0703
logger.error('Failed to generate statistics table. ' +
str(e), error=e, timestamp=False, terminal=True)
def show_details():
"""Print out details.
node transport address(es)
active user ID
active user ID in mnemonic form
"""
details = {
'Node Address': gethostname.get_hostname(),
'Public Key': onionrcrypto.pub_key.replace('=', ''),
@ -97,7 +133,14 @@ def show_details():
}
for detail in details:
logger.info('%s%s: \n%s%s\n' % (logger.colors.fg.lightgreen, detail, logger.colors.fg.green, details[detail]), terminal = True)
logger.info('%s%s: \n%s%s\n' % (logger.colors.fg.lightgreen,
detail, logger.colors.fg.green,
details[detail]), terminal=True)
show_details.onionr_help = "Shows relevant information for your Onionr install: node address, and active public key."
show_stats.onionr_help = "Shows statistics for your Onionr node. Slow if Onionr is not running"
show_details.onionr_help = "Shows relevant information " # type: ignore
show_details.onionr_help += "for your Onionr install: node " # type: ignore
show_details.onionr_help += "address, and active public key." # type: ignore
show_stats.onionr_help = "Shows statistics for your Onionr " # type: ignore
show_stats.onionr_help += "node. Slow if Onionr is not running" # type: ignore

43
src/onionrcommands/openwebinterface.py
View File

@ -1,9 +1,12 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Open the web interface properly into a web browser
'''
'''
"""
import webbrowser
import logger
from onionrutils import getclientapiserver
import config
"""
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
@ -16,33 +19,43 @@
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 webbrowser
import logger
from onionrutils import getclientapiserver
import config
"""
def get_url():
def get_url() -> str:
"""Build UI URL string and return it."""
try:
url = getclientapiserver.get_client_API_server()
except FileNotFoundError:
url = ""
logger.error('Onionr seems to not be running (could not get api host)', terminal=True)
logger.error(
'Onionr seems to not be running (could not get api host)',
terminal=True)
else:
url = 'http://%s/#%s' % (url, config.get('client.webpassword'))
logger.info('Onionr web interface URL: ' + url, terminal=True)
return url
get_url.onionr_help = "Shows the Onionr web interface URL with API key"
get_url.onionr_help = "Shows the Onionr " # type: ignore
get_url.onionr_help += "web interface URL with API key" # type: ignore
def open_home():
"""Command to open web interface URL in default browser."""
try:
url = getclientapiserver.get_client_API_server()
except FileNotFoundError:
logger.error('Onionr seems to not be running (could not get api host)', terminal=True)
logger.error(
'Onionr seems to not be running (could not get api host)',
terminal=True)
else:
url = 'http://%s/#%s' % (url, config.get('client.webpassword'))
logger.info('If Onionr does not open automatically, use this URL: ' + url, terminal=True)
logger.info(
'If Onionr does not open automatically, use this URL: ' + url,
terminal=True)
webbrowser.open_new_tab(url)
open_home.onionr_help = "Opens the Onionr web UI in the default browser. Node must be running."
open_home.onionr_help = "Opens the Onionr UI in the default " # type: ignore
open_home.onionr_help += "browser. Node must be running." # type: ignore

21
src/onionrcommands/parser/__init__.py
View File

@ -1,5 +1,4 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This module loads in the Onionr arguments and their help messages
"""
@ -28,10 +27,12 @@ from . import arguments, recommend
def plugin_command(cmd):
"""Build a plugin command function name."""
return f'on_{cmd}_cmd'
def register_plugin_commands(cmd) -> bool:
"""Find a plugin command hook and execute it for a given cmd."""
plugin_cmd = plugin_command(cmd)
for pl in onionrplugins.get_enabled_plugins():
pl = onionrplugins.get_plugin(pl)
@ -46,11 +47,10 @@ def _show_term(msg: str):
def register():
"""Registers commands and handles help command processing"""
"""Register commands and handles help command processing."""
def get_help_message(cmd: str,
default: str = 'No help available for this command'):
"""Return help message for a given command, supports plugin commands"""
"""Print help message for a given command, supports plugin commands."""
pl_cmd = plugin_command(cmd)
for pl in onionrplugins.get_enabled_plugins():
pl = onionrplugins.get_plugin(pl)
@ -61,7 +61,7 @@ def register():
pass
for i in arguments.get_arguments():
for alias in i:
for _ in i:
try:
return arguments.get_help(cmd)
except AttributeError:
@ -78,16 +78,19 @@ def register():
return
is_help_cmd = False
if cmd.replace('--', '').lower() == 'help': is_help_cmd = True
if cmd.replace('--', '').lower() == 'help':
is_help_cmd = True
try:
try:
if cmd not in ('start', 'details', 'show-details'):
os.chdir(os.environ['ORIG_ONIONR_RUN_DIR'])
except KeyError: pass
except KeyError:
pass
try:
arguments.get_func(cmd)()
except KeyboardInterrupt: pass
except KeyboardInterrupt:
pass
except onionrexceptions.NotFound:
if not register_plugin_commands(cmd) and not is_help_cmd:
recommend.recommend()

14
src/onionrcommands/parser/arguments.py
View File

@ -1,5 +1,4 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Sets CLI arguments for Onionr
"""
@ -38,8 +37,11 @@ from onionrutils import importnewblocks # func to import new blocks
def get_arguments() -> dict:
"""This is a function because we need to be able
to dynamically modify them with plugins"""
"""Return command argument dict, minus plugin cmds.
This is a function because we need to be able to
dynamically modify them with plugins
"""
args = {
('blacklist', 'blacklist-block', 'remove-block',
'removeblock', 'banblock', 'ban-block'): banblocks.ban_block,
@ -76,7 +78,7 @@ def get_arguments() -> dict:
def get_help(arg: str) -> str:
"""Returns the help info string from a given command"""
"""Return the help info string from a given command."""
arguments = get_arguments()
# Iterate the command alias tuples
for argument in arguments:
@ -87,7 +89,7 @@ def get_help(arg: str) -> str:
def get_func(argument: str) -> Callable:
"""Returns the function for a given command argument"""
"""Return the function for a given command argument."""
argument = argument.lower()
args = get_arguments()

23
src/onionrcommands/parser/recommend.py
View File

@ -1,10 +1,29 @@
"""Onionr - Private P2P Communication.
Try to provide recommendations for invalid Onionr commands
"""
import sys
from difflib import SequenceMatcher
import logger
from . import arguments
"""
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 recommend(print_default: bool = True):
"""Print out a recommendation for argv cmd if one is available."""
tried = sys.argv[1]
args = arguments.get_arguments()
print_message = 'Command not found:'
@ -15,5 +34,5 @@ def recommend(print_default: bool = True):
+ 'did you mean "{word}"?',
terminal=True)
return
if print_default: logger.error('%s "%s"' %
(print_message, tried), terminal=True)
if print_default:
logger.error('%s "%s"' % (print_message, tried), terminal=True)

114
src/onionrcommands/pubkeymanager.py
View File

@ -1,9 +1,22 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
This module defines user ID-related CLI commands
'''
'''
"""
import sys
import getpass
import unpaddedbase32
import niceware
import vanityonionr
import logger
import onionrexceptions
from onionrutils import stringvalidators, bytesconverter
import config
import keymanager
import onionrcrypto
from etc import onionrvalues
"""
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
@ -16,45 +29,48 @@
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, getpass
import unpaddedbase32
import niceware
import vanityonionr
import logger, onionrexceptions
from onionrutils import stringvalidators, bytesconverter
from onionrusers import onionrusers, contactmanager
import config
from coredb import keydb
import keymanager, onionrcrypto
from etc import onionrvalues
DETERMINISTIC_REQUIREMENT = onionrvalues.PASSWORD_LENGTH
def add_ID():
"""Command to create a new user ID key pair."""
key_manager = keymanager.KeyManager()
try:
sys.argv[2]
if not sys.argv[2].lower() == 'true': raise ValueError
except (IndexError, ValueError) as e:
sys.argv[2] # pylint: disable=W0104
if not sys.argv[2].lower() == 'true':
raise ValueError
except (IndexError, ValueError):
newID = key_manager.addKey()[0]
else:
logger.warn('Deterministic keys require random and long passphrases.', terminal=True)
logger.warn('If a good passphrase is not used, your key can be easily stolen.', terminal=True)
logger.warn('You should use a series of hard to guess words, see this for reference: https://www.xkcd.com/936/', terminal=True)
logger.warn(
'Deterministic keys require random and long passphrases.',
terminal=True)
logger.warn(
'If a good passphrase is not used, your key can be easily stolen.',
terminal=True)
logger.warn(
'You should use a series of hard to guess words, ' +
'see this for reference: https://www.xkcd.com/936/',
terminal=True)
try:
pass1 = getpass.getpass(prompt='Enter at least %s characters: ' % (DETERMINISTIC_REQUIREMENT,))
pass1 = getpass.getpass(
prompt='Enter at least %s characters: ' %
(DETERMINISTIC_REQUIREMENT,))
pass2 = getpass.getpass(prompt='Confirm entry: ')
except KeyboardInterrupt:
sys.exit(42)
if onionrcrypto.cryptoutils.safe_compare(pass1, pass2):
try:
logger.info('Generating deterministic key. This can take a while.', terminal=True)
logger.info(
'Generating deterministic key. This can take a while.',
terminal=True)
newID, privKey = onionrcrypto.generate_deterministic(pass1)
except onionrexceptions.PasswordStrengthError:
logger.error('Passphrase must use at least %s characters.' % (DETERMINISTIC_REQUIREMENT,), terminal=True)
logger.error('Passphrase must use at least %s characters.' % (
DETERMINISTIC_REQUIREMENT,), terminal=True)
sys.exit(1)
else:
logger.error('Passwords do not match.', terminal=True)
@ -63,13 +79,22 @@ def add_ID():
key_manager.addKey(pubKey=newID,
privKey=privKey)
except ValueError:
logger.error('That ID is already available, you can change to it with the change-id command.', terminal=True)
logger.error(
'That ID is already available, you can change to it ' +
'with the change-id command.', terminal=True)
return
logger.info('Added ID: %s' % (bytesconverter.bytes_to_str(newID),), terminal=True)
logger.info('Added ID: %s' %
(bytesconverter.bytes_to_str(newID),), terminal=True)
add_ID.onionr_help = "If the first argument is true, " # type: ignore
add_ID.onionr_help += "Onionr will show a deterministic " # type: ignore
add_ID.onionr_help += "generation prompt. Otherwise it will " # type: ignore
add_ID.onionr_help += "generate & save a new random key pair." # type: ignore
add_ID.onionr_help = "If the first argument is true, Onionr will show a deterministic generation prompt. Otherwise it will generate & save a new random key pair."
def change_ID():
"""Command to change active ID from argv or stdin."""
key_manager = keymanager.KeyManager()
try:
key = sys.argv[2]
@ -89,14 +114,24 @@ def change_ID():
else:
logger.warn('Invalid key %s' % (key,), terminal=True)
change_ID.onionr_help = "<pubkey>: Switches Onionr to use a different user ID key. You should immediately restart Onionr if it is running."
change_ID.onionr_help = "<pubkey>: Switches Onionr to " # type: ignore
change_ID.onionr_help += "use a different user ID key. " # type: ignore
change_ID.onionr_help += "You should immediately restart " # type: ignore
change_ID.onionr_help += "Onionr if it is running." # type: ignore
def add_vanity():
"""Command to generate menmonic vanity key pair."""
key_manager = keymanager.KeyManager()
tell = lambda tell: logger.info(tell, terminal=True)
def tell(tell):
return logger.info(tell, terminal=True)
words = ''
length = len(sys.argv) - 2
if length == 0: return
if length == 0:
return
for i in range(2, len(sys.argv)):
words += ' '
words += sys.argv[i]
@ -108,12 +143,19 @@ def add_vanity():
try:
vanity = vanityonionr.find_multiprocess(words)
except ValueError:
logger.warn('Vanity words must be valid english bip39', terminal=True)
logger.warn('Vanity words must be valid english bip39',
terminal=True)
else:
b32_pub = unpaddedbase32.b32encode(vanity[0])
tell('Found vanity address:\n' + niceware.bytes_to_passphrase(vanity[0]))
tell('Found vanity address:\n' +
niceware.bytes_to_passphrase(vanity[0]))
tell('Base32 Public key: %s' % (b32_pub.decode(),))
key_manager.addKey(b32_pub, unpaddedbase32.b32encode(vanity[1]))
except KeyboardInterrupt:
pass
add_vanity.onionr_help = "<space separated words> - Generates and stores an Onionr vanity address (see https://github.com/moreati/python-niceware/blob/master/niceware/wordlist.py)"
add_vanity.onionr_help = "<space separated words> - " # type: ignore
add_vanity.onionr_help += "Generates and stores an " # type: ignore
add_vanity.onionr_help += "Onionr vanity address " # type: ignore
add_vanity.onionr_help += "(see is.gd/YklHGe)" # type: ignore

32
src/onionrcommands/resetplugins.py
View File

@ -1,9 +1,13 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Reset default plugins from source
'''
'''
"""
import os
import shutil
from utils import identifyhome
import logger
"""
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
@ -16,21 +20,21 @@
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 os
import shutil
"""
from utils import identifyhome
from onionrsetup import setup_default_plugins
import logger
def reset():
"""Reinstalls Onionr default plugins"""
"""Reinstalls Onionr default plugins."""
home = identifyhome.identify_home()
plugin_dir = home + '/plugins/'
if not os.path.exists(home): return
if os.path.exists(plugin_dir): shutil.rmtree(plugin_dir)
if not os.path.exists(home):
return
if os.path.exists(plugin_dir):
shutil.rmtree(plugin_dir)
logger.info('Default plugins have been reset.', terminal=True)
reset.onionr_help = "reinstalls default Onionr plugins (e.g. mail). Should be done after git pulls or plugin modification."
reset.onionr_help = "reinstalls default Onionr plugins" # type: ignore
reset.onionr_help += "(e.g. mail). Should be done after " # type: ignore
reset.onionr_help += "git pulls or plugin modification." # type: ignore

43
src/onionrcommands/resettor.py
View File

@ -1,9 +1,13 @@
'''
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Command to delete the Tor data directory if its safe to do so
'''
'''
"""
import os
import shutil
import logger
from onionrutils import localcommand
from utils import identifyhome
"""
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
@ -16,27 +20,42 @@
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 os, shutil
import logger
from onionrutils import localcommand
from utils import identifyhome
"""
def __delete(directory):
tor_dir = '%s/%s/' % (identifyhome.identify_home(), directory)
if os.path.exists(tor_dir):
if localcommand.local_command('/ping') == 'pong!':
logger.warn('Cannot delete Tor data while Onionr is running', terminal=True)
logger.warn(
'Cannot delete Tor data while Onionr is running',
terminal=True)
else:
shutil.rmtree(tor_dir)
logger.info('Tor reset', terminal=True)
def reset_tor():
"""Delete tor data directory."""
__delete('tordata')
reset_tor.onionr_help = "Deletes Onionr's Tor data directory. Only do this as a last resort if you have serious Tor issues."
reset_tor.onionr_help = "Deletes Onionr's Tor data directory. " # type: ignore
reset_tor.onionr_help += "Only do this as a last resort if " # type: ignore
reset_tor.onionr_help += "you have serious Tor issues." # type: ignore
def reset_tor_key_pair():
"""Delete Tor HS key pair for our node."""
__delete('hs')
reset_tor_key_pair.onionr_help = "Delete's your Tor node address permanently. Note that through fingerprinting attackers may be able to know that your new generated node address belongs to the same node as the deleted one."
reset_tor_key_pair.onionr_help = "Delete's your Tor " # type: ignore
reset_tor_key_pair.onionr_help += "node address permanently. " # type: ignore
reset_tor_key_pair.onionr_help += "Note that through " # type: ignore
reset_tor_key_pair.onionr_help += "fingerprinting attackers " # type: ignore
reset_tor_key_pair.onionr_help += "may be able to know that " # type: ignore
reset_tor_key_pair.onionr_help += "your new generated node " # type: ignore
reset_tor_key_pair.onionr_help += "address belongs to " # type: ignore
reset_tor_key_pair.onionr_help += "the same node " # type: ignore
reset_tor_key_pair.onionr_help += "as the deleted one." # type: ignore

40
src/onionrcommands/restartonionr.py
View File

@ -1,8 +1,19 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Command to restart Onionr
"""
import time
import os
import subprocess # nosec
import platform
from etc import onionrvalues
from etc import cleanup
from onionrutils import localcommand
import logger
import filepaths
from . import daemonlaunch
"""
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
@ -17,29 +28,22 @@
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
import os
import subprocess
import platform
from etc import onionrvalues
from etc import cleanup
from onionrutils import localcommand
import logger
import filepaths
from . import daemonlaunch
SCRIPT_NAME = os.path.dirname(os.path.realpath(
__file__)) + f'/../../{onionrvalues.SCRIPT_NAME}'
SCRIPT_NAME = os.path.dirname(os.path.realpath(__file__)) + f'/../../{onionrvalues.SCRIPT_NAME}'
def restart():
"""Tell the Onionr daemon to restart."""
logger.info('Restarting Onionr', terminal=True)
# On platforms where we can, fork out to prevent locking
try:
pid = os.fork()
if pid != 0: return
except (AttributeError, OSError) as e:
if pid != 0:
return
except (AttributeError, OSError):
if platform.platform() != 'Windows':
logger.warn('Could not fork on restart')
@ -47,10 +51,12 @@ def restart():
while localcommand.local_command('ping', maxWait=8) == 'pong!':
time.sleep(0.3)
time.sleep(15)
while os.path.exists(filepaths.private_API_host_file) or os.path.exists(filepaths.daemon_mark_file):
while (os.path.exists(filepaths.private_API_host_file) or
(os.path.exists(filepaths.daemon_mark_file))):
time.sleep(1)
cleanup.delete_run_files()
subprocess.Popen([SCRIPT_NAME, 'start'])
restart.onionr_help = 'Gracefully restart Onionr'
restart.onionr_help = 'Gracefully restart Onionr' # type: ignore

24
src/onionrcommands/runtimetestcmd.py
View File

@ -1,6 +1,28 @@
"""Onionr - Private P2P Communication.
Command to tell daemon to do run time tests
"""
from coredb import 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/>.
"""
def do_runtime_test():
"""Send runtime test daemon queue command."""
daemonqueue.daemon_queue_add("runtimeTest")
do_runtime_test.onionr_help = "If Onionr is running, initialize run time tests (check logs)"
do_runtime_test.onionr_help = "If Onionr is running, " # type: ignore
do_runtime_test.onionr_help += "run runtime tests (check logs)" # type: ignore

45
src/onionrcommands/sitecreator.py
View File

@ -1,12 +1,31 @@
"""Onionr - Private P2P Communication.
Command to create Onionr mutli-page sites
"""
import sys
import getpass
from httpapi import onionrsitesapi
import onionrexceptions
import logger
from etc import onionrvalues
"""
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 create_multipage_site():
"""Command to create mutlipage sites with specified dir and password."""
error_encountered = False
try:
directory = sys.argv[2]
@ -16,8 +35,13 @@ def create_multipage_site():
passphrase = sys.argv[3]
except IndexError:
logger.warn('''It is critical that this passphrase is long.
If you want to update your site later you must remember the passphrase.''', terminal=True)
passphrase = getpass.getpass(f'Please enter a site passphrase of at least {onionrvalues.PASSWORD_LENGTH} characters.')
If you want to update your site later you must remember the passphrase.''',
terminal=True)
passphrase = getpass.getpass(
'Please enter a site passphrase of at least ' +
onionrvalues.PASSWORD_LENGTH + ' characters.')
confirm = getpass.getpass('Confirm passphrase:')
if passphrase != confirm:
logger.error('Passphrases do not match', terminal=True)
@ -25,14 +49,23 @@ If you want to update your site later you must remember the passphrase.''', term
if len(passphrase) < onionrvalues.PASSWORD_LENGTH:
error_encountered = True
logger.error(f'Passphrase must be at least {onionrvalues.PASSWORD_LENGTH} characters.', terminal=True)
logger.error(
f'Passphrase must be at least {onionrvalues.PASSWORD_LENGTH}' +
'characters.', terminal=True)
if error_encountered:
sys.exit(1)
results = onionrsitesapi.sitefiles.create_site(passphrase, directory=directory)
results = onionrsitesapi.sitefiles.create_site(
passphrase, directory=directory)
results = (results[0].replace('=', ''), results[1])
logger.info(f'Site address {results[0]}', terminal=True)
logger.info(f'Block for this version {results[1]}', terminal=True)
create_multipage_site.onionr_help = "[directory path (default relative)] - packages a whole directory and makes it available as an Onionr site."
create_multipage_site.onionr_help = "[directory path " # type: ignore
create_multipage_site.onionr_help += "(default relative)] " # type: ignore
create_multipage_site.onionr_help += "- packages a whole " # type: ignore
create_multipage_site.onionr_help += "directory and makes " # type: ignore
create_multipage_site.onionr_help += "it available as " # type: ignore
create_multipage_site.onionr_help += "an Onionr site." # type: ignore

33
src/onionrcommands/softreset.py
View File

@ -1,8 +1,15 @@
"""
Onionr - Private P2P Communication
"""Onionr - Private P2P Communication.
Command to soft-reset Onionr (deletes blocks)
"""
import os
import shutil
from onionrutils import localcommand
from coredb import dbfiles
import filepaths
from onionrplugins import onionrevents
import logger
"""
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
@ -17,14 +24,7 @@
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 os
import shutil
from onionrutils import localcommand
from coredb import dbfiles
import filepaths
from onionrplugins import onionrevents
import logger
def _ignore_not_found_delete(path):
try:
@ -32,9 +32,15 @@ def _ignore_not_found_delete(path):
except FileNotFoundError:
pass
def soft_reset():
"""Command to soft reset Onionr home data.
Onionr must not be running
"""
if localcommand.local_command('/ping') == 'pong!':
logger.warn('Cannot soft reset while Onionr is running', terminal=True)
logger.warn('Cannot soft reset while Onionr is running',
terminal=True)
return
path = filepaths.block_data_location
shutil.rmtree(path)
@ -43,4 +49,9 @@ def soft_reset():
onionrevents.event('softreset')
logger.info("Soft reset Onionr", terminal=True)
soft_reset.onionr_help = "Deletes Onionr blocks and their associated metadata, except for any exported block files. Does NOT delete data on other nodes in the network."
soft_reset.onionr_help = "Deletes Onionr blocks and their " # type: ignore
soft_reset.onionr_help += "associated metadata, except for " # type: ignore
soft_reset.onionr_help += "any exported block files. Does NOT " # type: ignore
soft_reset.onionr_help += "delete data on " # type: ignore
soft_reset.onionr_help += "other nodes in the network." # type: ignore

42
src/onionrcommands/version.py
View File

@ -1,13 +1,33 @@
"""Onionr - Private P2P Communication.
Command to show version info
"""
import platform
from utils import identifyhome
from etc import onionrvalues
import logger
def version(verbosity = 5, function = logger.info):
'''
Displays the Onionr version
'''
"""
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.
function('Onionr v%s (%s) (API v%s)' % (onionrvalues.ONIONR_VERSION, platform.machine(), onionrvalues.API_VERSION), terminal=True)
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 version(verbosity=5, function=logger.info):
"""Display the Onionr version."""
function('Onionr v%s (%s) (API v%s)' % (onionrvalues.ONIONR_VERSION,
platform.machine(),
onionrvalues.API_VERSION),
terminal=True)
if verbosity >= 1:
function(onionrvalues.ONIONR_TAGLINE, terminal=True)
if verbosity >= 2:
@ -15,7 +35,13 @@ def version(verbosity = 5, function = logger.info):
release = platform.release()
python_imp = platform.python_implementation()
python_version = platform.python_version()
function(f'{python_imp} {python_version} on {pf} {release}', terminal=True)
function('Onionr data dir: %s' % identifyhome.identify_home(), terminal=True)
function(
f'{python_imp} {python_version} on {pf} {release}',
terminal=True)
function('Onionr data dir: %s' %
identifyhome.identify_home(), terminal=True)
version.onionr_help = 'Shows environment details including Onionr version & data directory, OS and Python version'
version.onionr_help = 'Shows environment details including ' # type: ignore
version.onionr_help += 'Onionr version & data directory, ' # type: ignore
version.onionr_help += 'OS and Python version' # type: ignore