mirror of
https://github.com/ElementsProject/lightning.git
synced 2024-11-19 09:54:16 +01:00
77b7a74cd6
`urllib3` does not ship as built-in with any of the recent python releases, whereas `urllib` does, and for the uses we have, they are pretty much identical.
736 lines
27 KiB
Python
Executable File
736 lines
27 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from subprocess import Popen, PIPE
|
|
import sys
|
|
import json
|
|
import os
|
|
import argparse
|
|
from pathlib import Path, PosixPath
|
|
import shutil
|
|
import tempfile
|
|
from typing import Union
|
|
from urllib.parse import urlparse
|
|
from urllib.request import urlopen
|
|
|
|
|
|
repos = ['https://github.com/lightningd/plugins']
|
|
|
|
|
|
def py_entry_guesses(name):
|
|
return [name, f'{name}.py', '__init__.py']
|
|
|
|
|
|
def unsupported_entry(name):
|
|
return [f'{name}.go', f'{name}.sh']
|
|
|
|
|
|
class InstInfo:
|
|
def __init__(self, name, url, git_url):
|
|
self.name = name
|
|
self.repo = url # Used for 'git clone'
|
|
self.git_url = git_url # API access for github repos
|
|
self.entry = None
|
|
self.deps = None
|
|
self.subdir = None
|
|
self.commit = None
|
|
|
|
def __repr__(self):
|
|
return (f'InstInfo({self.name}, {self.repo}, {self.git_url}'
|
|
f'{self.entry}, {self.deps})')
|
|
|
|
def get_inst_details(self):
|
|
"""
|
|
Populate installation details from a github repo url.
|
|
Return True if all data is found.
|
|
"""
|
|
r = urlopen(self.git_url, timeout=5)
|
|
if r.status != 200:
|
|
return False
|
|
if 'git/tree' in self.git_url:
|
|
tree = json.loads(r.read().decode())['tree']
|
|
else:
|
|
tree = json.loads(r.read().decode())
|
|
entry_guesses = py_entry_guesses(self.name)
|
|
for g in entry_guesses:
|
|
for f in tree:
|
|
if f['path'] == g:
|
|
self.entry = g
|
|
break
|
|
if self.entry is not None:
|
|
break
|
|
if self.entry is None:
|
|
for g in unsupported_entry(self.name):
|
|
for f in tree:
|
|
if f['path'] == g:
|
|
# FIXME: This should be easier to implement
|
|
print(f'entrypoint {g} is not yet supported')
|
|
return False
|
|
dependency_info = ['requirements.txt', 'pyproject.toml']
|
|
for d in dependency_info:
|
|
for f in tree:
|
|
if f['path'] == d:
|
|
self.deps = d
|
|
break
|
|
if self.deps is not None:
|
|
break
|
|
if not self.entry:
|
|
return False
|
|
if not self.deps:
|
|
return False
|
|
return True
|
|
|
|
|
|
def create_dir(r: int, directory: PosixPath) -> bool:
|
|
"""Creation of a directory at path `d` with a maximum new dir depth `r`"""
|
|
if directory.exists():
|
|
return True
|
|
if r <= 0:
|
|
return False
|
|
if create_dir(r-1, directory.parent):
|
|
os.mkdir(directory, 0o777)
|
|
print(f'created directory {directory}')
|
|
assert directory.exists()
|
|
return True
|
|
|
|
|
|
def remove_dir(target: str) -> bool:
|
|
try:
|
|
shutil.rmtree(target)
|
|
return True
|
|
except NotADirectoryError:
|
|
print(f"Tried to remove directory {target} that does not exist.")
|
|
except PermissionError:
|
|
print(f"Permission denied removing dir: {target}")
|
|
return False
|
|
|
|
|
|
class Config():
|
|
"""A generic class for procuring, reading and editing config files"""
|
|
def obtain_config(self,
|
|
config_path: str,
|
|
default_text: str,
|
|
warn: bool = False) -> str:
|
|
"""Return a config file from the desired location. Create one with
|
|
default_text if it cannot be found."""
|
|
if isinstance(config_path, type(None)):
|
|
raise Exception("Generic config must be passed a config_path.")
|
|
assert isinstance(config_path, str)
|
|
# FIXME: warn if reckless dir exists, but conf not found
|
|
if Path(config_path).exists():
|
|
with open(config_path, 'r+') as f:
|
|
config_content = f.readlines()
|
|
return config_content
|
|
print(f'config file not found: {config_path}')
|
|
if warn:
|
|
confirm = input('press [Y] to create one now.\n').upper() == 'Y'
|
|
else:
|
|
confirm = True
|
|
if not confirm:
|
|
sys.exit(1)
|
|
parent_path = Path(config_path).parent
|
|
# Create up to one parent in the directory tree.
|
|
if create_dir(1, parent_path):
|
|
with open(self.conf_fp, 'w') as f:
|
|
f.write(default_text)
|
|
# FIXME: Handle write failure
|
|
return default_text
|
|
else:
|
|
verbose(f'could not create the parent directory {parent_path}')
|
|
raise FileNotFoundError('invalid parent directory')
|
|
|
|
def editConfigFile(self, addline: str, removeline: str):
|
|
remove_these_lines = []
|
|
with open(self.conf_fp, 'r') as reckless_conf:
|
|
original = reckless_conf.readlines()
|
|
empty_lines = []
|
|
for n, l in enumerate(original):
|
|
if l.strip() == removeline:
|
|
remove_these_lines.append(n)
|
|
continue
|
|
if l.strip() == '':
|
|
empty_lines.append(n)
|
|
if n-1 in empty_lines:
|
|
# The white space is getting excessive.
|
|
remove_these_lines.append(n)
|
|
continue
|
|
with open(self.conf_fp, 'w') as conf_write:
|
|
# no need to write if passed 'None'
|
|
line_exists = not bool(addline)
|
|
for n, l in enumerate(original):
|
|
if n not in remove_these_lines:
|
|
if n > 0:
|
|
conf_write.write(f'\n{l.strip()}')
|
|
else:
|
|
conf_write.write(l.strip())
|
|
if addline == l:
|
|
# addline is idempotent
|
|
line_exists = True
|
|
if not line_exists:
|
|
conf_write.write(f'\n{addline}')
|
|
|
|
def __init__(self, path: Union[str, None] = None,
|
|
default_text: Union[str, None] = None,
|
|
warn: bool = False):
|
|
assert path is not None
|
|
assert default_text is not None
|
|
self.conf_fp = path
|
|
self.content = self.obtain_config(self.conf_fp, default_text,
|
|
warn=warn)
|
|
|
|
|
|
class RecklessConfig(Config):
|
|
"""Reckless config (by default, specific to the bitcoin network only.)
|
|
This is inherited by the lightningd config and contains all reckless
|
|
maintained plugins."""
|
|
|
|
def enable_plugin(self, plugin_path: str):
|
|
"""Handle persistent plugin loading via config"""
|
|
self.editConfigFile(f'plugin={plugin_path}',
|
|
f'disable-plugin={plugin_path}')
|
|
|
|
def disable_plugin(self, plugin_path: str):
|
|
"""Handle persistent plugin disabling via config"""
|
|
self.editConfigFile(f'disable-plugin={plugin_path}',
|
|
f'plugin={plugin_path}')
|
|
|
|
def __init__(self, path: Union[str, None] = None,
|
|
default_text: Union[str, None] = None):
|
|
if path is None:
|
|
path = Path(LIGHTNING_DIR) / 'reckless' / 'bitcoin-reckless.conf'
|
|
if default_text is None:
|
|
default_text = (
|
|
'# This configuration file is managed by reckless to activate '
|
|
'and disable\n# reckless-installed plugins\n\n'
|
|
)
|
|
Config.__init__(self, path=str(path), default_text=default_text)
|
|
self.reckless_dir = Path(path).parent
|
|
|
|
|
|
class LightningBitcoinConfig(Config):
|
|
"""lightningd config specific to the bitcoin network. This is inherited by
|
|
the main lightningd config and in turn, inherits bitcoin-reckless.conf."""
|
|
|
|
def __init__(self, path: Union[str, None] = None,
|
|
default_text: Union[str, None] = None,
|
|
warn: bool = True):
|
|
if path is None:
|
|
path = Path(LIGHTNING_DIR).joinpath('bitcoin', 'config')
|
|
if default_text is None:
|
|
default_text = "# This config was autopopulated by reckless\n\n"
|
|
Config.__init__(self, path=str(path),
|
|
default_text=default_text, warn=warn)
|
|
|
|
|
|
class InferInstall():
|
|
"""Once a plugin is installed, we may need its directory and entrypoint"""
|
|
def __init__(self, name: str):
|
|
reck_contents = os.listdir(RECKLESS_CONFIG.reckless_dir)
|
|
if name[-3:] == '.py':
|
|
name = name[:-3]
|
|
if name in reck_contents:
|
|
self.dir = Path(RECKLESS_CONFIG.reckless_dir).joinpath(name)
|
|
else:
|
|
raise Exception(f"Could not find a reckless directory for {name}")
|
|
plug_dir = Path(RECKLESS_CONFIG.reckless_dir).joinpath(name)
|
|
for guess in py_entry_guesses(name):
|
|
for content in plug_dir.iterdir():
|
|
if content.name == guess:
|
|
self.entry = str(content)
|
|
self.name = guess
|
|
return
|
|
raise Exception(f'plugin entrypoint not found in {self.dir}')
|
|
|
|
|
|
def help_alias(targets: list):
|
|
if len(targets) == 0:
|
|
parser.print_help(sys.stdout)
|
|
else:
|
|
print('try "reckless {} -h"'.format(' '.join(targets)))
|
|
sys.exit(1)
|
|
|
|
|
|
def verbose(*args):
|
|
if not IS_VERBOSE:
|
|
return
|
|
print(*args)
|
|
|
|
|
|
def _search_repo(name: str, url: str) -> InstInfo:
|
|
"""look in given repo and, if found, populate InstInfo"""
|
|
# Remove api subdomain, subdirectories, etc.
|
|
repo = url.split('/')
|
|
while '' in repo:
|
|
repo.remove('')
|
|
repo_name = None
|
|
parsed_url = urlparse(url)
|
|
if 'github.com' not in parsed_url.netloc:
|
|
# FIXME: Handle non-github repos.
|
|
return False
|
|
if len(parsed_url.path.split('/')) < 2:
|
|
return False
|
|
start = 1
|
|
# Maybe we were passed an api.github.com/repo/<user> url
|
|
if 'api' in parsed_url.netloc:
|
|
start += 1
|
|
repo_user = parsed_url.path.split('/')[start]
|
|
repo_name = parsed_url.path.split('/')[start + 1]
|
|
|
|
# Get details from the github API.
|
|
api_url = f'https://api.github.com/repos/{repo_user}/{repo_name}/contents/'
|
|
plugins_cont = api_url
|
|
r = urlopen(plugins_cont, timeout=5)
|
|
if r.status != 200:
|
|
print("Plugin repository unavailable")
|
|
return False
|
|
# Repo is for this plugin
|
|
if repo_name == name:
|
|
MyPlugin = InstInfo(name,
|
|
f'https://github.com/{repo_user}/{repo_name}',
|
|
api_url)
|
|
if not MyPlugin.get_inst_details():
|
|
return False
|
|
return MyPlugin
|
|
# Repo contains multiple plugins?
|
|
for x in json.loads(r.read().decode()):
|
|
if x["name"] == name:
|
|
# Look for the rest of the install details
|
|
# These are in lightningd/plugins directly
|
|
if 'lightningd/plugins/' in x['html_url']:
|
|
MyPlugin = InstInfo(name,
|
|
'https://github.com/lightningd/plugins',
|
|
x['git_url'])
|
|
MyPlugin.subdir = x['name']
|
|
# submodules from another github repo
|
|
else:
|
|
MyPlugin = InstInfo(name, x['html_url'], x['git_url'])
|
|
# Submodule URLs are appended with /tree/<commit hash>
|
|
if MyPlugin.repo.split('/')[-2] == 'tree':
|
|
MyPlugin.commit = MyPlugin.repo.split('/')[-1]
|
|
MyPlugin.repo = MyPlugin.repo.split('/tree/')[0]
|
|
verbose(f'repo using commit: {MyPlugin.commit}')
|
|
if not MyPlugin.get_inst_details():
|
|
return False
|
|
return MyPlugin
|
|
return False
|
|
|
|
|
|
def _install_plugin(src: InstInfo) -> bool:
|
|
"""make sure the repo exists and clone it."""
|
|
verbose(f'Install requested from {src}.')
|
|
if RECKLESS_CONFIG is None:
|
|
print('error: reckless install directory unavailable')
|
|
sys.exit(2)
|
|
|
|
# FIXME: This request seems rather pointless
|
|
req = urlopen(src.repo, timeout=10)
|
|
if not req.status == 200:
|
|
print('plugin source repository unavailable')
|
|
sys.exit(1)
|
|
# Use a unique directory for each cloned repo.
|
|
clone_path = 'reckless-{}'.format(str(hash(os.times()))[-9:])
|
|
clone_path = Path(tempfile.gettempdir()) / clone_path
|
|
inst_path = Path(RECKLESS_CONFIG.reckless_dir) / src.name
|
|
if Path(clone_path).exists():
|
|
verbose(f'{clone_path} already exists - deleting')
|
|
shutil.rmtree(clone_path)
|
|
# clone git repository to /tmp/reckless-...
|
|
if ('http' in src.repo[:4]) or ('github.com' in src.repo):
|
|
# Ugly, but interactively handling stderr gets hairy.
|
|
if IS_VERBOSE:
|
|
git = Popen(['git', 'clone', src.repo, str(clone_path)],
|
|
stdout=PIPE)
|
|
else:
|
|
git = Popen(['git', 'clone', src.repo, str(clone_path)],
|
|
stdout=PIPE, stderr=PIPE)
|
|
git.wait()
|
|
if git.returncode != 0:
|
|
if git.stderr:
|
|
print(git.stderr.read().decode())
|
|
if Path(clone_path).exists():
|
|
remove_dir(clone_path)
|
|
print('Error: Failed to clone repo')
|
|
return False
|
|
plugin_path = clone_path
|
|
if src.subdir is not None:
|
|
plugin_path = Path(clone_path) / src.subdir
|
|
if src.commit:
|
|
verbose(f"Checking out commit {src.commit}")
|
|
checkout = Popen(['git', 'checkout', src.commit],
|
|
cwd=plugin_path, stdout=PIPE, stderr=PIPE)
|
|
checkout.wait()
|
|
if checkout.returncode != 0:
|
|
print(f'failed to checkout referenced commit {src.commit}')
|
|
return False
|
|
|
|
# Install dependencies via requirements.txt or pyproject.toml
|
|
mypip = 'pip3' if shutil.which('pip3') else 'pip'
|
|
if not shutil.which(mypip):
|
|
raise Exception(f'{mypip} not found in PATH')
|
|
install_methods = {
|
|
'requirements.txt': [mypip, 'install', '-r', 'requirements.txt'],
|
|
'pyproject.toml': [mypip, 'install', '-e', '.']
|
|
}
|
|
|
|
if src.deps is not None:
|
|
verbose(f'installing dependencies using {src.deps}')
|
|
procedure = install_methods[src.deps]
|
|
pip = Popen(procedure, cwd=plugin_path, stdout=PIPE)
|
|
pip.wait()
|
|
if pip.returncode == 0:
|
|
print('dependencies installed successfully')
|
|
else:
|
|
print('error encountered installing dependencies')
|
|
verbose(pip.stdout.read())
|
|
return False
|
|
test = Popen([Path(plugin_path).joinpath(src.entry)], cwd=plugin_path,
|
|
stdout=PIPE, stderr=PIPE, universal_newlines=True)
|
|
test_log = []
|
|
with test.stderr:
|
|
for line in test.stderr:
|
|
test_log.append(line.strip('\n'))
|
|
test.wait()
|
|
# FIXME: add noexec test/warning. Maybe try chmod entrypoint.
|
|
if test.returncode != 0:
|
|
verbose("plugin testing error:")
|
|
for line in test_log:
|
|
verbose(f' {line}')
|
|
print('plugin testing failed')
|
|
return False
|
|
|
|
# Find this cute little plugin a forever home
|
|
shutil.copytree(plugin_path, inst_path)
|
|
print(f'plugin installed: {inst_path}')
|
|
remove_dir(clone_path)
|
|
return True
|
|
|
|
|
|
def install(plugin_name: str):
|
|
"""downloads plugin from source repos, installs and activates plugin"""
|
|
assert isinstance(plugin_name, str)
|
|
src = search(plugin_name)
|
|
if src:
|
|
verbose(f'Retrieving {plugin_name} from {src.repo}')
|
|
if not _install_plugin(src):
|
|
print('installation aborted')
|
|
sys.exit(1)
|
|
inst_path = Path(RECKLESS_CONFIG.reckless_dir) / src.name / src.entry
|
|
RECKLESS_CONFIG.enable_plugin(inst_path)
|
|
enable(plugin_name)
|
|
|
|
|
|
def uninstall(plugin_name: str):
|
|
"""disables plugin and deletes the plugin's reckless dir"""
|
|
assert isinstance(plugin_name, str)
|
|
verbose(f'Uninstalling plugin {plugin_name}')
|
|
disable(plugin_name)
|
|
plugin_dir = Path(RECKLESS_CONFIG.reckless_dir) / plugin_name
|
|
verbose(f'looking for {plugin_dir}')
|
|
if remove_dir(plugin_dir):
|
|
print(f"{plugin_name} uninstalled successfully.")
|
|
|
|
|
|
def search(plugin_name: str) -> InstInfo:
|
|
"""searches plugin index for plugin"""
|
|
ordered_repos = RECKLESS_SOURCES
|
|
for r in RECKLESS_SOURCES:
|
|
# Search repos named after the plugin first
|
|
if r.split('/')[-1].lower() == plugin_name.lower():
|
|
ordered_repos.remove(r)
|
|
ordered_repos.insert(0, r)
|
|
for r in ordered_repos:
|
|
p = _search_repo(plugin_name, r)
|
|
if p:
|
|
print(f"found {p.name} in repo: {p.repo}")
|
|
verbose(f"entry: {p.entry}")
|
|
if p.subdir:
|
|
verbose(f'sub-directory: {p.subdir}')
|
|
return p
|
|
print(f'Unable to locate source for plugin {plugin_name}')
|
|
|
|
|
|
class RPCError(Exception):
|
|
"""lightning-cli fails to connect to lightningd RPC"""
|
|
def __init__(self, err):
|
|
self.err = err
|
|
|
|
def __str__(self):
|
|
return 'RPCError({self.err})'
|
|
|
|
|
|
class CLIError(Exception):
|
|
"""lightningd error response"""
|
|
def __init__(self, code, message):
|
|
self.code = code
|
|
self.message = message
|
|
|
|
def __str__(self):
|
|
return f'CLIError({self.code} {self.message})'
|
|
|
|
|
|
def lightning_cli(*args, timeout=15) -> dict:
|
|
# CLI commands will be added to any necessary options
|
|
cmd = LIGHTNING_CLI_CALL.copy()
|
|
cmd.extend(args)
|
|
clncli = Popen(cmd, stdout=PIPE, stderr=PIPE)
|
|
clncli.wait(timeout=timeout)
|
|
out = clncli.stdout.read().decode()
|
|
if len(out) > 0 and out[0] == '{':
|
|
# If all goes well, a json object is typically returned
|
|
out = json.loads(out.replace('\n', ''))
|
|
else:
|
|
# help, -V, etc. may not return json, so stash it here.
|
|
out = {'content': out}
|
|
if clncli.returncode == 0:
|
|
return out
|
|
if clncli.returncode == 1:
|
|
# RPC doesn't like our input
|
|
# output contains 'code' and 'message'
|
|
raise CLIError(out['code'], out['message'])
|
|
if clncli.returncode == 2:
|
|
# RPC not available - lightningd not running or using alternate config
|
|
err = clncli.stderr.read().decode()
|
|
raise RPCError(err)
|
|
|
|
|
|
def enable(plugin_name: str):
|
|
"""dynamically activates plugin and adds to config (persistent)"""
|
|
assert isinstance(plugin_name, str)
|
|
inst = InferInstall(plugin_name)
|
|
path = inst.entry
|
|
if not Path(path).exists():
|
|
print(f'cannot find installed plugin at expected path {path}')
|
|
sys.exit(1)
|
|
verbose(f'activating {plugin_name}')
|
|
try:
|
|
lightning_cli('plugin', 'start', path)
|
|
except CLIError as err:
|
|
if 'already registered' in err.message:
|
|
verbose(f'{inst.name} is already running')
|
|
else:
|
|
print(f'reckless: {inst.name} failed to start!')
|
|
raise err
|
|
except RPCError:
|
|
verbose('lightningd rpc unavailable. Skipping dynamic activation.')
|
|
RECKLESS_CONFIG.enable_plugin(path)
|
|
print(f'{plugin_name} enabled')
|
|
|
|
|
|
def disable(plugin_name: str):
|
|
"""reckless disable <plugin>
|
|
deactivates an installed plugin"""
|
|
assert isinstance(plugin_name, str)
|
|
inst = InferInstall(plugin_name)
|
|
path = inst.entry
|
|
if not Path(path).exists():
|
|
sys.stderr.write(f'Could not find plugin at {path}\n')
|
|
sys.exit(1)
|
|
verbose(f'deactivating {plugin_name}')
|
|
try:
|
|
lightning_cli('plugin', 'stop', path)
|
|
except CLIError as err:
|
|
if err.code == -32602:
|
|
verbose('plugin not currently running')
|
|
else:
|
|
print('lightning-cli plugin stop failed')
|
|
raise err
|
|
except RPCError:
|
|
verbose('lightningd rpc unavailable. Skipping dynamic deactivation.')
|
|
RECKLESS_CONFIG.disable_plugin(path)
|
|
print(f'{plugin_name} disabled')
|
|
|
|
|
|
def load_config(reckless_dir: Union[str, None] = None,
|
|
network: str = 'bitcoin') -> Config:
|
|
"""Initial directory discovery and config file creation."""
|
|
net_conf = None
|
|
# Does the lightning-cli already reference an explicit config?
|
|
try:
|
|
active_config = lightning_cli('listconfigs', timeout=3)
|
|
if 'conf' in active_config:
|
|
net_conf = LightningBitcoinConfig(path=active_config['conf'])
|
|
except RPCError:
|
|
pass
|
|
if reckless_dir is None:
|
|
reckless_dir = Path(LIGHTNING_DIR) / 'reckless'
|
|
else:
|
|
if not os.path.isabs(reckless_dir):
|
|
reckless_dir = Path.cwd() / reckless_dir
|
|
if LIGHTNING_CONFIG:
|
|
network_path = LIGHTNING_CONFIG
|
|
else:
|
|
network_path = Path(LIGHTNING_DIR) / network / 'config'
|
|
reck_conf_path = Path(reckless_dir) / f'{network}-reckless.conf'
|
|
if net_conf:
|
|
if str(network_path) != net_conf.conf_fp:
|
|
print('error: reckless configuration does not match lightningd:\n'
|
|
f'reckless network config path: {network_path}\n'
|
|
f'lightningd active config: {net_conf.conf_fp}')
|
|
sys.exit(1)
|
|
else:
|
|
# The network-specific config file (bitcoin by default)
|
|
net_conf = LightningBitcoinConfig(path=network_path)
|
|
# Reckless manages plugins here.
|
|
try:
|
|
reckless_conf = RecklessConfig(path=reck_conf_path)
|
|
except FileNotFoundError:
|
|
print('Error: reckless config file could not be written: ',
|
|
str(reck_conf_path))
|
|
sys.exit(1)
|
|
if not net_conf:
|
|
print('Error: could not load or create the network specific lightningd'
|
|
' config (default .lightning/bitcoin)')
|
|
sys.exit(1)
|
|
net_conf.editConfigFile(f'include {reckless_conf.conf_fp}', None)
|
|
return reckless_conf
|
|
|
|
|
|
def get_sources_file() -> str:
|
|
return Path(RECKLESS_DIR) / '.sources'
|
|
|
|
|
|
def sources_from_file() -> list:
|
|
sources_file = get_sources_file()
|
|
read_sources = []
|
|
with open(sources_file, 'r') as f:
|
|
for src in f.readlines():
|
|
if len(src.strip()) > 0:
|
|
read_sources.append(src.strip())
|
|
return read_sources
|
|
|
|
|
|
def loadSources() -> list:
|
|
"""Look for the repo sources file."""
|
|
sources_file = get_sources_file()
|
|
# This would have been created if possible
|
|
if not Path(sources_file).exists():
|
|
verbose('Warning: Reckless requires write access')
|
|
Config(path=str(sources_file),
|
|
default_text='https://github.com/lightningd/plugins')
|
|
return ['https://github.com/lightningd/plugins']
|
|
return sources_from_file()
|
|
|
|
|
|
def add_source(src: str):
|
|
"""Additional git repositories, directories, etc. are passed here."""
|
|
assert isinstance(src, str)
|
|
# Is it a file?
|
|
maybe_path = os.path.realpath(src)
|
|
if Path(maybe_path).exists():
|
|
# FIXME: This should handle either a directory or a git repo
|
|
if os.path.isdir(maybe_path):
|
|
print(f'Plugin source directory found: {maybe_path}')
|
|
elif 'github.com' in src:
|
|
my_file = Config(path=str(get_sources_file()),
|
|
default_text='https://github.com/lightningd/plugins')
|
|
my_file.editConfigFile(src, None)
|
|
|
|
|
|
def remove_source(src: str):
|
|
"""Remove a source from the sources file."""
|
|
assert isinstance(src, str)
|
|
if src in sources_from_file():
|
|
my_file = Config(path=get_sources_file(),
|
|
default_text='https://github.com/lightningd/plugins')
|
|
my_file.editConfigFile(None, src)
|
|
print('plugin source removed')
|
|
else:
|
|
print(f'source not found: {src}')
|
|
|
|
|
|
def list_source():
|
|
"""Provide the user with all stored source repositories."""
|
|
for src in sources_from_file():
|
|
print(src)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser()
|
|
# This default depends on the .lightning directory
|
|
parser.add_argument('-d', '--reckless-dir',
|
|
help='specify a data directory for reckless to use',
|
|
type=str, default=None)
|
|
parser.add_argument('-l', '--lightning',
|
|
help='lightning data directory (default:~/.lightning)',
|
|
type=str,
|
|
default=Path.home().joinpath('.lightning'))
|
|
parser.add_argument('-c', '--conf',
|
|
help=' config file used by lightningd',
|
|
type=str,
|
|
default=None)
|
|
parser.add_argument('-r', '--regtest', action='store_true')
|
|
parser.add_argument('-v', '--verbose', action='store_true')
|
|
cmd1 = parser.add_subparsers(dest='cmd1', help='command',
|
|
required=True)
|
|
|
|
install_cmd = cmd1.add_parser('install', help='search for and install a '
|
|
'plugin, then test and activate')
|
|
install_cmd.add_argument('targets', type=str, nargs='*')
|
|
install_cmd.set_defaults(func=install)
|
|
|
|
uninstall_cmd = cmd1.add_parser('uninstall', help='deactivate a plugin '
|
|
'and remove it from the directory')
|
|
uninstall_cmd.add_argument('targets', type=str, nargs='*')
|
|
uninstall_cmd.set_defaults(func=uninstall)
|
|
|
|
search_cmd = cmd1.add_parser('search', help='search for a plugin from '
|
|
'the available source repositories')
|
|
search_cmd.add_argument('targets', type=str, nargs='*')
|
|
search_cmd.set_defaults(func=search)
|
|
|
|
enable_cmd = cmd1.add_parser('enable', help='dynamically enable a plugin '
|
|
'and update config')
|
|
enable_cmd.add_argument('targets', type=str, nargs='*')
|
|
enable_cmd.set_defaults(func=enable)
|
|
disable_cmd = cmd1.add_parser('disable', help='disable a plugin')
|
|
disable_cmd.add_argument('targets', type=str, nargs='*')
|
|
disable_cmd.set_defaults(func=disable)
|
|
source_parser = cmd1.add_parser('source', help='manage plugin search '
|
|
'sources')
|
|
source_subs = source_parser.add_subparsers(dest='source_subs',
|
|
required=True)
|
|
list_parse = source_subs.add_parser('list', help='list available plugin '
|
|
'sources (repositories)')
|
|
list_parse.set_defaults(func=list_source)
|
|
source_add = source_subs.add_parser('add', help='add a source repository')
|
|
source_add.add_argument('targets', type=str, nargs='*')
|
|
source_add.set_defaults(func=add_source)
|
|
source_rem = source_subs.add_parser('remove', aliases=['rem', 'rm'],
|
|
help='remove a plugin source '
|
|
'repository')
|
|
source_rem.add_argument('targets', type=str, nargs='*')
|
|
source_rem.set_defaults(func=remove_source)
|
|
|
|
help_cmd = cmd1.add_parser('help', help='for contextual help, use '
|
|
'"reckless <cmd> -h"')
|
|
help_cmd.add_argument('targets', type=str, nargs='*')
|
|
help_cmd.set_defaults(func=help_alias)
|
|
|
|
args = parser.parse_args()
|
|
|
|
NETWORK = 'regtest' if args.regtest else 'bitcoin'
|
|
LIGHTNING_DIR = Path(args.lightning)
|
|
LIGHTNING_CLI_CALL = ['lightning-cli']
|
|
if NETWORK != 'bitcoin':
|
|
LIGHTNING_CLI_CALL.append(f'--network={NETWORK}')
|
|
if LIGHTNING_DIR != Path.home().joinpath('.lightning'):
|
|
LIGHTNING_CLI_CALL.append(f'--lightning-dir={LIGHTNING_DIR}')
|
|
if args.reckless_dir:
|
|
RECKLESS_DIR = args.reckless_dir
|
|
else:
|
|
RECKLESS_DIR = Path(LIGHTNING_DIR) / 'reckless'
|
|
LIGHTNING_CONFIG = args.conf
|
|
RECKLESS_CONFIG = load_config(reckless_dir=RECKLESS_DIR,
|
|
network=NETWORK)
|
|
RECKLESS_SOURCES = loadSources()
|
|
IS_VERBOSE = bool(args.verbose)
|
|
|
|
if 'targets' in args:
|
|
# FIXME: Catch missing argument
|
|
if args.func.__name__ == 'help_alias':
|
|
args.func(args.targets)
|
|
sys.exit(0)
|
|
for target in args.targets:
|
|
args.func(target)
|
|
else:
|
|
args.func()
|