Network Automation with Python + Discord Webhooks

Sayf A — Fri, 04 Jun 2021 20:49:39 +0000

What are we gonna Make?

In this epic tutorial we are going to be using discord webhooks and the power of python to receive notifications for when a device on our local network (at home) goes offline.

Why are we making this?

its a cool project
it uses the power of python and discord!
A tutorial for those wanting to flex your muscle on python and logic building.

Let's Roll!

Phase I - The Plan

This project is going to fully depend on logic since we are embarking on coding something that's totally hybrid, but hey!...who cares? Below you'll find the flow chart of\n how this algorithm will flow.

Libraries Needed

from discord_webhook import DiscordEmbed, DiscordWebhook 
from datetime import datetime 
import subprocess
import time 
import io

Phase II - Sending the Alert

Let's start off by preparing a function that when called upon sends a discord message via webhook, we wanna make sure that the message is crafted with the following content :

Missing device MAC Address
Missing device IP Address
Missing device Name

def send_alert(webhook_url, offline_device, offline_ip, offline_mac):
    embed = DiscordEmbed(title=f"⚠️ Lost Contact with {offline_device}", color="1ca1ed")
    webhook_object = DiscordWebhook(url=webhook_url)

    embed.add_embed_field(name="Device IP", value=f"{offline_ip}")
    embed.add_embed_field(name="Device MAC", value=f"{offline_mac}")

    webhook_object.add_embed(embed)
    response = webhook_object.execute()

the function will be called with 3 arguments given which are webhook_url (for sending discord messages), offline_device (the device missing), offline_ip (device ip that's missing) and offline_mac (Mac address of missing device). Reason why we coded this alert function is because we'll be utilising it in our network monitoring function coming up at the end, so hang in there!

Phase III - Gathering Devices on our Network

If you're following along then congrats 🎉🎉🎉 on making it to phase 3 of this epic tutorial! Now we are going to start making a function which uses a module from python's standard library called subprocess to perform a custom ARP scan.

def fetch_devices():
    network_data = {
        "device_name": [],
        "ip_address": [],
        "mac_address": []
    }

    process = subprocess.Popen(['arp', '-a'], stdout=subprocess.PIPE)
    arp_resp = [line.strip() for line in io.TextIOWrapper(process.stdout, encoding='utf-8')]

    for name in arp_resp:
        network_data["device_name"].append(name.split()[0])

    for ip in arp_resp:
        network_data["ip_address"].append(ip.split()[1])

    for mac in arp_resp:
        network_data["mac_address"].append(mac.split()[3])

    return network_data

the function first initialises a dictionary which has keys that have lists attached to it as values. We will then use subprocess to run the command

arp -a

and then move to read the output of the command executed. After that list comprehension comes into play in which we clean the lines of text coming from the parsed command output. Finally we use 3 different for loops to append the data found in the command output to the dictionary of lists and finally return the populated dictionary.

Phase IV - The Final Touch...

We have made the 2 blocks of code which will be relied on to perform actions which will lead our automation to working exactly as planned, this is the part where we code up the monitoring function which will utilise the above 2 functions we have finished up.

def monitor():
    network_patch = fetch_devices()

    while True:
        active_patch = fetch_devices()

        for name, ip, mac in zip(network_patch["device_name"], network_patch["ip_address"], network_patch["mac_address"]):
            if name in active_patch["device_name"]:
                print(f"[+] {name} is online...Swinging back for another run")
                time.sleep(2.5)
                continue 

            else:
                send_alert("DISCORD-WEBHOOK-URL", name, ip, mac)
                time.sleep(1.5)
                continue

The main theory!

First we start off by fetching all the active devices on our network, this will store the dictionary that we are expecting to be returned from our fetch_devices function. After that we begin with a while loop since we want the monitoring to be persistent, then we will call upon the fetch devices function on every iteration that the for loop goes through, this way we can then check if any name from our network_patch call exists or not within our active_patch call

Wrap up + Output

Finally our program is ready, your code should look like this :

import subprocess
from datetime import datetime 
from discord_webhook import DiscordEmbed, DiscordWebhook 
import time 
import io 


def send_alert(webhook_url, offline_device, offline_ip, offline_mac):
    embed = DiscordEmbed(title=f"⚠️ Lost Contact with {offline_device}", color="1ca1ed")
    webhook_object = DiscordWebhook(url=webhook_url)

    embed.add_embed_field(name="Device IP", value=f"{offline_ip}")
    embed.add_embed_field(name="Device MAC", value=f"{offline_mac}")

    webhook_object.add_embed(embed)
    response = webhook_object.execute()

def fetch_devices():
    network_data = {
        "device_name": [],
        "ip_address": [],
        "mac_address": []
    }

    process = subprocess.Popen(['arp', '-a'], stdout=subprocess.PIPE)
    arp_resp = [line.strip() for line in io.TextIOWrapper(process.stdout, encoding='utf-8')]

    for name in arp_resp:
        network_data["device_name"].append(name.split()[0])

    for ip in arp_resp:
        network_data["ip_address"].append(ip.split()[1])

    for mac in arp_resp:
        network_data["mac_address"].append(mac.split()[3])

    return network_data

def monitor():
    network_patch = fetch_devices()

    while True:
        active_patch = fetch_devices()

        for name, ip, mac in zip(network_patch["device_name"], network_patch["ip_address"], network_patch["mac_address"]):
            if name in active_patch["device_name"]:
                print(f"[+] {name} is online...Swinging back for another run")
                time.sleep(2.5)
                continue 

            else:
                send_alert("DISCORD-WEBHOOK-URL", name, ip, mac)
                time.sleep(1.5)
                continue 

if __name__ == "__main__":
    monitor()

Testing

To test the program go ahead and start it up and once it starts running make your phone or any device active on the network disconnect from your wifi, you should see something like this come up :

Hope you enjoyed the tutorial! Do leave your comments & questions below.

Here's a link to the Repository on Github
Happy coding!

DEV Community: Sayf A