# # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import abc import copy import re import string import typing as ty from nova import exception from nova.i18n import _ from nova import objects from nova.pci.request import PCI_REMOTE_MANAGED_TAG from nova.pci import utils from oslo_log import log as logging from oslo_utils import strutils MAX_VENDOR_ID = 0xFFFF MAX_PRODUCT_ID = 0xFFFF MAX_FUNC = 0x7 MAX_DOMAIN = 0xFFFF MAX_BUS = 0xFF MAX_SLOT = 0x1F ANY = '*' REGEX_ANY = '.*' LOG = logging.getLogger(__name__) PCISpecAddressType = ty.Union[ty.Dict[str, str], str] class PciAddressSpec(metaclass=abc.ABCMeta): """Abstract class for all PCI address spec styles This class checks the address fields of the pci.device_spec """ def __init__(self, pci_addr: str) -> None: self.domain = '' self.bus = '' self.slot = '' self.func = '' @abc.abstractmethod def match(self, pci_addr): pass def is_single_address(self) -> bool: return all([ all(c in string.hexdigits for c in self.domain), all(c in string.hexdigits for c in self.bus), all(c in string.hexdigits for c in self.slot), all(c in string.hexdigits for c in self.func)]) def _set_pci_dev_info( self, prop: str, maxval: int, hex_value: str ) -> None: a = getattr(self, prop) if a == ANY: return try: v = int(a, 16) except ValueError: raise exception.PciConfigInvalidSpec( reason=_("property %(property)s ('%(attr)s') does not parse " "as a hex number.") % {'property': prop, 'attr': a}) if v > maxval: raise exception.PciConfigInvalidSpec( reason=_("property %(property)s (%(attr)s) is greater than " "the maximum allowable value (%(max)X).") % {'property': prop, 'attr': a, 'max': maxval}) setattr(self, prop, hex_value % v) class PhysicalPciAddress(PciAddressSpec): """Manages the address fields for a fully-qualified PCI address. This function class will validate the address fields for a single PCI device. """ def __init__(self, pci_addr: PCISpecAddressType) -> None: try: # TODO(stephenfin): Is this ever actually a string? if isinstance(pci_addr, dict): self.domain = pci_addr['domain'] self.bus = pci_addr['bus'] self.slot = pci_addr['slot'] self.func = pci_addr['function'] else: self.domain, self.bus, self.slot, self.func = ( utils.get_pci_address_fields(pci_addr)) self._set_pci_dev_info('func', MAX_FUNC, '%1x') self._set_pci_dev_info('domain', MAX_DOMAIN, '%04x') self._set_pci_dev_info('bus', MAX_BUS, '%02x') self._set_pci_dev_info('slot', MAX_SLOT, '%02x') except (KeyError, ValueError): raise exception.PciDeviceWrongAddressFormat(address=pci_addr) def match(self, phys_pci_addr: PciAddressSpec) -> bool: conditions = [ self.domain == phys_pci_addr.domain, self.bus == phys_pci_addr.bus, self.slot == phys_pci_addr.slot, self.func == phys_pci_addr.func, ] return all(conditions) def __str__(self): return f'{self.domain}:{self.bus}:{self.slot}.{self.func}' class PciAddressGlobSpec(PciAddressSpec): """Manages the address fields with glob style. This function class will validate the address fields with glob style, check for wildcards, and insert wildcards where the field is left blank. """ def __init__(self, pci_addr: str) -> None: self.domain = ANY self.bus = ANY self.slot = ANY self.func = ANY dbs, sep, func = pci_addr.partition('.') if func: self.func = func.strip() self._set_pci_dev_info('func', MAX_FUNC, '%01x') if dbs: dbs_fields = dbs.split(':') if len(dbs_fields) > 3: raise exception.PciDeviceWrongAddressFormat(address=pci_addr) # If we got a partial address like ":00.", we need to turn this # into a domain of ANY, a bus of ANY, and a slot of 00. This code # allows the address bus and/or domain to be left off dbs_all = [ANY] * (3 - len(dbs_fields)) dbs_all.extend(dbs_fields) dbs_checked = [s.strip() or ANY for s in dbs_all] self.domain, self.bus, self.slot = dbs_checked self._set_pci_dev_info('domain', MAX_DOMAIN, '%04x') self._set_pci_dev_info('bus', MAX_BUS, '%02x') self._set_pci_dev_info('slot', MAX_SLOT, '%02x') def match(self, phys_pci_addr: PciAddressSpec) -> bool: conditions = [ self.domain in (ANY, phys_pci_addr.domain), self.bus in (ANY, phys_pci_addr.bus), self.slot in (ANY, phys_pci_addr.slot), self.func in (ANY, phys_pci_addr.func) ] return all(conditions) class PciAddressRegexSpec(PciAddressSpec): """Manages the address fields with regex style. This function class will validate the address fields with regex style. The validation includes check for all PCI address attributes and validate their regex. """ def __init__(self, pci_addr: dict) -> None: try: self.domain = pci_addr.get('domain', REGEX_ANY) self.bus = pci_addr.get('bus', REGEX_ANY) self.slot = pci_addr.get('slot', REGEX_ANY) self.func = pci_addr.get('function', REGEX_ANY) self.domain_regex = re.compile(self.domain) self.bus_regex = re.compile(self.bus) self.slot_regex = re.compile(self.slot) self.func_regex = re.compile(self.func) except re.error: raise exception.PciDeviceWrongAddressFormat(address=pci_addr) def match(self, phys_pci_addr: PciAddressSpec) -> bool: conditions = [ bool(self.domain_regex.match(phys_pci_addr.domain)), bool(self.bus_regex.match(phys_pci_addr.bus)), bool(self.slot_regex.match(phys_pci_addr.slot)), bool(self.func_regex.match(phys_pci_addr.func)) ] return all(conditions) class WhitelistPciAddress(object): """Manages the address fields of the whitelist. This class checks the address fields of the pci.device_spec configuration option, validating the address fields. Example configs: | [pci] | device_spec = {"address":"*:0a:00.*", | "physical_network":"physnet1"} | device_spec = {"address": {"domain": ".*", "bus": "02", "slot": "01", "function": "[0-2]"}, "physical_network":"net1"} | device_spec = {"vendor_id":"1137","product_id":"0071"} """ def __init__( self, pci_addr: PCISpecAddressType, is_physical_function: bool ) -> None: self.is_physical_function = is_physical_function self._init_address_fields(pci_addr) def _check_physical_function(self) -> None: if self.pci_address_spec.is_single_address(): self.is_physical_function = ( utils.is_physical_function( self.pci_address_spec.domain, self.pci_address_spec.bus, self.pci_address_spec.slot, self.pci_address_spec.func)) def _init_address_fields(self, pci_addr: PCISpecAddressType) -> None: self.pci_address_spec: PciAddressSpec if not self.is_physical_function: if isinstance(pci_addr, str): self.pci_address_spec = PciAddressGlobSpec(pci_addr) elif isinstance(pci_addr, dict): self.pci_address_spec = PciAddressRegexSpec(pci_addr) else: raise exception.PciDeviceWrongAddressFormat(address=pci_addr) self._check_physical_function() else: self.pci_address_spec = PhysicalPciAddress(pci_addr) def match(self, pci_addr: str, pci_phys_addr: ty.Optional[str]) -> bool: """Match a device to this PciAddress. Assume this is called with a ``pci_addr`` and ``pci_phys_addr`` reported by libvirt. No attempt is made to verify if ``pci_addr`` is a VF of ``pci_phys_addr``. :param pci_addr: PCI address of the device to match. :param pci_phys_addr: PCI address of the parent of the device to match (or None if the device is not a VF). """ # Try to match on the parent PCI address if the PciDeviceSpec is a # PF (sriov is available) and the device to match is a VF. This # makes it possible to specify the PCI address of a PF in the # pci.device_spec to match any of its VFs' PCI addresses. if self.is_physical_function and pci_phys_addr: pci_phys_addr_obj = PhysicalPciAddress(pci_phys_addr) if self.pci_address_spec.match(pci_phys_addr_obj): return True # Try to match on the device PCI address only. pci_addr_obj = PhysicalPciAddress(pci_addr) return self.pci_address_spec.match(pci_addr_obj) class PciDeviceSpec(PciAddressSpec): def __init__(self, dev_spec: ty.Dict[str, str]) -> None: # stored for better error reporting self.dev_spec_conf = copy.deepcopy(dev_spec) # the non tag fields (i.e. address, devname) will be removed by # _init_dev_details self.tags = dev_spec self._init_dev_details() def _address_obj(self) -> ty.Optional[WhitelistPciAddress]: address_obj = None if self.dev_name: address_str, pf = utils.get_function_by_ifname(self.dev_name) if not address_str: return None # Note(moshele): In this case we always passing a string # of the PF pci address address_obj = WhitelistPciAddress(address_str, pf) else: # use self.address address_obj = self.address return address_obj def _init_dev_details(self) -> None: self.vendor_id = self.tags.pop("vendor_id", ANY) self.product_id = self.tags.pop("product_id", ANY) self.dev_name = self.tags.pop("devname", None) self.address: ty.Optional[WhitelistPciAddress] = None # Note(moshele): The address attribute can be a string or a dict. # For glob syntax or specific pci it is a string and for regex syntax # it is a dict. The WhitelistPciAddress class handles both types. address = self.tags.pop("address", None) self.vendor_id = self.vendor_id.strip() self._set_pci_dev_info('vendor_id', MAX_VENDOR_ID, '%04x') self._set_pci_dev_info('product_id', MAX_PRODUCT_ID, '%04x') if address and self.dev_name: raise exception.PciDeviceInvalidDeviceName() if not self.dev_name: self.address = WhitelistPciAddress(address or '*:*:*.*', False) # PFs with remote_managed tags are explicitly not supported. If they # are tagged as such by mistake in the whitelist Nova will # raise an exception. The reason for excluding PFs is the lack of a way # for an instance to access the control plane at the remote side (e.g. # on a DPU) for managing the PF representor corresponding to the PF. address_obj = self._address_obj() self._remote_managed = strutils.bool_from_string( self.tags.get(PCI_REMOTE_MANAGED_TAG)) if self._remote_managed: if address_obj is None: # Note that this will happen if a netdev was specified in the # whitelist but it is not actually present on a system - in # this case Nova is not able to look up an address by # a netdev name. raise exception.PciDeviceRemoteManagedNotPresent() elif address_obj.is_physical_function: pf_addr = str(address_obj.pci_address_spec) vf_product_id = utils.get_vf_product_id_by_pf_addr(pf_addr) # VF vendor IDs have to match the corresponding PF vendor IDs # per the SR-IOV spec so we use it for matching here. pf_vendor_id, pf_product_id = utils.get_pci_ids_by_pci_addr( pf_addr) # Check the actual vendor ID and VF product ID of an assumed # VF (based on the actual PF). The VF product ID must match # the actual one if this is a VF device spec. if (self.product_id == vf_product_id and self.vendor_id in (pf_vendor_id, ANY)): pass elif (self.product_id in (pf_product_id, ANY) and self.vendor_id in (pf_vendor_id, ANY)): raise exception.PciDeviceInvalidPFRemoteManaged( address_obj.pci_address_spec) else: # The specified product and vendor IDs of what is supposed # to be a VF corresponding to the PF PCI address do not # match the actual ones for this PF. This means that the # whitelist is invalid. raise exception.PciConfigInvalidSpec( reason=_('the specified VF vendor ID %(vendor_id)s and' ' product ID %(product_id)s do not match the' ' expected VF IDs based on the corresponding' ' PF identified by PCI address %(pf_addr)s') % {'vendor_id': self.vendor_id, 'product_id': self.product_id, 'pf_addr': pf_addr}) def _ensure_remote_managed_dev_vpd_serial( self, dev_dict: ty.Dict[str, ty.Any]) -> bool: """Ensure the presence of a serial number field in PCI VPD. A card serial number extracted from PCI VPD is required to allow a networking backend to identify which remote host needs to program a given device. So if a device is tagged as remote_managed, it must have the card serial number or be filtered out. """ if not self._remote_managed: return True card_sn = dev_dict.get('capabilities', {}).get( 'vpd', {}).get('card_serial_number') # None or empty card_serial_number should be filtered out. That would # mean either no serial number in the VPD (if present at all) or SN is # an empty string which is not useful for device identification. return bool(card_sn) def match(self, dev_dict: ty.Dict[str, ty.Any]) -> bool: address_obj: ty.Optional[WhitelistPciAddress] = self._address_obj() if not address_obj: return False return all([ self.vendor_id in (ANY, dev_dict['vendor_id']), self.product_id in (ANY, dev_dict['product_id']), address_obj.match(dev_dict['address'], dev_dict.get('parent_addr')), self._ensure_remote_managed_dev_vpd_serial(dev_dict), ]) def match_pci_obj(self, pci_obj: 'objects.PciDevice') -> bool: dev_dict = { 'vendor_id': pci_obj.vendor_id, 'product_id': pci_obj.product_id, 'address': pci_obj.address, 'parent_addr': pci_obj.parent_addr, 'capabilities': { 'vpd': {'card_serial_number': pci_obj.card_serial_number}} } return self.match(dev_dict) def get_tags(self) -> ty.Dict[str, str]: return self.tags