core-lightning/tools/reckless

#!/usr/bin/env python3

import sys
import argparse
import copy
import datetime
from enum import Enum
import json
import logging
import os
from pathlib import Path, PosixPath
import shutil
from subprocess import Popen, PIPE, TimeoutExpired, run
import tempfile
import time
from typing import Union
from urllib.parse import urlparse
from urllib.request import urlopen


logging.basicConfig(
    level=logging.DEBUG,
    format='[%(asctime)s] %(levelname)s: %(message)s',
    handlers=[logging.StreamHandler(stream=sys.stdout)],
)


repos = ['https://github.com/lightningd/plugins']


def py_entry_guesses(name) -> list:
    return [name, f'{name}.py', '__init__.py']


def unsupported_entry(name) -> list:
    return [f'{name}.go', f'{name}.sh']


def entry_guesses(name: str) -> list:
    guesses = []
    for inst in INSTALLERS:
        for entry in inst.entries:
            guesses.append(entry.format(name=name))
    return guesses


class Installer:
    '''
    The identification of a plugin language, compiler or interpreter
    availability, and the install procedures.
    '''
    def __init__(self, name: str, mimetype: str,
                 exe: Union[str, None] = None,
                 compiler: Union[str, None] = None,
                 manager: Union[str, None] = None,
                 entry: Union[str, None] = None):
        self.name = name
        self.mimetype = mimetype
        self.entries = []
        if entry:
            self.entries.append(entry)
        self.exe = exe            # interpreter (if required)
        self.compiler = compiler  # compiler bin (if required)
        self.manager = manager    # dependency manager (if required)
        self.dependency_file = None
        self.dependency_call = None

    def __repr__(self):
        return (f'<Installer {self.name}: mimetype: {self.mimetype}, '
                f'exe: {self.exe}, manager: {self.manager}>')

    def executable(self) -> bool:
        '''Validate the necessary bins are available to execute the plugin.'''
        if self.exe:
            if shutil.which(self.exe):
                # This should arguably not be checked here.
                if self.manager:
                    if shutil.which(self.manager):
                        return True
                    return False
                return True
            return False
        return True

    def installable(self) -> bool:
        '''Validate the necessary compiler and package manager executables are
        available to install.  If these are defined, they are considered
        mandatory even though the user may have the requisite packages already
        installed.'''
        if self.compiler and not shutil.which(self.compiler):
            return False
        if self.manager and not shutil.which(self.manager):
            return False
        return True

    def add_entrypoint(self, entry: str):
        assert isinstance(entry, str)
        self.entries.append(entry)

    def get_entrypoints(self, name: str):
        guesses = []
        for entry in self.entries:
            guesses.append(entry.format(name=name))
        return guesses

    def add_dependency_file(self, dep: str):
        assert isinstance(dep, str)
        self.dependency_file = dep

    def add_dependency_call(self, call: list):
        if self.dependency_call is None:
            self.dependency_call = []
        self.dependency_call.append(call)

    def copy(self):
        return copy.deepcopy(self)


class InstInfo:
    def __init__(self, name: str, location: str, git_url: str):
        self.name = name
        self.source_loc = str(location)     # Used for 'git clone'
        self.git_url = git_url              # API access for github repos
        self.srctype = Source.get_type(location)
        self.entry = None                   # relative to source_loc or subdir
        self.deps = None
        self.subdir = None
        self.commit = None

    def __repr__(self):
        return (f'InstInfo({self.name}, {self.source_loc}, {self.git_url}, '
                f'{self.entry}, {self.deps}, {self.subdir})')

    def get_inst_details(self) -> bool:
        """Search the source_loc for plugin install details.
        This may be necessary if a contents api is unavailable.
        Extracts entrypoint and dependencies if searchable, otherwise
        matches a directory to the plugin name and stops."""
        if self.srctype == Source.DIRECTORY:
            assert Path(self.source_loc).exists()
            assert os.path.isdir(self.source_loc)
        target = SourceDir(self.source_loc, srctype=self.srctype)
        # Set recursion for how many directories deep we should search
        depth = 0
        if self.srctype in [Source.DIRECTORY, Source.LOCAL_REPO]:
            depth = 5
        elif self.srctype == Source.GITHUB_REPO:
            depth = 2

        def search_dir(self, sub: SourceDir, subdir: bool,
                       recursion: int) -> Union[SourceDir, None]:
            assert isinstance(recursion, int)
            # If unable to search deeper, resort to matching directory name
            if recursion < 1:
                if sub.name.lower() == self.name.lower():
                    # Partial success (can't check for entrypoint)
                    self.name = sub.name
                    return sub
                return None
            sub.populate()

            if sub.name.lower() == self.name.lower():
                # Directory matches the name we're trying to install, so check
                # for entrypoint and dependencies.
                for inst in INSTALLERS:
                    for g in inst.get_entrypoints(self.name):
                        found_entry = sub.find(g, ftype=SourceFile)
                        if found_entry:
                            break
                    # FIXME: handle a list of dependencies
                    found_dep = sub.find(inst.dependency_file,
                                         ftype=SourceFile)
                    if found_entry:
                        # Success!
                        if found_dep:
                            self.name = sub.name
                            self.entry = found_entry.name
                            self.deps = found_dep.name
                            return sub
                        logging.debug(f"missing dependency for {self}")
                        found_entry = None
            for file in sub.contents:
                if isinstance(file, SourceDir):
                    success = search_dir(self, file, True, recursion - 1)
                    if success:
                        return success
            return None

        result = search_dir(self, target, False, depth)
        if result:
            if result != target:
                if result.relative:
                    self.subdir = result.relative
            return True
        return False


