Netbox Sync Device Prefixes Task

task api name: sync_device_prefixes

The task derives IPv4 and IPv6 networks from live device interface addresses and reconciles those prefixes with NetBox. It is independent from sync_device_ip: it does not require NetBox interfaces and does not create or update IP address records.

How It Works

1. Resolve the selected devices and validate that they exist in NetBox.
2. Run Nornir parse_ttp with get="interfaces" for the selected devices.
3. Apply interface, prefix, and ignore-range filters.
4. Convert each address to its canonical network, for example 10.0.0.1/31 to 10.0.0.0/31.
5. Deduplicate observations, compare them with NetBox, and create missing or update out-of-sync prefixes.

NetBox interface presence is never checked. A prefix can therefore be synchronized before its source interface exists in NetBox.

Output

The result is global because the same prefix can be reported by several devices:

{
    "created": ["10.0.0.0/31"],
    "updated": ["10.0.1.0/31"],
    "in_sync": ["2001:db8::1/128"]
}

Dry-run returns the same structure without writing to NetBox. With with_approval=True, the prepared result is shown for approval before any prefixes are written.

VRF Handling

ignore_vrf=True is the default:

• Prefixes are matched by canonical prefix alone.
• Existing VRF associations are preserved.
• New prefixes are created without a VRF.

With ignore_vrf=False:

• The live interface VRF is resolved by name; a missing VRF is created.
• Prefixes are matched by canonical prefix and VRF.
• If the same prefix exists in another VRF, that object is preserved and a new prefix is created in the live VRF.
• global and default interface VRFs are treated as the global table.

Site Handling

ignore_site=True is the default:

• Device sites are not written to prefixes.
• Existing prefix scope/site associations are preserved.

With ignore_site=False:

• New prefixes are associated with the reporting device's site.
• Existing matching prefixes are updated when their site differs.
• When several devices report the same prefix identity, the site from the alphabetically first device name is authoritative.

Prefix identity is the prefix alone when VRFs are ignored, otherwise it is the prefix and resolved VRF together.

Filtering

Interface filters are applied to live data before prefix derivation. Prefix filters are applied to the derived canonical prefixes:

• filter_by_name — interface-name glob, such as Loopback*.
• filter_by_description — interface-description glob.
• filter_by_prefix — include prefixes contained within one IPv4 or IPv6 network.
• ignore_ranges — exclude derived prefixes fully contained within any supplied IPv4 or IPv6 network. A narrower ignored network does not exclude a broader live prefix containing it.

The default ignored ranges are:

127.0.0.0/8
224.0.0.0/24
fe80::/10
ff02::/16
::ffff:0:0/96
::1/128

Filters combine using intersection.

For example, ignore_ranges="10.3.15.33/32" does not exclude the live prefix 10.3.15.32/30, while ignore_ranges="10.3.15.0/24" does.

Deletion Behavior

The task never deletes prefixes. A prefix absent from the selected live device data is left unchanged in NetBox.

Branching Support

Pass branch=<name> to create or update prefixes in a NetBox Branching Plugin branch. The task creates the branch when it does not already exist.

Examples

CLIPython

Synchronize prefixes from one device:

nf#netbox sync prefixes devices ceos-spine-1

Preview prefixes derived from loopbacks:

nf#netbox sync prefixes devices ceos-spine-1 filter-by-name "Loopback*" dry-run

Associate prefixes with live VRFs and device sites:

nf#netbox sync prefixes devices ceos-spine-1 ignore-vrf false ignore-site false

Select devices with a Nornir filter:

nf#netbox sync prefixes FC spine

result = client.run_job(
    "netbox",
    "sync_device_prefixes",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "dry_run": True,
    },
)

result = client.run_job(
    "netbox",
    "sync_device_prefixes",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "ignore_vrf": False,
        "ignore_site": False,
        "filter_by_prefix": "10.0.0.0/8",
    },
)

NORFAB Netbox Sync Device Prefixes Command Shell Reference

nf# man tree netbox.sync.prefixes
root
└── netbox
    └── sync
        └── prefixes
            ├── timeout
            ├── workers
            ├── verbose-result
            ├── progress
            ├── instance
            ├── dry-run
            ├── with-approval
            ├── devices
            ├── ignore-ranges
            ├── ignore-vrf
            ├── ignore-site
            ├── filter-by-name
            ├── filter-by-description
            ├── filter-by-prefix
            ├── branch
            └── Nornir host filters

Python API Reference

Synchronize prefixes derived from live interface addresses into NetBox.

