Script IPAM GUI

[davidgib01]

Hello,

I have Netbox version 3.6.2

I managed to create a script in Python which works well when I launch it in CLI from the server.

Now I'm trying to integrate it into the web part in >Customization>Scripts but it doesn't work.
I think my syntax is bad but I just started python with Netbox so I don't have the logic right yet.

Could someone please advise me?

Here is my code that I am trying to integrate into the web part:

		from extras.scripts import *
		from ipcalc import *
		from requests import *
		from urllib3 import *
		from socket import *
		from json import *
		from netbox import NetBox
		from networkscan import *

		class MacFind(Script):
			class Meta:
			   name = "Scan IP"
			   description = "Scan IP"

			IPScan = StringVar(required=True)

			API_TOKEN = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
			HEADERS = {'Authorization': f'Token {API_TOKEN}', 'Content-Type': 'application/json', 'Accept': 'application/json'}
			netbox = NetBox(host='1.2.3.6', port=443, use_ssl=True, auth_token='XXXXXXXXXXXXXXXXXXXXXXXXXXXXXX')
			NB_URL = "https://1.2.3.6"

			def run(self, data, commit):

				if __name__ == '__main__':
					my_network_list = ['IPScan']
					mask = my_network_list[0].split('/')[1]
					for network in my_network_list:
						my_scan = networkscan.Networkscan(network)
						my_scan.run(     
						found_ip_in_network = []
						for address1 in my_scan.list_of_hosts_found:
							found_ip_in_network.append(str(address1)
						for ipaddress in ipcalc.Network(network):
							try:
								request_url = f"{NB_URL}/api/ipam/ip-addresses/?q={ipaddress}/"
								ipaddress1 = requests.get(request_url, headers=HEADERS, verify=False)
								ipaddress1.raise_for_status()
								netboxip = ipaddress1.json(     
								if netboxip['count'] == 0:
									if ipaddress in found_ip_in_network:
										netbox.ipam.create_ip_address(f"{ipaddress}/{mask}")
										try:
											name = socket.gethostbyaddr(str(ipaddress))
											name = name[0]
											netbox.ipam.update_ip(str(ipaddress), dns_name=name)
										except socket.error:
											pass
								else:
									if ipaddress in found_ip_in_network:
										existing_status = netboxip['results'][0]['status']['value']
										if existing_status == "reserved":
											netbox.ipam.update_ip(str(ipaddress), status="active")
											try:
												name = socket.gethostbyaddr(str(ipaddress))
												name = name[0]
												netbox.ipam.update_ip(str(ipaddress), dns_name=name)
											except socket.error:
												pass
										elif existing_status == "active":
											netbox.ipam.update_ip(str(ipaddress), status="active")
											try:
												name = socket.gethostbyaddr(str(ipaddress))
												name = name[0]
												netbox.ipam.update_ip(str(ipaddress), dns_name=name)
											except socket.error:
												pass
										else:
											netbox.ipam.update_ip(str(ipaddress), status="deprecated")
									else:
										existing_status = netboxip['results'][0]['status']['value']
										if existing_status == "active":
											netbox.ipam.update_ip(str(ipaddress), status="deprecated"     
							except requests.exceptions.RequestException as e:
								print(f"Request to Netbox failed: {e}")
							except json.decoder.JSONDecodeError as je:
								print(f"JSON decoding error: {je}")
							except TimeoutError as te:
								print(f"Timeout error: {te}"

[candlerb]

There are some sample custom scripts you can compare with here: https://github.com/netbox-community/customizations/tree/master/scripts

What error do you see?

You definitely need to remove the if __name__ == '__main__': test from the run function.

You seem to have some broken lines where a trailing parenthesis is missing, e.g.

• my_scan.run(
• found_ip_in_network.append(str(address1)
• netboxip = ipaddress1.json(
• print(f"Timeout error: {te}"

...but maybe this is just an artefact of how you're copy-pasting. You also should also not have the code indendent away from the left-hand side, as it appears to be.

More generally: you would not normally use the REST API within a custom script, you would use the Django ORM instead. This lets you benefit from proper database features like transactions and rollback - and you wouldn't have all this except socket.error nonsense which means you silently accept when things fail. But that would involve rewriting your script not to use the netbox.NetBox REST API client.

Stylistically I would avoid import *, especially where you have lots of them, as it's possible that a name imported in one module will overwrite a name from anther. But I realise it's more typing.

Happy debugging!

[davidgib01]

Thanks for your response, I'll look into it and post the solution.

[davidgib01]

I'm almost there, here's where I am:

import subprocess
import socket
from ipaddress import IPv4Network
from ipam.models import IPAddress
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates IPAM in NetBox based on active addresses"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        # Récupérer la plage réseau à scanner depuis la variable paramétrable
        network = data['network_to_scan']

        # Scanner le réseau
        network_range = IPv4Network(network)
        output = []

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            # Vérifier si l'adresse IP est active en effectuant un ping
            ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            try:
                ip_obj = IPAddress.objects.get(address=ip_str)
                if ping_cmd.returncode == 0:
                    if ip_obj.status == IPAddressStatusChoices.STATUS_RESERVED or ip_obj.status == IPAddressStatusChoices.STATUS_DEPRECATED:
                        ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE
                        ip_obj.save()
                        output.append(f"Address {ip_str} is now active.")

                        # Vérifier si on peut mettre à jour le nom DNS
                        try:
                            name = socket.gethostbyaddr(ip_str)[0]
                            if ip_obj.dns_name != name:
                                ip_obj.dns_name = name
                                ip_obj.save()
                                output.append(f"DNS updated for {ip_str}.")

                        except socket.error:
                            pass
                elif ping_cmd.returncode != 0 and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                    ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                    ip_obj.save()
                    output.append(f"Address {ip_str} is now deprecated.")

            except IPAddress.DoesNotExist:
                if ping_cmd.returncode == 0:
                    ip_obj = IPAddress.objects.create(address=ip_str, status=IPAddressStatusChoices.STATUS_ACTIVE)
                    output.append(f"Address {ip_str} is created and active.")

                    # Vérifier si on peut mettre à jour le nom DNS
                    try:
                        name = socket.gethostbyaddr(ip_str)[0]
                        ip_obj.dns_name = name
                        ip_obj.save()
                        output.append(f"DNS added for {ip_str}.")

                    except socket.error:
                        pass
                else:
                    pass

        return "\n".join(output) if output else "No actions performed."

But this part doesn't work:

                elif ping_cmd.returncode != 0 and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                    ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                    ip_obj.save()
                    output.append(f"Address {ip_str} is now deprecated.")

i try with else but i th same
i try with elif ping_cmd.returncode = 1 and not work

I would like the addresses which are present in netbox and which did not respond to the ping of my script to be passed to "deprecated" unless they are to "reserved"

[mtinberg]

I believe that you should call object.full_clean()​ before object.save()​ to run any validation steps which are implemented in code, save() only validates at the level of the SQL schema and might not trigger webhooks, changelog entries, custom validation or other business rules (I think). I was looking for the documentation reference in https://demo.netbox.dev/static/docs/customization/custom-scripts/ to confirm the details of my faulty recollection but didn't find it. — Mark Tinberg ***@***.***> Division of Information Technology-Network Services University of Wisconsin-Madison

…

________________________________ From: davidgib01 ***@***.***> Sent: Thursday, November 30, 2023 9:06 AM To: netbox-community/netbox ***@***.***> Cc: Subscribed ***@***.***> Subject: Re: [netbox-community/netbox] Script IPAM GUI (Discussion #14364) I'm almost there, here's where I am: import subprocess import socket from ipaddress import IPv4Network from ipam.models import IPAddress from ipam.choices import IPAddressStatusChoices from extras.scripts import Script, IPAddressWithMaskVar class NetworkScan(Script): class Meta: name = "Network IP Scan" description = "Scans a specified network, updates IPAM in NetBox based on active addresses" network_to_scan = IPAddressWithMaskVar( label="Network to Scan", description="Enter the network to scan (e.g., 192.168.1.0/24)", required=True ) def run(self, data, commit): # Récupérer la plage réseau à scanner depuis la variable paramétrable network = data['network_to_scan'] # Scanner le réseau network_range = IPv4Network(network) output = [] for ip_address in network_range.hosts(): ip_str = str(ip_address) # Vérifier si l'adresse IP est active en effectuant un ping ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE) try: ip_obj = IPAddress.objects.get(address=ip_str) if ping_cmd.returncode == 0: if ip_obj.status == IPAddressStatusChoices.STATUS_RESERVED or ip_obj.status == IPAddressStatusChoices.STATUS_DEPRECATED: ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE ip_obj.save() output.append(f"Address {ip_str} is now active.") # Vérifier si on peut mettre à jour le nom DNS try: name = socket.gethostbyaddr(ip_str)[0] if ip_obj.dns_name != name: ip_obj.dns_name = name ip_obj.save() output.append(f"DNS updated for {ip_str}.") except socket.error: pass elif ping_cmd.returncode != 0 and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE: ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED ip_obj.save() output.append(f"Address {ip_str} is now deprecated.") except IPAddress.DoesNotExist: if ping_cmd.returncode == 0: ip_obj = IPAddress.objects.create(address=ip_str, status=IPAddressStatusChoices.STATUS_ACTIVE) output.append(f"Address {ip_str} is created and active.") # Vérifier si on peut mettre à jour le nom DNS try: name = socket.gethostbyaddr(ip_str)[0] ip_obj.dns_name = name ip_obj.save() output.append(f"DNS added for {ip_str}.") except socket.error: pass else: pass return "\n".join(output) if output else "No actions performed." But this part doesn't work: elif ping_cmd.returncode != 0 and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE: ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED ip_obj.save() output.append(f"Address {ip_str} is now deprecated.") i try with else but i th same i try with elif ping_cmd.returncode = 1 and not work I would like the addresses which are present in netbox and which did not respond to the ping of my script to be passed to "deprecated" unless they are to "reserved" — Reply to this email directly, view it on GitHub<#14364 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AAS7UM3DOHHDPHSK2KXNL53YHCOIRAVCNFSM6AAAAAA76HHE7OVHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM3TOMJYHE2TS>. You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[candlerb]

Mark's correct. You need ip_obj.full_clean() before ip_obj.save(). This is a Djangoism and isn't mentioned in Netbox docs AFAIK.

If elif ping_cmd.returncode != 0 and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE: doesn't work, then I'd guess either the returncode or the existing ip_obj.status is not what you expect. Add some log calls to see what they actually are.

The result code I expect is 2:

>>> import subprocess
>>> ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', '1.2.3.4'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
>>> ping_cmd.returncode
2
>>>

Of course, there's no need to check that ping_cmd.returncode != 0 because the earlier branch of the if will be taken if ping_cmd.returncode == 0.

So you could just do:

                if ping_cmd.returncode == 0:
                    ... foo
                elif ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                    ... bar

[davidgib01]

I tried both methods but I just got a return of addresses that already exist, here is the code I tried:

import subprocess
import socket
from ipaddress import IPv4Network
from ipam.models import IPAddress
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates IPAM in NetBox based on active addresses"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        # Récupérer la plage réseau à scanner depuis la variable paramétrable
        network = data['network_to_scan']

        # Scanner le réseau
        network_range = IPv4Network(network)
        output = []

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            # Vérifier si l'adresse IP est active en effectuant un ping
            ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            try:
                ip_obj = IPAddress.objects.get(address=ip_str)

                # Validation complète avant la sauvegarde
                ip_obj.full_clean()

                if ping_cmd.returncode == 0:
                    if ip_obj.status == IPAddressStatusChoices.STATUS_RESERVED or ip_obj.status == IPAddressStatusChoices.STATUS_DEPRECATED:
                        ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE
                        ip_obj.save()
                        output.append(f"Address {ip_str} is now active.")

                        # Vérifier si on peut mettre à jour le nom DNS
                        try:
                            name = socket.gethostbyaddr(ip_str)[0]
                            if ip_obj.dns_name != name:
                                ip_obj.dns_name = name
                                ip_obj.save()
                                output.append(f"DNS updated for {ip_str}.")

                        except socket.error:
                            pass
                    elif ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                        output.append(f"Address {ip_str} is already active.")
                else:
                    if ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                        ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                        ip_obj.save()
                        output.append(f"Address {ip_str} is now deprecated due to ping failure.")

                ip_obj.save()  # Sauvegarde après le traitement

            except IPAddress.DoesNotExist:
                if ping_cmd.returncode == 0:
                    ip_obj = IPAddress.objects.create(address=ip_str, status=IPAddressStatusChoices.STATUS_ACTIVE) 
                    output.append(f"Address {ip_str} is created and active.")

                    # Vérifier si on peut mettre à jour le nom DNS
                    try:
                        name = socket.gethostbyaddr(ip_str)[0]
                        ip_obj.dns_name = name
                        ip_obj.save()
                        output.append(f"DNS added for {ip_str}.")

                    except socket.error:
                        pass
                else:
                    pass

        return "\n".join(output) if output else "No actions performed."

in return I have:

Address 1.2.3.6 is already active.
Address 1.2.3.7 is already active.
Address 1.2.3.8 is already active.
Address 1.2.3.11 is already active.
Address 1.2.3.254 is already active.

either my addresses which are already active and which respond to ping

To be sure I tried this:

import subprocess
from ipaddress import IPv4Network
from ipam.models import IPAddress
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates inactive IP addresses to deprecated status"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        network = data['network_to_scan']
        network_range = IPv4Network(network)
        output = []

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            output.append(f"Ping result for {ip_str}: Return code {ping_cmd.returncode}")

            try:
                # Vérifier si l'adresse IP existe dans l'IPAM
                ip_obj = IPAddress.objects.get(address=ip_str)

                if ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                    if ping_cmd.returncode != 0:
                        ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                        ip_obj.save()
                        output.append(f"Address {ip_str} is now deprecated.")

            except IPAddress.DoesNotExist:
                pass

        return "\n".join(output) if output else "No actions performed."

but I have no changes to my addresses which do not respond to ping and are already present in the IPAM

and with this version:

import subprocess
import socket
from ipaddress import IPv4Network
from ipam.models import IPAddress
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates IPAM in NetBox based on active addresses"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        # Récupérer la plage réseau à scanner depuis la variable paramétrable
        network = data['network_to_scan']

        # Scanner le réseau
        network_range = IPv4Network(network)
        output = []

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            # Vérifier si l'adresse IP est active en effectuant un ping
            ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            try:
                ip_obj = IPAddress.objects.get(address=ip_str)
                if ping_cmd.returncode == 0:
                    if ip_obj.status == IPAddressStatusChoices.STATUS_RESERVED or ip_obj.status == IPAddressStatusChoices.STATUS_DEPRECATED:
                        ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE
                        ip_obj.save()
                        output.append(f"Address {ip_str} is now active.")

                        # Vérifier si on peut mettre à jour le nom DNS
                        try:
                            name = socket.gethostbyaddr(ip_str)[0]
                            if ip_obj.dns_name != name:
                                ip_obj.dns_name = name
                                ip_obj.save()
                                output.append(f"DNS updated for {ip_str}.")

                        except socket.error:
                            pass
                elif ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                    ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                    ip_obj.save()
                    output.append(f"Address {ip_str} is now deprecated due to ping failure.")

            except IPAddress.DoesNotExist:
                if ping_cmd.returncode == 0:
                    ip_obj = IPAddress.objects.create(address=ip_str, status=IPAddressStatusChoices.STATUS_ACTIVE) 
                    output.append(f"Address {ip_str} is created and active.")

                    # Vérifier si on peut mettre à jour le nom DNS
                    try:
                        name = socket.gethostbyaddr(ip_str)[0]
                        ip_obj.dns_name = name
                        ip_obj.save()
                        output.append(f"DNS added for {ip_str}.")

                    except socket.error:
                        pass
                else:
                    pass

        return "\n".join(output) if output else "No actions performed."

i have:

No actions performed.

[candlerb]

I still suggest you put lots of log calls in, before and after each significant statement, to see what's happening. Keep the message f"Ping result for {ip_str}: Return code {ping_cmd.returncode}" and put a message inside each branch of every if.

Rather than appending to output, it's easier to call the logging functions provided, like self.log_info("My message"). This means you'll get the log messages even if the script aborts due to an exception. Log messages can contain Markdown formatting too, e.g. if you want clickable links.

I note that you're still missing ip_obj.full_clean() before every instance of ip_obj.save(). This is not optional, it's a serious bug if you leave it out.

[davidgib01]

I tried with a simpler script to do what you asked me (well if I understood correctly, I'm sorry but I'm really not good at python)

here is the script:

import subprocess
from ipaddress import IPv4Network
from ipam.models import IPAddress
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class DeprecatedActiveIPsInRange(Script):
    class Meta:
        name = "Mark Active IPs with Ping Return Code 1 in Range as Deprecated"
        description = "Marks active IP addresses with ping return code 1 in a given range as deprecated in the IPAM"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        network = data['network_to_scan']
        network_range = IPv4Network(network)

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

            self.log_info(f"Ping result for {ip_str}: Return code {ping_cmd.returncode}")

            if ping_cmd.returncode == 1:
                try:
                    ip_obj = IPAddress.objects.get(address=ip_str)
                    if ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                        ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                        ip_obj.full_clean()
                        ip_obj.save()
                        self.log_info(f"Address {ip_str} marked as deprecated due to ping failure.")
                except IPAddress.DoesNotExist:
                    pass
                except Exception as e:
                    self.log_failure(f"Error occurred for {ip_str}: {e}")

        return "Execution completed."

I got this kind of feedback:

1	Info	
Ping result for 1.2.3.1: Return code 1

2	Info	
Ping result for 1.2.3.2: Return code 1

3	Info	
Ping result for 1.2.3.3: Return code 1

4	Info	
Ping result for 1.2.3.4: Return code 1

5	Info	
Ping result for 1.2.3.5: Return code 1

6	Info	
Ping result for 1.2.3.6: Return code 0

7	Info	
Ping result for 1.2.3.7: Return code 0

8	Info	
Ping result for 1.2.3.8: Return code 0

9	Info	
Ping result for 1.2.3.9: Return code 1

10	Info	
Ping result for 1.2.3.10: Return code 1

....

Output
Execution completed.

But nothing changes. Active addresses which do not respond to ping therefore =1 are not deprecated

[candlerb]

                except IPAddress.DoesNotExist:
                    pass

My guess is this is where it's silently skipping. Did you try putting a log message here?

IP Addresses in Netbox are a strange thing. You cannot have "1.2.3.4" as an IP address, but you can have "1.2.3.4/24". Your log messages show that ip_str doesn't have the prefixlen.

nbshell is a good environment for testing this sort of thing interactively:

# /opt/netbox/venv/bin/python /opt/netbox/netbox/manage.py nbshell
### NetBox interactive shell (netbox3)
### Python 3.8.10 | Django 4.2.7 | NetBox 3.6.6
### lsmodels() will show available models. Use help(<model>) for more info.
>>> IPAddress.objects.get(address="10.12.0.1")
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "/opt/netbox/venv/lib/python3.8/site-packages/django/db/models/manager.py", line 87, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/opt/netbox/venv/lib/python3.8/site-packages/django/db/models/query.py", line 637, in get
    raise self.model.DoesNotExist(
ipam.models.ip.IPAddress.DoesNotExist: IPAddress matching query does not exist.
>>> IPAddress.objects.get(address="10.12.0.1/24")
<IPAddress: 10.12.0.1/24>
>>>

Now, you could synthesise it from the prefixlen of the netblock being scanned, but that wouldn't work if you were scanning a wider range which covers multiple prefixes.

Probably your best bet is something like this (which is not at all obvious, but inspired from search_contains in class PrefixFilterSet in netbox/ipam/models/filtersets.py)

>>> ip_str = "10.12.0.1"
>>> Prefix.objects.filter(prefix__net_contains_or_equals=ip_str)
<PrefixQuerySet [<Prefix: 10.12.0.0/16>, <Prefix: 10.12.0.0/24>]>
>>> pfs = Prefix.objects.filter(prefix__net_contains_or_equals=ip_str)
>>> sorted(pfs, key=lambda x: x.mask_length)
[<Prefix: 10.12.0.0/16>, <Prefix: 10.12.0.0/24>]
>>> pfx = sorted(pfs, key=lambda x: x.mask_length)[-1]
>>> pfx
<Prefix: 10.12.0.0/24>
>>> pfx.mask_length
24
>>> f"{ip_str}/{pfx.mask_length}"
'10.12.0.1/24'

That is: find the nearest enclosing Prefix object (the one with the longest prefix length) and use that to build a combined IP + prefix length as the address value to use.

[davidgib01]

in my example the IPs are modified, because even if it remains from the lab they are IPs used in my work

[candlerb]

It doesn't matter what the actual IPs are. Your log message shows

Ping result for 1.2.3.1: Return code 1

not

Ping result for 1.2.3.1/24: Return code 1

[davidgib01]

Good morning,

Thank you for your help, now I can put the IPs which do not respond to ping and which are in the active status of the IPAM in deprecated with this script:

import subprocess
from ipaddress import IPv4Network
from ipam.models import IPAddress, Prefix
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class DeprecatedActiveIPsInRange(Script):
    class Meta:
        name = "Mark Active IPs with Ping Return Code 1 in Range as Deprecated"
        description = "Marks active IP addresses with ping return code 1 in a given range as deprecated in the IPAM"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def find_enclosing_prefix(self, ip_str):
        prefixes = Prefix.objects.filter(prefix__net_contains_or_equals=ip_str)
        sorted_prefixes = sorted(prefixes, key=lambda x: x.prefix.prefixlen, reverse=True)
        if sorted_prefixes:
            return sorted_prefixes[0]
        return None

    def run(self, data, commit):
        network = data['network_to_scan']
        network_range = IPv4Network(network)

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)

            enclosing_prefix = self.find_enclosing_prefix(ip_str)
            if enclosing_prefix:
                ip_str_with_prefixlen = f"{ip_str}/{enclosing_prefix.prefix.prefixlen}"

                ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

                self.log_info(f"Ping result for {ip_str}: Return code {ping_cmd.returncode}")

                if ping_cmd.returncode == 1:
                    try:
                        ip_obj = IPAddress.objects.get(address=ip_str_with_prefixlen)
                        if ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                            ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                            ip_obj.full_clean()
                            ip_obj.save()
                            self.log_info(f"Address {ip_str} marked as deprecated due to ping failure.")
                    except IPAddress.DoesNotExist:
                        pass
                    except Exception as e:
                        self.log_failure(f"Error occurred for {ip_str}: {e}")

        return "Execution completed."

I then wanted to put it in my complete script but it doesn't really work, because it creates duplicate IP addresses for me, I mean it creates IP addresses that already exist in the IPAM.
Here is the script I tried:

import subprocess
import socket
from ipaddress import IPv4Network
from ipam.models import IPAddress, Prefix
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates IPAM in NetBox based on active addresses"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def find_enclosing_prefix(self, ip_str):
        prefixes = Prefix.objects.filter(prefix__net_contains_or_equals=ip_str)
        sorted_prefixes = sorted(prefixes, key=lambda x: x.prefix.prefixlen, reverse=True)
        if sorted_prefixes:
            return sorted_prefixes[0]
        return None

    def run(self, data, commit):
        # Récupérer la plage réseau à scanner depuis la variable paramétrable
        network = data['network_to_scan']

        # Scanner le réseau
        network_range = IPv4Network(network)
        output = []

        for ip_address in network_range.hosts():
            ip_str = str(ip_address)
            enclosing_prefix = self.find_enclosing_prefix(ip_str)
            
            if enclosing_prefix:
                ip_str_with_prefixlen = f"{ip_str}/{enclosing_prefix.prefix.prefixlen}"

                # Vérifier si l'adresse IP est active en effectuant un ping
                ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)

                try:
                    ip_obj = IPAddress.objects.get(address=ip_str_with_prefixlen)

                    # Validation complète avant la sauvegarde
                    ip_obj.full_clean()

                    if ping_cmd.returncode == 0:
                        if ip_obj.status == IPAddressStatusChoices.STATUS_RESERVED or ip_obj.status == IPAddressStatusChoices.STATUS_DEPRECATED:
                            ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE
                            ip_obj.save()
                            output.append(f"Address {ip_str} is now active.")

                            # Vérifier si on peut mettre à jour le nom DNS
                            try:
                                name = socket.gethostbyaddr(ip_str)[0]
                                if ip_obj.dns_name != name:
                                    ip_obj.dns_name = name
                                    ip_obj.save()
                                    output.append(f"DNS updated for {ip_str}.")

                            except socket.error:
                                pass
                        elif ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                            output.append(f"Address {ip_str} is already active.")
                    else:
                        if ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                            ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                            ip_obj.save()
                            output.append(f"Address {ip_str} is now deprecated due to ping failure.")

                    ip_obj.save()  # Sauvegarde après le traitement

                except IPAddress.DoesNotExist:
                    if ping_cmd.returncode == 0:
                        ip_obj = IPAddress.objects.create(address=ip_str_with_prefixlen, status=IPAddressStatusChoices.STATUS_ACTIVE) 
                        output.append(f"Address {ip_str} is created and active.")

                        # Vérifier si on peut mettre à jour le nom DNS
                        try:
                            name = socket.gethostbyaddr(ip_str)[0]
                            ip_obj.dns_name = name
                            ip_obj.save()
                            output.append(f"DNS added for {ip_str}.")

                        except socket.error:
                            pass
                    else:
                        pass

        return "\n".join(output) if output else "No actions performed."

The code is starting to be really complex for me and as soon as I try to correct it it's worse than before, I tried to separate the two treatments but it doesn't work, I must probably be doing it wrong

[candlerb]

You don't seem to have read what I said before. You must put full_clean() directly before every save(). It's no good calling full_clean(), then modifying an attribute, then calling save() - you will have bypassed all Django ORM data validation, and it will allow you to save junk in the database.

If you want to avoid saving duplicate IP addresses, then you need to call full_clean() in the right places. Also you need to set ENFORCE_GLOBAL_UNIQUE = True in configuration.py. Then Netbox should reject attempts to create duplicates.

Separately, there's the question of why your code even tries to create a duplicate IP address in the first place. I'm not going to attempt to debug it line by line - it's your code and that's for you to do. Also, you haven't shown any example of what these duplicate IP addresses look like in the database. If they are (e.g.) 1.2.3.4/24 and 1.2.3.4/26, then that's an artefact of how Netbox names IP addresses with the prefixlen included. You might look for 1.2.3.4/26, find it doesn't exist (even though, say, 1.2.3.4/24 exists), and try to create it. However if you have two instances of 1.2.3.4/24 then something is wrong with your code which checks the existence of a database entry.

Lastly, I would observe that you're trying to do something that Netbox isn't designed to do: see https://docs.netbox.dev/en/stable/introduction/#serve-as-a-source-of-truth

Of course, you're free to use open source software in whatever way you like, but what you're doing is counter to the design philosophy of Netbox. The "status" of an IP address in Netbox is supposed to describe how the address should respond. If for example address 1.2.3.4 has status Active, and you ping it and find it doesn't respond, then clearly there's a problem, and it needs to be escalated for analysis. It could be that the device or network has failed: the solution then is to fix the device or network. Or it could be that the device has been intentionally decommissioned, but a human has forgotten to change the state from Active to Deprecated; in that case, the status does need to be changed. But only a human can make that assessment.

Netbox is certainly not intended to be a portal of live network status, where status flips between Active and Deprecated depending on whether the device is up or down. You should use a monitoring system for that, of which there are many (e.g. Nagios/check_mk, LibreNMS, Prometheus/Grafana). Using Netbox as a source of information to configure your monitoring system is a fine use of it.

[davidgib01]

Thank you for your answers, I will take a look on my own. As I told you I don't know python I try to make do with what little I understand. For me an ipam must be able to scan a network at any time, and give me feedback on the IPs which are already in the database, this is the core function of an ipam as do others like PHPipam, solarwinds...

My cli script does it without problem so I thought I could adapt it to use it in the GUI part.

[davidgib01]

Thanks for your help. I found how to do this by separating the two processes with two separate scripts.
I'm carrying out some more tests and will post the two scripts for those who it will help

[davidgib01]

Here is my final code, I managed to put everything in one script.
I started from scratch with table logic, and the duplicate problem came although when creating new addresses it created them for me in /32 instead of using the mask of the scanned prefix:

import subprocess
from ipaddress import IPv4Network
from ipam.models import IPAddress, Prefix
from ipam.choices import IPAddressStatusChoices
from extras.scripts import Script, IPAddressWithMaskVar
import socket

class NetworkScan(Script):
    class Meta:
        name = "Network IP Scan"
        description = "Scans a specified network, updates IPAM in NetBox based on active addresses"

    network_to_scan = IPAddressWithMaskVar(
        label="Network to Scan",
        description="Enter the network to scan (e.g., 192.168.1.0/24)",
        required=True
    )

    def run(self, data, commit):
        network = data['network_to_scan']

        # Parse the network input to get the network object and its prefix length
        network_range = IPv4Network(network)
        network_prefixlen = network_range.prefixlen

        output = []
        ips_ok = set()
        ips_nok = set()

        # Collect IPs responding to ping and those not responding
        for ip_address in network_range.hosts():
            ip_str = str(ip_address)
            try:
                ping_cmd = subprocess.run(['ping', '-c', '1', '-W', '1', ip_str], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                if ping_cmd.returncode == 0:
                    ips_ok.add(ip_str)
                else:
                    ips_nok.add(ip_str)
            except subprocess.CalledProcessError:
                ips_nok.add(ip_str)

        prefix = Prefix.objects.get(prefix=network)

        for ip_obj in IPAddress.objects.filter(address__net_contained_or_equal=str(prefix.prefix)):
            ip_str = str(ip_obj.address.ip)
            
            if ip_str in ips_ok:
                if ip_obj.status in (IPAddressStatusChoices.STATUS_RESERVED, IPAddressStatusChoices.STATUS_DEPRECATED):
                    ip_obj.status = IPAddressStatusChoices.STATUS_ACTIVE
                    ip_obj.full_clean()
                    ip_obj.save()
                    output.append(f"Address {ip_str} is now active.")

                    # Mise à jour du nom DNS
                    try:
                        name = socket.gethostbyaddr(ip_str)[0]
                        if ip_obj.dns_name != name:
                            ip_obj.dns_name = name
                            ip_obj.save()
                            output.append(f"DNS updated for {ip_str}.")
                    except socket.error:
                        pass
            elif ip_str in ips_nok and ip_obj.status == IPAddressStatusChoices.STATUS_ACTIVE:
                ip_obj.status = IPAddressStatusChoices.STATUS_DEPRECATED
                ip_obj.full_clean()
                ip_obj.save()
                output.append(f"Address {ip_str} is now deprecated due to ping failure.")

        # Create new IP objects for IPs in ips_ok not present in the prefix
        for ip_str in ips_ok:
            if ip_str not in ips_nok and ip_str not in [str(ip_obj.address.ip) for ip_obj in IPAddress.objects.filter(address__net_contained_or_equal=str(prefix.prefix))]:
                ip_obj = IPAddress.objects.create(address=f"{ip_str}/{network_prefixlen}", status=IPAddressStatusChoices.STATUS_ACTIVE)
                output.append(f"Address {ip_str} is created and active.")

                # Mise à jour du nom DNS
                try:
                    name = socket.gethostbyaddr(ip_str)[0]
                    ip_obj.dns_name = name
                    ip_obj.save()
                    output.append(f"DNS added for {ip_str}.")
                except socket.error:
                    pass

        return "\n".join(output) if output else "No actions performed."

Thank you again for your help