Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
client = docker.from_env(version='auto')
ipam_pool = docker.types.IPAMPool(
subnet='172.28.0.0/16',
iprange='172.28.0.0/24',
gateway='172.28.0.254',
)
# We use the default container prefix so that the
# ``minidcos docker clean`` command cleans this up.
prefix = Docker().container_name_prefix
random = uuid.uuid4()
name = '{prefix}-network-{random}'.format(prefix=prefix, random=random)
network = client.networks.create(
name=name,
driver='bridge',
ipam=docker.types.IPAMConfig(pool_configs=[ipam_pool]),
attachable=False,
)
try:
yield network
finally:
network.remove()
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
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)
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},
)
def docker_network_add(name):
cli = get_docker_client(DOCKER_BASE_URL)
ipam_pool = IPAMPool(subnet=CALICO_NETWORK)
ipam_config = IPAMConfig(driver="calico-ipam", pool_configs=[ipam_pool])
result = cli.create_network(name, driver="calico", ipam=ipam_config)
logger.info("create docker network for app %s : %s" % (name, result))
raise DockerError("Failed to create network: %s." % e)
else:
self.log("No private network specified")
# public network
have_pub_net = self.exaconf.has_pub_net()
if have_pub_net:
pub_net_name = self.exaconf.get_pub_net_name()
try:
pub_net = self.exaconf.get_pub_net()
except EXAConf.EXAConfError as e:
raise DockerError("Failed to read EXAConf: %s" % e)
# create public network (if any)
self.log("Creating public network %s ('%s')..." % (pub_net, pub_net_name), no_nl=True)
ipam_pool = docker.types.IPAMPool(subnet = pub_net)
ipam_config = docker.types.IPAMConfig(pool_configs = [ipam_pool])
try:
net = self.client.create_network(pub_net_name, driver="bridge", ipam=ipam_config,
labels={"ClusterName":self.cluster_name, "Scope":"public"})
# add name and type (for usage by create_containers())
net['MyName'] = pub_net_name
net['MyScope'] = 'public'
created_networks.append(net)
self.log("successful")
if self.verbose:
print "Created the following network:"
pprint.pprint(net)
except docker.errors.APIError as e:
raise DockerError("Failed to create network: %s." % e)
else:
self.log("No public network specified.")
for container in network.containers:
self.log.debug("Disconnecting Docker network "
f"'{network.name}' from container"
f"'{container.name}'")
network.disconnect(container, force=True)
self.log.debug(f"Removing Docker network '{network.name}'")
network.remove()
network = None
except docker.errors.NotFound:
if not remove:
self.log.debug(f"Creating Docker network '{name}'")
subnet = str(IPNetwork(bridge_ipaddr + '/' +
str(netprefix)).cidr)
ipam_pool = docker.types.IPAMPool(subnet=subnet,
gateway=bridge_ipaddr)
ipam_config = docker.types.IPAMConfig(
pool_configs=[ipam_pool])
try:
network = self.client.networks.create(
name=name,
driver='bridge',
ipam=ipam_config,
options={'com.docker.network.bridge.name':
br_name})
except docker.errors.APIError as exc:
msg = (f"Failed to create network '{name}': {exc}")
self.log.error(msg)
raise UserException(msg)
return network
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)
if self.parameters.enable_ipv6 is not None:
params['enable_ipv6'] = self.parameters.enable_ipv6
if self.parameters.internal is not None:
params['internal'] = self.parameters.internal
if self.parameters.scope is not None:
params['scope'] = self.parameters.scope
if self.parameters.attachable is not None:
params['attachable'] = self.parameters.attachable
if self.parameters.labels:
params['labels'] = self.parameters.labels
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')
)