Untitled

 avatar
user_3431143
plain_text
a year ago
2.6 kB
1
Indexable
Never
import scapy.all as scapy 
import requests 
import json 
import time 
import threading

WEBHOOK_URL = "https://your-webhook-url.com"
JSON_FILE = "devices.json"
BUFFER_FILE = "buffer.json"

def get_mac(ip):
    arp_request = scapy.ARP(pdst=ip)
    broadcast = scapy.Ether(dst="ff:ff:ff:ff:ff:ff")
    arp_request_broadcast = broadcast/arp_request
    answered_list = scapy.srp(arp_request_broadcast, timeout=1, verbose=False)[0]         
    return answered_list[0][1].hwsrc if answered_list else None

def scan_network():
    devices = []

    ip_range = "192.168.1.0/24"  # Replace with your network IP range

    arp_request = scapy.ARP(pdst=ip_range)
    broadcast = scapy.Ether(dst="ff:ff:ff:ff:ff:ff")
    arp_request_broadcast = broadcast/arp_request
    answered_list = scapy.srp(arp_request_broadcast, timeout=1, verbose=False)[0]

    for item in answered_list:
        mac_address = item[1].hwsrc
        ip_address = item[1].psrc
        device = {"ip": ip_address, "mac": mac_address, "status": "online"}
        devices.append(device)

    # Save devices information to JSON file
    with open(JSON_FILE, 'w') as f:
        json.dump(devices, f)

    return devices

def send_webhook(devices):
    data = {"devices": devices}
    headers = {'Content-type': 'application/json'}

    response = requests.post(WEBHOOK_URL, data=json.dumps(data), headers=headers)

    if response.status_code == 200:
        print("Webhook sent successfully!")
    else:
        print(f"Failed to send webhook. Status code: {response.status_code}")

def check_internet():
    while True:
        try:
            response = requests.get('https://www.google.com')
            if response.status_code == 200:
                print("Internet is online.")
                with open(BUFFER_FILE, 'r') as f:
                    data = json.load(f)
                    if data:
                        send_webhook(data)
                        with open(BUFFER_FILE, 'w') as bf:
                            json.dump([], bf)
            time.sleep(60)
        except:
            print("Internet is offline.")
            time.sleep(10)

def buffer_data(devices):
    with open(BUFFER_FILE, 'r') as f:
        data = json.load(f)
        data.extend(devices)
    with open(BUFFER_FILE, 'w') as bf:
        json.dump(data, bf)

def run():
    devices = scan_network()
    buffer_data(devices)
    print(f"{len(devices)} devices found.")
    threading.Timer(5.0, run).start()

if __name__ == "__main__":
    print("Starting network scanner...")
    threading.Timer(0, check_internet).start()
    run()