Automating SR-IOV/VFIO bond creation on Oracle Compute Cloud@Customer and Private Cloud Appliance

oci compute instance launch \
–shape VM.PCAStandard1.1 \
–subnet-id ocid1.subnet…. \
–display-name “SR-IOV instance” \
–availability-domain AD-1 \
–source-details ‘{ “imageId”: “ocid1.image…”, “sourceType”: “image”}’ \
–user-data-file configure_vfio.sh \
–ssh-authorized-keys-file $HOME/.ssh/id_rsa.pub 

oci compute instance attach-vnic \
–instance-id ocid1.instance… \
–subnet-id ocid1.subnet… 

#!/bin/bash
# Copyright (c) 2024, 2025, Oracle and/or its affiliates.
# Version 1.3

# This script can be used as user-data for cloud-init to automate
# bond interface configuration on top of SR-IOV/VFIO interfaces.

config_script=”/usr/sbin/configure_vfio.py”
service_file=”configure_vfio.service”
this_archive=”$(/usr/bin/readlink -f “$0″)”
/usr/bin/sed ‘0,/^#BASE64_CONTENT#$/d’ “$this_archive” | \
  /usr/bin/base64 -d | /usr/bin/gunzip > “$config_script”
/usr/bin/chmod +x “$config_script”

cat > “/etc/systemd/system/$service_file” <<-EOF
[Unit]
Description=configure bonding with SR-IOV/VFIO virtual functions
After=network-online.target
Wants=network.target

[Service]
Type=oneshot
RemainAfterExit=yes
ExecStart=$config_script –debug

