cic-cli/cic/cmd/arg.py

# standard imports
import importlib
import logging
import os
import sys

# external imports
import chainlib.eth.cli
import cic.cmd.easy as cmd_easy
import cic.cmd.export as cmd_export
import cic.cmd.ext as cmd_ext
import cic.cmd.init as cmd_init
import cic.cmd.show as cmd_show
from chainlib.chain import ChainSpec
from cic.auth import PGPAuthCrypt
from cic.crypt.aes import AESCTREncrypt
from cic.http import HTTPSession, PGPClientSession

# local imports
from cic.notify import NotifyWriter

notifier = NotifyWriter()
logg = logging.getLogger(__name__)

script_dir = os.path.dirname(os.path.realpath(__file__))
data_dir = os.path.join(script_dir, "..", "data")
base_config_dir = os.path.join(data_dir, "config")


class NullWriter:
    def notify(self, v):
        pass

    def ouch(self, v):
        pass

    def write(self, v):
        sys.stdout.write(str(v))


class CmdCtrl:

    __cmd_alias = {
        "u": "user",
        "t": "tag",
    }

    __auth_for = [
        "user",
    ]

    def __init__(self, *_args, argv=None, _description=None, logger=None, **_kwargs):
        self.args(argv)

        self.logging(logger)

        self.module()

        self.load_config()

        self.notifier()

        self.auth()

        self.blockchain()

        self.remote_openers = {}
        if self.get("META_URL") is not None:
            auth_client_session = PGPClientSession(self.__auth)
            self.remote_openers["meta"] = HTTPSession(
                self.get("META_URL"),
                auth=auth_client_session,
                origin=self.config.get("META_HTTP_ORIGIN"),
            )

    def blockchain(self):
        self.chain_spec = ChainSpec.from_chain_str(self.config.get("CHAIN_SPEC"))
        self.rpc = chainlib.eth.cli.Rpc()
        self.__conn = self.rpc.connect_by_config(self.config)

    def args(self, argv):
        self.argparser = chainlib.eth.cli.ArgumentParser(
            chainlib.eth.cli.argflag_std_read
        )
        sub = self.argparser.add_subparsers()
        sub.dest = "command"

        sub_init = sub.add_parser("init", help="initialize new cic data directory")
        cmd_init.process_args(sub_init)

        sub_show = sub.add_parser(
            "show", help="display summary of current state of cic data directory"
        )
        cmd_show.process_args(sub_show)

        sub_export = sub.add_parser(
            "export", help="export cic data directory state to a specified target"
        )
        cmd_export.process_args(sub_export)

        sub_ext = sub.add_parser("ext", help="extension helpers")
        cmd_ext.process_args(sub_ext)

        sub_easy = sub.add_parser("easy", help="Easy Mode Contract Deployment")
        cmd_easy.process_args(sub_easy)

        self.cmd_args = self.argparser.parse_args(argv)

    def module(self):
        self.cmd_string = self.cmd_args.command
        cmd_string_translate = self.__cmd_alias.get(self.cmd_string)
        if cmd_string_translate is not None:
            self.cmd_string = cmd_string_translate

        if self.cmd_string is None:
            self.cmd_string = "none"
            self.argparser.print_help()
            exit(1)

        modname = f"cic.cmd.{self.cmd_string}"
        self.logger.debug(f"using module {modname}")
        self.cmd_mod = importlib.import_module(modname)

    def logging(self, logger):
        self.logger = logger
        if self.logger is None:
            self.logger = logging.getLogger()
        if self.cmd_args.vv:
            self.logger.setLevel(logging.DEBUG)
        elif self.cmd_args.v:
            self.logger.setLevel(logging.INFO)

    def load_config(self):
        override_dir = self.cmd_args.config
        if override_dir is None:
            p = os.environ.get("HOME")
            if p is not None:
                p = os.path.join(p, ".config", "cic", "cli")
                try:
                    os.stat(p)
                    override_dir = p
                    logg.info(
                        f"applying user config override from standard location: {p}"
                    )
                except FileNotFoundError:
                    pass
        extra_args = self.cmd_mod.extra_args()
        self.config = chainlib.eth.cli.Config.from_args(
            self.cmd_args,
            base_config_dir=base_config_dir,
            extra_args=extra_args,
            default_config_dir=override_dir,
        )

        self.config.add(False, "_SEQ")

        self.config.censor("AUTH_PASSPHRASE")

        self.logger.debug(f"loaded config:\n{self.config}")

    def auth(self):
        typ = self.get("AUTH_TYPE")
        if typ != "gnupg":
            raise NotImplementedError("Valid aut implementations are: gnupg")
        default_auth_db_path = None
        if os.environ.get("HOME") is not None:
            default_auth_db_path = os.path.join(
                os.environ["HOME"], ".local/share/cic/clicada"
            )
        auth_db_path = self.get("AUTH_DB_PATH", default_auth_db_path)
        self.__auth = PGPAuthCrypt(
            auth_db_path, self.get("AUTH_KEY"), self.get("AUTH_KEYRING_PATH")
        )
        self.__auth.get_secret(self.get("AUTH_PASSPHRASE"))
        self.encrypter = AESCTREncrypt(auth_db_path, self.__auth.secret)
        logg.debug(f"loaded auth: {self.__auth}")
        logg.debug(f"AUTH_PASSPHRASE: {self.get('AUTH_PASSPHRASE')}")
        logg.debug(f"AUTH_KEY: {self.get('AUTH_KEY')}")
        logg.debug(f"AUTH_DB_PATH: {self.get('AUTH_DB_PATH')}")
        logg.debug(f"AUTH_KEYRING_PATH: {self.get('AUTH_KEYRING_PATH')}")

    def get(self, k, default=None):
        r = self.config.get(k, default)
        if k in [
            "_FORCE",
        ]:
            if r is None:
                return False
            return self.config.true(k)
        return r

    def chain(self):
        return self.chain_spec

    def conn(self):
        return self.__conn

    def execute(self):
        self.cmd_mod.execute(self)

    def opener(self, k):
        return self.remote_openers[k]

    def notifier(self):
        if logg.root.level >= logging.WARNING:
            logging.disable()
            self.writer = notifier
        else:
            self.writer = NullWriter()

    def notify(self, v):
        self.writer.notify(v)

    def ouch(self, v):
        self.writer.ouch(v)
        print()

    def write(self, v):
        self.writer.write("")
        self.writer.write(v)
        print()