mds_plugin/bastion.py

# Copyright (c) 2021, 2024, Oracle and/or its affiliates.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License, version 2.0,
# as published by the Free Software Foundation.
#
# This program is designed to work with certain software (including
# but not limited to OpenSSL) that is licensed under separate terms, as
# designated in a particular file or component or in included license
# documentation.  The authors of MySQL hereby grant you an additional
# permission to link the program and your derivative works with the
# separately licensed software that they have either included with
# the program or referenced in the documentation.
#
# This program is distributed in the hope that it will be useful,  but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
# the GNU General Public License, version 2.0, for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA

"""Sub-Module to manage OCI Bastions"""

from code import interact
from mysqlsh.plugin_manager import plugin_function
from mds_plugin import core, configuration, compute, mysql_database_service
from mds_plugin import network

# cSpell:ignore vnics, vnic


def format_bastion_listing(items, current=None) -> str:
    """Formats a given list of objects in a human readable form

    Args:
        items: Either a list of objects or a single object
        current (str): OCID of the current item

    Returns:
       The db_systems formatted as str
    """

    # If a single db_system was given, wrap it in a list
    if not type(items) is list:
        items = [items]

    # return objects in READABLE text output
    out = ""
    id = 1
    for i in items:
        index = f"*{id:>3} " if current == i.id else f"{id:>4} "
        out += (index +
                core.fixed_len(i.name, 24, ' ', True) +
                core.fixed_len(i.lifecycle_state, 8, ' ') +
                core.fixed_len(f"{i.time_created:%Y-%m-%d %H:%M}", 16, '\n'))
        id += 1

    return out


def format_session_listing(items, current=None) -> str:
    """Formats a given list of objects in a human readable form

    Args:
        items: Either a list of objects or a single object
        current (str): OCID of the current item

    Returns:
       The db_systems formatted as str
    """

    # If a single db_system was given, wrap it in a list
    if not type(items) is list:
        items = [items]

    # return objects in READABLE text output
    out = ""
    id = 1
    for i in items:
        index = f"*{id:>3} " if current == i.id else f"{id:>4} "
        s_type = i.target_resource_details.session_type
        s_port = str(i.target_resource_details.target_resource_port)
        s_ip = i.target_resource_details.target_resource_private_ip_address
        s_time = f'{i.time_created:%Y%m%d-%H%M%S}'
        out += (index +
                core.fixed_len(i.display_name, 24, ' ', True) +
                core.fixed_len(i.lifecycle_state, 8, ' ') +
                core.fixed_len(s_type, 15, ' ') +
                core.fixed_len(s_ip, 15, ' ') +
                core.fixed_len(s_port, 15, ' ') +
                core.fixed_len(s_time, 15, '\n'))
        id += 1
        # try:
        #     out += "       " + i.ssh_metadata.get("command") + "\n"
        # except:
        #     pass

    return out


