def find_from_tuple(cls, link):
     """
     Find link by providing a tuple with two ip addresses or two mac addresses
     :param link: tuple with two string elements indicating source and destination (ip or mac addresses)
     :returns: Link object
     """
     try:
         a = link[0]
         b = link[1]
     except IndexError:
         raise ValueError('Expecting tuple with source and destination')
     # find interfaces
     if (valid_ipv4(a) and valid_ipv4(b)) or (valid_ipv6(a) and valid_ipv6(b)):
         try:
             a = Ip.objects.get(address=a).interface
             b = Ip.objects.get(address=b).interface
         except Ip.DoesNotExist as e:
             raise LinkDataNotFound(e)
     elif valid_mac(a) and valid_mac(b):
         try:
             a = Interface.objects.get(mac=a)
             b = Interface.objects.get(mac=b)
         except Interface.DoesNotExist as e:
             raise LinkDataNotFound(e)
     else:
         raise ValueError('Expecting valid ipv4, ipv6 or mac address')
     # find link with interfaces
     # inverse order is also ok
     q = Q(interface_a=a, interface_b=b) | Q(interface_a=b, interface_b=a)
     link = Link.objects.filter(q).first()
     if link is None:
         raise LinkNotFound('Link matching query does not exist',
                            interface_a=a,
                            interface_b=b)
     return link

 def validate(value):
     try:
         netaddr.valid_mac(value)
     except netaddr.AddrFormatError:
         error = 'value should be MAC format.'
         raise ValueError(error)
     else:
         return value

 def check_config(self):
     super(DP, self).check_config()
     test_config_condition(not isinstance(self.dp_id, int), (
         'dp_id must be %s not %s' % (int, type(self.dp_id))))
     test_config_condition(self.dp_id < 0 or self.dp_id > 2**64-1, (
         'DP ID %s not in valid range' % self.dp_id))
     test_config_condition(not netaddr.valid_mac(self.faucet_dp_mac), (
         'invalid MAC address %s' % self.faucet_dp_mac))
     test_config_condition(not (self.interfaces or self.interface_ranges), (
         'DP %s must have at least one interface' % self))
     # To prevent L2 learning from timing out before L3 can refresh
     test_config_condition(self.timeout < self.arp_neighbor_timeout, (
         'L2 timeout must be >= ARP timeout'))
     if self.cache_update_guard_time == 0:
         self.cache_update_guard_time = int(self.timeout / 2)
     if self.learn_jitter == 0:
         self.learn_jitter = int(max(math.sqrt(self.timeout) * 3, 1))
     if self.learn_ban_timeout == 0:
         self.learn_ban_timeout = self.learn_jitter
     if self.lldp_beacon:
         self._check_conf_types(self.lldp_beacon, self.lldp_beacon_defaults_types)
         test_config_condition('send_interval' not in self.lldp_beacon, (
             'lldp_beacon send_interval not set'))
         test_config_condition('max_per_interval' not in self.lldp_beacon, (
             'lldp_beacon max_per_interval not set'))
         self.lldp_beacon = self._set_unknown_conf(
             self.lldp_beacon, self.lldp_beacon_defaults_types)
         if self.lldp_beacon['system_name'] is None:
             self.lldp_beacon['system_name'] = self.name
     if self.stack:
         self._check_conf_types(self.stack, self.stack_defaults_types)
     if self.dot1x:
         self._check_conf_types(self.dot1x, self.dot1x_defaults_types)
     self._check_conf_types(self.table_sizes, self.default_table_sizes_types)

