<s> import os <EOL> import re <EOL> import eventlet <EOL> import netaddr <EOL> from neutron_lib import constants <EOL> from neutron_lib import exceptions <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> from oslo_utils import excutils <EOL> import six <EOL> from neutron . _i18n import _ , _LE <EOL> from neutron . agent . common import utils <EOL> from neutron . common import exceptions as n_exc <EOL> LOG = logging . getLogger ( __name__ ) <EOL> OPTS = [ <EOL> cfg . BoolOpt ( '<STR_LIT>' , <EOL> default = False , <EOL> help = _ ( '<STR_LIT>' ) ) , <EOL> ] <EOL> LOOPBACK_DEVNAME = '<STR_LIT>' <EOL> SYS_NET_PATH = '<STR_LIT>' <EOL> DEFAULT_GW_PATTERN = re . compile ( r"<STR_LIT>" ) <EOL> METRIC_PATTERN = re . compile ( r"<STR_LIT>" ) <EOL> DEVICE_NAME_PATTERN = re . compile ( r"<STR_LIT>" ) <EOL> def remove_interface_suffix ( interface ) : <EOL> """<STR_LIT>""" <EOL> return interface . partition ( "<STR_LIT:@>" ) [ <NUM_LIT:0> ] <EOL> class AddressNotReady ( exceptions . NeutronException ) : <EOL> message = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class SubProcessBase ( object ) : <EOL> def __init__ ( self , namespace = None , <EOL> log_fail_as_error = True ) : <EOL> self . namespace = namespace <EOL> self . log_fail_as_error = log_fail_as_error <EOL> try : <EOL> self . force_root = cfg . CONF . ip_lib_force_root <EOL> except cfg . NoSuchOptError : <EOL> self . force_root = False <EOL> def _run ( self , options , command , args ) : <EOL> if self . namespace : <EOL> return self . _as_root ( options , command , args ) <EOL> elif self . force_root : <EOL> return self . _execute ( options , command , args , run_as_root = True , <EOL> log_fail_as_error = self . log_fail_as_error ) <EOL> else : <EOL> return self . _execute ( options , command , args , <EOL> log_fail_as_error = self . log_fail_as_error ) <EOL> def _as_root ( self , options , command , args , use_root_namespace = False ) : <EOL> namespace = self . namespace if not use_root_namespace else None <EOL> return self . _execute ( options , command , args , run_as_root = True , <EOL> namespace = namespace , <EOL> log_fail_as_error = self . log_fail_as_error ) <EOL> @ classmethod <EOL> def _execute ( cls , options , command , args , run_as_root = False , <EOL> namespace = None , log_fail_as_error = True ) : <EOL> opt_list = [ '<STR_LIT>' % o for o in options ] <EOL> ip_cmd = add_namespace_to_cmd ( [ '<STR_LIT>' ] , namespace ) <EOL> cmd = ip_cmd + opt_list + [ command ] + list ( args ) <EOL> return utils . execute ( cmd , run_as_root = run_as_root , <EOL> log_fail_as_error = log_fail_as_error ) <EOL> def set_log_fail_as_error ( self , fail_with_error ) : <EOL> self . log_fail_as_error = fail_with_error <EOL> def get_log_fail_as_error ( self ) : <EOL> return self . log_fail_as_error <EOL> class IPWrapper ( SubProcessBase ) : <EOL> def __init__ ( self , namespace = None ) : <EOL> super ( IPWrapper , self ) . __init__ ( namespace = namespace ) <EOL> self . netns = IpNetnsCommand ( self ) <EOL> def device ( self , name ) : <EOL> return IPDevice ( name , namespace = self . namespace ) <EOL> def get_devices ( self , exclude_loopback = False ) : <EOL> retval = [ ] <EOL> if self . namespace : <EOL> output = utils . execute ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , self . namespace , <EOL> '<STR_LIT>' , SYS_NET_PATH , '<STR_LIT>' , '<STR_LIT:1>' , <EOL> '<STR_LIT>' , '<STR_LIT:l>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> run_as_root = True , <EOL> log_fail_as_error = self . log_fail_as_error <EOL> ) . split ( ) <EOL> else : <EOL> output = ( <EOL> i for i in os . listdir ( SYS_NET_PATH ) <EOL> if os . path . islink ( os . path . join ( SYS_NET_PATH , i ) ) <EOL> ) <EOL> for name in output : <EOL> if exclude_loopback and name == LOOPBACK_DEVNAME : <EOL> continue <EOL> retval . append ( IPDevice ( name , namespace = self . namespace ) ) <EOL> return retval <EOL> def get_device_by_ip ( self , ip ) : <EOL> """<STR_LIT>""" <EOL> if not ip : <EOL> return None <EOL> addr = IpAddrCommand ( self ) <EOL> devices = addr . get_devices_with_ip ( to = ip ) <EOL> if devices : <EOL> return IPDevice ( devices [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , namespace = self . namespace ) <EOL> def add_tuntap ( self , name , mode = '<STR_LIT>' ) : <EOL> self . _as_root ( [ ] , '<STR_LIT>' , ( '<STR_LIT>' , name , '<STR_LIT>' , mode ) ) <EOL> return IPDevice ( name , namespace = self . namespace ) <EOL> def add_veth ( self , name1 , name2 , namespace2 = None ) : <EOL> args = [ '<STR_LIT>' , name1 , '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , name2 ] <EOL> if namespace2 is None : <EOL> namespace2 = self . namespace <EOL> else : <EOL> self . ensure_namespace ( namespace2 ) <EOL> args += [ '<STR_LIT>' , namespace2 ] <EOL> self . _as_root ( [ ] , '<STR_LIT>' , tuple ( args ) ) <EOL> return ( IPDevice ( name1 , namespace = self . namespace ) , <EOL> IPDevice ( name2 , namespace = namespace2 ) ) <EOL> def add_macvtap ( self , name , src_dev , mode = '<STR_LIT>' ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , src_dev , '<STR_LIT:name>' , name , '<STR_LIT:type>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , mode ] <EOL> self . _as_root ( [ ] , '<STR_LIT>' , tuple ( args ) ) <EOL> return IPDevice ( name , namespace = self . namespace ) <EOL> def del_veth ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _as_root ( [ ] , '<STR_LIT>' , ( '<STR_LIT>' , name ) ) <EOL> def add_dummy ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _as_root ( [ ] , '<STR_LIT>' , ( '<STR_LIT>' , name , '<STR_LIT:type>' , '<STR_LIT>' ) ) <EOL> return IPDevice ( name , namespace = self . namespace ) <EOL> def ensure_namespace ( self , name ) : <EOL> if not self . netns . exists ( name ) : <EOL> ip = self . netns . add ( name ) <EOL> lo = ip . device ( LOOPBACK_DEVNAME ) <EOL> lo . link . set_up ( ) <EOL> else : <EOL> ip = IPWrapper ( namespace = name ) <EOL> return ip <EOL> def namespace_is_empty ( self ) : <EOL> return not self . get_devices ( exclude_loopback = True ) <EOL> def garbage_collect_namespace ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . namespace and self . netns . exists ( self . namespace ) : <EOL> if self . namespace_is_empty ( ) : <EOL> self . netns . delete ( self . namespace ) <EOL> return True <EOL> return False <EOL> def add_device_to_namespace ( self , device ) : <EOL> if self . namespace : <EOL> device . link . set_netns ( self . namespace ) <EOL> def add_vlan ( self , name , physical_interface , vlan_id ) : <EOL> cmd = [ '<STR_LIT>' , '<STR_LIT>' , physical_interface , '<STR_LIT:name>' , name , <EOL> '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT:id>' , vlan_id ] <EOL> self . _as_root ( [ ] , '<STR_LIT>' , cmd ) <EOL> return IPDevice ( name , namespace = self . namespace ) <EOL> def add_vxlan ( self , name , vni , group = None , dev = None , ttl = None , tos = None , <EOL> local = None , port = None , proxy = False ) : <EOL> cmd = [ '<STR_LIT>' , name , '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT:id>' , vni ] <EOL> if group : <EOL> cmd . extend ( [ '<STR_LIT>' , group ] ) <EOL> if dev : <EOL> cmd . extend ( [ '<STR_LIT>' , dev ] ) <EOL> if ttl : <EOL> cmd . extend ( [ '<STR_LIT>' , ttl ] ) <EOL> if tos : <EOL> cmd . extend ( [ '<STR_LIT>' , tos ] ) <EOL> if local : <EOL> cmd . extend ( [ '<STR_LIT>' , local ] ) <EOL> if proxy : <EOL> cmd . append ( '<STR_LIT>' ) <EOL> if port and len ( port ) == <NUM_LIT:2> : <EOL> cmd . extend ( [ '<STR_LIT:port>' , port [ <NUM_LIT:0> ] , port [ <NUM_LIT:1> ] ] ) <EOL> elif port : <EOL> raise n_exc . NetworkVxlanPortRangeError ( vxlan_range = port ) <EOL> self . _as_root ( [ ] , '<STR_LIT>' , cmd ) <EOL> return ( IPDevice ( name , namespace = self . namespace ) ) <EOL> @ classmethod <EOL> def get_namespaces ( cls ) : <EOL> output = cls . _execute ( [ ] , '<STR_LIT>' , ( '<STR_LIT:list>' , ) ) <EOL> return [ l . split ( ) [ <NUM_LIT:0> ] for l in output . splitlines ( ) ] <EOL> class IPDevice ( SubProcessBase ) : <EOL> def __init__ ( self , name , namespace = None ) : <EOL> super ( IPDevice , self ) . __init__ ( namespace = namespace ) <EOL> self . name = name <EOL> self . link = IpLinkCommand ( self ) <EOL> self . addr = IpAddrCommand ( self ) <EOL> self . route = IpRouteCommand ( self ) <EOL> self . neigh = IpNeighCommand ( self ) <EOL> def __eq__ ( self , other ) : <EOL> return ( other is not None and self . name == other . name <EOL> and self . namespace == other . namespace ) <EOL> def __str__ ( self ) : <EOL> return self . name <EOL> def exists ( self ) : <EOL> """<STR_LIT>""" <EOL> orig_log_fail_as_error = self . get_log_fail_as_error ( ) <EOL> self . set_log_fail_as_error ( False ) <EOL> try : <EOL> return bool ( self . link . address ) <EOL> except RuntimeError : <EOL> return False <EOL> finally : <EOL> self . set_log_fail_as_error ( orig_log_fail_as_error ) <EOL> def delete_addr_and_conntrack_state ( self , cidr ) : <EOL> """<STR_LIT>""" <EOL> self . addr . delete ( cidr ) <EOL> ip_str = str ( netaddr . IPNetwork ( cidr ) . ip ) <EOL> ip_wrapper = IPWrapper ( namespace = self . namespace ) <EOL> try : <EOL> ip_wrapper . netns . execute ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , ip_str ] , <EOL> check_exit_code = True , <EOL> extra_ok_codes = [ <NUM_LIT:1> ] ) <EOL> except RuntimeError : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , ip_str ) <EOL> try : <EOL> ip_wrapper . netns . execute ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , ip_str ] , <EOL> check_exit_code = True , <EOL> extra_ok_codes = [ <NUM_LIT:1> ] ) <EOL> except RuntimeError : <EOL> LOG . exception ( _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , ip_str ) <EOL> def _sysctl ( self , cmd ) : <EOL> """<STR_LIT>""" <EOL> cmd = [ '<STR_LIT>' , '<STR_LIT>' ] + cmd <EOL> ip_wrapper = IPWrapper ( self . namespace ) <EOL> try : <EOL> ip_wrapper . netns . execute ( cmd , run_as_root = True , <EOL> check_exit_code = True ) <EOL> except RuntimeError : <EOL> LOG . exception ( _LE ( "<STR_LIT>" ) , cmd ) <EOL> return <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> def disable_ipv6 ( self ) : <EOL> sysctl_name = re . sub ( r'<STR_LIT>' , '<STR_LIT:/>' , self . name ) <EOL> cmd = '<STR_LIT>' % sysctl_name <EOL> return self . _sysctl ( [ cmd ] ) <EOL> class IpCommandBase ( object ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def __init__ ( self , parent ) : <EOL> self . _parent = parent <EOL> def _run ( self , options , args ) : <EOL> return self . _parent . _run ( options , self . COMMAND , args ) <EOL> def _as_root ( self , options , args , use_root_namespace = False ) : <EOL> return self . _parent . _as_root ( options , <EOL> self . COMMAND , <EOL> args , <EOL> use_root_namespace = use_root_namespace ) <EOL> class IPRule ( SubProcessBase ) : <EOL> def __init__ ( self , namespace = None ) : <EOL> super ( IPRule , self ) . __init__ ( namespace = namespace ) <EOL> self . rule = IpRuleCommand ( self ) <EOL> class IpRuleCommand ( IpCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> @ staticmethod <EOL> def _make_canonical ( ip_version , settings ) : <EOL> """<STR_LIT>""" <EOL> def canonicalize_fwmark_string ( fwmark_mask ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( fwmark_mask , six . string_types ) : <EOL> iterable = fwmark_mask . split ( '<STR_LIT:/>' ) <EOL> else : <EOL> try : <EOL> iterable = iter ( fwmark_mask ) <EOL> except TypeError : <EOL> iterable = [ fwmark_mask ] <EOL> def to_i ( s ) : <EOL> if isinstance ( s , six . string_types ) : <EOL> return int ( s , <NUM_LIT:0> ) <EOL> return int ( s ) <EOL> integers = [ to_i ( x ) for x in iterable ] <EOL> if len ( integers ) == <NUM_LIT:1> : <EOL> integers . append ( <NUM_LIT> ) <EOL> return '<STR_LIT>' . format ( * integers ) <EOL> def canonicalize ( item ) : <EOL> k , v = item <EOL> if k == '<STR_LIT>' and v == '<STR_LIT:all>' : <EOL> return k , constants . IP_ANY [ ip_version ] <EOL> if k == '<STR_LIT>' : <EOL> return '<STR_LIT>' , v <EOL> if k == '<STR_LIT>' : <EOL> return k , canonicalize_fwmark_string ( v ) <EOL> return k , v <EOL> if '<STR_LIT:type>' not in settings : <EOL> settings [ '<STR_LIT:type>' ] = '<STR_LIT>' <EOL> return { k : str ( v ) for k , v in map ( canonicalize , settings . items ( ) ) } <EOL> def _parse_line ( self , ip_version , line ) : <EOL> parts = line . split ( ) <EOL> if not parts : <EOL> return { } <EOL> settings = { k : v for k , v in zip ( parts [ <NUM_LIT:1> : : <NUM_LIT:2> ] , parts [ <NUM_LIT:2> : : <NUM_LIT:2> ] ) } <EOL> settings [ '<STR_LIT>' ] = parts [ <NUM_LIT:0> ] [ : - <NUM_LIT:1> ] <EOL> if len ( parts ) % <NUM_LIT:2> == <NUM_LIT:0> : <EOL> settings [ '<STR_LIT:type>' ] = parts [ - <NUM_LIT:1> ] <EOL> return self . _make_canonical ( ip_version , settings ) <EOL> def list_rules ( self , ip_version ) : <EOL> lines = self . _as_root ( [ ip_version ] , [ '<STR_LIT>' ] ) . splitlines ( ) <EOL> return [ self . _parse_line ( ip_version , line ) for line in lines ] <EOL> def _exists ( self , ip_version , ** kwargs ) : <EOL> return kwargs in self . list_rules ( ip_version ) <EOL> def _make__flat_args_tuple ( self , * args , ** kwargs ) : <EOL> for kwargs_item in sorted ( kwargs . items ( ) , key = lambda i : i [ <NUM_LIT:0> ] ) : <EOL> args += kwargs_item <EOL> return tuple ( args ) <EOL> def add ( self , ip , ** kwargs ) : <EOL> ip_version = get_ip_version ( ip ) <EOL> kwargs . update ( { '<STR_LIT>' : ip } ) <EOL> canonical_kwargs = self . _make_canonical ( ip_version , kwargs ) <EOL> if not self . _exists ( ip_version , ** canonical_kwargs ) : <EOL> args_tuple = self . _make__flat_args_tuple ( '<STR_LIT>' , ** canonical_kwargs ) <EOL> self . _as_root ( [ ip_version ] , args_tuple ) <EOL> def delete ( self , ip , ** kwargs ) : <EOL> ip_version = get_ip_version ( ip ) <EOL> canonical_kwargs = self . _make_canonical ( ip_version , kwargs ) <EOL> args_tuple = self . _make__flat_args_tuple ( '<STR_LIT>' , ** canonical_kwargs ) <EOL> self . _as_root ( [ ip_version ] , args_tuple ) <EOL> class IpDeviceCommandBase ( IpCommandBase ) : <EOL> @ property <EOL> def name ( self ) : <EOL> return self . _parent . name <EOL> class IpLinkCommand ( IpDeviceCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def set_address ( self , mac_address ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT:address>' , mac_address ) ) <EOL> def set_allmulticast_on ( self ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def set_mtu ( self , mtu_size ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' , mtu_size ) ) <EOL> def set_up ( self ) : <EOL> return self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' ) ) <EOL> def set_down ( self ) : <EOL> return self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' ) ) <EOL> def set_netns ( self , namespace ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' , namespace ) ) <EOL> self . _parent . namespace = namespace <EOL> def set_name ( self , name ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT:name>' , name ) ) <EOL> self . _parent . name = name <EOL> def set_alias ( self , alias_name ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name , '<STR_LIT>' , alias_name ) ) <EOL> def delete ( self ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , self . name ) ) <EOL> @ property <EOL> def address ( self ) : <EOL> return self . attributes . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def state ( self ) : <EOL> return self . attributes . get ( '<STR_LIT:state>' ) <EOL> @ property <EOL> def mtu ( self ) : <EOL> return self . attributes . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def qdisc ( self ) : <EOL> return self . attributes . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def qlen ( self ) : <EOL> return self . attributes . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def alias ( self ) : <EOL> return self . attributes . get ( '<STR_LIT>' ) <EOL> @ property <EOL> def attributes ( self ) : <EOL> return self . _parse_line ( self . _run ( [ '<STR_LIT:o>' ] , ( '<STR_LIT>' , self . name ) ) ) <EOL> def _parse_line ( self , value ) : <EOL> if not value : <EOL> return { } <EOL> device_name , settings = value . replace ( "<STR_LIT:\\>" , '<STR_LIT>' ) . split ( '<STR_LIT:>>' , <NUM_LIT:1> ) <EOL> tokens = settings . split ( ) <EOL> keys = tokens [ : : <NUM_LIT:2> ] <EOL> values = [ int ( v ) if v . isdigit ( ) else v for v in tokens [ <NUM_LIT:1> : : <NUM_LIT:2> ] ] <EOL> retval = dict ( zip ( keys , values ) ) <EOL> return retval <EOL> class IpAddrCommand ( IpDeviceCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def add ( self , cidr , scope = '<STR_LIT>' ) : <EOL> net = netaddr . IPNetwork ( cidr ) <EOL> args = [ '<STR_LIT>' , cidr , <EOL> '<STR_LIT>' , scope , <EOL> '<STR_LIT>' , self . name ] <EOL> if net . version == <NUM_LIT:4> : <EOL> args += [ '<STR_LIT>' , str ( net [ - <NUM_LIT:1> ] ) ] <EOL> self . _as_root ( [ net . version ] , tuple ( args ) ) <EOL> def delete ( self , cidr ) : <EOL> ip_version = get_ip_version ( cidr ) <EOL> self . _as_root ( [ ip_version ] , <EOL> ( '<STR_LIT>' , cidr , <EOL> '<STR_LIT>' , self . name ) ) <EOL> def flush ( self , ip_version ) : <EOL> self . _as_root ( [ ip_version ] , ( '<STR_LIT>' , self . name ) ) <EOL> def get_devices_with_ip ( self , name = None , scope = None , to = None , <EOL> filters = None , ip_version = None ) : <EOL> """<STR_LIT>""" <EOL> options = [ ip_version ] if ip_version else [ ] <EOL> args = [ '<STR_LIT>' ] <EOL> if name : <EOL> args += [ name ] <EOL> if filters : <EOL> args += filters <EOL> if scope : <EOL> args += [ '<STR_LIT>' , scope ] <EOL> if to : <EOL> args += [ '<STR_LIT:to>' , to ] <EOL> retval = [ ] <EOL> for line in self . _run ( options , tuple ( args ) ) . split ( '<STR_LIT:\n>' ) : <EOL> line = line . strip ( ) <EOL> match = DEVICE_NAME_PATTERN . search ( line ) <EOL> if match : <EOL> device_name = remove_interface_suffix ( match . group ( <NUM_LIT:2> ) ) <EOL> continue <EOL> elif not line . startswith ( '<STR_LIT>' ) : <EOL> continue <EOL> parts = line . split ( "<STR_LIT:U+0020>" ) <EOL> if parts [ <NUM_LIT:0> ] == '<STR_LIT>' : <EOL> scope = parts [ <NUM_LIT:3> ] <EOL> else : <EOL> if parts [ <NUM_LIT:2> ] == '<STR_LIT>' : <EOL> scope = parts [ <NUM_LIT:5> ] <EOL> else : <EOL> scope = parts [ <NUM_LIT:3> ] <EOL> retval . append ( dict ( name = device_name , <EOL> cidr = parts [ <NUM_LIT:1> ] , <EOL> scope = scope , <EOL> dynamic = ( '<STR_LIT>' == parts [ - <NUM_LIT:1> ] ) , <EOL> tentative = ( '<STR_LIT>' in line ) , <EOL> dadfailed = ( '<STR_LIT>' == parts [ - <NUM_LIT:1> ] ) ) ) <EOL> return retval <EOL> def list ( self , scope = None , to = None , filters = None , ip_version = None ) : <EOL> """<STR_LIT>""" <EOL> return self . get_devices_with_ip ( <EOL> self . name , scope , to , filters , ip_version ) <EOL> def wait_until_address_ready ( self , address , wait_time = <NUM_LIT:30> ) : <EOL> """<STR_LIT>""" <EOL> def is_address_ready ( ) : <EOL> try : <EOL> addr_info = self . list ( to = address ) [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> raise AddressNotReady ( <EOL> address = address , <EOL> reason = _ ( '<STR_LIT>' ) ) <EOL> if not addr_info [ '<STR_LIT>' ] : <EOL> return True <EOL> if addr_info [ '<STR_LIT>' ] : <EOL> raise AddressNotReady ( <EOL> address = address , reason = _ ( '<STR_LIT>' ) ) <EOL> errmsg = _ ( "<STR_LIT>" <EOL> "<STR_LIT>" ) % wait_time <EOL> utils . utils . wait_until_true ( <EOL> is_address_ready , timeout = wait_time , sleep = <NUM_LIT> , <EOL> exception = AddressNotReady ( address = address , reason = errmsg ) ) <EOL> class IpRouteCommand ( IpDeviceCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def __init__ ( self , parent , table = None ) : <EOL> super ( IpRouteCommand , self ) . __init__ ( parent ) <EOL> self . _table = table <EOL> def table ( self , table ) : <EOL> """<STR_LIT>""" <EOL> return IpRouteCommand ( self . _parent , table ) <EOL> def _table_args ( self , override = None ) : <EOL> if override : <EOL> return [ '<STR_LIT>' , override ] <EOL> return [ '<STR_LIT>' , self . _table ] if self . _table else [ ] <EOL> def _dev_args ( self ) : <EOL> return [ '<STR_LIT>' , self . name ] if self . name else [ ] <EOL> def add_gateway ( self , gateway , metric = None , table = None ) : <EOL> ip_version = get_ip_version ( gateway ) <EOL> args = [ '<STR_LIT:replace>' , '<STR_LIT:default>' , '<STR_LIT>' , gateway ] <EOL> if metric : <EOL> args += [ '<STR_LIT>' , metric ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( table ) <EOL> self . _as_root ( [ ip_version ] , tuple ( args ) ) <EOL> def _run_as_root_detect_device_not_found ( self , * args , ** kwargs ) : <EOL> try : <EOL> return self . _as_root ( * args , ** kwargs ) <EOL> except RuntimeError as rte : <EOL> with excutils . save_and_reraise_exception ( ) as ctx : <EOL> if "<STR_LIT>" in str ( rte ) : <EOL> ctx . reraise = False <EOL> raise n_exc . DeviceNotFoundError ( device_name = self . name ) <EOL> def delete_gateway ( self , gateway , table = None ) : <EOL> ip_version = get_ip_version ( gateway ) <EOL> args = [ '<STR_LIT>' , '<STR_LIT:default>' , <EOL> '<STR_LIT>' , gateway ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( table ) <EOL> self . _run_as_root_detect_device_not_found ( [ ip_version ] , tuple ( args ) ) <EOL> def _parse_routes ( self , ip_version , output , ** kwargs ) : <EOL> for line in output . splitlines ( ) : <EOL> parts = line . split ( ) <EOL> route = { k : v for k , v in zip ( parts [ <NUM_LIT:1> : : <NUM_LIT:2> ] , parts [ <NUM_LIT:2> : : <NUM_LIT:2> ] ) } <EOL> route [ '<STR_LIT>' ] = parts [ <NUM_LIT:0> ] <EOL> if route [ '<STR_LIT>' ] == '<STR_LIT:default>' : <EOL> route [ '<STR_LIT>' ] = constants . IP_ANY [ ip_version ] <EOL> if self . name : <EOL> route [ '<STR_LIT>' ] = self . name <EOL> if self . _table : <EOL> route [ '<STR_LIT>' ] = self . _table <EOL> route . update ( kwargs ) <EOL> yield route <EOL> def list_routes ( self , ip_version , ** kwargs ) : <EOL> args = [ '<STR_LIT:list>' ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( ) <EOL> for k , v in kwargs . items ( ) : <EOL> args += [ k , v ] <EOL> output = self . _run ( [ ip_version ] , tuple ( args ) ) <EOL> return [ r for r in self . _parse_routes ( ip_version , output , ** kwargs ) ] <EOL> def list_onlink_routes ( self , ip_version ) : <EOL> routes = self . list_routes ( ip_version , scope = '<STR_LIT>' ) <EOL> return [ r for r in routes if '<STR_LIT:src>' not in r ] <EOL> def add_onlink_route ( self , cidr ) : <EOL> self . add_route ( cidr , scope = '<STR_LIT>' ) <EOL> def delete_onlink_route ( self , cidr ) : <EOL> self . delete_route ( cidr , scope = '<STR_LIT>' ) <EOL> def get_gateway ( self , scope = None , filters = None , ip_version = None ) : <EOL> options = [ ip_version ] if ip_version else [ ] <EOL> args = [ '<STR_LIT:list>' ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( ) <EOL> if filters : <EOL> args += filters <EOL> retval = None <EOL> if scope : <EOL> args += [ '<STR_LIT>' , scope ] <EOL> route_list_lines = self . _run ( options , tuple ( args ) ) . split ( '<STR_LIT:\n>' ) <EOL> default_route_line = next ( ( x . strip ( ) for x in <EOL> route_list_lines if <EOL> x . strip ( ) . startswith ( '<STR_LIT:default>' ) ) , None ) <EOL> if default_route_line : <EOL> retval = dict ( ) <EOL> gateway = DEFAULT_GW_PATTERN . search ( default_route_line ) <EOL> if gateway : <EOL> retval . update ( gateway = gateway . group ( <NUM_LIT:1> ) ) <EOL> metric = METRIC_PATTERN . search ( default_route_line ) <EOL> if metric : <EOL> retval . update ( metric = int ( metric . group ( <NUM_LIT:1> ) ) ) <EOL> return retval <EOL> def add_route ( self , cidr , via = None , table = None , ** kwargs ) : <EOL> ip_version = get_ip_version ( cidr ) <EOL> args = [ '<STR_LIT:replace>' , cidr ] <EOL> if via : <EOL> args += [ '<STR_LIT>' , via ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( table ) <EOL> for k , v in kwargs . items ( ) : <EOL> args += [ k , v ] <EOL> self . _run_as_root_detect_device_not_found ( [ ip_version ] , tuple ( args ) ) <EOL> def delete_route ( self , cidr , via = None , table = None , ** kwargs ) : <EOL> ip_version = get_ip_version ( cidr ) <EOL> args = [ '<STR_LIT>' , cidr ] <EOL> if via : <EOL> args += [ '<STR_LIT>' , via ] <EOL> args += self . _dev_args ( ) <EOL> args += self . _table_args ( table ) <EOL> for k , v in kwargs . items ( ) : <EOL> args += [ k , v ] <EOL> self . _run_as_root_detect_device_not_found ( [ ip_version ] , tuple ( args ) ) <EOL> class IPRoute ( SubProcessBase ) : <EOL> def __init__ ( self , namespace = None , table = None ) : <EOL> super ( IPRoute , self ) . __init__ ( namespace = namespace ) <EOL> self . name = None <EOL> self . route = IpRouteCommand ( self , table = table ) <EOL> class IpNeighCommand ( IpDeviceCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def add ( self , ip_address , mac_address ) : <EOL> ip_version = get_ip_version ( ip_address ) <EOL> self . _as_root ( [ ip_version ] , <EOL> ( '<STR_LIT:replace>' , ip_address , <EOL> '<STR_LIT>' , mac_address , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , self . name ) ) <EOL> def delete ( self , ip_address , mac_address ) : <EOL> ip_version = get_ip_version ( ip_address ) <EOL> self . _as_root ( [ ip_version ] , <EOL> ( '<STR_LIT>' , ip_address , <EOL> '<STR_LIT>' , mac_address , <EOL> '<STR_LIT>' , self . name ) ) <EOL> def show ( self , ip_version ) : <EOL> options = [ ip_version ] <EOL> return self . _as_root ( options , <EOL> ( '<STR_LIT>' , <EOL> '<STR_LIT>' , self . name ) ) <EOL> def flush ( self , ip_version , ip_address ) : <EOL> """<STR_LIT>""" <EOL> self . _as_root ( [ ip_version ] , ( '<STR_LIT>' , '<STR_LIT:to>' , ip_address ) ) <EOL> class IpNetnsCommand ( IpCommandBase ) : <EOL> COMMAND = '<STR_LIT>' <EOL> def add ( self , name ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , name ) , use_root_namespace = True ) <EOL> wrapper = IPWrapper ( namespace = name ) <EOL> wrapper . netns . execute ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> return wrapper <EOL> def delete ( self , name ) : <EOL> self . _as_root ( [ ] , ( '<STR_LIT>' , name ) , use_root_namespace = True ) <EOL> def execute ( self , cmds , addl_env = None , check_exit_code = True , <EOL> log_fail_as_error = True , extra_ok_codes = None , <EOL> run_as_root = False ) : <EOL> ns_params = [ ] <EOL> kwargs = { '<STR_LIT>' : run_as_root } <EOL> if self . _parent . namespace : <EOL> kwargs [ '<STR_LIT>' ] = True <EOL> ns_params = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , self . _parent . namespace ] <EOL> env_params = [ ] <EOL> if addl_env : <EOL> env_params = ( [ '<STR_LIT>' ] + <EOL> [ '<STR_LIT>' % pair for pair in addl_env . items ( ) ] ) <EOL> cmd = ns_params + env_params + list ( cmds ) <EOL> return utils . execute ( cmd , check_exit_code = check_exit_code , <EOL> extra_ok_codes = extra_ok_codes , <EOL> log_fail_as_error = log_fail_as_error , ** kwargs ) <EOL> def exists ( self , name ) : <EOL> output = self . _parent . _execute ( <EOL> [ '<STR_LIT:o>' ] , '<STR_LIT>' , [ '<STR_LIT:list>' ] , <EOL> run_as_root = cfg . CONF . AGENT . use_helper_for_ns_read ) <EOL> for line in [ l . split ( ) [ <NUM_LIT:0> ] for l in output . splitlines ( ) ] : <EOL> if name == line : <EOL> return True <EOL> return False <EOL> def vlan_in_use ( segmentation_id , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> ip_wrapper = IPWrapper ( namespace = namespace ) <EOL> interfaces = ip_wrapper . netns . execute ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:list>" ] , <EOL> check_exit_code = True ) <EOL> return '<STR_LIT>' % segmentation_id in interfaces <EOL> def vxlan_in_use ( segmentation_id , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> ip_wrapper = IPWrapper ( namespace = namespace ) <EOL> interfaces = ip_wrapper . netns . execute ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:list>" ] , <EOL> check_exit_code = True ) <EOL> return '<STR_LIT>' % segmentation_id in interfaces <EOL> def device_exists ( device_name , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> return IPDevice ( device_name , namespace = namespace ) . exists ( ) <EOL> def device_exists_with_ips_and_mac ( device_name , ip_cidrs , mac , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> device = IPDevice ( device_name , namespace = namespace ) <EOL> if mac != device . link . address : <EOL> return False <EOL> device_ip_cidrs = [ ip [ '<STR_LIT>' ] for ip in device . addr . list ( ) ] <EOL> for ip_cidr in ip_cidrs : <EOL> if ip_cidr not in device_ip_cidrs : <EOL> return False <EOL> except RuntimeError : <EOL> return False <EOL> else : <EOL> return True <EOL> def get_routing_table ( ip_version , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> ip_wrapper = IPWrapper ( namespace = namespace ) <EOL> table = ip_wrapper . netns . execute ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' % ip_version , '<STR_LIT>' ] , <EOL> check_exit_code = True ) <EOL> routes = [ ] <EOL> route_lines = ( line . split ( ) for line in table . split ( '<STR_LIT:\n>' ) if line . strip ( ) ) <EOL> for route in route_lines : <EOL> network = route [ <NUM_LIT:0> ] <EOL> data = dict ( route [ i : i + <NUM_LIT:2> ] for i in range ( <NUM_LIT:1> , len ( route ) , <NUM_LIT:2> ) ) <EOL> routes . append ( { '<STR_LIT>' : network , <EOL> '<STR_LIT>' : data . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : data . get ( '<STR_LIT>' ) } ) <EOL> return routes <EOL> def ensure_device_is_ready ( device_name , namespace = None ) : <EOL> dev = IPDevice ( device_name , namespace = namespace ) <EOL> dev . set_log_fail_as_error ( False ) <EOL> try : <EOL> dev . link . set_up ( ) <EOL> except RuntimeError : <EOL> return False <EOL> return True <EOL> def iproute_arg_supported ( command , arg ) : <EOL> command += [ '<STR_LIT>' ] <EOL> stdout , stderr = utils . execute ( command , check_exit_code = False , <EOL> return_stderr = True , log_fail_as_error = False ) <EOL> return any ( arg in line for line in stderr . split ( '<STR_LIT:\n>' ) ) <EOL> def _arping ( ns_name , iface_name , address , count ) : <EOL> arping_cmd = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , iface_name , '<STR_LIT:-c>' , count , <EOL> '<STR_LIT>' , <NUM_LIT> * count , address ] <EOL> try : <EOL> ip_wrapper = IPWrapper ( namespace = ns_name ) <EOL> ip_wrapper . netns . execute ( arping_cmd , check_exit_code = True ) <EOL> except Exception : <EOL> msg = _LE ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> LOG . exception ( msg , { '<STR_LIT>' : address , <EOL> '<STR_LIT>' : iface_name , <EOL> '<STR_LIT>' : ns_name } ) <EOL> def send_ip_addr_adv_notif ( ns_name , iface_name , address , config ) : <EOL> """<STR_LIT>""" <EOL> count = config . send_arp_for_ha <EOL> def arping ( ) : <EOL> _arping ( ns_name , iface_name , address , count ) <EOL> if count > <NUM_LIT:0> and netaddr . IPAddress ( address ) . version == <NUM_LIT:4> : <EOL> eventlet . spawn_n ( arping ) <EOL> def add_namespace_to_cmd ( cmd , namespace = None ) : <EOL> """<STR_LIT>""" <EOL> return [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , namespace ] + cmd if namespace else cmd <EOL> def get_ip_version ( ip_or_cidr ) : <EOL> return netaddr . IPNetwork ( ip_or_cidr ) . version <EOL> def get_ipv6_lladdr ( mac_addr ) : <EOL> return '<STR_LIT>' % netaddr . EUI ( mac_addr ) . ipv6_link_local ( ) </s>
