Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if len(sys.argv) != 7:
raise RuntimeError('Invalid number of arguments provided to the script. Consult the script header for required arguments')
instance_display_name = sys.argv[1]
compartment_id = sys.argv[2]
availability_domain = sys.argv[3]
cidr_block = sys.argv[4]
ssh_public_key_path = os.path.expandvars(os.path.expanduser(sys.argv[5]))
kms_key_id = sys.argv[6]
# Default config file and profile
config = oci.config.from_file()
compute_client = oci.core.ComputeClient(config)
virtual_network_client = oci.core.VirtualNetworkClient(config)
vcn = None
subnet = None
internet_gateway = None
try:
vcn = create_vcn(virtual_network_client, compartment_id, cidr_block)
subnet = create_subnet(virtual_network_client, vcn, availability_domain)
internet_gateway = create_internet_gateway(virtual_network_client, vcn)
image = get_image(compute_client, 'Oracle Linux', '7.5', 'VM.Standard2.1')
with open(ssh_public_key_path, mode='r') as file:
ssh_key = file.read()
# We can use instance metadata to specify the SSH keys to be included in the
# ~/.ssh/authorized_keys file for the default user on the instance via the special "ssh_authorized_keys" key.
# ---------- assign provided arguments
compartment_id = args.compartment_id
vcn_id = args.vcn_id
availability_domain = args.availability_domain
display_name = args.display_name
# vnic_id = args.vnic_id
subnet_id = args.subnet_id
image_id = args.image_id
# ---------- read config from file
config = oci.config.from_file()
compute_client = oci.core.ComputeClient(config)
virtual_network_client = oci.core.VirtualNetworkClient(config)
# Create Network Security Group objects
print("==================")
print("Creating network security group")
print("==================")
new_nsg = create_nsg(virtual_network_client, vcn_id, display_name)
nsg_id = new_nsg.id
print("==================")
print("Created network security group {}".format(nsg_id))
print("==================")
time.sleep(2)
print("==================")
print("Creating second network security group")
print("==================")
new_nsg_2 = create_nsg(virtual_network_client, vcn_id, "Second network security group")
nsg_id_2 = new_nsg_2.id
[oci.core.models.Subnet.LIFECYCLE_STATE_TERMINATED]
)
print('Deleted Subnet 2')
virtual_network_client_composite_ops.delete_vcn_and_wait_for_state(
vcn.id,
[oci.core.models.Vcn.LIFECYCLE_STATE_TERMINATED]
)
print('Deleted VCN')
# Default config file and profile
config = oci.config.from_file()
load_balancer_client = oci.load_balancer.LoadBalancerClient(config)
load_balancer_client_composite_ops = oci.load_balancer.LoadBalancerClientCompositeOperations(load_balancer_client)
virtual_network_client = oci.core.VirtualNetworkClient(config)
virtual_network_client_composite_ops = oci.core.VirtualNetworkClientCompositeOperations(virtual_network_client)
if len(sys.argv) != 4:
raise RuntimeError('This script needs to be provided a compartment ID and two availability domains')
compartment_id = sys.argv[1]
first_ad = sys.argv[2]
second_ad = sys.argv[3]
vcn_and_subnets = create_vcn_and_subnets(virtual_network_client_composite_ops, compartment_id, first_ad, second_ad)
# Load Balancer operations return work requests so when using composite operations we have to wait for the state of the
# work request (e.g. for it to succeed) rather than the state of the load balancer. However, as a convenience, when the
# composite operation completes we'll return information on the load balancer (if possible) rather than the work
# request
get_load_balancer_response = load_balancer_client_composite_ops.create_load_balancer_and_wait_for_state(
response_data = virtual_network_client.create_service_gateway_and_wait_for_state(
create_sgw_details,
wait_for_states=[oci.core.models.ServiceGateway.LIFECYCLE_STATE_AVAILABLE]
).data
print("Created Service Gateway %s" % response_data.id)
# ---------- read config from file
config = oci.config.from_file()
compute_client = oci.core.ComputeClient(config)
# Create Virtual Network Client with configuration for composite operations
virtual_network_client = oci.core.VirtualNetworkClientCompositeOperations(
oci.core.VirtualNetworkClient(config)
)
# create Service Gateway
create_service_gateway(
virtual_network_client,
compartment_id,
vcn_id,
display_name
)
self.tenancy = tenancy or self.ipa_config['tenancy']
self.signing_key_fingerprint = (
signing_key_fingerprint
or self.ipa_config['signing_key_fingerprint']
)
self.signing_key_file = (
signing_key_file or self.ipa_config['signing_key_file']
)
config = self._get_config()
self.compute_client = oci.core.ComputeClient(config)
self.compute_composite_client = \
oci.core.ComputeClientCompositeOperations(
self.compute_client
)
self.vnet_client = oci.core.VirtualNetworkClient(config)
self.vnet_composite_client = \
oci.core.VirtualNetworkClientCompositeOperations(
self.vnet_client
)
except oci.exceptions.ServiceError as e:
attempts += 1
if e.status == 409 and attempts < 5:
time.sleep(50)
else:
raise
composite_virtual_network.delete_vcn_and_wait_for_state(
vcn.id,
[oci.core.models.Vcn.LIFECYCLE_STATE_TERMINATED]
)
config = oci.config.from_file()
iam_client = oci.identity.IdentityClient(config)
file_storage_client = oci.file_storage.FileStorageClient(config)
virtual_network_client = oci.core.VirtualNetworkClient(config)
if len(sys.argv) != 5:
raise RuntimeError('This script expects an argument of the compartment OCID '
'and availability domain where the file system will be created. '
'It also expects defined tag namespace/key.')
# The first argument is the name of the script, so start the index at 1
compartment_id = sys.argv[1]
availability_domain = sys.argv[2]
namespace = sys.argv[3]
defined_key = sys.argv[4]
# Here we apply a retry strategy to the call to ride out any throttles, timeouts or intermittent 500s (internal server
# errors). The retry strategy will also make requests with an opc-retry-token that it generates.
#
# If you do not use the retry_strategy (or have an alternate way of retrying you wish to use instead) we still
def DeleteReservedIPs(config, compartment):
AllItems = []
object = oci.core.VirtualNetworkClient(config)
print("Getting all Reserved IP objects")
items = oci.pagination.list_call_get_all_results(object.list_public_ips, scope="REGION",compartment_id=compartment.id, lifetime="RESERVED").data
for item in items:
if (item.lifecycle_state != "TERMINATED"):
AllItems.append(item)
print("- {} - {}".format(item.display_name, item.lifecycle_state))
itemsPresent = True
if itemsPresent:
count = 0
for item in AllItems:
try:
itemstatus = object.get_public_ip(public_ip_id=item.id).data
def DeleteDRGs(config, compartment):
AllItems = []
object = oci.core.VirtualNetworkClient(config)
print ("Getting DRGs for {}".format(compartment.name))
AllItems = oci.pagination.list_call_get_all_results(object.list_drgs, compartment_id=compartment.id).data
itemsPresent = True
while itemsPresent:
count = 0
for item in AllItems:
try:
itemstatus = object.get_drg_attachment(drg_attachment_id=item.id).data
if itemstatus.lifecycle_state != "TERMINATED":
if itemstatus.lifecycle_state != "TERMINATING":
try:
print("Deleting: {}".format(itemstatus.display_name))
object.delete_drg(drg_id=itemstatus.id)
except:
def DeleteServiceGateways(config, compartment, vcn):
AllItems = []
object = oci.core.VirtualNetworkClient(config)
print("Getting all Service Gateway objects")
items = oci.pagination.list_call_get_all_results(object.list_service_gateways,compartment_id=compartment.id, vcn_id=vcn.id).data
for item in items:
if (item.lifecycle_state != "TERMINATED"):
AllItems.append(item)
print("- {} - {}".format(item.display_name, item.lifecycle_state))
itemsPresent = True
if itemsPresent:
count = 0
for item in AllItems:
try:
itemstatus = object.get_service_gateway(service_gateway_id=item.id).data
def DeleteSecurityLists(config, compartment, vcn):
AllItems = []
object = oci.core.VirtualNetworkClient(config)
print ("Getting SecurityLists for {}".format(vcn.display_name))
items = oci.pagination.list_call_get_all_results(object.list_security_lists, compartment_id=compartment.id,vcn_id=vcn.id).data
for item in items:
if (item.lifecycle_state != "TERMINATED"):
AllItems.append(item)
print("- {} - {}".format(item.display_name, item.lifecycle_state))
itemsPresent = True
while itemsPresent:
count = 0
for item in AllItems:
try:
itemstatus = object.get_security_list(security_list_id=item.id).data
if itemstatus.lifecycle_state != "TERMINATED":