def validate_config(options):
    config = ConfigParser.ConfigParser()
    config.readfp(options.config)
    if not config.has_section(options.env):
        error("Environment {} does not exist in file {}".format(options.env, options.config.name))
    env = dict(config.items(options.env))

    if not valid_mac(env["fuel_mac"]):
        error("Wrong MAC address for Fuel node: {}".format(env["fuel_mac"]))

    for key in ["fuel_ip", "fuel_netmask", "fuel_gw", "fuel_control_ip"]:
        if not valid_ipv4(env[key]):
            error("Wrong IP address ({}) for {} parameter".format(env[key], key))

    ip = IPNetwork(env["fuel_ip"] + "/" + env["fuel_netmask"])
    gw = IPAddress(env["fuel_gw"])
    if gw not in ip:
        error("Gateway address is not within network range")

    for path in [
        env["image_mount_dir"],
        env["ks_files_dir"],
        env["ks_patch"],
        env["pxe_config_dir"],
        env["esxi_private_key_file"],
    ]:
        if not os.path.exists(path):
            error("Path {} does not exist or is not accessible".format(path))

    env["pxe_config_file"] = os.path.join(env["pxe_config_dir"], "01-{}".format(env["fuel_mac"].replace(":", "-")))
    return env

 def __post_load_macs(self):
     """
     Called by post_load() to load internal representation of macs
     """
     self.macs = self.link_data('macs', dict)
     for k, v in self.macs.items():
         old_k = k
         k = k.lower()
         if netaddr.valid_mac(k) is False:
             self.app.log.error("data:mac: Invalid key format '{}: File {}".format(k, self.path))
             continue
         if v is not None:
             v = v.lower()
             if netaddr.valid_mac(k) is False:
                 self.app.log.error("data:mac: Invalid value format '{}: File {}".format(k, self.path))
                 continue
         del self.macs[old_k]
         self.macs[k] = v

def _validate_mac_address(data, valid_values=None):
    try:
        valid_mac = netaddr.valid_mac(_validate_no_whitespace(data))
        if valid_mac is False:
            msg = _("'%s' is not a valid MAC address") % data
            LOG.debug(msg)
            return msg
    except AttributeError as ex:
        msg = _("MAC address must be string: %s") % ex
        LOG.exception(msg)
        return msg

 def test_list_virtual_interfaces(self):
     # Positive test:Should be able to GET the virtual interfaces list
     # for a given server_id
     resp, output = self.client.list_virtual_interfaces(self.server_id)
     self.assertEqual(200, resp.status)
     self.assertNotEqual(output, None)
     virt_ifaces = output
     self.assertNotEqual(0, len(virt_ifaces["virtual_interfaces"]), "Expected virtual interfaces, got 0 interfaces.")
     for virt_iface in virt_ifaces["virtual_interfaces"]:
         mac_address = virt_iface["mac_address"]
         self.assertTrue(netaddr.valid_mac(mac_address), "Invalid mac address detected.")

 def test_list_virtual_interfaces(self):
     # Positive test:Should be able to GET the virtual interfaces list
     # for a given server_id
     output = self.client.list_virtual_interfaces(self.server_id)
     self.assertIsNotNone(output)
     virt_ifaces = output
     self.assertNotEqual(0, len(virt_ifaces['virtual_interfaces']),
                         'Expected virtual interfaces, got 0 interfaces.')
     for virt_iface in virt_ifaces['virtual_interfaces']:
         mac_address = virt_iface['mac_address']
         self.assertTrue(netaddr.valid_mac(mac_address),
                         "Invalid mac address detected.")

def _validate_mac_address(data, valid_values=None):
    try:
        valid_mac = netaddr.valid_mac(_validate_no_whitespace(data))
    except Exception:
        valid_mac = False
    # TODO(arosen): The code in this file should be refactored
    # so it catches the correct exceptions. _validate_no_whitespace
    # raises AttributeError if data is None.
    if not valid_mac:
        msg = _("'%s' is not a valid MAC address") % data
        LOG.debug(msg)
        return msg