def create_dir(directory: PosixPath) -> bool:
    try:
        Path(directory).mkdir(parents=False, exist_ok=True)
        return True
    # Okay if directory already exists
    except FileExistsError:
        return True
    # Parent directory missing
    except FileNotFoundError:
        return False


def remove_dir(directory: str) -> bool:
    try:
        shutil.rmtree(directory)
        return True
    except NotADirectoryError:
        print(f"Tried to remove directory {directory} that does not exist.")
    except PermissionError:
        print(f"Permission denied removing dir: {directory}")
    return False


class Source(Enum):
    DIRECTORY = 1
    LOCAL_REPO = 2
    GITHUB_REPO = 3
    OTHER_URL = 4
    UNKNOWN = 5

    @classmethod
    def get_type(cls, source: str):
        if Path(os.path.realpath(source)).exists():
            if os.path.isdir(os.path.realpath(source)):
                # returns 0 if git repository
                proc = run(['git', '-C', source, 'rev-parse'],
                           cwd=os.path.realpath(source), stdout=PIPE,
                           stderr=PIPE, text=True, timeout=3)
                if proc.returncode == 0:
                    return cls(2)
                return cls(1)
        if 'github.com' in source.lower():
            return cls(3)
        if 'http://' in source.lower() or 'https://' in source.lower():
            return cls(4)
        return cls(5)


class SourceDir():
    """Structure to search source contents."""
    def __init__(self, location: str, srctype: Source = None, name: str = None,
                 relative: str = None):
        self.location = str(location)
        if name:
            self.name = name
        else:
            self.name = Path(location).name
        self.contents = []
        self.srctype = srctype
        self.prepopulated = False
        self.relative = relative  # location relative to source

    def populate(self):
        """populates contents of the directory at least one level"""
        if self.prepopulated:
            return
        if not self.srctype:
            self.srctype = Source.get_type(self.location)
        # logging.debug(f"populating {self.srctype} {self.location}")
        if self.srctype == Source.DIRECTORY:
            self.contents = populate_local_dir(self.location)
        elif self.srctype == Source.LOCAL_REPO:
            self.contents = populate_local_repo(self.location)
        elif self.srctype == Source.GITHUB_REPO:
            self.contents = populate_github_repo(self.location)
        else:
            raise Exception("populate method undefined for {self.srctype}")
        # Ensure the relative path of the contents is inherited.
        for c in self.contents:
            if self.relative is None:
                c.relative = c.name
            else:
                c.relative = str(Path(self.relative) / c.name)

    def find(self, name: str, ftype: type = None) -> str:
        """Match a SourceFile or SourceDir to the provided name
        (case insentive) and return its filename."""
        assert isinstance(name, str)
        if len(self.contents) == 0:
            return None
        for c in self.contents:
            if ftype and not isinstance(c, ftype):
                continue
            if c.name.lower() == name.lower():
                return c
        return None

    def __repr__(self):
        return f"<SourceDir: {self.name} ({self.location})>"

    def __eq__(self, compared):
        if isinstance(compared, str):
            return self.name == compared
        if isinstance(compared, SourceDir):
            return (self.name == compared.name and
                    self.location == compared.location)
        return False