[Install]
WantedBy=multi-user.target
EOF
/usr/bin/systemctl enable “$service_file”
/usr/bin/systemctl start “$service_file”
exit 0
#BASE64_CONTENT#
H4sICLPZp2gCA2NvbmZpZ3VyZV92ZmlvLnB5AMU7a3PburHf9StQpjMhW4lycs7JNOqoU8V2enXr
2B47x7f3qi4Lk6DEY77Kh2xVo//eXQAkAT5kJ+3M9UwiElgsFovFPsE3v5qWeTZ9COIpi7ck3RWb
JP5hNHpDTpN0lwXrTUFM1yLvT97/OMb/fxqTq4y6ISM09qZJRoIiJ9T3gzCgBcttGHnHsjxIYvLO
BkR+lkTklxxegyhNsoKECfVgRM4bHf4mgFJabMLgoYK7hlfRkZcPaZa4LM+rvqyMEUPT4UDLmFwv
r8/HQHiUhqxg3rXoE0iKXRrE6wrBRZAXY3IWuPD/It7dFhksKy2AahqOyS0rxCCarVOa5awatsjW
ZcTi4hobMwFTZiFQbbMsQ2YIuP/6+vX6HBvG5OebC/6kAWfsHyXLiwr8RryOsTtJWSxgN0WR2m4Y
wHwq3huWp0mcMwGU1bS5sOwAtyWHRke+CSA3CUPm4upqDsY0Yl5RphVIkFLPyxQWB6kTs+IpyR7H
+Cx7UTC+bhh5ZLuc+LDgAl7++/bqkiRlkZYFSXzy9yAlk18IJfnfR8vPl4sv52ROjMDHKY3Rl8Xt
1/MbbIloXrDMGC3Ozm7Ob2+xSc4i2pzl5eerqtUJYj8xRhdXp4sLbAsTl4bG6O4zAjkXy9uv2Lr1
EcwJYXcBdnn55wpFGMSPAgOHh/YzThRCQ4dnjD5dXZ451zfnn5d/kZgA+A1ZXm8/oJwBL4A1Mbk6
XYJUMOIHz8wjRQLLnH74cQTNzvL67oNz+/Ony/OvEpFzcX4JyKAfMG3jwCURK6hHC8oZOPrifFmc
OrhWwQ93ASsFFsHw5R3gw9Y0C7ZwsJYptt/dyNZtkBUlDW+A6ywTfTCzc7oUqATBp4GXfQJGPfKh
HxzJ6HPO6iDdflgIdrOcA1wuT50lZwuSugSejH7++tn5HbbAw+R3xuj2/BTYtLj5X0FGI0Smccvc
JPZotgNqxsTkOJzAg2ejER/DskafFheLy9Nz54ZT+kBDGrtsksG6q56/XGldzwn0jUZuSEE672Bf
Tv31bETgDxC7kYewXInlqMUCYIbs43KTb5InCeUbezHigPI58Ugl8ggjRuU7kMnILUINMeKtewTg
G1ASjB/RfDadPj092Y8si1loJ9l66iXu9CxxuaqgeOym8iiBBpo+AJvg1y6eC45JvjtR4rEcptw3
DJqR/WFMFLZAg/EcBYWzofnGSZMwcHfGDMSb7lj2/rc/GIcDx+kxn5Zh4ai4AbWCiYNV0ujgZuUO
aB9cMi4K1vTuw0f7/U8/2vJ3mqTudPt+ykENffiGUQ90PhJvLEqwIFnwT75upO0Tg/OSSaNhHOo1
O1FRwoiPJycnAht95vSC6ipB483JDyeS06hwoiCIwIZsaVgykrOCKx8EPyHQjNoOTgKJOQ+DnLz/
6SQCW9TMJYbPsaNn/9ZBsSkfbEADFg3pnCZuMPmFbukk9x6nD2HyMBXqavoQuROAA2zTPHOhFWQD
AafKYICZgkY4s7HjzcX7d3xKQOk5GVs/AxmNhjYz42/m6mTykU7+uZj838S5/+3Knt1bettveNv+
h/GhBWv92rA49jgCjLHveEEG+NF4msaUFe70UojeFxrTNdCPQLYHgypBIQ7oVhApx8xZ6I81eZz3
CdIYRj2U6/lnGubwklcH3wnSuaogzMskhn7837LEgeVHDGaxW4KpvuqAfCqA4L96lzoxQKivNWDg
i5HN9PgHOjUuTOM38IeuRIyOAfIlWJewMaj6CfZJztYTOiB9AQg8A4blBY2LgAu54wcs9HKzBV1t
iI9Gea634SZNifHx46SedoLT2vhqiK35Y466w4VDtkm8erOC3GnUqSl/ZyQvMotM/gCMTMJmrUW2
0xfeHWpp/RkryiwmX7Oy2Qb27LK0IHd48rgn0+IlaOVRazyXDEXAjvMNOaPIB2xZV0TipEDr2+nI
dWIyGoCz1pBq+oaGJirB6XpgoDLgnw/atIvwoOw6kELjndmRNqvFVZ/T1wGzpQ0koKr6AZr90DH2
rsX4lBQbMpkg1glgBf8b3oJ0UpmxJBUuXrVK8I23gcfdlDVIEctswxokvNZOdkQLd2MOrsZ6Bam+
gTobPQp0mM5m5O1+EN/hbU0w7DD6k0B3ChYDNDyQcoxm/TAcY+9riO5A4J9YSY0V/EFSH7q+RTVT
vriuznzWsYPvcKUFOyS8fzObadEA6m1vJxRBs1Zh0TN7jecyVDSU0IC+UIF76DuQCjPJaVHC6vaZ
jYSUObgg0sDPlL1A+wuhwRYXl9kSwA7ARwJN2Kduwft6PMzAH9xqB4yT3QveRxuHbkkEUgL+PWcz
9tt5GgYFtnQoadBj93F+59RnIjw1gRsiSOQ6FmPHFYaO97O24mtCWjO3uCYjDFQhWd0r2hAsDhr/
CM6vNLnga87qMDQvPPDq58J4wgsElvyFz92Obbtqc9DcwSRWm149gEYQwTzTqukQPzUl4gdI3jD3
cY6GQvUkQM4c1fEWun2IaW5a2cUOT+y2+17TgzF+dx0cibJjbmoL+DZ1wtOtPNfj9IHUge8DNML2
NW0icp9XQbsgt+tJ14dmrgPIVmvYSD+BM1plAkw5nyXCenEKujKNYs+9SgFgZzCJaYEsuGDSTB7F
WZ1BRyRGd3vamkdTOV28gm02ey5YtZvq1vBBVtu9qDMmuFDWJ7+jrmo+v7mBUBG2tUAPro6tf765
QPUlOGf7ZRjihoAaA7bAAYU+Bi9GD8JKz5H92zF5a/+SBLHpv0W9RfZbUOiq0mNtpWcddJSdNVap
oMEl/lsrMjqnQmxE3wE4LvccpD6aPaeme/6SDHSSvkXmlrOL80pgRHcqDM0tmiPz2eLdz9ht1tmH
MWlyIvgsEyH4yHMf+FAlOizLGmszPrLdPKTRg0fJ80z1dZ9XNaZ7Zcxx/c91WQwhn09BQTrcK+Je
InXNpn2mcHAMMazb+OJH7MQq4IsPcPENLmRQwJkjMzWWHSZPLDMtMp8j7ur1/kXCt75C+wC530+k
mnXDzuP0gJeGHnZDxYx4MGFPvCInrgE5L6o03pjsDxZvqVN4nC/cwzdaUs6DfoUDws7+p/kgDki9
Pou75GIHReLTQhcqK3JU6aaSYbTuW/SyZ5gfpUsnXNINp8Z5lchVKeyVp6+kxQ7tbPexyj4q/To5
nARLniVpV0KwXC28fK9OZn3hJno3rxr+rnd4C9JOk9RUyOHdmKnEZa8eVmJn7kUCSfiM2njF1vMI
4ZYn/qq4Bv14PwnhIKKCxqE52dAt4w5+DrMQUF91YnEPrDlwayKNSUOMddByML37Xr/zIXybx816
2s6+9J9wmbomBs+78ql48hOT4TyHttcnOBAveYqN8XcOFgkysq8JPHw3qjJ99dBmNoIJxSaix/zi
4XvQvDg57i7xwzLfwN5tO9QbqoVRgxS5QShz1V7pAt31heWvKikwu3oyi0Ron2MabtyS8UEFjAF1
vRgcwzAJLtI56kpqIF0n9qZFdG0uyj89YZlYe+sY1E/VqW2vpGrv8W57l6Jltipz04XssSao/Gi2
znuypFbHZBjEkM4jBr3oOxqa79hNO3Hm6FlWNHeVa6kQ5ILPVzB103WNoJgE3QBUK+hof21pGlnt
BKgqmi8fLJBUcUCUw6VpxpYPDtp1lzJOa5XvV48zb4FTjom8buLuIKap1qL44s0i3pCzIKcPISj1
iwWo6UTkZaR0KEIdRczDcna4I9RHpcaZzuvHhV2joy669U5Gq4S7b0wxnMZaEVZ8plhj4wn3qcKC
aT2slWmsmjGthUljU/gTTTtGdU4BcZWM6MC5yAKwdeRX4AWdGPqZaoY9ZSBFYhxAjYkMB3scJm6Y
XvaVblmxgqm7ntK+x7oOOxqKcT+0qImB2IakF70gThVQ1BCEsJrhb5EhVqpj1JV1zBNbNF4zGeFr
5amWAqtxiiqj4ukd9nErLOQ5rwpc5rV1cntylYqngyBHXRQgMPTitwXxAxQfwL6lQcjFnh8tPnEi
avX6qqqEuN6qOypNnQRET6avAi+TqgeOe3DEvtRuLq0qz1gzVCK1YMUr/PdW43ILlHVY5IhYYHVv
NeESH1MJkeSxghTpk3ktY2pYqxMcG/cQ0+te6tavqwd7vYOMpSFqkj3OfRBaUCykOiE98To3TjWn
G0anWRCJ9LJkOIbTgsevtuzuBkU515NZ4poAtDWXPUzEvWqCbMWyShQ2TdMmsaOJA7BCDq+i7cN0
Lyax04z5wTN49uAftYi29JOHWtOpb6OI5IHwHbRLDFwG9B17FYUqeiBv+ObGUULllmF5SM7af0jU
4sD37550pTB78lIxqldqYbzcGJlnuUebMViSec05kFdixPEdFJ5Dk1T6oB36Y5vansQuU6wemupM
sJH0xd1D++gWolAtNArlAqWQcrA6TIYtfW1ViRs4jphLqqTXemX68n8WN5fLyz/NIADGgpbI+aml
pkofD1WYyANzaYnX08CK5KjyU5pxvd6X3CxkSe5tXh183CNMdkq6B/OXg1IwpMZbWkARg3qEuql1
TfpVm/RNGzSfA7oaaY3x/2mLCPtHSUO8Nta3QaIqzV1SmufBOmaewmH0hvuDn4P96o3TFK2UAkzA
doW4ZUNfPgb9eqhfBx9l2atMxpErB0fSouhniTTF0aTosIMbrEQAe388fyuAxsQwBjOAqsvrhozG
jhfkwJCYuQXzuPuVKwajFc8f8YW7TvA3+78Km7ShKvuGxiZZCvawnrehYaKi1eRQGYIMVTG8lB/p
jTs9FgqPS8FUO7FVgQMzmVxtbVdNyeG+VbJo9FYcOWUc8TtUXjW0YUyn1+wwRI7p4pk05PRxBdoV
piBUH088EdciITCWX2GSlZ5m4BgzHcAbVrDewm13EXWVaPBA6Dpz31FqkS09YROvt2L134B/77rp
GlxxVMeL6t2pnrBXuPJzcOUn7+6rKLhq/r3RW+qMtMbDqK8Sqk1bR+EaKC/ngyFQmlU+Du9Dc5Sd
Kj+jbojwUWZKquIqDnfES0ieoCZDuwMWIGZPLAMvEdf8UIrrPhu6xV40G5dfZNgAIFmTqGBRWuyk
EsYbnqtHtsO13f81rnd8IsKTfG6MRn13bZSra8ga+DF7Nf7x0TpneyRZA1QyF+oS9BSGuDvqat7l
8ZPZdourM6Zh6eQy673qhufaQJDXKNkys8LclZ+h+H40DHR8/tbwLknwpNCjrv/Imo/uByhedUsO
+7e/r+vlKN/7SJbLow5nrcNf4yYN1az5+Cl8ibYyRt2v7u6wtdBueePtIryZQPSLst3qTPscf49J
5uYFPwSoLIy+rFoa51W8Vlum1u7KctxATW+ocqdU9LUCccuYN3pKYWctMJ2ph5L5w8qw/2yIIhaK
k9ObruivZmoLH6qjtkuWwwRIrdUKzI/cjmnn4+2+DOYRBvdh1sH1rVWvV/WP/o/y54Ud0nMtA7Wp
HpnpSa18y0y9yTEk+jV48ds2OVyL5XI1WHNpjFc3vURcLKQSEp63icvvL9vfxJqBlNBraZZxrz6D
Nfs3paeTztJGNgoQLwe2rml/u4gOXCsyrZdwKtqX/9/Wte0q1WA4NTB6NMI14pcclZnhH/3hdxT6
B3+mx3IXnE3c/blRJAkP5WlZJBFGnIB316Rtq5qbuM93ezNZXt1N7z4vr4j8eIv4ZSw+ypMeq5gV
jTWWsPi8jX9tTCYS4QTLXUqR2d0k3HGTn0bpdcUGTH7KUYP1ftpRQ29YmM7r2/Oi6AZrLfHaH3qa
cjR4JTOyP4ayKocPL9GY8K/EJhN+EREeqSsYnBcJHJcCQhVjLAgyWMzrGOKrEPHJoWSeeqoQt7PO
8Hppz4TQk5Spwtlb7Xa3hMqRpkUYEv3GR93dvWX/tGE8w8TB9Cvj/DtBzL3JTM0wyW3WVLf9aw7g
ffoqS8Qv2G+S0Kv8f3VS4xsmaT4iqOfpu/IuPr+SiGUpWTKY/4gqslQrrc9ztC+DEK5KNo85JjWd
NOpci5UCphXI53xY34dJvIM/tr5O0tLmNio1MVfnE5fWDU3tdmZ12XI0CvDbKTTvjsMvojkO6hDH
kZVYoVBG/wKWlkbpYj0AAA==