def validate_mac_address(data, valid_values=None):
    try:
        valid_mac = netaddr.valid_mac(validate_no_whitespace(data))
    except Exception:
        valid_mac = False

    if valid_mac:
        valid_mac = (not netaddr.EUI(data) in
                     map(netaddr.EUI, constants.INVALID_MAC_ADDRESSES))
    # TODO(arosen): The code in this file should be refactored
    # so it catches the correct exceptions. validate_no_whitespace
    # raises AttributeError if data is None.
    if not valid_mac:
        msg = _("'%s' is not a valid MAC address") % data
        LOG.debug(msg)
        return msg

    def check_config(self):
        super(VLAN, self).check_config()
        test_config_condition(not self.vid_valid(self.vid), 'invalid VID %s' % self.vid)
        test_config_condition(not netaddr.valid_mac(self.faucet_mac), (
            'invalid MAC address %s' % self.faucet_mac))

        test_config_condition(
            self.acl_in and self.acls_in, 'found both acl_in and acls_in, use only acls_in')
        if self.acl_in and not isinstance(self.acl_in, list):
            self.acls_in = [self.acl_in,]
            self.acl_in = None
        if self.acls_in:
            for acl in self.acls_in:
                test_config_condition(
                    not isinstance(acl, (int, str)), 'acl names must be int or str')

        if self.max_hosts:
            if not self.proactive_arp_limit:
                self.proactive_arp_limit = 2 * self.max_hosts
            if not self.proactive_nd_limit:
                self.proactive_nd_limit = 2 * self.max_hosts

        if self.faucet_vips:
            self.faucet_vips = frozenset([
                self._check_ip_str(ip_str, ip_method=ipaddress.ip_interface)
                for ip_str in self.faucet_vips])

        if self.routes:
            test_config_condition(not isinstance(self.routes, list), 'invalid VLAN routes format')
            try:
                self.routes = [route['route'] for route in self.routes]
            except TypeError:
                raise InvalidConfigError('%s is not a valid routes value' % self.routes)
            except KeyError:
                pass
            for route in self.routes:
                test_config_condition(not isinstance(route, dict), 'invalid VLAN route format')
                test_config_condition('ip_gw' not in route, 'missing ip_gw in VLAN route')
                test_config_condition('ip_dst' not in route, 'missing ip_dst in VLAN route')
                ip_gw = self._check_ip_str(route['ip_gw'])
                ip_dst = self._check_ip_str(route['ip_dst'], ip_method=ipaddress.ip_network)
                test_config_condition(
                    ip_gw.version != ip_dst.version,
                    'ip_gw version does not match the ip_dst version')
                self.add_route(ip_dst, ip_gw)

    def test_list_virtual_interfaces(self):
        # Positive test:Should be able to GET the virtual interfaces list
        # for a given server_id

        if CONF.service_available.neutron:
            with testtools.ExpectedException(exceptions.BadRequest):
                self.client.list_virtual_interfaces(self.server['id'])
        else:
            output = self.client.list_virtual_interfaces(self.server['id'])
            virt_ifaces = output['virtual_interfaces']
            self.assertNotEmpty(virt_ifaces,
                                'Expected virtual interfaces, got 0 '
                                'interfaces.')
            for virt_iface in virt_ifaces:
                mac_address = virt_iface['mac_address']
                self.assertTrue(netaddr.valid_mac(mac_address),
                                "Invalid mac address detected. mac address: %s"
                                % mac_address)

 def serialize(self, _payload=None, _prev=None):
     opt_buf = bytearray()
     if self.options is not None:
         opt_buf = self.options.serialize()
     if netaddr.valid_mac(self.chaddr):
         chaddr = addrconv.mac.text_to_bin(self.chaddr)
     else:
         chaddr = self.chaddr
     self.hlen = len(chaddr)
     return struct.pack(self._DHCP_PACK_STR, self.op, self.htype, self.hlen,
                        self.hops, self.xid, self.secs, self.flags,
                        addrconv.ipv4.text_to_bin(self.ciaddr),
                        addrconv.ipv4.text_to_bin(self.yiaddr),
                        addrconv.ipv4.text_to_bin(self.siaddr),
                        addrconv.ipv4.text_to_bin(self.giaddr),
                        chaddr,
                        self.sname.encode('ascii'),
                        self.boot_file.encode('ascii')) + opt_buf

	def get(self, mac, full):
		"""
		json response from www.macvendorlookup.com:

		{u'addressL1': u'1 Infinite Loop',
		u'addressL2': u'',
		u'addressL3': u'Cupertino CA 95014',
		u'company': u'Apple',
		u'country': u'UNITED STATES',
		u'endDec': u'202412195315711',
		u'endHex': u'B817C2FFFFFF',
		u'startDec': u'202412178538496',
		u'startHex': u'B817C2000000',
		u'type': u'MA-L'}
		"""
		unknown = {'company': 'unknown'}
		if not valid_mac(mac):
			print('Error: the mac addr {} is not valid'.format(mac))
			return

		try:
			r = requests.get('http://www.macvendorlookup.com/api/v2/' + mac)
		except requests.exceptions.HTTPError as e:
			print ("HTTPError:", e.message)
			return unknown

		if r.status_code == 204:  # no content found, bad MAC addr
			print ('ERROR: Bad MAC addr:', mac)
			return unknown
		elif r.headers['content-type'] != 'application/json':
			print ('ERROR: Wrong content type:', r.headers['content-type'])
			return unknown

		a = {}

		try:
			if full: a = r.json()[0]
			else: a['company'] = r.json()[0]['company']
			# print 'GOOD:',r.status_code,r.headers,r.ok,r.text,r.reason
		except:
			print ('ERROR:', r.status_code, r.headers, r.ok, r.text, r.reason)
			a = unknown

		return a

    def test_list_virtual_interfaces(self):
        # Positive test:Should be able to GET the virtual interfaces list
        # for a given server_id

        if CONF.service_available.neutron:
            # TODO(mriedem): After a microversion implements the API for
            # neutron, a 400 should be a failure for nova-network and neutron.
            with testtools.ExpectedException(exceptions.BadRequest):
                self.client.list_virtual_interfaces(self.server['id'])
        else:
            output = self.client.list_virtual_interfaces(self.server['id'])
            virt_ifaces = output
            self.assertNotEmpty(virt_ifaces['virtual_interfaces'],
                                'Expected virtual interfaces, got 0 '
                                'interfaces.')
            for virt_iface in virt_ifaces['virtual_interfaces']:
                mac_address = virt_iface['mac_address']
                self.assertTrue(netaddr.valid_mac(mac_address),
                                "Invalid mac address detected. mac address: %s"
                                % mac_address)