class SourceFile():
    def __init__(self, location: str):
        self.location = str(location)
        self.name = Path(location).name

    def __repr__(self):
        return f"<SourceFile: {self.name} ({self.location})>"

    def __eq__(self, compared):
        if isinstance(compared, str):
            return self.name == compared
        if isinstance(compared, SourceFile):
            return (self.name == compared.name and
                    self.location == compared.location)
        return False


def populate_local_dir(path: str) -> list:
    assert Path(os.path.realpath(path)).exists()
    contents = []
    for c in os.listdir(path):
        fullpath = Path(path) / c
        if os.path.isdir(fullpath):
            # Inheriting type saves a call to test if it's a git repo
            contents.append(SourceDir(fullpath, srctype=Source.DIRECTORY))
        else:
            contents.append(SourceFile(fullpath))
    return contents


def populate_local_repo(path: str) -> list:
    assert Path(os.path.realpath(path)).exists()
    basedir = SourceDir('base')

    def populate_source_path(parent, mypath):
        """`git ls-tree` lists all files with their full path.
        This populates all intermediate directories and the file."""
        parentdir = parent
        if mypath == '.':
            logging.debug(' asked to populate root dir')
            return
        # reverse the parents
        pdirs = mypath
        revpath = []
        child = parentdir
        while pdirs.parent.name != '':
            revpath.append(pdirs.parent.name)
            pdirs = pdirs.parent
        for p in reversed(revpath):
            child = parentdir.find(p)
            if child:
                parentdir = child
            else:
                child = SourceDir(p, srctype=Source.LOCAL_REPO)
                child.prepopulated = True
                parentdir.contents.append(child)
                parentdir = child
        newfile = SourceFile(mypath.name)
        child.contents.append(newfile)

    # FIXME: Pass in tag or commit hash
    ver = 'HEAD'
    git_call = ['git', '-C', path, 'ls-tree', '--full-tree', '-r',
                '--name-only', ver]
    proc = run(git_call, stdout=PIPE, stderr=PIPE, text=True, timeout=5)
    if proc.returncode != 0:
        logging.debug(f'ls-tree of repo {path} failed')
        return None
    for filepath in proc.stdout.splitlines():
        populate_source_path(basedir, Path(filepath))
    return basedir.contents


def populate_github_repo(url: str) -> list:
    # FIXME: This probably contains leftover cruft.
    repo = url.split('/')
    while '' in repo:
        repo.remove('')
    repo_name = None
    parsed_url = urlparse(url)
    if 'github.com' not in parsed_url.netloc:
        return None
    if len(parsed_url.path.split('/')) < 2:
        return None
    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'{API_GITHUB_COM}/repos/{repo_user}/{repo_name}/contents/'

    git_url = api_url
    if "api.github.com" in git_url:
        # This lets us redirect to handle blackbox testing
        git_url = (API_GITHUB_COM + git_url.split("api.github.com")[-1])
    r = urlopen(git_url, timeout=5)
    if r.status != 200:
        return False
    if 'git/tree' in git_url:
        tree = json.loads(r.read().decode())['tree']
    else:
        tree = json.loads(r.read().decode())
    contents = []
    for sub in tree:
        if 'type' in sub and 'name' in sub and 'git_url' in sub:
            if sub['type'] == 'dir':
                new_sub = SourceDir(sub['git_url'], srctype=Source.GITHUB_REPO,
                                    name=sub['name'])
                contents.append(new_sub)
            elif sub['type'] == 'file':
                new_file = SourceFile(sub['name'])
                contents.append(new_file)
    return contents


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(parent_path):
            with open(self.conf_fp, 'w') as f:
                f.write(default_text)
                # FIXME: Handle write failure
                return default_text
        else:
            logging.debug('could not create the parent directory ' +
                          parent_path)
            raise FileNotFoundError('invalid parent directory')

    def editConfigFile(self, addline: Union[str, None],
                       removeline: Union[str, None]):
        """Idempotent function to add and/or remove a single line each."""
        remove_these_lines = []
        with open(self.conf_fp, 'r') as reckless_conf:
            original = reckless_conf.readlines()
            empty_lines = []
            write_required = False
            for n, l in enumerate(original):
                if removeline and l.strip() == removeline.strip():
                    write_required = True
                    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
            if not addline and not write_required:
                return
            # No write necessary if addline is already in config.
            if addline and not write_required:
                for line in original:
                    if line.strip() == addline.strip():
                        return
            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 and addline.strip() == l.strip():
                            # 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)
        reck_contents_lower = {}
        for f in reck_contents:
            reck_contents_lower.update({f.lower(): f})

        def match_name(name) -> str:
            for tier in range(0, 10):
                # Look for each installers preferred entrypoint format first
                for inst in INSTALLERS:
                    fmt = inst.entries[tier]
                    if '{name}' in fmt:
                        pre = fmt.split('{name}')[0]
                        post = fmt.split('{name}')[-1]
                        if name.startswith(pre) and name.endswith(post):
                            return name.lstrip(pre).rstrip(post)
                    else:
                        if fmt == name:
                            return name
            return name

        name = match_name(name)
        if name.lower() in reck_contents_lower.keys():
            actual_name = reck_contents_lower[name.lower()]
            self.dir = Path(RECKLESS_CONFIG.reckless_dir).joinpath(actual_name)
        else:
            raise Exception(f"Could not find a reckless directory for {name}")
        plug_dir = Path(RECKLESS_CONFIG.reckless_dir).joinpath(actual_name)
        for guess in entry_guesses(actual_name):
            for content in plug_dir.iterdir():
                if content.name == guess:
                    self.entry = str(content)
                    self.name = actual_name
                    return
        raise Exception(f'plugin entrypoint not found in {self.dir}')


