Netbox Create IP Task¤

task api name: create_ip

Task to create next available IP from prefix or get existing IP address.

Netbox service create_ip task integrated with Nornir service and can be called using netbox.create_ip Jinja2 filter, allowing to allocate IP addresses in Netbox on the fly while rendering configuration templates.

Branching Support¤

Create IP task is branch aware and can create IP addresses within the branch. Netbox Branching Plugin need to be installed on Netbox instance.

NORFAB Netbox Create IP Command Shell Reference¤

NorFab shell supports these command options for Netbox create_ip task:

nf#man tree netbox.create.ip
root
└── netbox:    Netbox service
    └── create:    Create objects in Netbox
            ├── timeout:    Job timeout
            ├── 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:    Do not commit to database
            ├── *prefix:    Prefix to allocate IP address from, can also provide prefix name or filters
            ├── device:    Device name to associate IP address with
            ├── interface:    Device interface name to associate IP address with
            ├── description:    IP address description
            ├── vrf:    VRF to associate with IP address
            ├── tags:    Tags to add to IP address
            ├── dns_name:    IP address DNS name
            ├── tenant:    Tenant name to associate with IP address
            ├── comments:    IP address comments field
            ├── role:    IP address functional role
            └── branch:    Branching plugin branch name to use
nf#

Python API Reference¤

Allocate the next available IP address from a given subnet.

This task finds or creates an IP address in NetBox, updates its metadata, optionally links it to a device/interface, and supports a dry run mode for previewing changes.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	The prefix from which to allocate the IP address, could be: IPv4 prefix string e.g. 10.0.0.0/24 IPv6 prefix string e.g. 2001::/64 Prefix description string to filter by Dictionary with prefix filters to feed `pynetbox` get method e.g. `{"prefix": "10.0.0.0/24", "site__name": "foo"}`	required
`description`	`str`	A description for the allocated IP address.	`None`
`device`	`str`	The device associated with the IP address.	`None`
`interface`	`str`	The interface associated with the IP address.	`None`
`vrf`	`str`	The VRF (Virtual Routing and Forwarding) instance.	`None`
`tags`	`list`	A list of tags to associate with the IP address.	`None`
`dns_name`	`str`	The DNS name for the IP address.	`None`
`tenant`	`str`	The tenant associated with the IP address.	`None`
`comments`	`str`	Additional comments for the IP address.	`None`
`instance`	`str`	The NetBox instance to use.	`None`
`dry_run`	`bool`	If True, do not actually allocate the IP address.	`False`
`branch`	`str`	Branch name to use, need to have branching plugin installed, automatically creates branch if it does not exist in Netbox.	`None`
`mask_len`	`int`	mask length to use for IP address on creation or to update existing IP address. On new IP address creation will create child subnet of `mask_len` within parent `prefix`, new subnet not created for existing IP addresses. `mask_len` argument ignored on dry run and ip allocated from parent prefix directly.	`None`
`create_peer_ip`	`bool`	If True creates IP address for link peer - remote device interface connected to requested device and interface	`True`

Returns:

Name	Type	Description
`dict`	`Result`	A dictionary containing the result of the IP allocation.

Tasks execution follow these steps:

Tries to find an existing IP in NetBox matching the device/interface/description. If found, uses it; otherwise, proceeds to create a new IP.
If prefix is a string, determines if it’s an IP network or a description. Builds a filter dictionary for NetBox queries, optionally including VRF.
Queries NetBox for the prefix using the constructed filter.
If dry_run is True, fetches the next available IP but doesn’t create it.
If not a dry run, creates the next available IP in the prefix.
Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags) if provided and different from current values. Handles interface assignment and can set the IP as primary for the device.
If changes were made and not a dry run, saves the IP and device updates to NetBox.

