Netbox Sync Device IP Task

task api name: sync_device_ip

The Netbox Sync Device IP Task synchronizes IP address assignments from live network devices into NetBox using a normalized desired/current state model and bulk-reconciliation. The task computes an explicit action plan — create, update, or mark in-sync — and applies it in a single pair of bulk API calls.

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 IPv4 and IPv6 addresses per interface.
2. Fetch NetBox state — Retrieve all existing IP address objects from NetBox that match any discovered address value.
3. Reconcile — Compare live addresses against NetBox records and classify each IP as:
   • create — no reusable candidate remains after interface and VRF selection
   • update — a reusable record is unassigned or needs a managed role or VRF change
   • in_sync — the address is on the target interface and no managed change is needed

NetBox candidates are searched by host address without using the prefix length. When several records have the same address, interface assignment takes priority over VRF and unassigned-record fallback matching.

Roles are synchronized from the normalized TTP interface data. The ip_address_role value is written to NetBox when present, including roles such as secondary, anycast, and vip. The task also applies these fallbacks:

• Interfaces whose name starts with loopback or lo → role loopback
• Addresses that fall within any configured anycast_ranges prefix → role anycast
• All other addresses → no role (standard unicast)

Configured anycast_ranges take precedence over the parsed role. A parsed role takes precedence over loopback-name inference.

Parsed live anycast addresses extend the task's anycast ranges with IPv4 /32 or IPv6 /128 host networks before reconciliation. These discovered ranges apply across VRFs for this task invocation and correct overlapping Netbox records to the anycast role without requiring configured ranges.

Addresses with a parsed vrrp, glbp, hsrp, or carp role are excluded from this task. The sync_vrrp task exclusively creates or reuses VRRP addresses and assigns them to the appropriate NetBox FHRP group. Matching IP records assigned to any non-interface NetBox object are also excluded so this task cannot reassign or duplicate records owned by another synchronization domain. Truly unassigned records remain eligible for reuse.

1. Client submits an on-demand request to the NorFab Netbox worker to sync device IP addresses
2. Netbox worker sends a job request to the Nornir service to fetch live interface data from devices
3. Nornir service collects IP address data from the network using parse_ttp
4. Nornir returns normalized interface and IP data to the Netbox worker
5. Netbox worker reconciles live addresses against NetBox and applies bulk create/update operations

Output

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

{
    "<device>": {
        "created": ["10.0.0.1/31", "192.168.1.1/32"],
        "updated": ["10.0.0.3/31"],
        "in_sync": ["10.0.0.5/31", "2001:db8::1/128"]
    }
}

The top-level diff field always contains the per-device reconciliation plan, including when every address is already synchronized:

{
    "<device>": {
        "create": [],
        "update": {},
        "delete": [],
        "in_sync": ["10.0.0.5/31", "2001:db8::1/128"]
    }
}

The task does not delete IP addresses, so delete is always empty.

Dry-run mode (dry_run=True) returns what would be created or updated without making any changes to NetBox.

With Approval — Pass with_approval=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_approval are True, dry-run logic ignored.

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

Filtering

IP addresses and interfaces can be scoped before reconciliation. All filters are applied at the live-data collection phase, so excluded addresses are completely ignored — they are neither created nor updated.

• filter_by_name — glob pattern matched against interface names, e.g. "Loopback*" or "Ethernet[1-4]". Interfaces not matching are skipped entirely.
• filter_by_description — glob pattern matched against interface descriptions, e.g. "uplink*". Interfaces not matching are skipped.
• filter_by_prefix — CIDR prefix such as "10.0.0.0/8". Only IP addresses whose host address falls within the prefix are included; supports both IPv4 and IPv6.
• filter_by_ip — glob pattern matched against the host portion of the IP address (without prefix length), e.g. "10.0.1.*". Only matching addresses are included.

Multiple filters combine as intersection — all specified conditions must be satisfied for an IP to be included.

Interface Mapping

Use interface_map to translate live interface names to their NetBox names. Rules are evaluated in order and the first matching device name, device type, and literal interface-name substring wins. Mapping occurs before filter_by_name, role inference, and NetBox interface lookup. Use the same map for sync_device_interfaces and sync_device_ip so IP addresses are assigned to the interfaces created by interface sync.

Rules can be supplied inline or through an nf:// YAML file:

- device_name: "fn-ceos-sp-*"
  device_type: "Arista *"
  match: "Loopback"
  replace: "NetBoxLoopback"

Existing IP Selection

For each live address, the task uses this order:

1. Reuse the matching IP already assigned to the target interface.
2. If no interface match exists, apply the ignore_vrf rule.
3. Reuse the first matching unassigned IP, if available.
4. Otherwise, create the IP or report a conflict according to its role.

Scenario	ignore_vrf=True (default)	ignore_vrf=False
Same IP is assigned to the target interface	Reuse it and leave its VRF unchanged. For example, an IP discovered in BLUE but stored in RED remains in RED.	Reuse it and update its VRF to the VRF of the IP discovered from the live device. The existing record ID and assignment are preserved.
Same IP is unassigned in NetBox	Reuse the first unassigned record across all VRFs, assign it to the interface, and leave its VRF unchanged.	Reuse the first unassigned record in the VRF of the IP discovered from the live device and assign it to the interface.
Only an unassigned IP in another VRF exists	Reuse it and preserve its current VRF.	Ignore it and create a record in the VRF of the IP discovered from the live device.
Same non-shared IP is assigned to another interface	Report a duplicate conflict.	Report a conflict when it is in the same VRF as the IP discovered from the live device. A record in another VRF does not match, so a new record is created in the discovered IP's VRF.
Same IP is assigned to a non-interface object	Exclude the live address from IP synchronization.	Exclude the live address when the existing record is in the discovered IP's VRF.
Same anycast or VIP IP is assigned to another interface	Create another record with the shared role for the target interface without a VRF.	Create another record with the shared role for the target interface in the VRF of the IP discovered from the live device.
No matching IP exists	Create and assign an IP without a VRF.	Create and assign an IP in the VRF of the IP discovered from the live device.

Anycast and VIP Support

Addresses within anycast_ranges receive the anycast role. An existing anycast record on the target interface is reused, so repeated syncs do not add another copy to that interface. If the same anycast address belongs to another interface, NetBox receives a separate record for the target interface. The same assignment behavior applies to addresses with the vip role parsed from live data or found on an existing NetBox IP. An unassigned VIP record can be reused once; subsequent interfaces receive separate records. Repeated syncs reuse the record already assigned to each target interface.

Note

If the same IP already has the anycast role in NetBox and is assigned to another interface, the IP discovered from the live device automatically receives the anycast role. This applies even when anycast_ranges is not provided.

Set anycast_ranges to a prefix string, list of prefixes, or an nf:// URL to a YAML file containing a list of prefixes:

anycast_ranges = "10.0.250.0/24"
anycast_ranges = ["10.0.250.0/24", "2001:db8:ffff::/48"]
anycast_ranges = "nf://netbox/anycast_ranges.yaml"

- 10.0.250.0/24
- 2001:db8:ffff::/48

Duplicate IP Guard

Before executing bulk writes the task checks that no IP address without the anycast or vip role appears more than once across the combined create and update payloads. If a duplicate is detected, all copies are removed from the payload and an error is recorded in the result, preventing NetBox from receiving conflicting assignments in a single request.

Deletion Behavior

By default, all discovered live IP addresses are managed by this task. Addresses present in NetBox but absent from live devices are left untouched. There is no explicit deletion mechanism — instead, unused addresses 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.

Bulk Request Batching

List-based NetBox writes are sent as sequential requests containing at most batch_size objects. The default is 1000; set any integer greater than zero to tune the request size. Each batch emits matching progress event and log messages. If a request fails, the task stops and returns the results recorded for earlier successful batches.

Examples

CLIPython

Sync IP addresses for a list of devices:

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

Preview changes without writing to NetBox (dry run):

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

Sync only loopback interface addresses:

nf#netbox sync ip-addresses devices ceos-spine-1 filter-by-name "Loopback*"

Map live interface names before filtering and assignment:

nf#netbox sync ip-addresses devices fn-ceos-sp-1 interface-map nf://netbox/interface_map.yaml filter-by-name "NetBoxLoopback*"

Sync only addresses within a specific prefix:

nf#netbox sync ip-addresses devices ceos-spine-1 filter-by-prefix "10.3.0.0/16"

Classify addresses in an anycast range and sync all devices:

nf#netbox sync ip-addresses devices ceos-spine-1 ceos-spine-2 ceos-leaf-1 anycast-ranges 10.0.250.0/24

Sync IPs with discovered interface VRFs:

nf#netbox sync ip-addresses devices ceos-spine-1 ignore-vrf false

Sync into a NetBox branch:

nf#netbox sync ip-addresses devices ceos-spine-1 ceos-spine-2 branch sprint-42-ips

Sync using Nornir host filters instead of explicit device names:

nf#netbox sync ip-addresses FC spine

from norfab.core.nfapi import NorFab

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

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

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

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

# restrict sync to addresses within a specific prefix
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "filter_by_prefix": "10.3.0.0/16",
    },
)

# restrict sync to addresses matching a glob pattern
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "filter_by_ip": "10.3.4.*",
    },
)

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

# classify addresses in anycast ranges and sync all fabric devices
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2", "ceos-leaf-1", "ceos-leaf-2"],
        "anycast_ranges": "10.0.250.0/24",
    },
)

# multiple anycast ranges (IPv4 and IPv6)
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "anycast_ranges": ["10.0.250.0/24", "2001:db8:ffff::/48"],
    },
)

# associate IPs with discovered interface VRFs
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1"],
        "ignore_vrf": False,
    },
)

# map live interface names to their NetBox names
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["fn-ceos-sp-1"],
        "interface_map": "nf://netbox/interface_map.yaml",
        "filter_by_name": "NetBoxLoopback*",
    },
)

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

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

# combine filters: loopback interfaces within a specific prefix, dry run
result = client.run_job(
    "netbox",
    "sync_device_ip",
    workers="any",
    kwargs={
        "devices": ["ceos-spine-1", "ceos-spine-2"],
        "filter_by_name": "Loopback*",
        "filter_by_prefix": "10.3.4.0/24",
        "dry_run": True,
    },
)

nf.destroy()

NORFAB Netbox Sync Device IP Command Shell Reference

NorFab shell supports these command options for the sync_device_ip task:

nf# man tree netbox.sync.ip-addresses
root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── ip-addresses:    Sync device IP 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
            ├── anycast-ranges:    IP prefix(es) used to classify addresses as anycast role
            ├── ignore-vrf:    Ignore discovered interface VRFs during IP sync
            ├── interface-map:    Ordered rules mapping live interface names to preferred NetBox names
            ├── filter-by-name:    Glob pattern to restrict sync by interface name, e.g. 'Loopback*'
            ├── filter-by-description:    Glob pattern to restrict sync by interface description
            ├── filter-by-prefix:    CIDR prefix to restrict sync to addresses within it, e.g. '10.0.0.0/8'
            ├── filter-by-ip:    Glob pattern to restrict sync by IP host address, e.g. '10.0.1.*'
            ├── 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 IP 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 IP addresses per interface.
2. Fetch NetBox state: Retrieve existing IP address objects from NetBox.
3. Reconcile: Create new IP address objects, update existing unassigned or stale ones, and mark already-correct entries as in-sync.

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

{
    "<device>": {
        "created": ["10.0.0.1/31", ...],
        "updated": ["10.0.0.3/31", ...],
        "in_sync": ["10.0.0.5/31", ...]
    }
}

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

Side Effects

