Merge pull request #14 from jiri-otoupal/refactor-improvements

Refactor and improvements

abst/__version__.py

@@ -10,7 +10,7 @@
     "CLI Command making OCI Bastion and kubernetes usage simple and fast"
 )
 
-__version__ = "2.1.5"
+__version__ = "2.1.6"
 __author__ = "Jiri Otoupal"
 __author_email__ = "jiri-otoupal@ips-database.eu"
 __license__ = "MIT"

abst/bastion_support/oci_bastion.py

@@ -257,6 +257,7 @@ def load_response(self, res):
 
     def create_bastion_forward_port_session(self, creds):
         ssh_key_path = self.get_ssh_pub_key_path(creds)
+        cfg = Bastion.load_config()
 
         if "region" not in creds.keys():
             rich.print("Missing region, will use profile default")
@@ -266,12 +267,17 @@ def create_bastion_forward_port_session(self, creds):
             rich.print("[red]SSH key has invalid path[/red]")
             exit(1)
 
+        ssh_pub_path = creds.get(ssh_key_path, cfg.get("ssh-pub-path", "No Public key supplied"))
+
+        if ssh_pub_path == "No Public key supplied":
+            raise Exception(ssh_pub_path)
+
         res = self.__create_bastion_session_port_forward(creds["bastion-id"],
                                                          creds["target-ip"],
                                                          f'{creds["default-name"]}-ctx-'
                                                          f'{self.get_print_name()}',
                                                          int(creds["target-port"]),
-                                                         ssh_key_path,
+                                                         ssh_pub_path,
                                                          int(creds["ttl"]), False, creds.get("region", None))
         try:
             trs = Bastion.parse_response(res)
@@ -280,10 +286,11 @@ def create_bastion_forward_port_session(self, creds):
             logging.debug(f"Added session id of {self.context_name}")
         except:
             pass
-        return creds["host"], creds["target-ip"], creds["target-port"], creds["ssh-pub-path"], res
+        return creds["host"], creds["target-ip"], creds["target-port"], ssh_pub_path, res
 
     def create_bastion_ssh_session_managed(self, creds):
         ssh_key_path = self.get_ssh_pub_key_path(creds)
+        cfg = Bastion.load_config()
 
         if "region" not in creds.keys():
             rich.print("Missing region, will use profile default")
@@ -293,13 +300,18 @@ def create_bastion_ssh_session_managed(self, creds):
             rich.print("[red]SSH key has invalid path[/red]")
             exit(1)
 
+        ssh_pub_path = creds.get(ssh_key_path, cfg.get("ssh-pub-path", "No Public key supplied"))
+
+        if ssh_pub_path == "No Public key supplied":
+            raise Exception(ssh_pub_path)
+
         try:
             res = self.__create_bastion_ssh_session_managed(creds["bastion-id"],
                                                             creds["resource-id"],
                                                             f'{creds["default-name"]}-ctx-'
                                                             f'{self.get_print_name()}',
                                                             creds["resource-os-username"],
-                                                            ssh_key_path,
+                                                            ssh_pub_path,
                                                             int(creds["ttl"]), False, creds.get("region", None)
                                                             )
             try:
@@ -368,11 +380,22 @@ def load_json(cls, path=default_creds_path) -> dict:
 
     @classmethod
     def create_default_location(cls):
+
         Path(default_creds_path.parent).mkdir(exist_ok=True)
         Path(default_contexts_location).mkdir(exist_ok=True)
 
         if not default_conf_path.exists():
             cls.write_creds_json(default_conf_contents, default_conf_path)
+        else:
+            something_changed = False
+            data = Bastion.load_config()
+            for key, value in default_conf_contents.items():
+                if key not in data.keys():
+                    data[key] = value
+                    something_changed = True
+            if something_changed:
+                cls.write_creds_json(default_conf_contents, default_conf_path)
+                rich.print("[green]Updated .abst/config file[/green]")
 
     @classmethod
     def write_creds_json(cls, td: dict, path: Path):