@plugin_function('mds.list.bastions', shell=True, cli=True, web=True)
def list_bastions(**kwargs):
    """Lists bastions

    This function will list all bastions of the compartment with the
    given compartment_id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        compartment_id (str): OCID of the parent compartment
        valid_for_db_system_id (str): OCID of the db_system_id the bastions
            needs to be valid for and therefore are in the same subnet
        config (dict): An OCI config object or None
        config_profile (str): The name of an OCI config profile
        interactive (bool): Indicates whether to execute in interactive mode
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
        Based on return_type
    """

    compartment_id = kwargs.get("compartment_id")
    valid_for_db_system_id = kwargs.get("valid_for_db_system_id")

    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get(
        "return_type",  # In interactive mode, default to formatted str return
        core.RETURN_STR if interactive else core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        current_bastion_id = configuration.get_current_bastion_id(
            config=config)

        import oci.exceptions
        try:
            # Initialize the  Object Store client
            bastion_client = core.get_oci_bastion_client(config=config)

            # List the bastions
            bastions = bastion_client.list_bastions(
                compartment_id=compartment_id).data

            # Filter out all deleted bastions
            bastions = [b for b in bastions if b.lifecycle_state != "DELETED"]

            # Filter out all bastions that are not valid for the given DbSystem
            if valid_for_db_system_id:
                # Just consider active bastions here
                bastions = [
                    b for b in bastions if b.lifecycle_state == "ACTIVE"]
                valid_bastions = []
                db_system = mysql_database_service.get_db_system(
                    db_system_id=valid_for_db_system_id,
                    config=config, interactive=False,
                    return_python_object=True)
                for b in bastions:
                    bastion = bastion_client.get_bastion(bastion_id=b.id).data
                    if bastion.target_subnet_id == db_system.subnet_id:
                        valid_bastions.append(bastion)
                bastions = valid_bastions

            # Add the is_current field
            current_bastion_id = configuration.get_current_bastion_id(
                config=config)

            for b in bastions:
                # Add the is_current field to the object, also adding it to
                # the swagger_types so oci.util.to_dict() does include it
                setattr(b, "is_current", b.id == current_bastion_id)
                b.swagger_types["is_current"] = "bool"
                # bastions[i] = b

            if len(bastions) < 1 and interactive:
                print("This compartment contains no bastions.")
                bastions = None

            return core.oci_object(
                oci_object=bastions,
                return_type=return_type,
                format_function=format_bastion_listing,
                current=current_bastion_id)
        except oci.exceptions.ServiceError as e:
            if raise_exceptions:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except Exception as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.get.bastion', shell=True, cli=True, web=True)
def get_bastion(**kwargs):
    """Gets a Bastion with the given id

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        bastion_name (str): The new name bastion
        bastion_id (str): OCID of the bastion.
        await_state (str): Await the given lifecycle state, ACTIVE, DELETED, ..
        ignore_current (bool): Whether the current bastion should be ignored
        fallback_to_any_in_compartment (bool): Whether to lookup bastion in
            compartment
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        interactive (bool): Indicates whether to execute in interactive mode
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       None
    """

    bastion_name = kwargs.get("bastion_name")
    bastion_id = kwargs.get("bastion_id")
    await_state = kwargs.get("await_state")
    ignore_current = kwargs.get("ignore_current", False)
    fallback_to_any_in_compartment = kwargs.get(
        "fallback_to_any_in_compartment", False)

    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get("return_type", core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        if not ignore_current and bastion_name is None:
            bastion_id = configuration.get_current_bastion_id(
                bastion_id=bastion_id, config=config)

        import oci.exceptions
        try:
            # Initialize the Bastion client
            bastion_client = core.get_oci_bastion_client(config=config)

            if not interactive and not bastion_id and not bastion_name:
                raise ValueError('No bastion_id nor bastion_name given.')

            bastion = None

            if bastion_id:
                bastion = bastion_client.get_bastion(
                    bastion_id=bastion_id).data
                if not bastion:
                    raise ValueError('The bastion with the given OCID '
                                     f'{bastion_id} was not found.')

            if not bastion and (bastion_name or interactive):
                # List the Bastion of the current compartment
                bastions = bastion_client.list_bastions(
                    compartment_id=compartment_id).data

                # Filter out all deleted compartments
                bastions = [
                    u for u in bastions if u.lifecycle_state != "DELETED"]

                if len(bastions) == 0:
                    if interactive:
                        print("No Bastions available in this compartment.")
                    return

                # If a bastion name was given, look it up
                if bastion_name:
                    for b in bastions:
                        if b.name == bastion_name:
                            bastion = bastion_client.get_bastion(
                                bastion_id=b.id).data
                            break

                    if bastion is None and not interactive:
                        raise ValueError(f"Bastion {bastion_name} was not "
                                         "found in this compartment.")

                # Fallback to first in compartment
                if fallback_to_any_in_compartment:
                    bastion = bastion_client.get_bastion(
                        bastion_id=bastions[0].id).data

            if not bastion and interactive:
                # If the user_name was not given or found, print out the list
                bastion_list = format_bastion_listing(items=bastions)
                if bastion_name is None:
                    print(bastion_list)

                # Let the user choose from the list
                selected_bastion = core.prompt_for_list_item(
                    item_list=bastions,
                    prompt_caption=("Please enter the name or index "
                                    "of the Bastion: "),
                    item_name_property="name", given_value=bastion_name)

                if selected_bastion:
                    bastion = bastion_client.get_bastion(
                        bastion_id=selected_bastion.id).data

            if bastion and await_state:
                import time
                if interactive:
                    print(f'Waiting for Bastion to reach '
                          f'{await_state} state...')

                bastion_id = bastion.id

                # Wait for the Bastion Session to reach state await_state
                cycles = 0
                while cycles < 48:
                    bastion = bastion_client.get_bastion(
                        bastion_id=bastion_id).data
                    if bastion.lifecycle_state == await_state:
                        break
                    else:
                        time.sleep(5)
                        s = "." * (cycles + 1)
                        if interactive:
                            print(f'Waiting for Bastion to reach '
                                  f'{await_state} state...{s}')
                    cycles += 1

                if bastion.lifecycle_state != await_state:
                    raise Exception("Bastion did not reach the state "
                                    f"{await_state} within 4 minutes.")

            return core.oci_object(
                oci_object=bastion,
                return_type=return_type,
                format_function=format_bastion_listing)
        except oci.exceptions.ServiceError as e:
            if interactive:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
            return
    except Exception as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.create.bastion', shell=True, cli=True, web=True)
def create_bastion(**kwargs):
    """Creates a Bastion

    Args:
        **kwargs: Additional options

    Keyword Args:
        bastion_name (str): The new name of the compartment.
        db_system_id (str): OCID of the DbSystem.
        client_cidr (str): The client CIDR, defaults to "0.0.0.0/0"
        max_session_ttl_in_seconds (int): The maximum amount of time that any
            session on the bastion can remain active, defaults to 10800
        target_subnet_id (str): The OCID of the subnet, defaults to the
            subnet of the db_system if db_system_id is given
        await_active_state (bool): Await the ACTIVE lifecycle state before
            returning
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        ignore_current (bool): Whether the current DbSystem should be ignored
        interactive (bool): Whether exceptions are raised
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       The id of the Bastion Session, None in interactive mode
    """

    bastion_name = kwargs.get("bastion_name")
    db_system_id = kwargs.get("db_system_id")
    client_cidr = kwargs.get("client_cidr", "0.0.0.0/0")
    max_session_ttl_in_seconds = kwargs.get(
        "max_session_ttl_in_seconds", 10800)
    target_subnet_id = kwargs.get("target_subnet_id")
    await_active_state = kwargs.get("await_active_state", False)

    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")
    ignore_current = kwargs.get("ignore_current", False)

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get(
        "return_type",  # In interactive mode, default to formatted str return
        core.RETURN_STR if interactive else core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        current_compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        if not ignore_current:
            db_system_id = configuration.get_current_db_system_id(
                db_system_id=db_system_id, config=config)

        import oci.bastion.models
        import oci.mysql.models
        import oci.exceptions
        try:
            # Initialize the Bastion client
            bastion_client = core.get_oci_bastion_client(config=config)
            db_system = None

            if db_system_id:
                db_system = mysql_database_service.get_db_system(
                    db_system_id=db_system_id,
                    config=config, interactive=False,
                    return_python_object=True)
                if not db_system:
                    raise ValueError("No db_system found with the given id. "
                                     "Operation cancelled.")
            elif interactive:
                for_db_system = core.prompt(
                    "Should the new Bastion be used to connect to "
                    "a MySQL DB System? [Y/n]: ",
                    options={'defaultValue': 'y'})
                if not for_db_system:
                    raise ValueError("Operation cancelled.")

                if for_db_system.lower() == 'y':
                    db_system = mysql_database_service.get_db_system(
                        compartment_id=current_compartment_id,
                        config=config, interactive=interactive,
                        return_python_object=True)
                    if not db_system:
                        raise ValueError("Operation cancelled.")
            else:
                raise ValueError("No db_system_id given. "
                                 "Operation cancelled.")

            # Check if the db_system already has a Bastion set in the
            # freeform_tags
            # if db_system and db_system.freeform_tags.get('bastion_id'):
            #     bastion = None
            #     # Check if that bastion still exists
            #     try:
            #         print("Check if that bastion still exists")
            #         bastion = get_bastion(
            #             bastion_id=db_system.freeform_tags.get('bastion_id'),
            #             return_type="OBJ",
            #             config=config, interactive=False)
            #     except ValueError:
            #         # If not, remove that old bastion id from the freeform_tags
            #         db_system.freeform_tags.pop('bastion_id', None)
            #         mysql_database_service.update_db_system(
            #             db_system_id=db_system.id,
            #             new_freeform_tags=db_system.freeform_tags,
            #             config=config, interactive=False)

            #     # If the assigned bastion does exist, error out
            #     if bastion and bastion.lifecycle_state == \
            #         oci.bastion.models.Bastion.LIFECYCLE_STATE_ACTIVE:
            #         raise ValueError(
            #             "The given MySQL DB System already has a Bastion "
            #             "assigned. Please remove 'bastion_id' from the "
            #             "freeform_tags to create a new Bastion for this "
            #             "DB System. Operation cancelled.")

            # If a db_system was given, take the compartment_id from there
            if not compartment_id and db_system:
                compartment_id = db_system.compartment_id
            elif not compartment_id:
                compartment_id = current_compartment_id

            if not bastion_name:
                if db_system:
                    from datetime import datetime
                    bastion_name = (
                        "Bastion" +
                        datetime.now().strftime("%y%m%d%H%M"))

                elif interactive:
                    bastion_name = core.prompt(
                        'Please enter a name for this new Bastion: ')
                    if not bastion_name:
                        raise ValueError("Operation cancelled.")
            if not bastion_name:
                raise ValueError("No bastion_name given. "
                                 "Operation cancelled.")

            if not target_subnet_id:
                if db_system:
                    target_subnet_id = db_system.subnet_id
                elif interactive:
                    # Get private subnet
                    subnet = network.get_subnet(
                        public_subnet=False,
                        compartment_id=compartment_id, config=config,
                        interactive=interactive)
                    if subnet is None:
                        print("Operation cancelled.")
                        return
                    target_subnet_id = subnet.id
                else:
                    raise ValueError("No target_subnet_id given. "
                                     "Operation cancelled.")

            bastion_details = oci.bastion.models.CreateBastionDetails(
                bastion_type="standard",
                client_cidr_block_allow_list=[client_cidr],
                compartment_id=compartment_id,
                max_session_ttl_in_seconds=max_session_ttl_in_seconds,
                name=bastion_name,
                target_subnet_id=target_subnet_id
            )

            # Create the new bastion
            new_bastion = bastion_client.create_bastion(
                create_bastion_details=bastion_details).data

            # Update the db_system freeform_tags to hold the assigned bastion
            # if db_system:
            #     print("Update the db_system freeform_tags to hold the assigned bastion ")
            #     db_system.freeform_tags["bastion_id"] = new_bastion.id

            #     mysql_database_service.update_db_system(
            #         db_system_id=db_system.id,
            #         new_freeform_tags=db_system.freeform_tags,
            #         config=config, interactive=False)

            if new_bastion and await_active_state:
                import time
                if interactive:
                    print(f'Waiting for Bastion to reach '
                          f'ACTIVE state...')

                bastion_id = new_bastion.id

                # Wait for the Bastion Session to reach state await_state
                cycles = 0
                while cycles < 60:
                    bastion = bastion_client.get_bastion(
                        bastion_id=bastion_id).data
                    if bastion.lifecycle_state == "ACTIVE":
                        break
                    else:
                        time.sleep(5)
                        s = "." * (cycles + 1)
                        if interactive:
                            print(f'Waiting for Bastion to reach '
                                  f'ACTIVE state...{s}')
                    cycles += 1

                if bastion.lifecycle_state != "ACTIVE":
                    raise Exception("Bastion did not reach the state "
                                    f"ACTIVE within 5 minutes.")

                return core.oci_object(
                    oci_object=bastion,
                    return_type=return_type,
                    format_function=lambda b: print(
                        f"Bastion {b.name} has been created."))

            else:
                return core.oci_object(
                    oci_object=new_bastion,
                    return_type=return_type,
                    format_function=lambda b: print(
                        f"Bastion {b.name} is being "
                        f"created. Use mds.list.bastions() to check "
                        "it's provisioning state.\n"))

        except oci.exceptions.ServiceError as e:
            if raise_exceptions:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except Exception as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.delete.bastion', shell=True, cli=True, web=True)
def delete_bastion(**kwargs):
    """Deletes a Bastion with the given id

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        bastion_name (str): The new name bastion
        bastion_id (str): OCID of the bastion.
        await_deletion (bool): Whether to wait till the bastion reaches
            lifecycle state DELETED
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        ignore_current (bool): Whether the current bastion should be ignored
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       None
    """

    bastion_name = kwargs.get("bastion_name")
    bastion_id = kwargs.get("bastion_id")
    await_deletion = kwargs.get("await_deletion")

    compartment_id = kwargs.get("compartment_id")
    ignore_current = kwargs.get("ignore_current", True)
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)

    import mysqlsh

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        if not ignore_current and bastion_name is None:
            bastion_id = configuration.get_current_bastion_id(
                bastion_id=bastion_id, config=config)

        import oci.identity
        import oci.mysql
        try:
            bastion = get_bastion(
                bastion_name=bastion_name,
                bastion_id=bastion_id, compartment_id=compartment_id,
                config=config, ignore_current=ignore_current,
                interactive=interactive, return_type=core.RETURN_OBJ)
            if not bastion:
                raise ValueError(
                    "No bastion given or found. "
                    "Operation cancelled.")
            bastion_id = bastion.id

            if interactive:
                # Prompt the user for specifying a compartment
                prompt = mysqlsh.globals.shell.prompt(
                    f"Are you sure you want to delete the Bastion "
                    f"{bastion.name} [yes/NO]: ",
                    {'defaultValue': 'no'}).strip().lower()
                if prompt != "yes":
                    raise Exception("Deletion aborted.\n")

            # Initialize the Bastion client
            bastion_client = core.get_oci_bastion_client(config=config)

            bastion_client.delete_bastion(bastion_id=bastion_id)

            # Get db_system client
            db_system_client = core.get_oci_db_system_client(config=config)

            # Update db_systems in the compartment and remove the bastion_id
            # db_systems = mysql_database_service.list_db_systems(
            #     compartment_id=bastion.compartment_id,
            #     config=config, interactive=False)
            # for db_system in db_systems:
            #     if "bastion_id" in db_system.get("freeform_tags"):
            #         if db_system.get("freeform_tags").get("bastion_id") == \
            #                 bastion_id:
            #             # Remove the "bastion_id" key from the dict
            #             db_system.get("freeform_tags").pop("bastion_id")

            #             # Update the db_system
            #             update_details = oci.mysql.models.UpdateDbSystemDetails(
            #                 freeform_tags=db_system.get("freeform_tags"))
            #             db_system_client.update_db_system(
            #                 db_system.get("id"), update_details)

            # If the current bastion has been deleted, clear it
            if configuration.get_current_bastion_id(
                    config=config) == bastion_id:
                configuration.set_current_bastion(bastion_id='')

            # If the function should wait till the bastion reaches the DELETED
            # lifecycle state
            if await_deletion:
                import time
                if interactive:
                    print('Waiting for Bastion to be deleted...')

                # Wait for the Bastion Session to be ACTIVE
                cycles = 0
                while cycles < 48:
                    bastion = bastion_client.get_bastion(
                        bastion_id=bastion_id).data
                    if bastion.lifecycle_state == "DELETED":
                        break
                    else:
                        time.sleep(5)
                        s = "." * (cycles + 1)
                        if interactive:
                            print(f'Waiting for Bastion to be deleted...{s}')
                    cycles += 1

                if bastion.lifecycle_state != "DELETED":
                    raise Exception("Bastion did not reach the DELETED "
                                    "state within 4 minutes.")
                if interactive:
                    print(
                        f"Bastion '{bastion.name}' was deleted successfully.")
            elif interactive:
                print(f"Bastion '{bastion.name}' is being deleted.")

        except oci.exceptions.ServiceError as e:
            if interactive:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except (Exception) as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.list.bastionSessions', shell=True, cli=True, web=True)
def list_sessions(**kwargs):
    """Lists bastion sessions

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        bastion_id (str): OCID of the bastion.
        ignore_current (bool): Whether to not default to the current bastion.
        compartment_id (str): OCID of the parent compartment.
        config (object): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        interactive (bool): Whether output is more descriptive
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
        A list of dicts representing the bastions
    """

    bastion_id = kwargs.get("bastion_id")
    ignore_current = kwargs.get("ignore_current", False)
    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get(
        "return_type",  # In interactive mode, default to formatted str return
        core.RETURN_STR if interactive else core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        current_bastion_id = configuration.get_current_bastion_id(
            config=config)

        import oci.exceptions
        try:
            # Initialize the  Object Store client
            bastion_client = core.get_oci_bastion_client(config=config)

            if not bastion_id and not ignore_current:
                bastion_id = current_bastion_id

            bastion = get_bastion(
                bastion_id=bastion_id,
                compartment_id=compartment_id, config=config,
                ignore_current=ignore_current, interactive=interactive,
                return_type=core.RETURN_OBJ)

            sessions = bastion_client.list_sessions(bastion_id=bastion.id).data

            # Filter out all deleted sessions
            # bastions = [s for s in sessions if s.lifecycle_state != "DELETED"]

            ext_sessions = []
            for s in sessions:
                if s.lifecycle_state != "DELETED":
                    ext_sessions.append(
                        bastion_client.get_session(session_id=s.id).data)
                else:
                    ext_sessions.append(s)

            if len(ext_sessions) < 1 and interactive:
                print("This bastion contains no sessions.")

            return core.oci_object(
                oci_object=ext_sessions,
                return_type=return_type,
                format_function=format_session_listing)
        except oci.exceptions.ServiceError as e:
            if raise_exceptions:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except Exception as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.get.bastionSession', shell=True, cli=True, web=True)
def get_session(**kwargs):
    """Gets a Bastion Session with the given id

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        session_name (str): The name of the session.
        session_id (str): OCID of the session.
        bastion_id (str): OCID of the bastion.
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        ignore_current (bool): Whether the current bastion should be ignored
        interactive (bool): Whether exceptions are raised
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       None
    """

    session_name = kwargs.get("session_name")
    session_id = kwargs.get("session_id")
    bastion_id = kwargs.get("bastion_id")
    compartment_id = kwargs.get("compartment_id")
    ignore_current = kwargs.get("ignore_current", False)
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get("return_type", core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)

        import oci.exceptions
        try:
            # Initialize the Bastion client
            bastion_client = core.get_oci_bastion_client(config=config)

            if session_id:
                return bastion_client.get_session(session_id=session_id).data

            if not bastion_id and interactive:
                bastion_id = get_bastion(
                    compartment_id=compartment_id,
                    config=config, ignore_current=ignore_current,
                    interactive=interactive, return_type=core.RETURN_OBJ).id
            if not bastion_id:
                raise Exception("No bastion_id given. "
                                "Cancelling operation.")

            # List the Bastion of the current compartment
            sessions = bastion_client.list_sessions(bastion_id=bastion_id).data

            if len(sessions) == 0:
                print("No Bastions available in this compartment.")
                return

            # If the user_name was not given or found, print out the list
            sessions_list = format_session_listing(sessions)
            if session_name is None:
                print(sessions_list)

            # Let the user choose from the list
            selected_session = core.prompt_for_list_item(
                item_list=sessions,
                prompt_caption=("Please enter the name or index "
                                "of the Bastion Session: "),
                item_name_property="display_name", given_value=session_name)

            session = bastion_client.get_session(
                session_id=selected_session.id).data

            return core.oci_object(
                oci_object=session,
                return_type=return_type,
                format_function=format_session_listing)
        except oci.exceptions.ServiceError as e:
            if raise_exceptions:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except (ValueError, oci.exceptions.ClientError) as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.create.bastionSession', shell=True, cli=True, web=True)
def create_session(**kwargs):
    """Creates a Bastion Session for the given bastion_id

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Additional options

    Keyword Args:
        bastion_name (str): The new name of the compartment.
        bastion_id (str): OCID of the Bastion.
        fallback_to_any_in_compartment (bool): Fallback to first bastion in
            the given compartment
        session_type (str): The type of the session, either MANAGED_SSH or
            PORT_FORWARDING
        session_name (str): The name of the session.
        target_id (str): OCID of the session target.
        target_ip (str): The TCP/IP address of the session target.
        target_port (str): The TCP/IP Port of the session target.
        target_user (str): The user account on the session target.
        ttl_in_seconds (int): Time to live for the session, max 10800.
        public_key_file (str): The filename of a public SSH key.
        public_key_content (str): The public SSH key.
        await_creation (bool): Whether to wait till the bastion reaches
            lifecycle state ACTIVE
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        ignore_current (bool): Whether the current Bastion should be ignored
        interactive (bool): Whether exceptions are raised
        return_type (str): "STR" will return a formatted string, "DICT" will
            return the object converted to a dict structure and "OBJ" will
            return the OCI Python object for internal plugin usage
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       The id of the Bastion Session, None in interactive mode
    """

    bastion_name = kwargs.get("bastion_name")
    bastion_id = kwargs.get("bastion_id")
    session_type = kwargs.get("session_type")
    session_name = kwargs.get("session_name")
    target_id = kwargs.get("target_id")
    target_ip = kwargs.get("target_ip")
    target_port = kwargs.get("target_port")
    target_user = kwargs.get("target_user")
    ttl_in_seconds = kwargs.get("ttl_in_seconds", 10800)
    public_key_file = kwargs.get("public_key_file")
    public_key_content = kwargs.get("public_key_content")

    await_creation = kwargs.get("await_creation")

    compartment_id = kwargs.get("compartment_id")
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")
    ignore_current = kwargs.get("ignore_current", False)
    fallback_to_any_in_compartment = kwargs.get(
        "fallback_to_any_in_compartment", False)

    interactive = kwargs.get("interactive", core.get_interactive_default())
    return_type = kwargs.get(
        "return_type",  # In interactive mode, default to formatted str return
        core.RETURN_STR if interactive else core.RETURN_DICT)
    raise_exceptions = kwargs.get(
        "raise_exceptions",  # On internal call (RETURN_OBJ), raise exceptions
        True if return_type == core.RETURN_OBJ else not interactive)

    from datetime import datetime

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config,
            config_profile=config_profile, interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        if not ignore_current and not bastion_name:
            bastion_id = configuration.get_current_bastion_id(
                bastion_id=bastion_id, config=config)

        # Initialize the Bastion client
        bastion_client = core.get_oci_bastion_client(config=config)

        if not bastion_id and (bastion_name or interactive):
            bastion_id = get_bastion(
                bastion_name=bastion_name,
                fallback_to_any_in_compartment=fallback_to_any_in_compartment,
                compartment_id=compartment_id,
                config=config, ignore_current=ignore_current,
                interactive=interactive, return_type=core.RETURN_OBJ).id

        if not bastion_id:
            raise ValueError("No bastion_id or bastion_name specified.")

        if session_type == "MANAGED_SSH":
            default_name = f'COMPUTE-{datetime.now():%Y%m%d-%H%M%S}'
        else:
            default_name = f'MDS-{datetime.now():%Y%m%d-%H%M%S}'

        if not session_name and interactive:
            session_name = core.prompt(
                "Please specify a name for the Bastion Session "
                f"({default_name}): ",
                {'defaultValue': default_name}).strip()

        if not session_type and interactive:
            session_type = core.prompt_for_list_item(
                item_list=["MANAGED_SSH", "PORT_FORWARDING"],
                prompt_caption=(
                    "Please select the Bastion Session type "
                    "(PORT_FORWARDING): "),
                print_list=True,
                prompt_default_value="PORT_FORWARDING")

        if not session_type:
            raise ValueError("No session_type given. "
                             "Operation cancelled.")

        if session_type == "MANAGED_SSH":
            if not target_id and interactive:
                instance = compute.get_instance(
                    compartment_id=compartment_id, config=config,
                    interactive=interactive,
                    return_python_object=True)
                if not instance:
                    raise Exception("No target_id specified."
                                    "Cancelling operation")

                target_id = instance.id

            if target_id and not target_ip:
                vnics = compute.list_vnics(
                    instance_id=target_id,
                    compartment_id=compartment_id, config=config,
                    interactive=False,
                    return_python_object=True)
                for vnic in vnics:
                    if vnic.private_ip:
                        target_ip = vnic.private_ip
                        break
                if not target_ip:
                    raise ValueError(
                        'No private IP found for this compute instance.')

            if not target_port:
                target_port = 22

            if not target_user:
                target_user = "opc"
        else:
            if not target_ip and interactive:
                db_system = mysql_database_service.get_db_system(
                    compartment_id=compartment_id, config=config,
                    return_python_object=True)
                if not db_system:
                    raise Exception("No target_id specified."
                                    "Cancelling operation")

                endpoint = db_system.endpoints[0]
                target_ip = endpoint.ip_address
                if not target_port:
                    target_port = endpoint.port

        if not target_ip:
            raise Exception("No target_ip specified."
                            "Cancelling operation")

        if not target_port:
            raise Exception("No target_port specified."
                            "Cancelling operation")

        if session_type == "MANAGED_SSH":
            if not target_id:
                raise Exception("No target_id specified."
                                "Cancelling operation")
            if not target_user:
                raise Exception("No target_user specified."
                                "Cancelling operation")

        # Convert Unix path to Windows
        import os.path
        ssh_key_location = os.path.abspath(os.path.expanduser("~/.ssh"))

        if not public_key_file and not public_key_content and interactive:
            from os import listdir

            files = [f for f in listdir(ssh_key_location)
                     if os.path.isfile(os.path.join(ssh_key_location, f)) and
                     f.lower().endswith(".pub")]

            public_key_file = core.prompt_for_list_item(
                item_list=files, print_list=True,
                prompt_caption="Please select a public SSH Key: ")
            if not public_key_file:
                raise Exception("No public SSH Key specified."
                                "Cancelling operation")

        default_key_file = os.path.join(ssh_key_location, "id_rsa_oci_tunnel.pub")
        if not public_key_file and os.path.exists(default_key_file):
            public_key_file = default_key_file

        if public_key_file and not public_key_content:
            with open(os.path.join(
                    ssh_key_location, public_key_file), 'r') as file:
                public_key_content = file.read()

        if not public_key_content:
            raise Exception("No public SSH Key specified. "
                            "Cancelling operation")

        import oci.identity
        import oci.bastion
        import hashlib
        try:
            if session_type == "MANAGED_SSH":
                target_details = oci.bastion.models.\
                    CreateManagedSshSessionTargetResourceDetails(
                        target_resource_id=target_id,
                        target_resource_port=target_port,
                        target_resource_private_ip_address=target_ip,
                        target_resource_operating_system_user_name=target_user)
            else:
                target_details = oci.bastion.models.\
                    CreatePortForwardingSessionTargetResourceDetails(
                        target_resource_id=target_id,
                        target_resource_port=target_port,
                        target_resource_private_ip_address=target_ip)

            if not session_name:
                # Calculate unique fingerprint based on all params
                params = (
                    target_id + bastion_id +
                    config_profile + public_key_content +
                    f'{target_ip}:{target_port}')

                fingerprint = hashlib.md5(params.encode('utf-8')).hexdigest()
                session_name = f'MDS-{fingerprint}'

            # Check if a session with this fingerprinted name already exists
            sessions = bastion_client.list_sessions(
                bastion_id=bastion_id,
                display_name=session_name,
                session_lifecycle_state='ACTIVE').data
            if len(sessions) == 1:
                # If so, return that session
                return core.oci_object(
                    oci_object=sessions[0],
                    return_type=return_type,
                    format_function=format_session_listing)

            if session_type == "MANAGED_SSH":
                # Check if Bastion Plugin is already enabled
                ia_client = core.get_oci_instance_agent_client(config)
                # cSpell:ignore instanceagent
                bastion_plugin = ia_client.get_instance_agent_plugin(
                    instanceagent_id=target_id,
                    compartment_id=compartment_id,
                    plugin_name="Bastion").data
                if bastion_plugin.status != "RUNNING":
                    # TODO: use UpdateInstanceDetails to enabled the bastion
                    # service
                    raise Exception(
                        'Please enabled the Bastion plugin on '
                        'this instance.')

            session_details = oci.bastion.models.CreateSessionDetails(
                bastion_id=bastion_id,
                display_name=session_name,
                key_details=oci.bastion.models.PublicKeyDetails(
                    public_key_content=public_key_content),
                key_type="PUB",
                session_ttl_in_seconds=ttl_in_seconds,
                target_resource_details=target_details
            )
            new_session = bastion_client.create_session(session_details).data

            if await_creation:
                import time
                # Wait for the Bastion Session to be ACTIVE
                cycles = 0
                while cycles < 24:
                    bastion_session = bastion_client.get_session(
                        session_id=new_session.id).data
                    if bastion_session.lifecycle_state == "ACTIVE":
                        # TODO: Report bug to the Bastion dev team.
                        # Ask them to only switch lifecycle_state to ACTIVE
                        # if the Bastion Session can actually accept connections
                        time.sleep(5)
                        break
                    else:
                        time.sleep(5)
                        s = "." * (cycles + 1)
                        if interactive:
                            print(f'Waiting for Bastion Session to become '
                                  f'active...{s}')
                    cycles += 1

                bastion_session = bastion_client.get_session(
                    session_id=new_session.id).data
                if bastion_session.lifecycle_state != "ACTIVE":
                    raise Exception("Bastion Session did not reach ACTIVE "
                                    "state within 2 minutes.")

            return core.oci_object(
                oci_object=new_session,
                return_type=return_type,
                format_function=lambda s: print(
                    f"Bastion Session {s.display_name} is being\n"
                    f" created. Use mds.list.bastionSessions() to check "
                    "it's provisioning state.\n"))

        except oci.exceptions.ServiceError as e:
            if raise_exceptions:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except Exception as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')


@plugin_function('mds.delete.bastionSession', shell=True, cli=True, web=True)
def delete_session(**kwargs):
    """Deletes a Bastion Session with the given id

    If no id is given, it will prompt the user for the id.

    Args:
        **kwargs: Optional parameters

    Keyword Args:
        session_name (str): The name of the bastion session
        session_id (str): The id of the bastion session
        bastion_name (str): The name of the bastion.
        bastion_id (str): OCID of the bastion.
        compartment_id (str): OCID of the compartment.
        config (dict): An OCI config object or None.
        config_profile (str): The name of an OCI config profile
        ignore_current (bool): Whether the current bastion should be ignored
        interactive (bool): Indicates whether to execute in interactive mode
        raise_exceptions (bool): If set to true exceptions are raised

    Returns:
       None
    """

    session_name = kwargs.get("session_name")
    session_id = kwargs.get("session_id")
    bastion_name = kwargs.get("bastion_name")
    bastion_id = kwargs.get("bastion_id")
    compartment_id = kwargs.get("compartment_id")
    ignore_current = kwargs.get("ignore_current", False)
    config = kwargs.get("config")
    config_profile = kwargs.get("config_profile")

    interactive = kwargs.get("interactive", core.get_interactive_default())
    raise_exceptions = kwargs.get("raise_exceptions", not interactive)

    import mysqlsh

    # Get the active config and compartment
    try:
        config = configuration.get_current_config(
            config=config, config_profile=config_profile,
            interactive=interactive)
        compartment_id = configuration.get_current_compartment_id(
            compartment_id=compartment_id, config=config)
        if not ignore_current and bastion_name is None:
            bastion_id = configuration.get_current_bastion_id(
                bastion_id=bastion_id, config=config)

        import oci.exceptions
        try:
            if not session_id:
                bastion = get_bastion(
                    bastion_name=bastion_name,
                    bastion_id=bastion_id, compartment_id=compartment_id,
                    config=config, ignore_current=ignore_current,
                    interactive=interactive, return_type=core.RETURN_OBJ)
                if not bastion:
                    raise ValueError(
                        "No bastion given or found. "
                        "Operation cancelled.")

                session = get_session(
                    session_name=session_name, session_id=session_id,
                    bastion_id=bastion.id, compartment_id=compartment_id,
                    config=config, ignore_current=ignore_current,
                    interactive=interactive, return_type=core.RETURN_OBJ)
                if not session:
                    raise ValueError(
                        "No bastion session given or found. "
                        "Operation cancelled.")

                session_id = session.id

            if interactive:
                # Prompt the user for specifying a compartment
                prompt = mysqlsh.globals.shell.prompt(
                    f"Are you sure you want to delete the bastion session? "
                    f"[yes/NO]: ",
                    {'defaultValue': 'no'}).strip().lower()
                if prompt != "yes":
                    raise Exception("Deletion aborted.\n")

            # Initialize the Bastion client
            bastion_client = core.get_oci_bastion_client(config=config)

            bastion_client.delete_session(session_id=session_id)

            if interactive:
                print(f"The bastion session is being deleted.")

        except oci.exceptions.ServiceError as e:
            if interactive:
                raise
            print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
    except (Exception) as e:
        if raise_exceptions:
            raise
        print(f'ERROR: {e}')