python3pip = Installer('python3pip', 'text/x-python', exe='python3',
                       manager='pip', entry='{name}.py')
python3pip.add_entrypoint('{name}')
python3pip.add_entrypoint('__init__.py')
python3pip.add_dependency_file('requirements.txt')
python3pip.add_dependency_call(['pip', 'install', '-r', 'requirements.txt'])

python3pip3 = python3pip.copy()
python3pip3.manager = 'pip3'
python3pip3.dependency_call = [['pip3', 'install', '-r', 'requirements.txt']]

# Nodejs plugin installer
nodejs = Installer('nodejs', 'application/javascript', exe='node',
                   manager='npm', entry='{name}.js')
nodejs.add_entrypoint('{name}')
nodejs.add_dependency_call(['npm', 'install', '--omit=dev'])
nodejs.add_dependency_file('package.json')

INSTALLERS = {python3pip, python3pip3, nodejs}


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 _source_search(name: str, source: str) -> Union[InstInfo, None]:
    """Identify source type, retrieve contents, and populate InstInfo
    if the relevant contents are found."""
    root_dir = SourceDir(source)
    source = InstInfo(name, root_dir.location, None)
    if source.get_inst_details():
        return source
    return None


def _git_clone(src: InstInfo, dest: Union[PosixPath, str]) -> bool:
    print(f'cloning {src.srctype} {src}')
    if src.srctype == Source.GITHUB_REPO:
        assert 'github.com' in src.source_loc
        source = f"{GITHUB_COM}" + src.source_loc.split("github.com")[-1]
    elif src.srctype in [Source.LOCAL_REPO, Source.OTHER_URL]:
        source = src.source_loc
    else:
        return False
    git = run(['git', 'clone', source, str(dest)], stdout=PIPE, stderr=PIPE,
              text=True, check=False, timeout=60)
    if git.returncode != 0:
        for line in git.stderr:
            logging.debug(line)
        if Path(dest).exists():
            remove_dir(str(dest))
        print('Error: Failed to clone repo')
        return False
    return True


def get_temp_reckless_dir() -> PosixPath:
    random_dir = 'reckless-{}'.format(str(hash(os.times()))[-9:])
    new_path = Path(tempfile.gettempdir()) / random_dir
    return new_path


def add_installation_metadata(installed: InstInfo,
                              original_request: InstInfo):
    """Document the install request and installation details for use when
    updating the plugin."""
    install_dir = Path(installed.source_loc)
    assert install_dir.is_dir()
    data = ('installation date\n'
            f'{datetime.date.today().isoformat()}\n'
            'installation time\n'
            f'{int(time.time())}\n'
            'original source\n'
            f'{original_request.source_loc}\n'
            'requested commit\n'
            f'{original_request.commit}\n'
            'installed commit\n'
            f'{installed.commit}\n')
    with open(install_dir / '.metadata', 'w') as metadata:
        metadata.write(data)