#!/usr/bin/env python3

# Copyright (c) 2024, 2025, Oracle and/or its affiliates.
# Version 1.3

from json import loads as json_loads
from pathlib import Path
from subprocess import run as subprocess_run, PIPE, CompletedProcess
from typing import List, Dict, AnyStr, Optional, Set
from argparse import ArgumentParser
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from http.client import HTTPResponse
from re import compile as re_compile
from collections import namedtuple
from ipaddress import ip_network, ip_address

# The keys for the JSON output of `ip -j a s`
IFNAME = “ifname”
MASTER = “master”
ADDRESS = “address”
ADDR_INFO = “addr_info”
LOCAL = “local”
VFINFO_LIST = “vfinfo_list”
LINKINFO = “linkinfo”
INFO_KIND = “info_kind”
BOND_PREFIX = “vfio”
# IPv6 subnets in OCI are fixed to a /64
OCI_IPV6_SUBNET_PREFIX_LEN = 64

# vnic metadata keys
M_MAC_ADDR = “macAddr”
M_PRIV_IP = “privateIp”
M_VR_IP = “virtualRouterIp”
M_SUB_CIDR = “subnetCidrBlock”
M_V6_ADDRESSES = “ipv6Addresses”
M_VNIC_ID = “vnicId”

UTF_8 = “UTF-8”
SECONDARY_IP = namedtuple(“SecondaryIp”, (“vnic_id”, “ip_address”))
BALANCE_RR = “balance-rr”
BALANCE_XOR = “balance-xor”

