in mds_plugin/bastion.py [0:0]
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}')