def _checkout_commit(orig_src: InstInfo,
                     cloned_src: InstInfo,
                     cloned_path: PosixPath):
    # Check out and verify commit/tag if source was a repository
    if orig_src.srctype in [Source.LOCAL_REPO, Source.GITHUB_REPO,
                            Source.OTHER_URL]:
        if orig_src.commit:
            logging.debug(f"Checking out {orig_src.commit}")
            checkout = Popen(['git', 'checkout', orig_src.commit],
                             cwd=str(cloned_path),
                             stdout=PIPE, stderr=PIPE)
            checkout.wait()
            if checkout.returncode != 0:
                print('failed to checkout referenced '
                      f'commit {orig_src.commit}')
                return None
        else:
            logging.debug("using latest commit of default branch")

        # Log the commit we actually used (for installation metadata)
        git = run(['git', 'rev-parse', 'HEAD'], cwd=str(cloned_path),
                  stdout=PIPE, stderr=PIPE, text=True, check=False, timeout=60)
        if git.returncode == 0:
            head_commit = git.stdout.splitlines()[0]
            logging.debug(f'checked out HEAD: {head_commit}')
            cloned_src.commit = head_commit
        else:
            logging.debug(f'unable to collect commit: {git.stderr}')
    else:
        if orig_src.commit:
            logging.warning("unable to checkout commit/tag on non-repository "
                            "source")
        return cloned_path

    if cloned_src.subdir is not None:
        return Path(cloned_src.source_loc) / cloned_src.subdir
    return cloned_path


def _install_plugin(src: InstInfo) -> Union[InstInfo, None]:
    """make sure the repo exists and clone it."""
    logging.debug(f'Install requested from {src}.')
    if RECKLESS_CONFIG is None:
        print('error: reckless install directory unavailable')
        sys.exit(2)

    # Use a unique directory for each cloned repo.
    tmp_path = get_temp_reckless_dir()
    if not create_dir(tmp_path):
        logging.debug(f'failed to create {tmp_path}')
        return None
    clone_path = tmp_path / 'clone'
    if not create_dir(tmp_path):
        logging.debug(f'failed to create {clone_path}')
        return None
    # we rename the original repo here.
    plugin_path = clone_path / src.name
    inst_path = Path(RECKLESS_CONFIG.reckless_dir) / src.name
    if Path(clone_path).exists():
        logging.debug(f'{clone_path} already exists - deleting')
        shutil.rmtree(clone_path)
    if src.srctype == Source.DIRECTORY:
        logging.debug(("copying local directory contents from"
                       f" {src.source_loc}"))
        create_dir(clone_path)
        shutil.copytree(src.source_loc, plugin_path)
    elif src.srctype in [Source.LOCAL_REPO, Source.GITHUB_REPO,
                         Source.OTHER_URL]:
        # clone git repository to /tmp/reckless-...
        if not _git_clone(src, plugin_path):
            return None
    # FIXME: Validate path was cloned successfully.
    # Depending on how we accessed the original source, there may be install
    # details missing. Searching the cloned repo makes sure we have it.
    cloned_src = _source_search(src.name, str(clone_path))
    logging.debug(f'cloned_src: {cloned_src}')
    if not cloned_src:
        logging.debug('failed to find plugin after cloning repo.')
        return None

    # If a specific commit or tag was requested, check it out now.
    plugin_path = _checkout_commit(src, cloned_src, plugin_path)
    if not plugin_path:
        return None

    # Find a suitable installer
    INSTALLER = None
    for inst_method in INSTALLERS:
        if not (inst_method.installable() and inst_method.executable()):
            continue
        if inst_method.dependency_file is not None:
            if inst_method.dependency_file not in os.listdir(plugin_path):
                continue
        logging.debug(f"using installer {inst_method.name}")
        INSTALLER = inst_method
        break
    if not INSTALLER:
        logging.debug('Could not find a suitable installer method.')
        return None
    if not cloned_src.entry:
        # The plugin entrypoint may not be discernable prior to cloning.
        # Need to search the newly cloned directory, not the original
        cloned_src.source_loc = plugin_path

    # Relocate plugin to a staging directory prior to testing
    staging_path = tmp_path / src.name / 'source'
    shutil.copytree(str(plugin_path), staging_path)
    staged_src = cloned_src
    # Because the source files are copied to a 'source' directory, the
    # get_inst_details function no longer works. (dir must match plugin name)
    # Set these manually instead.
    staged_src.source_loc = str(staging_path.parent)
    staged_src.srctype = Source.DIRECTORY
    staged_src.subdir = None
    # Create symlink in staging tree to redirect to the plugins entrypoint
    Path(staging_path.parent / cloned_src.entry).\
        symlink_to(staging_path / cloned_src.entry)

    # try it out
    if INSTALLER.dependency_call:
        for call in INSTALLER.dependency_call:
            logging.debug(f"Install: invoking '{' '.join(call)}'")
            if logging.root.level < logging.WARNING:
                pip = Popen(call, cwd=staging_path, text=True)
            else:
                pip = Popen(call, cwd=staging_path, stdout=PIPE, stderr=PIPE,
                            text=True)
            pip.wait()
            # FIXME: handle output of multiple calls

        if pip.returncode == 0:
            print('dependencies installed successfully')
        else:
            print('error encountered installing dependencies')
            if pip.stdout:
                logging.debug(pip.stdout.read())
            return None
    test_log = []
    try:
        test = run([Path(staging_path).joinpath(cloned_src.entry)],
                   cwd=str(staging_path), stdout=PIPE, stderr=PIPE,
                   text=True, timeout=3)
        for line in test.stderr:
            test_log.append(line.strip('\n'))
        returncode = test.returncode
    except TimeoutExpired:
        # If the plugin is still running, it's assumed to be okay.
        returncode = 0
    if returncode != 0:
        logging.debug("plugin testing error:")
        for line in test_log:
            logging.debug(f'  {line}')
        print('plugin testing failed')
        return None

    # Find this cute little plugin a forever home
    shutil.copytree(str(staging_path), inst_path)
    add_installation_metadata(staged_src, src)
    print(f'plugin installed: {inst_path}')
    remove_dir(clone_path)
    return staged_src


