Netbox Sync MAC Addresses Task

task api name: sync_mac_addresses

The Netbox Sync MAC Addresses Task synchronizes interface MAC addresses from live network devices into NetBox. The task collects current MAC address state from devices, compares it against existing NetBox MAC address objects, and applies create or update operations to bring NetBox into alignment.

How It Works

The task follows a three-step pipeline:

1. Collect live state — Run a Nornir parse_ttp job against the target devices to collect live MAC addresses per interface.
2. Fetch NetBox state — Retrieve existing MAC address objects from NetBox for the discovered MACs.
3. Reconcile — Compare live versus NetBox state and apply changes:
   • Create — MAC not present in NetBox at all: create a new MAC address object assigned to the correct interface.
   • Update — MAC exists in NetBox but has no assigned_object (unassigned): assign it to the correct interface.
   • In sync — MAC already assigned to the correct interface: no action needed.
   • Error — MAC exists in NetBox and is assigned to a different interface: report a conflict error without modifying the record.

Output

Both dry-run and live-run return the same structure, keyed by device name:

{
    "<device>": {
        "created": ["aa:bb:cc:dd:ee:01", "aa:bb:cc:dd:ee:02"],
        "updated": ["aa:bb:cc:dd:ee:03"],
        "in_sync": ["aa:bb:cc:dd:ee:04"]
    }
}

In dry-run mode (dry_run=True) the lists reflect what would happen — no changes are written to NetBox.

With Review — Pass with_review=True to use interactive NFCLI workflow. Sync task displays its preview, and waits for approval before applying changes. Declining at that point will return dry-run result.

Note

When both dry-run and with_review are True, dry-run logic ignored.

In live-run mode (dry_run=False, default) the lists reflect what was actually done.

Conflict errors (MAC assigned to a different interface) are reported in res["errors"] and do not appear in any action list.

Filtering

MAC address collection can be scoped using glob patterns so that only a subset of interfaces and MACs is considered:

• filter_by_name — match interface names, e.g. "Loopback*" or "Ethernet[1-4]"
• filter_by_description — match interface descriptions, e.g. "uplink*" or "p2p*"
• filter_by_mac — match MAC address strings, e.g. "aa:bb:cc:*"

All filters are applied before the reconciliation step. Interfaces or MACs that do not match are completely ignored — they are neither created nor updated.

Special Handling

Duplicate MAC Handling

NetBox permits multiple MAC address records with the same MAC value. The task handles this safely:

• If a MAC exists in NetBox with an assigned interface that matches the live data — reported as in_sync.
• If a MAC exists in NetBox with an assigned interface that differs from the live data — reported as a conflict error; the record is not modified.
• If a MAC exists in NetBox but is unassigned — the record is updated to point at the correct interface.
• If both an assigned (conflicting) and an unassigned copy of the same MAC exist in NetBox, the assigned (conflicting) entry takes precedence and a conflict error is raised.

Deletion Behavior

By default, all discovered live MAC addresses are managed by this task. MACs present in NetBox but absent from live devices are left untouched. There is no explicit deletion mechanism — instead, unused MAC records are managed through standard NetBox administrative procedures.

Branching Support

The task is branch-aware and can push changes into a NetBox branch. The Netbox Branching Plugin must be installed. Specify the branch parameter; the branch is created automatically if it does not already exist.

Examples

CLIPython

Sync MAC addresses for a list of devices:

nf#netbox sync mac-addresses devices ceos-spine-1 ceos-spine-2

Preview changes without writing to NetBox (dry run):

nf#netbox sync mac-addresses devices ceos-spine-1 dry-run

Restrict sync to Ethernet interfaces only:

nf#netbox sync mac-addresses devices ceos-spine-1 filter-by-name "Ethernet*"

Restrict sync to interfaces whose description matches a glob pattern:

nf#netbox sync mac-addresses devices ceos-spine-1 filter-by-description "uplink*"

Restrict sync to a specific MAC prefix:

nf#netbox sync mac-addresses devices ceos-spine-1 filter-by-mac "aa:bb:cc:*"