def mac_address(mac, for_item=True):
    """
    Validate as an Eternet mac address.

    @param: str mac (mac address)
    @returns: str mac or CX
    """
    if not isinstance(mac, basestring):
        raise CX("Invalid input, mac must be a string")
    else:
        mac = mac.lower().strip()

    if for_item is True:
        # this value has special meaning for items
        if mac == "random":
            return mac

    if not netaddr.valid_mac(mac):
        raise CX("Invalid mac address format (%s)" % mac)

    return mac

def validate_mac_address(data, valid_values=None):
    """Validate data is a MAC address.

    :param data: The data to validate.
    :param valid_values: Not used!
    :returns: None if the data is a valid MAC address, otherwise a human
    readable message as to why validation failed.
    """
    try:
        valid_mac = netaddr.valid_mac(validate_no_whitespace(data))
    except Exception:
        valid_mac = False

    if valid_mac:
        valid_mac = (not netaddr.EUI(data) in
                     map(netaddr.EUI, constants.INVALID_MAC_ADDRESSES))
    # TODO(arosen): The code in this file should be refactored
    # so it catches the correct exceptions. validate_no_whitespace
    # raises AttributeError if data is None.
    if not valid_mac:
        msg = _("'%s' is not a valid MAC address") % data
        LOG.debug(msg)
        return msg

def test_valid_mac():
    assert valid_mac('00-B0-D0-86-BB-F7')
    assert not valid_mac('00-1B-77-49-54-FD-12-34')