abst/cli_commands/config_cli/commands.py

@@ -0,0 +1,107 @@
+import click
+import rich
+from InquirerPy import inquirer
+
+from abst.bastion_support.oci_bastion import Bastion
+from abst.cfg_func import __upgrade
+from abst.config import default_contexts_location
+from abst.utils.misc_funcs import setup_calls
+from abst.tools import get_context_path
+
+
+@click.group(help="Group of commands for operations with config")
+def config():
+    pass
+
+
+@config.command("generate", help="Will generate sample json and overwrite changes")
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def generate(debug, context_name):
+    setup_calls(debug)
+
+    path = get_context_path(context_name)
+
+    Bastion.create_default_location()
+    td = Bastion.generate_sample_dict()
+    creds_path = Bastion.write_creds_json(td, path)
+    print(
+        f"Sample credentials generated, please fill 'creds.json' in {creds_path} with "
+        f"your credentials for this to work, you can use 'abst json fill "
+        f"{context_name if context_name else ''}'"
+    )
+
+
+@config.command(
+    "fill", help="Fills Json config with credentials you enter interactively"
+)
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def fill(debug, context_name):
+    setup_calls(debug)
+
+    path = get_context_path(context_name)
+
+    if not path.exists():
+        rich.print("Generating sample Creds file")
+        Bastion.create_default_location()
+        td = Bastion.generate_sample_dict()
+        Bastion.write_creds_json(td, path)
+
+    if not default_contexts_location.exists():
+        rich.print("Generating contexts location")
+        Bastion.create_default_location()
+
+    rich.print(f"[green]Filling {str(path)}")
+    rich.print("Please fill field one by one as displayed")
+    n_dict = dict()
+
+    creds_json_ro = Bastion.load_json(path)
+
+    for key, value in creds_json_ro.items():
+        n_dict[key] = inquirer.text(
+            message=f"{key.capitalize()}:", default=value
+        ).execute()
+    rich.print("\n[red]New json looks like this:[/red]")
+    rich.print_json(data=n_dict)
+    if inquirer.confirm(message="Write New Json ?", default=False).execute():
+        Bastion.write_creds_json(n_dict, path)
+        rich.print("[green]Wrote changes[/green]")
+    else:
+        rich.print("[red]Fill interrupted, nothing changed[/red]")
+
+
+@config.command(
+    "locate", help="Locates Json config with credentials you enter interactively"
+)
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def locate(debug, context_name):
+    setup_calls(debug)
+
+    path = get_context_path(context_name)
+
+    if path.exists():
+        rich.print(f"[green]Config file location: {path.absolute()}[/green]")
+    else:
+        rich.print(
+            f"[red]Config does not exist yet, future location"
+            f" {path.absolute()}[/red]"
+        )
+
+
+@config.command(
+    "upgrade", help="Locates Json config with credentials you enter interactively"
+)
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def upgrade(debug, context_name):
+    setup_calls(debug)
+
+    path = get_context_path(context_name)
+
+    if not path.exists():
+        rich.print("[green]No config to upgrade[/green]")
+        return
+
+    __upgrade(context_name, path)

abst/cli_commands/context/commands.py

@@ -0,0 +1,51 @@
+import logging
+from pathlib import Path
+
+import click
+import pyperclip
+import rich
+
+from abst.config import default_contexts_location, share_excluded_keys
+from abst.utils.misc_funcs import get_context_data, setup_calls
+
+
+@click.group(help="Contexts commands")
+def context():
+    pass
+
+
+@context.command("list", help="Will list all contexts in ~/.abst/context/ folder")
+@click.option("--debug", is_flag=True, default=False)
+def _list(debug=False):
+    setup_calls(debug)
+    rich.print("Contexts:")
+    for file in Path(default_contexts_location).iterdir():
+        rich.print(f"    {file.name.replace('.json', '')}")
+
+
+@context.command(help="Will display JSON format of context")
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("name")
+def display(name, debug=False):
+    setup_calls(debug)
+    data = get_context_data(name)
+    if data is None:
+        return
+    rich.print_json(data=data)
+
+
+@context.command(help="Will print context without local paths and put it in clipboard for sharing")
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("name")
+def share(name, debug=False):
+    setup_calls(debug)
+
+    data = get_context_data(name)
+    if data is None:
+        return
+    for key in share_excluded_keys:
+        data.pop(key)
+    data["default-name"] = "!YOUR NAME!"
+    rich.print_json(data=data)
+    logging.info("Data transmitted into clipboard")
+    pyperclip.copy(str(data))

