Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_there_and_back(self):
require_user('root')
# wait until the previous test's side effects are gone
time.sleep(2)
#
fd = open('/proc/self/ns/net', 'r')
foo = str(uuid4())
#
# please notice, that IPRoute / IPDB, started in a netns, will continue
# to work in a given netns even if the process changes to another netns
#
with IPRoute() as ip:
links_main1 = set([x.get('index', None) for x in ip.get_links()])
netnsmod.setns(foo)
with IPRoute() as ip:
links_foo = set([x.get('index', None) for x in ip.get_links()])
netnsmod.setns(fd)
with IPRoute() as ip:
links_main2 = set([x.get('index', None) for x in ip.get_links()])
assert links_main1 == links_main2
assert links_main1 != links_foo
netnsmod.remove(foo)
fd.close()
ensures to move it back in original namespace or kills it if we fail to
move back in original namespace.
"""
if not namespace:
yield
return
org_netns_fd = os.open(PROCESS_NETNS, os.O_RDONLY)
pynetns.setns(namespace)
try:
yield
finally:
try:
# NOTE(cby): this code is not executed only if we fail to
# move in target namespace
pynetns.setns(org_netns_fd)
except Exception as e:
msg = _('Failed to move back in original netns: %s') % e
LOG.critical(msg)
raise BackInNamespaceExit(msg)
def _enable_ipv6(netns):
# Docker disables IPv6 for --net=none containers
# TODO(apuimedo) remove when it is no longer the case
try:
netns = utils.convert_netns(netns)
path = utils.convert_netns('/proc/self/ns/net')
self_ns_fd = open(path)
pyroute2.netns.setns(netns)
path = utils.convert_netns('/proc/sys/net/ipv6/conf/all/disable_ipv6')
with open(path, 'w') as disable_ipv6:
disable_ipv6.write('0')
except Exception:
raise
finally:
pyroute2.netns.setns(self_ns_fd)
def _enable_ipv6(netns):
# Docker disables IPv6 for --net=none containers
# TODO(apuimedo) remove when it is no longer the case
try:
path = utils.convert_netns('/proc/self/ns/net')
self_ns_fd = open(path)
pyroute2.netns.setns(netns)
path = utils.convert_netns('/proc/sys/net/ipv6/conf/all/disable_ipv6')
with open(path, 'w') as disable_ipv6:
disable_ipv6.write('0')
except Exception:
raise
finally:
pyroute2.netns.setns(self_ns_fd)
def Server(trnsp_in, trnsp_out, netns=None):
def stop_server(signum, frame):
Server.run = False
Server.run = True
signal.signal(signal.SIGTERM, stop_server)
try:
if netns is not None:
netnsmod.setns(netns)
ipr = IPRoute()
lock = ipr._sproxy.lock
ipr._s_channel = ProxyChannel(trnsp_out, 'broadcast')
except Exception as e:
trnsp_out.send({'stage': 'init',
'error': e})
return 255
inputs = [ipr.fileno(), trnsp_in.fileno()]
broadcasts = {ipr.fileno(): ipr}
outputs = []
# all is OK so far
trnsp_out.send({'stage': 'init',
'uname': config.uname,
'error': None})
def _enable_ipv6(netns):
# Docker disables IPv6 for --net=none containers
# TODO(apuimedo) remove when it is no longer the case
try:
netns = utils.convert_netns(netns)
path = utils.convert_netns('/proc/self/ns/net')
self_ns_fd = open(path)
pyroute2.netns.setns(netns)
path = utils.convert_netns('/proc/sys/net/ipv6/conf/all/disable_ipv6')
with open(path, 'w') as disable_ipv6:
disable_ipv6.write('0')
except Exception:
raise
finally:
pyroute2.netns.setns(self_ns_fd)