def validate_endpoint(config, combined_id, endpoint):
    """
    Ensures that the supplied endpoint is valid. Once this routine has returned
    successfully, we know that all required fields are present and have valid
    values.

    Has the side-effect of putting IP and MAC addresses in canonical form in
    the input dict.

    :param config: configuration structure
    :param combined_id: EndpointId object
    :param endpoint: endpoint dictionary as read from etcd
    :raises ValidationFailed
    """
    issues = []

    if not isinstance(endpoint, dict):
        raise ValidationFailed("Expected endpoint to be a dict.")

    if not VALID_ID_RE.match(combined_id.endpoint):
        issues.append("Invalid endpoint ID '%r'." % combined_id.endpoint)

    if "state" not in endpoint:
        issues.append("Missing 'state' field.")
    elif endpoint["state"] not in ("active", "inactive"):
        issues.append("Expected 'state' to be one of active/inactive.")

    for field in ["name", "mac"]:
        if field not in endpoint:
            issues.append("Missing '%s' field." % field)
        elif not isinstance(endpoint[field], StringTypes):
            issues.append("Expected '%s' to be a string; got %r." %
                          (field, endpoint[field]))
        elif field == "mac":
            if not netaddr.valid_mac(endpoint.get("mac")):
                issues.append("Invalid MAC address")
            else:
                endpoint["mac"] = canonicalise_mac(endpoint.get("mac"))

    if "profile_id" in endpoint:
        if "profile_ids" not in endpoint:
            endpoint["profile_ids"] = [endpoint["profile_id"]]
        del endpoint["profile_id"]

    if "profile_ids" not in endpoint:
        issues.append("Missing 'profile_id(s)' field.")
    else:
        for value in endpoint["profile_ids"]:
            if not isinstance(value, StringTypes):
                issues.append("Expected profile IDs to be strings.")
                break

            if not VALID_ID_RE.match(value):
                issues.append("Invalid profile ID '%r'." % value)

    if ("name" in endpoint and isinstance(endpoint['name'], StringTypes)
        and combined_id.host == config.HOSTNAME
        and not endpoint["name"].startswith(config.IFACE_PREFIX)):
        # Only test the interface for local endpoints - remote hosts may have
        # a different interface prefix.
        issues.append("Interface %r does not start with %r." %
                      (endpoint["name"], config.IFACE_PREFIX))

    if "labels" in endpoint:
        _validate_label_dict(issues, endpoint["labels"])

    for version in (4, 6):
        nets = "ipv%d_nets" % version
        if nets not in endpoint:
            endpoint[nets] = []
        else:
            canonical_nws = []
            nets_list = endpoint.get(nets, [])
            if not isinstance(nets_list, list):
                issues.append("%s should be a list." % nets)
            else:
                for ip in nets_list:
                    if not validate_cidr(ip, version):
                        issues.append("IP address %r is not a valid "
                                      "IPv%d CIDR." % (ip, version))
                        break
                    else:
                        canonical_nws.append(canonicalise_cidr(ip, version))
                endpoint[nets] = canonical_nws

        n_key = nat_key(version)
        nat_maps = endpoint.get(n_key, None)
        if nat_maps is not None:
            if isinstance(nat_maps, list):
                canonical_nm = []
                for nat_map in nat_maps:
                    canonical = {}
                    for t in "int", "ext":
                        canonical[t] = None
                        ip = nat_map.get("%s_ip" % t, None)
                        if ip:
                            if validate_ip_addr(ip, version):
                                canonical[t] = canonicalise_ip(ip, version)
                            else:
                                issues.append("%s_ip (%r) is not a valid IPv%d"
#.........这里部分代码省略.........

 def get_link(cls, source, target, topology=None):
     """
     Find link between source and target, (or vice versa, order is irrelevant).
     :param source: ip or mac addresses
     :param target: ip or mac addresses
     :param topology: optional topology relation
     :returns: Link object
     :raises: LinkNotFound
     """
     a = source
     b = target
     # ensure parameters are coherent
     if not (valid_ipv4(a) and valid_ipv4(b)) and not (valid_ipv6(a) and valid_ipv6(b)) and not (valid_mac(a) and valid_mac(b)):
         raise ValueError('Expecting valid ipv4, ipv6 or mac address')
     # get interfaces
     a = cls._get_link_interface(a)
     b = cls._get_link_interface(b)
     # raise LinkDataNotFound if an interface is not found
     not_found = []
     if a is None:
         not_found.append(source)
     if b is None:
         not_found.append(target)
     if not_found:
         msg = 'the following interfaces could not be found: {0}'.format(', '.join(not_found))
         raise LinkDataNotFound(msg)
     # find link with interfaces
     # inverse order is also ok
     q = (Q(interface_a=a, interface_b=b) | Q(interface_a=b, interface_b=a))
     # add topology to lookup
     if topology:
         q = q & Q(topology=topology)
     link = Link.objects.filter(q).first()
     if link is None:
         raise LinkNotFound('Link matching query does not exist',
                            interface_a=a,
                            interface_b=b,
                            topology=topology)
     return link