class VfioCfg:
    ip_cmd = “/usr/sbin/ip”
    ip_addr_show_cmd = f”{ip_cmd} -j -d address show”
    systemctl_cmd = “/usr/bin/systemctl”
    # See https://www.kernel.org/doc/Documentation/networking/bonding.txt
    bonding_modes = {BALANCE_RR: {}, BALANCE_XOR: {“xmit_hash_policy”: “layer2+3”}}
    default_bonding_mode = BALANCE_XOR
    metadata_vnics_url = “http://169.254.169.254/opc/v2/vnics”
    metadata_headers = {“Authorization”: “Bearer Oracle”}
    bond_mtu = 9000
    max_bond_count = 30
    # The miimon value set for bond0 on compute nodes is 250ms.
    bond_miimon = 250
    # See https://github.com/oracle/oci-java-sdk/blob/master/bmc-common/src/main/java/com/oracle/bmc/OCID.java#L21
    ocid_regx = re_compile(r”^([0-9a-zA-Z-_]+[.:])([0-9a-zA-Z-_]*[.:]){3,}([0-9a-zA-Z-_]+)$”)
    nm_conf_dir = Path(“/etc/NetworkManager/conf.d”)

    def __init__(self, bonding_mode=default_bonding_mode, debug=False, secondary_ip=SECONDARY_IP(None, None)):
        self.bonding_mode = bonding_mode
        self.debug = debug
        self.secondary_ip = secondary_ip
        if debug:
            print(“**** running configure_vfio ****”)
        self._validate_instantiation_fields()
        self.nm_conf_file = self.nm_conf_dir / “99-configure-vfio.conf”

    @staticmethod
    def is_ip_address(address: str) -> bool:
        try:
            ip_address(address)
            return True
        except ValueError:
            pass
        return False

    def _validate_instantiation_fields(self):
        if self.bonding_mode not in self.bonding_modes:
            raise ValueError(f”bonding_mode must be one of: {self.bonding_modes}”)
        if any(self.secondary_ip):
            if not self.secondary_ip.vnic_id or not self.secondary_ip.ip_address:
                raise ValueError(“Both –vnic-id and –ip-address options must be provided together.”)
            if not self.ocid_regx.match(self.secondary_ip.vnic_id):
                raise ValueError(f”The VNIC OCID: ‘{self.secondary_ip.vnic_id}’ must be in the proper format.”)
            if not self.is_ip_address(self.secondary_ip.ip_address):
                raise ValueError(
                    f”The secondary IP address: ‘{self.secondary_ip.ip_address}’ must be in the proper format.”
                )

    @staticmethod
    def _print_response(r: HTTPResponse, body: str):
        url = r.geturl()
        print(f”**** {url} response satus: {r.status}, headers:”)
        for k, v in r.headers.items():
            print(f”{k}:”, v)
        if body:
            print(f”**** {url} response body:”)
            for line in body.splitlines():
                print(line)

    @staticmethod
    def _safe_loads(s: AnyStr) -> List[Dict]:
        return json_loads(s) if s else []

    def _run_command(self, cmd: AnyStr, stdout=None, stderr=None) -> CompletedProcess:
        if self.debug:
            print(cmd)
        return subprocess_run(cmd.split(), stdout=stdout, stderr=stderr, check=True)

    def _get_ip_addr_show(self) -> List[Dict]:
        cp = self._run_command(self.ip_addr_show_cmd, stdout=PIPE)
        return self._safe_loads(cp.stdout)

    def _get_vnics_metadata(self) -> List[Dict]:
        result = []
        request = Request(self.metadata_vnics_url, headers=self.metadata_headers)
        try:
            with urlopen(request) as response:
                body = response.read().decode(UTF_8)
                if self.debug:
                    self._print_response(response, body)
                result.extend(self._safe_loads(body))
        except HTTPError as e:
            print(
                f”ERROR getting metadata URL: {request.full_url}, reason: {e}, “
                f”headers: {‘, ‘.join(f'{k}: {v}’ for k, v in e.headers.items())}”
            )
        except URLError as e:
            print(f”ERROR getting metadata URL: {request.full_url}, reason: {e}”)
        return result

    def _get_vnics(self) -> List[Dict]:
        vnics = self._get_vnics_metadata()
        return sorted(
            (v for v in vnics if all(v.get(x) for x in (M_VNIC_ID, M_MAC_ADDR, M_PRIV_IP, M_VR_IP, M_SUB_CIDR))),
            key=lambda x: ip_address(x[M_PRIV_IP]),
        )

    @staticmethod
    def _get_interfaces_matching_mac(interfaces: List[Dict], mac: str) -> List[Dict]:
        return [i for i in interfaces if i.get(ADDRESS).lower() == mac.lower()]

    @staticmethod
    def _get_vf_interfaces(interfaces: List[Dict]) -> List[Dict]:
        return [i for i in interfaces if VFINFO_LIST in i]

    @staticmethod
    def _is_bond(interface: dict) -> bool:
        return interface.get(LINKINFO, {}).get(INFO_KIND) == “bond”

    def _get_bond_interfaces(self, interfaces: List[Dict]) -> List[Dict]:
        return [i for i in interfaces if self._is_bond(i) and i.get(IFNAME).startswith(BOND_PREFIX)]

    def _get_existing_bond_interface(self, all_interfaces: List[Dict], mac: str) -> Optional[dict]:
        bond_interfaces = self._get_bond_interfaces(self._get_interfaces_matching_mac(all_interfaces, mac))

        if len(bond_interfaces) == 0:
            return None

        if len(bond_interfaces) == 1:
            return bond_interfaces.pop()

        bond_names = [b[IFNAME] for b in bond_interfaces]
        raise SystemError(f”the following bonds have the same MAC address {mac}: {‘,’.join(bond_names)}”)

    def _bond_interface(self, interface_name: str, bond_name: str):
        commands = (
            f”{self.ip_cmd} link set {interface_name} down”,
            f”{self.ip_cmd} link set {interface_name} master {bond_name}”,
            f”{self.ip_cmd} link set {interface_name} up”,
            f”{self.ip_cmd} link set {bond_name} mtu {self.bond_mtu}”,
            f”{self.ip_cmd} link set {bond_name} up”,
            f”{self.ip_cmd} addr flush dev {interface_name}”,
        )
        for command in commands:
            self._run_command(command)

    def _add_interfaces_to_bond(self, interfaces: List[Dict], bond_interface: dict) -> bool:
        is_interface_bonded = False
        for interface in interfaces:
            if not interface.get(MASTER):
                self._bond_interface(interface[IFNAME], bond_interface[IFNAME])
                is_interface_bonded = True
        return is_interface_bonded

    def _get_bonding_args(self, bonding_mode):
        return ” “.join(f”{k} {v}” for k, v in self.bonding_modes.get(bonding_mode, {}).items())

    def _create_bond(self, bond_name: str, mac: str):
        bonding_args = self._get_bonding_args(self.bonding_mode)
        self._run_command(
            f”{self.ip_cmd} link add dev {bond_name} address {mac} “
            f”type bond miimon {self.bond_miimon} mode {self.bonding_mode} {bonding_args}”
        )
        # Disable SLAAC on the bonded interface immediately after creating it.
        accept_ra = Path(f”/proc/sys/net/ipv6/conf/{bond_name}/accept_ra”)
        if accept_ra.is_file() and accept_ra.read_text(UTF_8).strip() != “0”:
            accept_ra.write_text(“0”, UTF_8)

    def _get_bond_names(self, interfaces: List[Dict]) -> Set[str]:
        return {b[IFNAME] for b in self._get_bond_interfaces(interfaces)}

    def _get_next_bond_name(self, all_interfaces: List[Dict]) -> str:
        all_bond_names = self._get_bond_names(all_interfaces)
        for n in range(self.max_bond_count):
            bond_name = f”{BOND_PREFIX}{n}”
            if bond_name not in all_bond_names:
                return bond_name
        raise SystemError(f”couldn’t find an available bond name out of max_bond_count: {self.max_bond_count}”)

    def _configure_ip(self, cidr: str, device: dict) -> bool:
        existing_addresses = {ip_address(i[LOCAL]) for i in device.get(ADDR_INFO, []) if i.get(LOCAL)}

        if ip_address(cidr.split(“/”)[0]) in existing_addresses:
            return False
        self._run_command(f”{self.ip_cmd} addr replace {cidr} dev {device[IFNAME]}”)
        return True

    def configure_primary_ip(self, vnic: dict, bond_interface: dict) -> bool:
        changes = []
        subnet = ip_network(vnic[M_SUB_CIDR])
        changes.append(self._configure_ip(f”{vnic[M_PRIV_IP]}/{subnet.prefixlen}”, bond_interface))
        for ipv6_address in vnic.get(M_V6_ADDRESSES, []):
            changes.append(self._configure_ip(f”{ipv6_address}/{OCI_IPV6_SUBNET_PREFIX_LEN}”, bond_interface))
        return any(changes)

    def _configure_secondary_ip(self, vnic: dict, bond_interface: dict) -> bool:
        if not all(self.secondary_ip):
            return False
        if vnic[M_VNIC_ID] != self.secondary_ip.vnic_id:
            return False
        subnets = {ip_network(vnic[M_SUB_CIDR])}
        v6_addresses = vnic.get(M_V6_ADDRESSES, [])
        subnets.update({ip_network(f”{a}/{OCI_IPV6_SUBNET_PREFIX_LEN}”, strict=False) for a in v6_addresses})
        if not any(ip_address(self.secondary_ip.ip_address) in s for s in subnets):
            print(
                f”WARNING: Not setting secondary IP: {self.secondary_ip.ip_address} because it isn’t part of “
                f”the VNIC’s subnet CIDRs: {subnets}”
            )
            return False
        addresses = {ip_address(vnic[M_PRIV_IP])}
        addresses.update({ip_address(a) for a in v6_addresses})
        if any(ip_address(self.secondary_ip.ip_address) == a for a in addresses):
            print(
                f”WARNING: Not setting secondary IP: {self.secondary_ip.ip_address} because it is equal to “
                f”one of the assigned addresses on {bond_interface[IFNAME]}.”
            )
            return False

        for subnet in (s for s in subnets if ip_address(self.secondary_ip.ip_address) in s):
            return self._configure_ip(f”{self.secondary_ip.ip_address}/{subnet.prefixlen}”, bond_interface)
        return False

    @staticmethod
    def _get_all_masters(interfaces: List[Dict]) -> Set[str]:
        return {i[MASTER] for i in interfaces if i.get(MASTER, “”).startswith(BOND_PREFIX)}

    def _clean_disconnected_bonds(self, vnics: List[Dict], all_interfaces: List[Dict]):
        all_bonds = self._get_bond_names(all_interfaces)
        all_masters = self._get_all_masters(all_interfaces)
        orphan_bonds = all_bonds – all_masters
        for orphan_bond in orphan_bonds:
            self._run_command(f”{self.ip_cmd} link del dev {orphan_bond}”)

        vnic_macs = {v[M_MAC_ADDR] for v in vnics}
        nm_unmanaged_macs = self._get_nm_unmanaged_macs()
        orphan_macs = nm_unmanaged_macs – vnic_macs
        for orphan_mac in orphan_macs:
            self._disable_nm_for_vfio_vnics(orphan_mac, is_delete=True)

    def _get_nm_unmanaged_macs(self) -> Set[str]:
        return (
            {
                m.replace(“mac:”, “”, 1)
                for m in self.nm_conf_file.read_text(UTF_8).split(“=”)[-1].strip().split(“;”)
                if m
            }
            if self.nm_conf_file.is_file()
            else set()
        )

    def _disable_nm_for_vfio_vnics(self, vnic_mac: str, is_delete=False):
        # Only do something on newer distributions having the NM config dir.
        empty_config = “[keyfile]\nunmanaged-devices=”

        if not self.nm_conf_dir.is_dir():
            return

        if not self.nm_conf_file.is_file():
            self.nm_conf_file.write_text(empty_config, UTF_8)

        mac_addresses = self._get_nm_unmanaged_macs()

        if vnic_mac in mac_addresses:
            if is_delete:
                mac_addresses.remove(vnic_mac)
            else:
                return
        else:
            if is_delete:
                return
            mac_addresses.add(vnic_mac)

        if mac_addresses:
            self.nm_conf_file.write_text(f”{empty_config}{‘;’.join(f’mac:{m}’ for m in mac_addresses)}\n”, UTF_8)
        elif self.nm_conf_file.is_file():
            self.nm_conf_file.unlink()

        self._run_command(f”{self.systemctl_cmd} reload NetworkManager”)

    def _bond_vnics(self, vnics: List[Dict], all_interfaces: List[Dict]):
        for vnic in vnics:
            vnic_mac = vnic[M_MAC_ADDR]
            matching_interfaces = self._get_interfaces_matching_mac(self._get_vf_interfaces(all_interfaces), vnic_mac)

            if matching_interfaces:
                self._disable_nm_for_vfio_vnics(vnic_mac)
                config_changes = []
                bond_interface = self._get_existing_bond_interface(all_interfaces, vnic_mac)
                if not bond_interface:
                    self._create_bond(self._get_next_bond_name(all_interfaces), vnic_mac)
                    all_interfaces = self._get_ip_addr_show()
                    bond_interface = self._get_existing_bond_interface(all_interfaces, vnic_mac)

                config_changes.append(self._add_interfaces_to_bond(matching_interfaces, bond_interface))

                config_changes.append(self.configure_primary_ip(vnic, bond_interface))

                # Configure secondary IPs because it can be done in a second invocation.
                config_changes.append(self._configure_secondary_ip(vnic, bond_interface))

                if any(config_changes):
                    all_interfaces = self._get_ip_addr_show()

        return all_interfaces

    def run(self):
        all_interfaces = self._get_ip_addr_show()
        vnics = self._get_vnics()
        all_interfaces = self._bond_vnics(vnics, all_interfaces)
        self._clean_disconnected_bonds(vnics, all_interfaces)

def main():
    parser = ArgumentParser(description=”tool to automatically configure bonding with SR-IOV/VFIO virtual functions”)
    parser.add_argument(
        “–bonding-mode”,
        choices=VfioCfg.bonding_modes,
        default=VfioCfg.default_bonding_mode,
        help=f”bonding mode to use, the default is: {VfioCfg.default_bonding_mode}”,
    )
    parser.add_argument(“-d”, “–debug”, action=”store_true”, help=”enable debug output”)
    secondary_ip_arg_grp = parser.add_argument_group(
        “Secondary IP arguments”, “All the following arguments must be provided when adding a secondary IP to a VNIC.”
    )
    secondary_ip_arg_grp.add_argument(“–vnic-id”, help=”OCID of the VNIC holding the secondary IP”)
    secondary_ip_arg_grp.add_argument(“–ip-address”, help=”secondary IP address value”)
    args = parser.parse_args()

    secondary_ip = SECONDARY_IP(args.vnic_id, args.ip_address)
    try:
        VfioCfg(bonding_mode=args.bonding_mode, debug=args.debug, secondary_ip=secondary_ip).run()
    except ValueError as e:
        print(f”ERROR: {e}”)

if __name__ == “__main__”:
    main()