Source code in norfab\workers\netbox_worker.py

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()}
)
def create_ip(
    self,
    job: Job,
    prefix: Union[str, dict],
    device: Union[None, str] = None,
    interface: Union[None, str] = None,
    description: Union[None, str] = None,
    vrf: Union[None, str] = None,
    tags: Union[None, list] = None,
    dns_name: Union[None, str] = None,
    tenant: Union[None, str] = None,
    comments: Union[None, str] = None,
    role: Union[None, str] = None,
    status: Union[None, str] = None,
    is_primary: Union[None, bool] = None,
    instance: Union[None, str] = None,
    dry_run: Union[None, bool] = False,
    branch: Union[None, str] = None,
    mask_len: Union[None, int] = None,
    create_peer_ip: Union[None, bool] = True,
) -> Result:
    """
    Allocate the next available IP address from a given subnet.

    This task finds or creates an IP address in NetBox, updates its metadata,
    optionally links it to a device/interface, and supports a dry run mode for
    previewing changes.

    Args:
        prefix (str): The prefix from which to allocate the IP address, could be:

            - IPv4 prefix string e.g. 10.0.0.0/24
            - IPv6 prefix string e.g. 2001::/64
            - Prefix description string to filter by
            - Dictionary with prefix filters to feed `pynetbox` get method
                e.g. `{"prefix": "10.0.0.0/24", "site__name": "foo"}`

        description (str, optional): A description for the allocated IP address.
        device (str, optional): The device associated with the IP address.
        interface (str, optional): The interface associated with the IP address.
        vrf (str, optional): The VRF (Virtual Routing and Forwarding) instance.
        tags (list, optional): A list of tags to associate with the IP address.
        dns_name (str, optional): The DNS name for the IP address.
        tenant (str, optional): The tenant associated with the IP address.
        comments (str, optional): Additional comments for the IP address.
        instance (str, optional): The NetBox instance to use.
        dry_run (bool, optional): If True, do not actually allocate the IP address.
        branch (str, optional): Branch name to use, need to have branching plugin
            installed, automatically creates branch if it does not exist in Netbox.
        mask_len (int, optional): mask length to use for IP address on creation or to
            update existing IP address. On new IP address creation will create child
            subnet of `mask_len` within parent `prefix`, new subnet not created for
            existing IP addresses. `mask_len` argument ignored on dry run and ip allocated
            from parent prefix directly.
        create_peer_ip (bool, optional): If True creates IP address for link peer -
            remote device interface connected to requested device and interface

    Returns:
        dict: A dictionary containing the result of the IP allocation.

    Tasks execution follow these steps:

    1. Tries to find an existing IP in NetBox matching the device/interface/description.
        If found, uses it; otherwise, proceeds to create a new IP.

    2. If prefix is a string, determines if it’s an IP network or a description.
        Builds a filter dictionary for NetBox queries, optionally including VRF.

    3. Queries NetBox for the prefix using the constructed filter.

    4. If dry_run is True, fetches the next available IP but doesn’t create it.

    5. If not a dry run, creates the next available IP in the prefix.

    6. Updates IP attributes (description, VRF, tenant, DNS name, comments, role, tags)
        if provided and different from current values. Handles interface assignment and
        can set the IP as primary for the device.

    7. If changes were made and not a dry run, saves the IP and device updates to NetBox.
    """
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:create_ip", result={}, resources=[instance])
    tags = tags or []
    has_changes = False
    nb_ip = None
    nb_device = None
    create_peer_ip_data = {}
    nb = self._get_pynetbox(instance, branch=branch)

    # source parent prefix from Netbox
    if isinstance(prefix, str):
        # try converting prefix to network, if fails prefix is not an IP network
        try:
            network = ipaddress.ip_network(prefix)
            is_network = True
        except:
            is_network = False
        if is_network is True and vrf:
            prefix = {"prefix": prefix, "vrf__name": vrf}
        elif is_network is True:
            prefix = {"prefix": prefix}
        elif is_network is False and vrf:
            prefix = {"description": prefix, "vrf__name": vrf}
        elif is_network is False:
            prefix = {"description": prefix}
    nb_prefix = nb.ipam.prefixes.get(**prefix)
    if not nb_prefix:
        raise NetboxAllocationError(
            f"Unable to source parent prefix from Netbox - {prefix}"
        )
    parent_prefix_len = int(str(nb_prefix).split("/")[1])

    # try to source existing IP from netbox
    if device and interface and description:
        nb_ip = nb.ipam.ip_addresses.get(
            device=device,
            interface=interface,
            description=description,
            parent=str(nb_prefix),
        )
    elif device and interface:
        nb_ip = nb.ipam.ip_addresses.get(
            device=device, interface=interface, parent=str(nb_prefix)
        )
    elif description:
        nb_ip = nb.ipam.ip_addresses.get(
            description=description, parent=str(nb_prefix)
        )

    # create new IP address
    if not nb_ip:
        # check if interface has link peer that has IP within parent prefix
        if device and interface:
            connection = self.get_connections(
                job=job,
                devices=[device],
                interface_regex=interface,
                instance=instance,
                include_virtual=True,
            )
            if interface in connection.result[device]:
                peer = connection.result[device][interface]
                # do not process breakout cables
                if isinstance(peer["remote_interface"], list):
                    peer["remote_interface"] = None
                # try to source peer ip subnet
                nb_peer_ip = None
                if peer["remote_device"] and peer["remote_interface"]:
                    nb_peer_ip = nb.ipam.ip_addresses.get(
                        device=peer["remote_device"],
                        interface=peer["remote_interface"],
                        parent=str(nb_prefix),
                    )
                # try to source peer ip subnet
                nb_peer_prefix = None
                if nb_peer_ip:
                    peer_ip = ipaddress.ip_interface(nb_peer_ip.address)
                    nb_peer_prefix = nb.ipam.prefixes.get(
                        prefix=str(peer_ip.network),
                        vrf__name=vrf,
                    )
                elif create_peer_ip and peer["remote_interface"]:
                    create_peer_ip_data = {
                        "device": peer["remote_device"],
                        "interface": peer["remote_interface"],
                        "vrf": vrf,
                        "branch": branch,
                        "tenant": tenant,
                        "dry_run": dry_run,
                        "tags": tags,
                        "status": status,
                        "create_peer_ip": False,
                    }
                # use peer subnet to create IP address
                if nb_peer_prefix:
                    nb_prefix = nb_peer_prefix
                    mask_len = None  # cancel subnet creation
                    job.event(
                        f"Using link peer '{peer['remote_device']}:{peer['remote_interface']}' "
                        f"prefix '{nb_peer_prefix}' to create IP address"
                    )
        # if mask_len provided create new subnet
        if mask_len and not dry_run and mask_len != parent_prefix_len:
            if mask_len < parent_prefix_len:
                raise ValueError(
                    f"Mask length '{mask_len}' must be longer then '{parent_prefix_len}' prefix length"
                )
            prefix_status = status
            if prefix_status not in ["active", "reserved", "deprecated"]:
                prefix_status = None
            child_subnet = self.create_prefix(
                job=job,
                parent=str(nb_prefix),
                prefixlen=mask_len,
                vrf=vrf,
                tags=tags,
                tenant=tenant,
                status=prefix_status,
                instance=instance,
                branch=branch,
            )
            prefix = {"prefix": child_subnet.result["prefix"]}
            if vrf:
                prefix["vrf__name"] = vrf
            nb_prefix = nb.ipam.prefixes.get(**prefix)

            if not nb_prefix:
                raise NetboxAllocationError(
                    f"Unable to source child prefix of mask length "
                    f"'{mask_len}' from '{prefix}' parent prefix"
                )
        # execute dry run on new IP
        if dry_run is True:
            nb_ip = nb_prefix.available_ips.list()[0]
            print(nb_prefix.available_ips.list()[0])
            ret.status = "unchanged"
            ret.dry_run = True
            ret.result = {
                "address": str(nb_ip),
                "description": description,
                "vrf": vrf,
                "device": device,
                "interface": interface,
            }
            # add branch to results
            if branch is not None:
                ret.result["branch"] = branch
            return ret
        # create new IP
        else:
            nb_ip = nb_prefix.available_ips.create()
            job.event(
                f"Created '{nb_ip}' IP address for '{device}:{interface}' within '{nb_prefix}' prefix"
            )
        ret.status = "created"
    else:
        job.event(f"Using existing IP address {nb_ip}")
        ret.status = "updated"

    # update IP address parameters
    if description and description != nb_ip.description:
        nb_ip.description = description
        has_changes = True
    if vrf and vrf != nb_ip.vrf:
        nb_ip.vrf = {"name": vrf}
        has_changes = True
    if tenant and tenant != nb_ip.tenant:
        nb_ip.tenant = {"name": tenant}
        has_changes = True
    if dns_name and dns_name != nb_ip.dns_name:
        nb_ip.dns_name = dns_name
        has_changes = True
    if comments and comments != nb_ip.comments:
        nb_ip.comments = comments
        has_changes = True
    if role and role != nb_ip.role:
        nb_ip.role = role
        has_changes = True
    if tags and not any(t in nb_ip.tags for t in tags):
        for t in tags:
            if t not in nb_ip.tags:
                nb_ip.tags.append({"name": t})
                has_changes = True
    if device and interface:
        nb_interface = nb.dcim.interfaces.get(device=device, name=interface)
        if not nb_interface:
            raise NetboxAllocationError(
                f"Unable to source '{device}:{interface}' interface from Netbox"
            )
        if (
            hasattr(nb_ip, "assigned_object")
            and nb_ip.assigned_object != nb_interface.id
        ):
            nb_ip.assigned_object_id = nb_interface.id
            nb_ip.assigned_object_type = "dcim.interface"
            if is_primary is not None:
                nb_device = nb.dcim.devices.get(name=device)
                nb_device.primary_ip4 = nb_ip.id
            has_changes = True
    if mask_len and not str(nb_ip).endswith(f"/{mask_len}"):
        address = str(nb_ip).split("/")[0]
        nb_ip.address = f"{address}/{mask_len}"
        has_changes = True

    # save IP address into Netbox
    if dry_run:
        ret.status = "unchanged"
        ret.dry_run = True
    elif has_changes:
        nb_ip.save()
        job.event(f"Updated '{str(nb_ip)}' IP address parameters")
        # make IP primary for device
        if is_primary is True and nb_device:
            nb_device.save()
    else:
        ret.status = "unchanged"

    # form and return results
    ret.result = {
        "address": str(nb_ip),
        "description": str(nb_ip.description),
        "vrf": str(nb_ip.vrf) if not vrf else nb_ip.vrf["name"],
        "device": device,
        "interface": interface,
    }
    # add branch to results
    if branch is not None:
        ret.result["branch"] = branch

    # create IP address for peer
    if create_peer_ip and create_peer_ip_data:
        job.event(
            f"Creating IP address for link peer '{create_peer_ip_data['device']}:{create_peer_ip_data['interface']}'"
        )
        peer_ip = self.create_ip(
            **create_peer_ip_data, prefix=str(nb_prefix), job=job
        )
        if peer_ip.failed == False:
            ret.result["peer"] = peer_ip.result

    return ret