Sync MAC addresses into a NetBox branch:

nf#netbox sync mac-addresses devices ceos-spine-1 ceos-spine-2 branch sprint-42-macs

Sync using Nornir host filters instead of explicit device names:

nf#netbox sync mac-addresses FC spine

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
nf.start()
client = nf.make_client()

# sync MAC addresses for specific devices
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
    },
)

# dry run — preview creates/updates without writing
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "dry_run": True,
    },
)

# restrict sync to Ethernet interfaces only
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_name": "Ethernet*",
    },
)

# restrict sync to interfaces with a specific description pattern
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_description": "uplink*",
    },
)

# restrict sync to a specific MAC prefix
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_mac": "aa:bb:cc:*",
    },
)

# sync into a NetBox branch
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "branch": "sprint-42-macs",
    },
)

# use Nornir host filters instead of explicit device names
result = client.run_job(
    "netbox",
    "sync_mac_addresses",
    workers="any",
    kwargs={
        "FC": "spine",
    },
)

nf.destroy()

NORFAB Netbox Sync MAC Addresses Command Shell Reference

NorFab shell supports these command options for the sync_mac_addresses task:

nf# man tree netbox.sync.mac-addresses
root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── mac-addresses:    Sync device MAC addresses with NetBox
            ├── timeout:    Job timeout in seconds
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Return reconciliation plan without pushing changes to NetBox
            ├── devices:    List of NetBox device names to sync
            ├── filter-by-name:    Glob pattern to restrict sync by interface name, e.g. 'Ethernet*'
            ├── filter-by-description:    Glob pattern to restrict sync by interface description
            ├── filter-by-mac:    Glob pattern to restrict sync by MAC address, e.g. 'aa:bb:*'
            ├── branch:    Branching plugin branch name to push changes into
            ├── FO:    Filter Nornir hosts using Filter Object
            ├── FB:    Filter Nornir hosts by name using Glob Patterns
            ├── FH:    Filter Nornir hosts by hostname
            ├── FC:    Filter Nornir hosts by name containment
            ├── FR:    Filter Nornir hosts by name using Regular Expressions
            ├── FG:    Filter Nornir hosts by group
            ├── FP:    Filter Nornir hosts by hostname using IP Prefix
            ├── FL:    Filter Nornir hosts by names list
            ├── FM:    Filter Nornir hosts by platform
            └── FN:    Negate the Nornir host filter match
nf#

Python API Reference

Synchronize MAC addresses from live devices into NetBox.

The task follows a three-step pipeline:

1. Collect live state: Run a Nornir parse_ttp get interfaces job against devices to collect live MAC addresses per interface.
2. Fetch NetBox state: Retrieve existing MAC address objects from NetBox.
3. Reconcile: Create new MAC address objects or update existing unassigned ones to point at the correct interface.

Dry-run mode (dry_run=True): returns the reconciliation plan without making any changes. Result is keyed by device name::

{
    "<device>": {
        "created": ["aa:bb:cc:dd:ee:01", ...],
        "updated": ["aa:bb:cc:dd:ee:02", ...],
        "in_sync": ["aa:bb:cc:dd:ee:03", ...]
    }
}

Live-run mode (dry_run=False, default): applies changes and returns the same structure showing what was done.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
instance	str	The NetBox instance name to use.	None
dry_run	bool	If True, no changes will be made to NetBox.	False
with_review	bool	Preview changes, ask for review, then apply them.	False
timeout	int	Timeout in seconds for the Nornir parse_ttp job.	600
devices	list	List of device names to sync.	None
branch	str	NetBox branch name to use.	None
filter_by_name	str	Glob pattern to restrict which interfaces are included by name, e.g. 'Loopback*' or 'Eth*'.	None
filter_by_description	str	Glob pattern to restrict which interfaces are included by description, e.g. 'uplink*'.	None
filter_by_mac	str	Glob pattern to restrict which MAC addresses are included, e.g. 'aa:bb:*'.	None
**kwargs	Any	Additional Nornir host filter keyword arguments passed to parse_ttp (e.g. FL, FC, FB).	{}

Returns:

Name	Type	Description
Result	Result	Per-device action summary with created, updated, and in_sync MAC address lists.

Source code in norfab\workers\netbox_worker\interfaces_tasks.py

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncMacAddressesInput,
    output=SyncMacAddressesResult,
    mcp={
        "annotations": {
            "title": "Sync MAC Addresses",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_mac_addresses(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    with_review: bool = False,
    timeout: int = 600,
    devices: Union[None, list] = None,
    branch: str = None,
    filter_by_name: Union[None, str] = None,
    filter_by_description: Union[None, str] = None,
    filter_by_mac: Union[None, str] = None,
    **kwargs: Any,
) -> Result:
    """
    Synchronize MAC addresses from live devices into NetBox.

    The task follows a three-step pipeline:

    1. **Collect live state**: Run a Nornir ``parse_ttp`` get interfaces job against
       devices to collect live MAC addresses per interface.
    2. **Fetch NetBox state**: Retrieve existing MAC address objects from NetBox.
    3. **Reconcile**: Create new MAC address objects or update existing unassigned
       ones to point at the correct interface.

    **Dry-run mode** (``dry_run=True``): returns the reconciliation plan without
    making any changes. Result is keyed by device name::

    ```
    {
        "<device>": {
            "created": ["aa:bb:cc:dd:ee:01", ...],
            "updated": ["aa:bb:cc:dd:ee:02", ...],
            "in_sync": ["aa:bb:cc:dd:ee:03", ...]
        }
    }
    ```

    **Live-run mode** (``dry_run=False``, default): applies changes and returns
    the same structure showing what was done.

    Args:
        job: NorFab Job object containing relevant metadata.
        instance (str, optional): The NetBox instance name to use.
        dry_run (bool, optional): If True, no changes will be made to NetBox.
        with_review (bool, optional): Preview changes, ask for review, then apply them.
        timeout (int, optional): Timeout in seconds for the Nornir parse_ttp job.
        devices (list, optional): List of device names to sync.
        branch (str, optional): NetBox branch name to use.
        filter_by_name (str, optional): Glob pattern to restrict which interfaces
            are included by name, e.g. ``'Loopback*'`` or ``'Eth*'``.
        filter_by_description (str, optional): Glob pattern to restrict which
            interfaces are included by description, e.g. ``'uplink*'``.
        filter_by_mac (str, optional): Glob pattern to restrict which MAC addresses
            are included, e.g. ``'aa:bb:*'``.
        **kwargs: Additional Nornir host filter keyword arguments passed to
            ``parse_ttp`` (e.g. ``FL``, ``FC``, ``FB``).

    Returns:
        Result: Per-device action summary with ``created``, ``updated``, and
            ``in_sync`` MAC address lists.
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_mac_addresses",
        result={},
        resources=[instance],
        dry_run=dry_run,
    )
    nb = self._get_pynetbox(instance, branch=branch)
    log.info(
        f"{self.name} - Sync MAC addresses: Processing {len(devices)} device(s) in '{instance}'"
    )

    # source additional hosts from Nornir filters
    if kwargs:
        job.event("resolving devices from Nornir filters")
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)
        job.event(
            f"resolved {len(nornir_hosts)} device(s) from Nornir filters, "
            f"{len(devices)} total device(s) selected"
        )

    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    job.event(
        f"syncing MAC addresses for {len(devices)} devices, dry_run={dry_run}"
    )

    # filter out devices not defined in NetBox
    job.event(f"validating {len(devices)} device(s) exist in NetBox")
    nb_devices_data = {
        d.name: {"id": d.id, "name": d.name}
        for d in self.bulk_filter(
            nb.dcim.devices, "name", devices, fields="id,name"
        )
    }
    for d in list(devices):
        if d not in nb_devices_data:
            msg = f"{d} - device not found in Netbox"
            log.error(msg)
            job.event(msg, severity="ERROR")
            ret.errors.append(msg)
            devices.remove(d)
    if not devices:
        job.event(
            "no valid NetBox devices remain after validation", severity="ERROR"
        )
        ret.failed = True
        return ret
    job.event(f"validated {len(devices)} device(s) in NetBox")

    # gather live interface data from Nornir parse_ttp
    job.event(f"retrieving live interfaces for {len(devices)} devices")
    parse_data = self.client.run_job(
        "nornir",
        "parse_ttp",
        kwargs={"get": "interfaces", "FL": devices},
        workers="all",
        timeout=timeout,
    )

    # collect all discovered MAC addresses applying interface and MAC filters
    job.event("collecting live MAC address candidates")
    all_mac_live: dict = {}  # {mac: {"device": ..., "interface": ...}}
    for wname, wdata in parse_data.items():
        if wdata.get("failed"):
            msg = f"{wname} - failed to parse interface data from devices"
            log.warning(msg)
            job.event(msg, severity="WARNING")
            continue
        for device_name, host_interfaces in wdata["result"].items():
            for data in host_interfaces:
                intf_name = data["name"]
                intf_description = data["description"]
                mac = data["mac_address"]
                if not mac:
                    continue
                if filter_by_name and not fnmatch.fnmatch(
                    intf_name, filter_by_name
                ):
                    continue
                if (
                    filter_by_description
                    and intf_description
                    and not fnmatch.fnmatch(intf_description, filter_by_description)
                ):
                    continue
                if filter_by_mac and not fnmatch.fnmatch(mac, filter_by_mac):
                    continue
                all_mac_live[mac] = {
                    "device": device_name,
                    "interface": intf_name,
                }

    if not all_mac_live:
        log.info(
            f"{self.name} - Sync MAC addresses: no MAC addresses found in live data"
        )
        job.event("no MAC addresses found in live data")
        return ret
    job.event(f"collected {len(all_mac_live)} live MAC address candidate(s)")

    # fetch interfaces data from NetBox to resolve interface IDs
    job.event("fetching NetBox interfaces to resolve MAC assignments")
    nb_interfaces_result = self.get_interfaces(
        job=job,
        instance=instance,
        branch=branch,
        devices=devices,
        ip_addresses=False,
        cache="refresh",
    )
    if nb_interfaces_result.errors:
        job.event(
            "failed to fetch NetBox interfaces for MAC sync", severity="ERROR"
        )
        ret.errors.extend(nb_interfaces_result.errors)
        ret.failed = True
        return ret

    # fetch existing MAC address objects from NetBox
    # NetBox allows duplicate MAC entries; prefer assigned entries over
    # unassigned ones so that a conflicting assignment is not silently
    # overwritten by a later unassigned copy during dict construction.
    job.event(
        f"fetching {len(all_mac_live)} matching MAC address object(s) from NetBox"
    )
    nb_macs: dict = {}
    for _m in self.bulk_filter(
        nb.dcim.mac_addresses,
        "mac_address",
        list(all_mac_live.keys()),
        fields="id,mac_address,assigned_object",
    ):
        _mac = _m.mac_address.lower()
        _entry = {
            "id": _m.id,
            "device": (
                _m.assigned_object.device.name if _m.assigned_object else None
            ),
            "interface": _m.assigned_object.name if _m.assigned_object else None,
        }
        # keep the entry if we haven't seen this MAC yet, or if the new
        # entry is assigned (has an interface) and the stored one is not
        if _mac not in nb_macs or (
            _entry["interface"] is not None and nb_macs[_mac]["interface"] is None
        ):
            nb_macs[_mac] = _entry
    job.event(
        f"retrieved {len(nb_macs)} matching MAC address object(s) from NetBox"
    )

    # per-device result tracking
    device_results = {
        device_name: {
            "created": [],
            "updated": [],
            "in_sync": [],
        }
        for device_name in devices
    }
    ret.result = device_results

    bulk_update_mac = []
    bulk_create_mac = []

    # process and compare live MACs versus NetBox MACs
    job.event("calculating MAC address sync actions")
    for mac, mac_data in all_mac_live.items():
        device_name = mac_data["device"]
        intf_name = mac_data["interface"]
        nb_raw = nb_interfaces_result.result.get(device_name, {})
        if intf_name not in nb_raw:
            msg = f"{device_name}:{intf_name} - interface not found in NetBox, skipping MAC {mac}"
            log.warning(msg)
            continue
        # MAC already assigned to a different interface
        if (
            nb_macs.get(mac, {}).get("interface")
            and nb_macs[mac]["interface"] != intf_name
        ):
            exist_intf = nb_macs[mac]["interface"]
            exist_device = nb_macs[mac]["device"]
            msg = (
                f"{device_name}:{intf_name} - {mac} already assigned to "
                f"a different interface {exist_device}:{exist_intf}"
            )
            log.error(msg)
            ret.errors.append(msg)
            job.event(msg, severity="ERROR")
            continue
        # MAC already assigned to correct interface
        elif mac in nb_macs and nb_macs[mac]["interface"] == intf_name:
            device_results[device_name]["in_sync"].append(mac)
        # update existing MAC if it is not associated with any interface
        elif mac in nb_macs and nb_macs[mac]["interface"] is None:
            if dry_run is True:
                device_results[device_name]["updated"].append(mac)
            else:
                bulk_update_mac.append(
                    {
                        "id": nb_macs[mac]["id"],
                        "mac_address": mac,
                        "assigned_object_type": "dcim.interface",
                        "assigned_object_id": nb_raw[intf_name]["id"],
                    }
                )
        # create new MAC address entry
        else:
            if dry_run is True:
                device_results[device_name]["created"].append(mac)
            else:
                bulk_create_mac.append(
                    {
                        "mac_address": mac,
                        "assigned_object_type": "dcim.interface",
                        "assigned_object_id": nb_raw[intf_name]["id"],
                    }
                )

    job.event(
        f"MAC address sync actions: {len(bulk_create_mac)} create, "
        f"{len(bulk_update_mac)} update"
    )

    if with_review:
        mac_preview_result = {
            device_name: {
                "created": list(device_result["created"]),
                "updated": list(device_result["updated"]),
                "in_sync": list(device_result["in_sync"]),
            }
            for device_name, device_result in device_results.items()
        }
        for m in bulk_create_mac:
            device_name = all_mac_live[m["mac_address"]]["device"]
            mac_preview_result[device_name]["created"].append(m["mac_address"])
        for m in bulk_update_mac:
            device_name = all_mac_live[m["mac_address"]]["device"]
            mac_preview_result[device_name]["updated"].append(m["mac_address"])
        # request confirmation from user
        if not review_sync_task_result(job, "MAC address sync", mac_preview_result):
            ret.status = "skipped"
            ret.result = mac_preview_result
            ret.dry_run = True
            ret.messages.append("review declined; changes were not applied")
            return ret
    elif dry_run is True:
        job.event(
            "dry-run requested, returning MAC address sync plan without changes"
        )
        ret.dry_run = True
        return ret

    if bulk_create_mac:
        job.event(f"creating {len(bulk_create_mac)} MAC address(es)")
        try:
            nb.dcim.mac_addresses.create(bulk_create_mac)
            job.event(f"created {len(bulk_create_mac)} MAC addresses")
            for m in bulk_create_mac:
                device_name = all_mac_live[m["mac_address"]]["device"]
                device_results[device_name]["created"].append(m["mac_address"])
        except Exception as e:
            msg = f"failed to bulk create MAC addresses: {e}"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")
    else:
        job.event("no MAC addresses to create")
    if bulk_update_mac:
        job.event(f"updating {len(bulk_update_mac)} MAC address(es)")
        try:
            nb.dcim.mac_addresses.update(bulk_update_mac)
            job.event(f"updated {len(bulk_update_mac)} MAC addresses")
            for m in bulk_update_mac:
                device_name = all_mac_live[m["mac_address"]]["device"]
                device_results[device_name]["updated"].append(m["mac_address"])
        except Exception as e:
            msg = f"failed to bulk update MAC addresses: {e}"
            ret.errors.append(msg)
            log.error(msg)
            job.event(msg, severity="ERROR")
    else:
        job.event("no MAC addresses to update")

    job.event("MAC address sync complete")
    return ret