1. Parsed live anycast addresses extend the task's anycast ranges with IPv4 /32 or IPv6 /128 host networks, regardless of VRF. Overlapping NetBox IP records have their role updated to anycast.
2. Parsed TTP IP roles are synchronized to NetBox, except for FHRP addresses using the vrrp, glbp, hsrp, or carp roles.
3. Addresses with the anycast or vip role may be assigned to multiple interfaces, with a separate NetBox IP record for each interface.

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_approval	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
anycast_ranges	list	IP prefix(es) used to classify IP addresses as anycast role, or an nf:// or git://<remote>/ YAML file containing a list of prefixes, e.g. '10.3.250.0/24'.	None
ignore_ranges	list	Prefixes to ignore IP addresses for, includes by default 127.0.0.0/8, 224.0.0.0/24 and others	None
ignore_vrf	bool	If True, discovered interface VRFs are ignored for IP writes. Existing VRF-scoped IPs are still matched by address and their VRF is left unchanged. Defaults to True.	True
interface_map	Union[None, str, list]	Ordered interface rename rules, or an nf:// YAML file containing them. Mapping is applied before interface filtering and NetBox interface lookup. The first matching rule wins.	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_prefix	str	IP prefix to restrict which IP addresses are included, e.g. '10.0.0.0/8'. Skips IPs not within this prefix.	None
filter_by_ip	str	Glob pattern to restrict which IP addresses are included, e.g. '10.0.*'.	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 IP address lists. Result.diff contains the corresponding create, update, delete, and in_sync reconciliation plan; delete is always empty because this task does not delete IP addresses.

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncDeviceIpInput,
    output=SyncDeviceIpResult,
    mcp={
        "annotations": {
            "title": "Sync Device IP Addresses",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_device_ip(
    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,
    anycast_ranges: Union[None, list] = None,
    ignore_ranges: Union[None, str, list] = None,
    ignore_vrf: bool = True,
    interface_map: Union[None, str, list] = None,
    filter_by_name: Union[None, str] = None,
    filter_by_description: Union[None, str] = None,
    filter_by_prefix: Union[None, str] = None,
    filter_by_ip: Union[None, str] = None,
    batch_size: int = 1000,
    **kwargs: Any,
) -> Result:
    """
    Synchronize IP 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 IP addresses per interface.
    2. **Fetch NetBox state**: Retrieve existing IP address objects from NetBox.
    3. **Reconcile**: Create new IP address objects, update existing unassigned or
       stale ones, and mark already-correct entries as in-sync.

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

    ```
    {
        "<device>": {
            "created": ["10.0.0.1/31", ...],
            "updated": ["10.0.0.3/31", ...],
            "in_sync": ["10.0.0.5/31", ...]
        }
    }
    ```

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

    **Side Effects**

    1. Parsed live anycast addresses extend the task's anycast ranges with
        IPv4 /32 or IPv6 /128 host networks, regardless of VRF. Overlapping
        NetBox IP records have their role updated to `anycast`.
    2. Parsed TTP IP roles are synchronized to NetBox, except for FHRP
        addresses using the `vrrp`, `glbp`, `hsrp`, or `carp` roles.
    3. Addresses with the `anycast` or `vip` role may be assigned to multiple
        interfaces, with a separate NetBox IP record for each interface.

    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_approval (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.
        anycast_ranges (list, optional): IP prefix(es) used to classify
            IP addresses as anycast role, or an ``nf://`` or ``git://<remote>/`` YAML
            file containing a list of prefixes, e.g. ``'10.3.250.0/24'``.
        ignore_ranges (list, optional): Prefixes to ignore IP addresses for, includes
            by default 127.0.0.0/8, 224.0.0.0/24 and others
        ignore_vrf (bool, optional): If True, discovered interface VRFs are ignored
            for IP writes. Existing VRF-scoped IPs are still matched by address
            and their VRF is left unchanged. Defaults to True.
        interface_map: Ordered interface rename rules, or an ``nf://`` YAML
            file containing them. Mapping is applied before interface filtering
            and NetBox interface lookup. The first matching rule wins.
        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_prefix (str, optional): IP prefix to restrict which IP addresses
            are included, e.g. ``'10.0.0.0/8'``. Skips IPs not within this prefix.
        filter_by_ip (str, optional): Glob pattern to restrict which IP addresses
            are included, e.g. ``'10.0.*'``.
        **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`` IP address lists. ``Result.diff`` contains the
            corresponding ``create``, ``update``, ``delete``, and ``in_sync``
            reconciliation plan; ``delete`` is always empty because this task
            does not delete IP addresses.
    """
    devices = devices or []
    if self.is_url(interface_map):
        interface_map = TypeAdapter(list[InterfaceMapRule]).validate_python(
            yaml.safe_load(self.fetch_file(interface_map, raise_on_fail=True)) or []
        )
    interface_map = [
        rule.model_dump() if hasattr(rule, "model_dump") else dict(rule)
        for rule in interface_map or []
    ]
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_device_ip",
        result=SyncActionSummaryMap().model_dump(),
        resources=[instance],
        diff={},
        dry_run=dry_run,
    )
    nb = self._get_pynetbox(instance, branch=branch, job=job)
    log.info(
        f"{self.name} - Sync device IP: Processing {len(devices)} device(s) in '{instance}'"
    )

    # pre-process anycast ranges
    if self.is_url(anycast_ranges):
        anycast_ranges = yaml.safe_load(
            self.fetch_file(anycast_ranges, raise_on_fail=True)
        )
    if isinstance(anycast_ranges, str):
        anycast_ranges = [anycast_ranges]
    anycast_nets = [
        ipaddress.ip_network(str(pfx), strict=False) for pfx in anycast_ranges or []
    ]
    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
    )

    # 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 IP 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,
            "device_type": d.device_type.model,
        }
        for d in self.bulk_filter(
            nb.dcim.devices,
            name=devices,
            fields="id,name,device_type",
        )
    }
    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 and filter live interface data from Nornir parse_ttp
    normalised_live_all = collect_live_interface_ip_data(
        client=self.client,
        job=job,
        ret=ret,
        devices=devices,
        timeout=timeout,
        interface_map=interface_map,
        device_types={
            name: data["device_type"] for name, data in nb_devices_data.items()
        },
        filter_by_name=filter_by_name,
        filter_by_description=filter_by_description,
    )

    # fetch interfaces data from NetBox to resolve interface IDs
    job.event("fetching NetBox interfaces to resolve IP 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 IP sync", severity="ERROR")
        ret.errors.extend(nb_interfaces_result.errors)
        ret.failed = True
        return ret

    # per-device result tracking
    device_results = {
        device_name: SyncActionSummary().model_dump() for device_name in devices
    }
    ret.result = device_results
    ret.diff = {
        device_name: {
            "create": [],
            "update": {},
            "delete": [],
            "in_sync": [],
        }
        for device_name in devices
    }

    # collect all discovered IP addresses
    job.event("collecting live IP address candidates")
    all_ip_live = []
    for device_name, interfaces in normalised_live_all.items():
        if device_name not in nb_devices_data:
            continue
        nb_raw = nb_interfaces_result.result.get(device_name, {})
        for intf_name, intf_data in interfaces.items():
            if intf_name not in nb_raw:
                continue
            for ip_data in (intf_data.get("ipv4_addresses") or []) + (
                intf_data.get("ipv6_addresses") or []
            ):
                ip = ip_data["ip"]
                parsed_ip_role = ip_data.get("ip_address_role") or None
                if parsed_ip_role in ("vrrp", "glbp", "hsrp", "carp"):
                    continue
                host_address = ipaddress.ip_interface(str(ip)).ip
                if filter_prefix_net and host_address not in filter_prefix_net:
                    continue
                if filter_by_ip and not fnmatch.fnmatch(
                    str(host_address), filter_by_ip
                ):
                    continue
                if any(host_address in network for network in ignore_nets):
                    continue
                vrf = (
                    None
                    if ignore_vrf
                    else resolve_vrf(intf_data["vrf"], nb, job, ret, self.name)
                )
                resolved_ip_role = resolve_ip_role(ip, intf_name, anycast_nets)
                # Preserve explicit roles parsed from the device, while anycast
                # detection remains authoritative for configured anycast ranges.
                if parsed_ip_role and resolved_ip_role != "anycast":
                    resolved_ip_role = parsed_ip_role
                if resolved_ip_role == "anycast":
                    anycast_nets.append(
                        ipaddress.ip_network(
                            f"{host_address}/{host_address.max_prefixlen}"
                        )
                    )
                ip_live = {
                    "device": device_name,
                    "interface": intf_name,
                    "address": ip,
                    "vrf": vrf,
                    "role": resolved_ip_role,
                    "assigned_object_type": "dcim.interface",
                    "assigned_object_id": nb_raw[intf_name]["id"],
                }
                all_ip_live.append(ip_live)

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

    # fetch existing IP addresses from NetBox
    job.event(f"fetching {len(all_ip_live)} matching IP address(es) from NetBox")
    nb_ips = [
        {
            "id": ip.id,
            "address": ip.address,
            "vrf": ip.vrf.id if ip.vrf else None,
            "role": str(ip.role).lower(),
            "assigned_object_type": ip.assigned_object_type,
            "assigned_object_id": (
                ip.assigned_object.id if ip.assigned_object else None
            ),
            "device": (
                ip.assigned_object.device.name
                if ip.assigned_object
                and ip.assigned_object_type == "dcim.interface"
                else None
            ),
            "interface": (
                ip.assigned_object.name
                if ip.assigned_object
                and ip.assigned_object_type == "dcim.interface"
                else None
            ),
        }
        for ip in self.bulk_filter(
            endpoint=nb.ipam.ip_addresses,
            address=[i["address"].split("/")[0] for i in all_ip_live],
            fields=(
                "id,address,vrf,role,assigned_object,"
                "assigned_object_type,assigned_object_id"
            ),
        )
    ]
    job.event(f"retrieved {len(nb_ips)} matching IP address object(s) from NetBox")

    # process IP addresses
    job.event("calculating IP address sync actions")
    bulk_update_ip = {}  # {(device, intf, ip): {ip data}}
    bulk_create_ip = {}  # {(device, intf, ip): {ip data}}

    for ip_live in all_ip_live:
        device_name = ip_live.pop("device")
        intf_name = ip_live.pop("interface")
        key = (device_name, intf_name, ip_live["address"])
        # find existing NetBox IPs of same value
        ip_live_no_mask = ip_live["address"].split("/")[0]
        # do IP comparison ignoring mask, in case live and netbox IPs have mask mismatch
        matching_nb_ips = [
            i for i in nb_ips if i["address"].startswith(f"{ip_live_no_mask}/")
        ]
        # Prefer the matching IP already assigned to the target interface.
        for nb_ip in matching_nb_ips:
            if (
                nb_ip["assigned_object_type"] == ip_live["assigned_object_type"]
                and nb_ip["assigned_object_id"] == ip_live["assigned_object_id"]
            ):
                matching_nb_ips = [nb_ip]
                break
        else:
            # Restrict fallback matches to the live VRF when VRFs are enforced.
            if not ignore_vrf:
                matching_nb_ips = [
                    i for i in matching_nb_ips if i["vrf"] == ip_live["vrf"]
                ]
        # Exclude records owned by non-interface objects while retaining valid
        # interface-assigned and unassigned candidates for reconciliation.
        eligible_nb_ips = [
            i
            for i in matching_nb_ips
            if i["assigned_object_id"] is None
            or i["assigned_object_type"] == "dcim.interface"
        ]
        if matching_nb_ips and not eligible_nb_ips:
            continue
        matching_nb_ips = eligible_nb_ips
        # no existing IP found, create it
        if not matching_nb_ips:
            bulk_create_ip[key] = _ip_payload(ip_live, ignore_vrf)
            continue
        # check if existing IP already assigned to same interface
        for nb_ip in matching_nb_ips:
            if nb_ip["assigned_object_id"]:
                # ip already assigned to same interface
                if (
                    nb_ip["assigned_object_type"] == ip_live["assigned_object_type"]
                    and nb_ip["assigned_object_id"] == ip_live["assigned_object_id"]
                ):
                    # Preserve a shared role already assigned in NetBox.
                    if (
                        nb_ip["role"] in ("anycast", "vip")
                        and ip_live["role"] != "anycast"
                    ):
                        msg = f"found existing NetBox IP with '{nb_ip['role']}' role {nb_ip['address']}, assigning {nb_ip['role']} role to live IP"
                        log.info(msg)
                        job.event(msg)
                        ip_live["role"] = nb_ip["role"]
                    # check if vrf or role need an update
                    if any(
                        nb_ip[k] != ip_live[k]
                        for k in ["role", "vrf"]
                        if (k == "vrf" and not ignore_vrf) or ip_live[k]
                    ):
                        payload = _ip_payload(ip_live, ignore_vrf)
                        payload["id"] = nb_ip["id"]
                        bulk_update_ip[key] = payload
                    else:
                        device_results[device_name]["in_sync"].append(
                            ip_live["address"]
                        )
                    break
        # no IP assigned to same interface
        else:
            # if an unassigned IP exists, update it instead of creating a duplicate
            nb_ip = None
            for candidate_nb_ip in matching_nb_ips:
                # Prefer reassigning an available IP over creating a duplicate.
                if not candidate_nb_ip["assigned_object_id"]:
                    nb_ip = candidate_nb_ip
                    break
            if nb_ip:
                if (
                    nb_ip["role"] in ("anycast", "vip")
                    and ip_live["role"] != "anycast"
                ):
                    ip_live["role"] = nb_ip["role"]
                if (
                    ip_live["assigned_object_id"] is None
                    and nb_ip["role"] == ip_live["role"]
                    and (ignore_vrf or nb_ip["vrf"] == ip_live["vrf"])
                ):
                    device_results[device_name]["in_sync"].append(
                        ip_live["address"]
                    )
                    continue
                payload = _ip_payload(ip_live, ignore_vrf)
                payload["id"] = nb_ip["id"]
                bulk_update_ip[key] = payload

                # Reflect the planned assignment in the local NetBox snapshot so
                # that another live interface cannot reuse the same unassigned
                # record before the bulk update is applied.
                nb_ip["assigned_object_type"] = ip_live["assigned_object_type"]
                nb_ip["assigned_object_id"] = ip_live["assigned_object_id"]
                nb_ip["device"] = device_name
                nb_ip["interface"] = intf_name
                nb_ip["role"] = ip_live["role"]
                if not ignore_vrf:
                    nb_ip["vrf"] = ip_live["vrf"]
                continue

            for nb_ip in matching_nb_ips:
                # existing NB IP already assigned to an interface
                if nb_ip["assigned_object_id"]:
                    # Inherit a shared role from an existing NetBox assignment.
                    if (
                        nb_ip["role"] in ("anycast", "vip")
                        and ip_live["role"] != "anycast"
                    ):
                        msg = f"found existing NetBox IP with '{nb_ip['role']}' role {nb_ip['address']}, using {nb_ip['role']} role for live IP"
                        log.info(msg)
                        job.event(msg)
                        nb_ip_resolved_role = nb_ip["role"]
                        ip_live["role"] = nb_ip["role"]
                    # attempt to resolve IP role
                    else:
                        nb_ip_resolved_role = (
                            resolve_ip_role(
                                nb_ip["address"], nb_ip["interface"], anycast_nets
                            )
                            or nb_ip["role"]
                        )
                    # add existing IP to updates if its role needs correcting
                    if nb_ip["role"] != nb_ip_resolved_role:
                        nb_ip_key = (
                            nb_ip["device"],
                            nb_ip["interface"],
                            nb_ip["address"],
                        )
                        bulk_update_ip[nb_ip_key] = {
                            "id": nb_ip["id"],
                            "role": nb_ip_resolved_role,
                        }
                    # Shared addresses need a record for each interface.
                    if nb_ip_resolved_role == ip_live["role"] and ip_live[
                        "role"
                    ] in ("anycast", "vip"):
                        bulk_create_ip[key] = _ip_payload(ip_live, ignore_vrf)
                        break
                    # A non-shared address assigned elsewhere is a conflict.
                    if nb_ip_resolved_role not in ("anycast", "vip"):
                        msg = (
                            f"duplicate non anycast, non vip ip found, live IP {device_name}:{intf_name}->{ip_live['address']}, "
                            f"overlaps with Netbox IP {nb_ip['device']}:{nb_ip['interface']}->{nb_ip['address']}"
                        )
                        log.error(msg)
                        ret.errors.append(msg)
                        job.event(msg, severity="ERROR")
                        break

    job.event(
        f"IP address sync actions: {len(bulk_create_ip)} create, "
        f"{len(bulk_update_ip)} update"
    )

    # make sure all devices present in results
    for key in list(bulk_create_ip) + list(bulk_update_ip):
        device_results.setdefault(
            key[0],
            SyncActionSummary().model_dump(),
        )

    # check that update and create payloads have no non-shared duplicate IPs
    # Netbox has a bug allowing to create duplicate IPs in single create request
    job.event("checking IP address payloads for duplicate non-shared addresses")
    ip_address_seen = {}  # {address: [key, ...]}
    for key, ip_data in {**bulk_create_ip, **bulk_update_ip}.items():
        addr = str(ipaddress.ip_interface(key[2]).ip)
        role = ip_data.get("role") or ""
        if role not in ("anycast", "vip"):
            ip_address_seen.setdefault(addr, []).append(key)
    for addr, dup_keys in ip_address_seen.items():
        if len(dup_keys) > 1:
            for key in dup_keys:
                device_name, intf_name, _ = key
                msg = (
                    f"found duplicate non-anycast IP {addr} in payload for "
                    f"{device_name}:{intf_name}, skipping"
                )
                log.error(msg)
                ret.errors.append(msg)
                job.event(msg, severity="ERROR")
                bulk_create_ip.pop(key, None)
                bulk_update_ip.pop(key, None)

    # Build the public diff after payload validation so it exactly describes
    # the actions that remain eligible for application.
    full_diff = {
        device_name: {
            "create": [],
            "update": {},
            "delete": [],
            "in_sync": sorted(set(actions["in_sync"])),
        }
        for device_name, actions in device_results.items()
    }
    nb_ips_by_id = {ip_data["id"]: ip_data for ip_data in nb_ips}
    for key in bulk_create_ip:
        full_diff[key[0]]["create"].append(key[2])
    for key, payload in bulk_update_ip.items():
        current = dict(nb_ips_by_id.get(payload["id"], {}))
        if current.get("role") == "none":
            current["role"] = None
        full_diff[key[0]]["update"][key[2]] = {
            field: {"old_value": current.get(field), "new_value": value}
            for field, value in payload.items()
            if field != "id" and current.get(field) != value
        }
    for actions in full_diff.values():
        actions["create"] = sorted(set(actions["create"]))
        actions["update"] = {
            address: actions["update"][address]
            for address in sorted(actions["update"])
        }
    ret.diff = full_diff

    if dry_run is True:
        job.event(
            "dry-run requested, returning IP address sync plan without changes"
        )
        for key in bulk_create_ip:
            device_name = key[0]
            device_results[device_name]["created"].append(key[2])
        for key in bulk_update_ip:
            device_name = key[0]
            device_results[device_name]["updated"].append(key[2])
        ret.dry_run = True
        return ret
    if not sync_diff_has_changes(full_diff):
        job.event("no IP address sync changes required")
        return ret
    if with_approval:
        ip_preview_result = copy.deepcopy(device_results)
        for key in bulk_create_ip:
            device_name = key[0]
            ip_preview_result[device_name]["created"].append(key[2])
        for key in bulk_update_ip:
            device_name = key[0]
            ip_preview_result[device_name]["updated"].append(key[2])
        if not review_sync_task_result(job, "IP address sync", ip_preview_result):
            ret.status = "skipped"
            ret.result = ip_preview_result
            ret.dry_run = True
            ret.messages.append("review declined; changes were not applied")
            return ret
    # update first, since existing IPs might change role to anycast
    if bulk_update_ip:
        job.event(f"updating {len(bulk_update_ip)} IP address(es)")
        update_items = list(bulk_update_ip.items())
        total_batches = (len(update_items) + batch_size - 1) // batch_size
        for batch_start in range(0, len(update_items), batch_size):
            batch = update_items[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            msg = f"updating IP address batch {batch_number}/{total_batches} ({len(batch)} address(es))"
            job.event(msg)
            log.info(msg)
            try:
                nb.ipam.ip_addresses.update([payload for _, payload in batch])
            except Exception as exc:
                msg = f"failed to update IP address batch {batch_number}/{total_batches}: {exc}"
                ret.errors.append(msg)
                ret.failed = True
                log.error(msg)
                job.event(msg, severity="ERROR")
                return ret
            for key, _ in batch:
                device_name = key[0]
                ip_address = key[2]
                device_results[device_name]["updated"].append(ip_address)
        job.event(f"updated {len(bulk_update_ip)} IP addresses")
    else:
        job.event("no IP addresses to update")

    # create new IPs next
    if bulk_create_ip:
        job.event(f"creating {len(bulk_create_ip)} IP address(es)")
        create_items = list(bulk_create_ip.items())
        total_batches = (len(create_items) + batch_size - 1) // batch_size
        for batch_start in range(0, len(create_items), batch_size):
            batch = create_items[batch_start : batch_start + batch_size]
            batch_number = batch_start // batch_size + 1
            msg = f"creating IP address batch {batch_number}/{total_batches} ({len(batch)} address(es))"
            job.event(msg)
            log.info(msg)
            try:
                nb.ipam.ip_addresses.create([payload for _, payload in batch])
            except Exception as exc:
                msg = f"failed to create IP address batch {batch_number}/{total_batches}: {exc}"
                ret.errors.append(msg)
                ret.failed = True
                log.error(msg)
                job.event(msg, severity="ERROR")
                return ret
            for key, _ in batch:
                device_name = key[0]
                ip_address = key[2]
                device_results[device_name]["created"].append(ip_address)
        job.event(f"created {len(bulk_create_ip)} IP addresses")
    else:
        job.event("no IP addresses to create")

    job.event("IP address sync complete")

    return ret