Prefixes are collected from the Nornir TTP interfaces getter without requiring corresponding NetBox interfaces. Existing prefixes are matched by prefix alone when ignore_vrf=True and by prefix plus VRF otherwise. Site association is left unchanged by default. When ignore_site=False, the site of the alphabetically first device reporting a prefix is used.

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object containing relevant metadata.	required
instance	Union[None, str]	NetBox instance name.	None
dry_run	bool	Return the reconciliation plan without writing to NetBox.	False
with_approval	bool	Preview changes and ask for review before applying them.	False
timeout	int	Timeout in seconds for Nornir parsing and host resolution.	600
devices	Union[None, list]	Explicit device names to collect prefixes from.	None
branch	str	NetBox Branching plugin branch name.	None
ignore_ranges	Union[None, str, list]	Exclude derived prefixes fully contained in these ranges.	None
ignore_vrf	bool	Ignore live VRFs and preserve existing prefix VRFs.	True
ignore_site	bool	Ignore device sites and preserve existing prefix scopes.	True
filter_by_name	Union[None, str]	Interface-name glob filter.	None
filter_by_description	Union[None, str]	Interface-description glob filter.	None
filter_by_prefix	Union[None, str]	Include prefixes within this network only.	None
**kwargs	Any	Nornir host filter arguments.	{}

Returns:

Name	Type	Description
Result	Result	Global created, updated, and in_sync prefix lists.

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncDevicePrefixesInput,
    output=SyncDevicePrefixesResult,
    mcp={
        "annotations": {
            "title": "Sync Device Prefixes",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_device_prefixes(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    with_approval: bool = False,
    timeout: int = 600,
    devices: Union[None, list] = None,
    branch: str = None,
    ignore_ranges: Union[None, str, list] = None,
    ignore_vrf: bool = True,
    ignore_site: bool = True,
    filter_by_name: Union[None, str] = None,
    filter_by_description: Union[None, str] = None,
    filter_by_prefix: Union[None, str] = None,
    **kwargs: Any,
) -> Result:
    """Synchronize prefixes derived from live interface addresses into NetBox.

    Prefixes are collected from the Nornir TTP ``interfaces`` getter without
    requiring corresponding NetBox interfaces. Existing prefixes are matched
    by prefix alone when ``ignore_vrf=True`` and by prefix plus VRF otherwise.
    Site association is left unchanged by default. When ``ignore_site=False``,
    the site of the alphabetically first device reporting a prefix is used.

    Args:
        job: NorFab Job object containing relevant metadata.
        instance: NetBox instance name.
        dry_run: Return the reconciliation plan without writing to NetBox.
        with_approval: Preview changes and ask for review before applying them.
        timeout: Timeout in seconds for Nornir parsing and host resolution.
        devices: Explicit device names to collect prefixes from.
        branch: NetBox Branching plugin branch name.
        ignore_ranges: Exclude derived prefixes fully contained in these ranges.
        ignore_vrf: Ignore live VRFs and preserve existing prefix VRFs.
        ignore_site: Ignore device sites and preserve existing prefix scopes.
        filter_by_name: Interface-name glob filter.
        filter_by_description: Interface-description glob filter.
        filter_by_prefix: Include prefixes within this network only.
        **kwargs: Nornir host filter arguments.

    Returns:
        Result: Global ``created``, ``updated``, and ``in_sync`` prefix lists.
    """
    devices = list(devices or [])
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_prefixes",
        result={"created": [], "updated": [], "in_sync": []},
        resources=[instance],
        dry_run=dry_run,
    )
    nb = self._get_pynetbox(instance, branch=branch, job=job)

    if kwargs:
        job.event("resolving devices from Nornir filters")
        devices.extend(self.get_nornir_hosts(kwargs, timeout))
    devices = sorted(set(devices))
    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    nb_devices = {
        device.name: device.site.id
        for device in self.bulk_filter(
            nb.dcim.devices,
            name=devices,
            fields="name,site",
        )
    }
    for device_name in [name for name in devices if name not in nb_devices]:
        msg = f"{device_name} - device not found in Netbox"
        log.error(msg)
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
    devices = [name for name in devices if name in nb_devices]
    if not devices:
        ret.failed = True
        return ret

    normalised_live_all = collect_live_interface_ip_data(
        client=self.client,
        job=job,
        devices=devices,
        timeout=timeout,
        filter_by_name=filter_by_name,
        filter_by_description=filter_by_description,
    )

    job.event("collecting live prefix candidates")
    ignore_ranges = ignore_ranges or DEFAULT_IGNORE_RANGES
    if isinstance(ignore_ranges, str):
        ignore_ranges = [ignore_ranges]
    ignore_nets = [
        ipaddress.ip_network(str(prefix), strict=False) for prefix in ignore_ranges
    ]
    filter_prefix_net = (
        ipaddress.ip_network(filter_by_prefix, strict=False)
        if filter_by_prefix
        else None
    )
    desired_prefixes = {}
    for device_name in sorted(normalised_live_all):
        if device_name not in nb_devices:
            continue
        site_id = nb_devices[device_name]
        for interface_data in normalised_live_all[device_name].values():
            vrf_name = interface_data["vrf"]
            if ignore_vrf:
                vrf_name = None
            elif vrf_name and vrf_name.lower() in ["global", "default"]:
                vrf_name = None
            for address in (interface_data.get("ipv4_addresses") or []) + (
                interface_data.get("ipv6_addresses") or []
            ):
                prefix_net = ipaddress.ip_interface(str(address)).network
                if any(
                    prefix_net.version == ignore_net.version
                    and prefix_net.subnet_of(ignore_net)
                    for ignore_net in ignore_nets
                ):
                    continue
                if filter_prefix_net and (
                    prefix_net.version != filter_prefix_net.version
                    or not prefix_net.subnet_of(filter_prefix_net)
                ):
                    continue
                prefix = str(prefix_net)
                key = prefix if ignore_vrf else (prefix, vrf_name)
                desired_prefixes.setdefault(
                    key,
                    {"prefix": prefix, "vrf": vrf_name, "site": site_id},
                )

    if not desired_prefixes:
        job.event("no prefixes found in live data")
        return ret
    job.event(f"collected {len(desired_prefixes)} unique prefix candidate(s)")

    nb_prefixes = self.bulk_filter(
        nb.ipam.prefixes,
        prefix=sorted({item["prefix"] for item in desired_prefixes.values()}),
        fields="id,prefix,vrf,scope_type,scope_id",
    )
    create_prefixes = {}
    update_prefixes = {}
    for key, desired in desired_prefixes.items():
        matching = [
            prefix
            for prefix in nb_prefixes
            if prefix.prefix == desired["prefix"]
            and (
                ignore_vrf
                or (prefix.vrf.name if prefix.vrf else None) == desired["vrf"]
            )
        ]
        if not matching:
            payload = {"prefix": desired["prefix"]}
            if not ignore_site:
                payload.update(
                    {"scope_type": "dcim.site", "scope_id": desired["site"]}
                )
            create_prefixes[key] = payload
            continue

        current = min(matching, key=lambda prefix: prefix.id)
        current_site = (
            getattr(current, "scope_id", None)
            if getattr(current, "scope_type", None) == "dcim.site"
            else None
        )
        if not ignore_site and current_site != desired["site"]:
            update_prefixes[key] = {
                "id": current.id,
                "scope_type": "dcim.site",
                "scope_id": desired["site"],
            }
        else:
            ret.result["in_sync"].append(desired["prefix"])

    preview = copy.deepcopy(ret.result)
    preview["created"].extend(
        desired_prefixes[key]["prefix"] for key in create_prefixes
    )
    preview["updated"].extend(
        desired_prefixes[key]["prefix"] for key in update_prefixes
    )
    for action in preview:
        preview[action] = sorted(set(preview[action]))
    if with_approval:
        if not review_sync_task_result(job, "prefix sync", preview):
            ret.status = "skipped"
            ret.result = preview
            ret.dry_run = True
            ret.messages.append("review declined; changes were not applied")
            return ret
    elif dry_run:
        ret.result = preview
        ret.dry_run = True
        return ret

    vrf_ids = {}
    for key in list(create_prefixes):
        vrf_name = desired_prefixes[key]["vrf"]
        if vrf_name is None:
            continue
        if vrf_name not in vrf_ids:
            vrf_ids[vrf_name] = resolve_vrf(vrf_name, nb, job, ret, self.name)
        if vrf_ids[vrf_name] is None:
            create_prefixes.pop(key)
            continue
        create_prefixes[key]["vrf"] = vrf_ids[vrf_name]

    if update_prefixes:
        nb.ipam.prefixes.update(list(update_prefixes.values()))
        ret.result["updated"].extend(
            desired_prefixes[key]["prefix"] for key in update_prefixes
        )
        job.event(f"updated {len(update_prefixes)} prefix(es)")
    if create_prefixes:
        nb.ipam.prefixes.create(list(create_prefixes.values()))
        ret.result["created"].extend(
            desired_prefixes[key]["prefix"] for key in create_prefixes
        )
        job.event(f"created {len(create_prefixes)} prefix(es)")

    for action in ret.result:
        ret.result[action] = sorted(set(ret.result[action]))

    job.event("device prefix sync complete")
    return ret