Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#
# The 8 IP addresses in the IPAM Pool are:
# * 172.28.0.0 (reserved because this is the subnet identifier)
# * 172.28.0.1 (reserved because this is the gateway address)
# * 172.28.0.2 (available)
# * 172.28.0.3 (available)
# * 172.28.0.4 (available)
# * 172.28.0.5 (reserved because we reserve this with `aux_addresses`)
# * 172.28.0.6 (reserved because we reserve this with `aux_addresses`)
# * 172.28.0.7 (reserved because this is the broadcast address)
client = docker.from_env(version='auto')
aux_addresses = {
'reserved_address_0': '172.28.0.5',
'reserved_address_1': '172.28.0.6',
}
ipam_pool = docker.types.IPAMPool(
subnet='172.28.0.0/29',
iprange='172.28.0.0/29',
gateway='172.28.0.1',
aux_addresses=aux_addresses,
)
network = client.networks.create(
name='dcos-e2e-network-{random}'.format(random=uuid.uuid4()),
driver='bridge',
ipam=docker.types.IPAMConfig(pool_configs=[ipam_pool]),
attachable=False,
)
try:
yield network
finally:
network.remove()
def create_docker_network(self) -> DockerNetworkInfo:
self.remove_container(self.test_container_name)
if self.db_container_name is not None:
self.remove_container(self.db_container_name)
self.remove_network(self.network_name)
network = self._client.networks.create(
name=self.network_name,
driver="bridge",
)
network_info = self.get_network_info(reused=False)
subnet = network_info.subnet
gateway = network_info.gateway
ipam_pool = docker.types.IPAMPool(
subnet=subnet,
gateway=gateway
)
ipam_config = docker.types.IPAMConfig(
pool_configs=[ipam_pool]
)
self.remove_network(self.network_name) # TODO race condition possible, add retry
network = self._client.networks.create(
name=self.network_name,
driver="bridge",
ipam=ipam_config
)
return network_info
def create_network(network_name, subnet, gateway):
ipam_pool = docker.types.IPAMPool(
subnet=subnet, # '192.168.0.0/16',
gateway=gateway # '192.168.0.254'
)
ipam_config = docker.types.IPAMConfig(
pool_configs=[ipam_pool]
)
client.networks.create(
network_name,
driver="bridge",
ipam=ipam_config
)
def network_containers_routerless(self, containers):
""" If there is no router, all containers are added to the same network. """
client = docker.from_env()
network_name = f'{self.untrusted_user}_routerless_network'
# Assumes untrustedXX naming scheme, where XX is a number
untrusted_num = int(self.untrusted_user.replace('untrusted','')) + 100
subnet = 1
ip_address_start = f'10.{untrusted_num}.{subnet}'
ipam_pool = docker.types.IPAMPool(subnet=f'{ip_address_start}.0/24')
ipam_config = docker.types.IPAMConfig(pool_configs=[ipam_pool])
#create the global network
# TODO: Can fail on ip conflict.
network = client.networks.create(network_name, driver='bridge', ipam=ipam_config, internal=True)
client.close()
host = 2
for container in containers:
ip_address = f'{ip_address_start}.{host}'
network.connect(container.container, ipv4_address=ip_address, aliases=[container.name,])
container.set_ip_address(network_name, ip_address)
host+=1
self.networks.append(network)
pool = {}
# subnet
pool["subnet"] = args["subnet"]
del args["subnet"]
# iprange
pool["iprange"] = args["iprange"]
del args["iprange"]
# gateway
pool["gateway"] = args["gateway"]
del args["gateway"]
# create ipam
ipam_pool = IPAMPool(**pool)
ipam_config = IPAMConfig(
driver=args["ipamdriver"],
pool_configs=[ipam_pool],
options=dict(args["ipamopt"]) if args["ipamopt"] else None
)
del args["ipamdriver"]
del args["ipamopt"]
if "None" in str(ipam_config):
return None
else:
return ipam_config
def create_network(self):
if not self.existing_network:
params = dict(
driver=self.parameters.driver,
options=self.parameters.driver_options,
)
ipam_pools = []
if self.parameters.ipam_config:
for ipam_pool in self.parameters.ipam_config:
if LooseVersion(docker_version) >= LooseVersion('2.0.0'):
ipam_pools.append(IPAMPool(**ipam_pool))
else:
ipam_pools.append(utils.create_ipam_pool(**ipam_pool))
if self.parameters.ipam_driver or self.parameters.ipam_driver_options or ipam_pools:
# Only add ipam parameter if a driver was specified or if IPAM parameters
# were specified. Leaving this parameter away can significantly speed up
# creation; on my machine creation with this option needs ~15 seconds,
# and without just a few seconds.
if LooseVersion(docker_version) >= LooseVersion('2.0.0'):
params['ipam'] = IPAMConfig(driver=self.parameters.ipam_driver,
pool_configs=ipam_pools,
options=self.parameters.ipam_driver_options)
else:
params['ipam'] = utils.create_ipam_config(driver=self.parameters.ipam_driver,
pool_configs=ipam_pools)
def create_ipam_config_from_dict(ipam_dict):
if not ipam_dict:
return None
return IPAMConfig(
driver=ipam_dict.get('driver') or 'default',
pool_configs=[
IPAMPool(
subnet=config.get('subnet'),
iprange=config.get('ip_range'),
gateway=config.get('gateway'),
aux_addresses=config.get('aux_addresses'),
)
for config in ipam_dict.get('config', [])
],
options=ipam_dict.get('options')
)
def _get_network(self) -> docker.models.networks.Network:
"""Get HassIO network."""
try:
return self.docker.networks.get(DOCKER_NETWORK)
except docker.errors.NotFound:
_LOGGER.info("Can't find Hass.io network, create new network")
ipam_pool = docker.types.IPAMPool(
subnet=str(DOCKER_NETWORK_MASK),
gateway=str(self.gateway),
iprange=str(DOCKER_NETWORK_RANGE),
)
ipam_config = docker.types.IPAMConfig(pool_configs=[ipam_pool])
return self.docker.networks.create(
DOCKER_NETWORK,
driver="bridge",
ipam=ipam_config,
enable_ipv6=False,
options={"com.docker.network.bridge.name": DOCKER_NETWORK},
)
network.attrs["IPAM"]["Config"][0]["Subnet"])
if cidr.network_address.packed[0] == INITIAL_CIDR_FIRST_OCTET \
and cidr.network_address.packed[1] >= INITIAL_CIDR_SECOND_OCTET:
if cidr > highest_cidr:
highest_cidr = cidr
# take the highest cidr and add 256 bits, so that if the highest subnet was 172.33.2.0, the new subnet is 172.33.3.0
next_cidr = ipaddress.ip_network(
(highest_cidr.network_address + 256).exploded + "/24")
if next_cidr.network_address.packed[0] > INITIAL_CIDR_FIRST_OCTET:
raise Exception("No more possible subnet addresses exist")
self.log.info("Create network {} with subnet {}".format(
name, next_cidr.exploded))
ipam_pool = docker.types.IPAMPool(subnet=next_cidr.exploded,
gateway=(next_cidr.network_address + 1).exploded)
ipam_config = docker.types.IPAMConfig(pool_configs=[ipam_pool])
return client.networks.create(name, ipam=ipam_config, labels=self.default_labels)