abst/cli_commands/cp_cli/commands.py

@@ -0,0 +1,42 @@
+import os
+
+import click
+import rich
+
+from abst.utils.misc_funcs import setup_calls
+
+
+@click.group("cp", help="Copy commands for special usage")
+def cp():
+    pass
+
+
+@cp.command("login", help="Login to docker and helm registry")
+@click.argument("secret_name")
+@click.argument("source_namespace")
+@click.argument("target_namespace")
+@click.option("--debug", is_flag=True, default=False)
+def cp_secret(
+        secret_name: str,
+        target_namespace: str,
+        source_namespace: str = "default",
+        debug=False,
+):
+    """
+    Copy Secret in current cluster from source namespace to target
+    @param secret_name: Secret Name
+    @param target_namespace: Target Namespace name
+    @param source_namespace: Source Namespace name
+    @return:
+    :param debug:
+    """
+    setup_calls(debug)
+    try:
+        rich.print("Trying Copy secret")
+        os.system(
+            f"kubectl get secret {secret_name} --namespace={source_namespace} -o yaml | sed "
+            f"'s/namespace: .*/namespace: {target_namespace}/' | kubectl apply -f -"
+        )
+    except FileNotFoundError:
+        rich.print("[red]kubectl not found on this machine[/red]")
+        return

abst/cli_commands/create_cli/commands.py

@@ -0,0 +1,74 @@
+from time import sleep
+
+import click
+
+from abst.bastion_support.oci_bastion import Bastion
+from abst.utils.misc_funcs import setup_calls, print_eligible
+
+
+@click.group(help="Group of commands for creating Bastion sessions")
+def create():
+    pass
+
+
+@create.command(
+    "forward",
+    help="Creates and connects to Bastion session indefinitely until terminated by user",
+)
+@click.option("--shell", is_flag=True, default=False)
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def fullauto_forward(shell, debug, context_name):
+    """Creates and connects to bastion sessions
+    automatically until terminated"""
+
+    setup_calls(debug)
+
+    if context_name == "?":
+        print_eligible("required only for Port Forward session")
+        return
+
+    if context_name is None:
+        conf = Bastion.load_config()
+        used_name = conf["used_context"]
+    else:
+        used_name = context_name
+
+    while True:
+        Bastion(used_name, region=
+        Bastion.load_json(Bastion.get_creds_path_resolve(context_name)).get("region",
+                                                                            None)).create_forward_loop(
+            shell=shell)
+
+        sleep(1)
+
+
+@create.command(
+    "managed",
+    help="Creates and connects to Bastion session indefinitely until terminated by user",
+)
+@click.option("--shell", is_flag=True, default=False)
+@click.option("--debug", is_flag=True, default=False)
+@click.argument("context-name", default=None, required=False)
+def fullauto_managed(shell, debug, context_name):
+    """Creates and connects to bastion sessions
+    automatically until terminated"""
+    setup_calls(debug)
+
+    if context_name == "?":
+        print_eligible("required only for Managed SSH session")
+        return
+
+    if context_name is None:
+        conf = Bastion.load_config()
+        used_name = conf["used_context"]
+    else:
+        used_name = context_name
+
+    while True:
+        Bastion(used_name, region=
+        Bastion.load_json(Bastion.get_creds_path_resolve(context_name)).get("region",
+                                                                            None)).create_managed_loop(
+            shell=shell)
+
+        sleep(1)