122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
from flask import Flask, render_template, request, jsonify
|
|
from wakeonlan import send_magic_packet
|
|
import json, os, subprocess, re, socket, concurrent.futures
|
|
|
|
app = Flask(__name__)
|
|
DEVICES_FILE = "devices.json"
|
|
|
|
def load_devices():
|
|
if os.path.exists(DEVICES_FILE):
|
|
with open(DEVICES_FILE, "r") as f:
|
|
return json.load(f)
|
|
return []
|
|
|
|
def save_devices(devices):
|
|
with open(DEVICES_FILE, "w") as f:
|
|
json.dump(devices, f, indent=2)
|
|
|
|
def ping(ip):
|
|
"""Ping a device to check if it is online (Windows)."""
|
|
try:
|
|
result = subprocess.run(
|
|
["ping", "-n", "1", "-w", "500", ip],
|
|
capture_output=True, timeout=2
|
|
)
|
|
return result.returncode == 0
|
|
except:
|
|
return False
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
@app.route("/wake/<mac>", methods=["POST"])
|
|
def wake(mac):
|
|
try:
|
|
send_magic_packet(mac)
|
|
return jsonify({"status": "success", "message": "Magic Packet sent!"})
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
|
|
@app.route("/devices", methods=["GET"])
|
|
def get_devices():
|
|
return jsonify(load_devices())
|
|
|
|
@app.route("/devices", methods=["POST"])
|
|
def add_device():
|
|
data = request.json
|
|
devices = load_devices()
|
|
devices.append({
|
|
"name": data["name"],
|
|
"mac": data["mac"],
|
|
"ip": data.get("ip", "")
|
|
})
|
|
save_devices(devices)
|
|
return jsonify({"status": "success"})
|
|
|
|
@app.route("/devices/<int:idx>", methods=["DELETE"])
|
|
def delete_device(idx):
|
|
devices = load_devices()
|
|
if 0 <= idx < len(devices):
|
|
devices.pop(idx)
|
|
save_devices(devices)
|
|
return jsonify({"status": "success"})
|
|
return jsonify({"status": "error"}), 404
|
|
|
|
@app.route("/status", methods=["GET"])
|
|
def get_status():
|
|
"""Check online status of all devices in parallel."""
|
|
devices = load_devices()
|
|
def check(d):
|
|
ip = d.get("ip", "")
|
|
if not ip:
|
|
return None
|
|
return ping(ip)
|
|
with concurrent.futures.ThreadPoolExecutor() as executor:
|
|
results = list(executor.map(check, devices))
|
|
return jsonify(results)
|
|
|
|
@app.route("/scan", methods=["POST"])
|
|
def scan_network():
|
|
"""Scan the local network for active devices using nmap."""
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
s.connect(("8.8.8.8", 80))
|
|
local_ip = s.getsockname()[0]
|
|
s.close()
|
|
subnet = ".".join(local_ip.split(".")[:3]) + ".0/24"
|
|
|
|
result = subprocess.check_output(
|
|
f"nmap -sn {subnet}",
|
|
shell=True, stderr=subprocess.DEVNULL
|
|
).decode(errors="ignore")
|
|
|
|
arp_output = subprocess.check_output("arp -a", shell=True).decode(errors="ignore")
|
|
|
|
found = []
|
|
nmap_hosts = re.findall(r"Nmap scan report for (.+)\r?\nHost is up", result)
|
|
|
|
for host in nmap_hosts:
|
|
ip_match = re.search(r"\((\d+\.\d+\.\d+\.\d+)\)", host)
|
|
ip = ip_match.group(1) if ip_match else host.strip()
|
|
name = re.sub(r"\s*\(.*?\)", "", host).strip()
|
|
|
|
arp_match = re.search(
|
|
rf"{re.escape(ip)}\s+([0-9a-f]{{2}}(?:[:-][0-9a-f]{{2}}){{5}})",
|
|
arp_output, re.IGNORECASE
|
|
)
|
|
if arp_match:
|
|
mac = arp_match.group(1).replace("-", ":").upper()
|
|
found.append({
|
|
"name": name if name != ip else f"Device ({ip})",
|
|
"mac": mac,
|
|
"ip": ip
|
|
})
|
|
|
|
return jsonify({"status": "success", "devices": found})
|
|
except Exception as e:
|
|
return jsonify({"status": "error", "message": str(e)}), 500
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=5000, debug=True)
|