def install(plugin_name: str):
    """downloads plugin from source repos, installs and activates plugin"""
    assert isinstance(plugin_name, str)
    # Specify a tag or commit to checkout by adding @<tag> to plugin name
    if '@' in plugin_name:
        logging.debug("testing for a commit/tag in plugin name")
        name, commit = plugin_name.split('@', 1)
    else:
        name = plugin_name
        commit = None
    logging.debug(f"Searching for {name}")
    src = search(name)
    if src:
        src.commit = commit
        logging.debug(f'Retrieving {src.name} from {src.source_loc}')
        installed = _install_plugin(src)
        if not installed:
            print('installation aborted')
            sys.exit(1)

        # Match case of the containing directory
        for dirname in os.listdir(RECKLESS_CONFIG.reckless_dir):
            if dirname.lower() == installed.name.lower():
                inst_path = Path(RECKLESS_CONFIG.reckless_dir)
                inst_path = inst_path / dirname / installed.entry
                RECKLESS_CONFIG.enable_plugin(inst_path)
                enable(installed.name)
                return
        print(('dynamic activation failed: '
               f'{installed.name} not found in reckless directory'))
        sys.exit(1)


def uninstall(plugin_name: str):
    """disables plugin and deletes the plugin's reckless dir"""
    assert isinstance(plugin_name, str)
    logging.debug(f'Uninstalling plugin {plugin_name}')
    disable(plugin_name)
    inst = InferInstall(plugin_name)
    if not Path(inst.entry).exists():
        print(f'cannot find installed plugin at expected path {inst.entry}')
        sys.exit(1)
    logging.debug(f'looking for {str(Path(inst.entry).parent)}')
    if remove_dir(str(Path(inst.entry).parent)):
        print(f"{inst.name} uninstalled successfully.")


def search(plugin_name: str) -> Union[InstInfo, None]:
    """searches plugin index for plugin"""
    ordered_sources = RECKLESS_SOURCES

    for src in RECKLESS_SOURCES:
        # Search repos named after the plugin before collections
        if Source.get_type(src) == Source.GITHUB_REPO:
            if src.split('/')[-1].lower() == plugin_name.lower():
                ordered_sources.remove(src)
                ordered_sources.insert(0, src)
    # Check locally before reaching out to remote repositories
    for src in RECKLESS_SOURCES:
        if Source.get_type(src) in [Source.DIRECTORY, Source.LOCAL_REPO]:
            ordered_sources.remove(src)
            ordered_sources.insert(0, src)
    for source in ordered_sources:
        srctype = Source.get_type(source)
        if srctype == Source.UNKNOWN:
            logging.debug(f'cannot search {srctype} {source}')
            continue
        if srctype in [Source.DIRECTORY, Source.LOCAL_REPO,
                       Source.GITHUB_REPO, Source.OTHER_URL]:
            found = _source_search(plugin_name, source)
        if not found:
            continue
        print(f"found {found.name} in source: {found.source_loc}")
        logging.debug(f"entry: {found.entry}")
        if found.subdir:
            logging.debug(f'sub-directory: {found.subdir}')
        return found
    logging.debug("Search exhausted all sources")
    return None


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(*cli_args, timeout: int = 15) -> dict:
    """Interfaces with Core-Lightning via CLI using any configured options."""
    cmd = LIGHTNING_CLI_CALL.copy()
    cmd.extend(cli_args)
    clncli = run(cmd, stdout=PIPE, stderr=PIPE, check=False, timeout=timeout)
    out = clncli.stdout.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'])
    # RPC may not be available - i.e., lightningd not running, using
    # alternate config.
    err = clncli.stderr.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)
    logging.debug(f'activating {plugin_name}')
    try:
        lightning_cli('plugin', 'start', path)
    except CLIError as err:
        if 'already registered' in err.message:
            logging.debug(f'{inst.name} is already running')
        else:
            print(f'reckless: {inst.name} failed to start!')
            raise err
    except RPCError:
        logging.debug(('lightningd rpc unavailable. '
                       'Skipping dynamic activation.'))
    RECKLESS_CONFIG.enable_plugin(path)
    print(f'{inst.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)
    logging.debug(f'deactivating {plugin_name}')
    try:
        lightning_cli('plugin', 'stop', path)
    except CLIError as err:
        if err.code == -32602:
            logging.debug('plugin not currently running')
        else:
            print('lightning-cli plugin stop failed')
            raise err
    except RPCError:
        logging.debug(('lightningd rpc unavailable. '
                       'Skipping dynamic deactivation.'))
    RECKLESS_CONFIG.disable_plugin(path)
    print(f'{inst.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)['configs']
        if 'conf' in active_config:
            net_conf = LightningBitcoinConfig(path=active_config['conf']
                                              ['value_str'])
    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 str(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 load_sources() -> 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():
        logging.debug('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():
        if os.path.isdir(maybe_path):
            default_repo = 'https://github.com/lightningd/plugins'
            my_file = Config(path=str(get_sources_file()),
                             default_text=default_repo)
            my_file.editConfigFile(src, None)
    elif 'github.com' in src or 'http://' in src or 'https://' in src:
        my_file = Config(path=str(get_sources_file()),
                         default_text='https://github.com/lightningd/plugins')
        my_file.editConfigFile(src, None)
    else:
        print(f'failed to add source {src}')


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('--network',
                        help="specify a network to use (default: bitcoin)",
                        type=str)
    parser.add_argument('-v', '--verbose', action="store_const",
                        dest="loglevel", const=logging.DEBUG,
                        default=logging.WARNING)
    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'
    SUPPORTED_NETWORKS = ['bitcoin', 'regtest', 'liquid', 'liquid-regtest',
                          'litecoin', 'signet', 'testnet']
    if args.network:
        if args.network in SUPPORTED_NETWORKS:
            NETWORK = args.network
        else:
            print(f"Error: {args.network} network not supported")
    LIGHTNING_DIR = Path(args.lightning)
    # This env variable is set under CI testing
    LIGHTNING_CLI_CALL = [os.environ.get('LIGHTNING_CLI')]
    if LIGHTNING_CLI_CALL == [None]:
        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 = load_sources()
    API_GITHUB_COM = 'https://api.github.com'
    GITHUB_COM = 'https://github.com'
    # Used for blackbox testing to avoid hitting github servers
    if 'REDIR_GITHUB_API' in os.environ:
        API_GITHUB_COM = os.environ['REDIR_GITHUB_API']
    if 'REDIR_GITHUB' in os.environ:
        GITHUB_COM = os.environ['REDIR_GITHUB']
    logging.root.setLevel(args.loglevel)

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