netmon.py

#!/usr/bin/env python3
"""
rpi_netmon.py
Raspberry Pi Network Monitoring System Indonesia
Single-file monitor:
 - Ping sweep (discovery)
 - Port checks (concurrency)
 - Bandwidth monitoring per network interface
 - SQLite state & event log
 - Minimal HTTP JSON status endpoint (aiohttp)
 - Telegram alerts (optional)

Dependencies:
 pip install psutil aiohttp aiosqlite requests

Run:
 python3 rpi_netmon.py
"""

import asyncio
import aiosqlite
import socket
import time
import ipaddress
import psutil
import requests
from aiohttp import web
from datetime import datetime

# -------------------------
# CONFIGURATION - edit this
# -------------------------
CONFIG = {
    "ip_ranges": ["192.168.1.0/24"],     # list of CIDRs to scan
    "scan_interval": 60,                 # seconds between full sweeps
    "port_check": [22, 80, 443, 3389],   # ports to check on discovered hosts
    "port_timeout": 1.5,                 # seconds per port connect attempt
    "concurrency": 200,                  # concurrency for ping/port tasks
    "bandwidth_check_interval": 5,       # seconds to sample interface counters
    "bandwidth_threshold_bytes_s": 2_000_000,  # alert if > ~2MB/s
    "telegram_bot_token": "",            # fill if you want alerts
    "telegram_chat_id": "",              # fill chat id
    "http_listen": ("0.0.0.0", 8080),    # status HTTP server
    "database": "netmon.db",
    "hostname": "RPI-NETMON-ID",
}
# -------------------------

# Basic utilities
def now_ts():
    return int(time.time())

def ts_human(ts=None):
    if ts is None:
        ts = now_ts()
    return datetime.fromtimestamp(ts).isoformat(sep=' ', timespec='seconds')

# Telegram helper (simple)
def telegram_send(text: str):
    token = CONFIG["telegram_bot_token"]
    chat = CONFIG["telegram_chat_id"]
    if not token or not chat:
        return False
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = {"chat_id": chat, "text": text}
    try:
        r = requests.post(url, json=payload, timeout=5)
        return r.ok
    except Exception as e:
        print("Telegram send error:", e)
        return False

# Database init
async def init_db(dbpath):
    db = await aiosqlite.connect(dbpath)
    await db.execute("""
    CREATE TABLE IF NOT EXISTS hosts (
        ip TEXT PRIMARY KEY,
        last_seen INTEGER,
        is_up INTEGER,
        last_change INTEGER
    )
    """)
    await db.execute("""
    CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts INTEGER,
        ip TEXT,
        type TEXT,
        detail TEXT
    )
    """)
    await db.commit()
    return db

# Ping using system ping (cross-platform-ish)
async def ping_host(ip, timeout=1.0):
    # Use simple socket connect on port 80? That would fail if host blocks ports.
    # Instead, call system ping for reliability (requires ping in PATH).
    # Use different flags for Windows vs Linux — on Pi it's Linux.
    proc = await asyncio.create_subprocess_exec(
        "ping", "-c", "1", "-W", str(int(timeout)), str(ip),
        stdout=asyncio.subprocess.DEVNULL,
        stderr=asyncio.subprocess.DEVNULL
    )
    try:
        await asyncio.wait_for(proc.wait(), timeout=timeout+1)
    except asyncio.TimeoutError:
        proc.kill()
        return False
    return proc.returncode == 0

# Async port check via open_connection
async def check_port(ip, port, timeout):
    try:
        reader, writer = await asyncio.wait_for(asyncio.open_connection(ip, port), timeout=timeout)
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass
        return True
    except Exception:
        return False

# Discovery: iterate addresses in CIDR and ping
async def discover_ips(cidr_list, semaphore, timeout=1.0):
    ips = []
    for cidr in cidr_list:
        try:
            net = ipaddress.ip_network(cidr, strict=False)
            for ip in net.hosts():
                ips.append(str(ip))
        except Exception as e:
            print("Invalid CIDR:", cidr, e)
    results = {}

    async def worker(ip):
        async with semaphore:
            ok = await ping_host(ip, timeout)
            results[ip] = ok

    tasks = [asyncio.create_task(worker(ip)) for ip in ips]
    await asyncio.gather(*tasks)
    return results  # dict ip -> bool

# Host state update & event generation
async def update_host_state(db, ip, is_up):
    async with db.execute("SELECT is_up FROM hosts WHERE ip = ?", (ip,)) as cur:
        row = await cur.fetchone()
    ts = now_ts()
    if row is None:
        # insert
        await db.execute("INSERT INTO hosts (ip, last_seen, is_up, last_change) VALUES (?, ?, ?, ?)",
                         (ip, ts if is_up else None, int(is_up), ts))
        if is_up:
            await db.execute("INSERT INTO events (ts, ip, type, detail) VALUES (?, ?, ?, ?)",
                             (ts, ip, "up", "first_seen"))
            telegram_send(f"[{CONFIG['hostname']}] Host UP: {ip} at {ts_human(ts)}")
    else:
        prev = row[0]
        if prev != int(is_up):
            # state changed
            await db.execute("UPDATE hosts SET is_up = ?, last_seen = ?, last_change = ? WHERE ip = ?",
                             (int(is_up), ts if is_up else None, ts, ip))
            ev_type = "up" if is_up else "down"
            await db.execute("INSERT INTO events (ts, ip, type, detail) VALUES (?, ?, ?, ?)",
                             (ts, ip, ev_type, "state_change"))
            telegram_send(f"[{CONFIG['hostname']}] Host {ev_type.upper()}: {ip} at {ts_human(ts)}")
        else:
            if is_up:
                await db.execute("UPDATE hosts SET last_seen = ? WHERE ip = ?", (ts, ip))
    await db.commit()

# Port scanning for a host (returns dict port->bool)
async def scan_ports_for_host(ip, ports, timeout, semaphore):
    results = {}
    async def worker(port):
        async with semaphore:
            ok = await check_port(ip, port, timeout)
            results[port] = ok
    tasks = [asyncio.create_task(worker(p)) for p in ports]
    await asyncio.gather(*tasks)
    return results

# Bandwidth monitor using psutil
class BandwidthMonitor:
    def __init__(self, interfaces=None):
        # interfaces: list of interface names, None = monitor all
        self.ifaces = interfaces
        self.last = psutil.net_io_counters(pernic=True)
        self.last_ts = time.time()

    def sample(self):
        cur = psutil.net_io_counters(pernic=True)
        now = time.time()
        res = {}
        for nic, counters in cur.items():
            if self.ifaces and nic not in self.ifaces:
                continue
            prev = self.last.get(nic)
            if prev:
                dt = now - self.last_ts
                if dt <= 0:
                    continue
                bytes_sent_s = (counters.bytes_sent - prev.bytes_sent) / dt
                bytes_recv_s = (counters.bytes_recv - prev.bytes_recv) / dt
            else:
                bytes_sent_s = 0.0
                bytes_recv_s = 0.0
            res[nic] = {"tx_B_s": int(bytes_sent_s), "rx_B_s": int(bytes_recv_s)}
        self.last = cur
        self.last_ts = now
        return res

# HTTP API server handlers
async def make_app(db, host_cache):
    app = web.Application()
    routes = web.RouteTableDef()

    @routes.get("/status")
    async def status(req):
        # return host list and bandwidth
        hosts = []
        async with db.execute("SELECT ip, last_seen, is_up, last_change FROM hosts") as cur:
            async for row in cur:
                hosts.append({
                    "ip": row[0],
                    "last_seen": ts_human(row[1]) if row[1] else None,
                    "is_up": bool(row[2]),
                    "last_change": ts_human(row[3]) if row[3] else None
                })
        return web.json_response({"hostname": CONFIG["hostname"], "hosts": hosts, "bandwidth": host_cache.get("bw", {})})

    @routes.get("/events")
    async def events(req):
        limit = int(req.query.get("limit", "50"))
        items = []
        async with db.execute("SELECT ts, ip, type, detail FROM events ORDER BY ts DESC LIMIT ?", (limit,)) as cur:
            async for row in cur:
                items.append({"ts": ts_human(row[0]), "ip": row[1], "type": row[2], "detail": row[3]})
        return web.json_response({"events": items})

    app.add_routes(routes)
    return app

# Main monitor loop
async def monitor_loop():
    db = await init_db(CONFIG["database"])
    sem_scan = asyncio.Semaphore(CONFIG["concurrency"])
    sem_ports = asyncio.Semaphore(int(CONFIG["concurrency"] / 5) or 10)
    bw_monitor = BandwidthMonitor()
    host_cache = {"bw": {}}

    # Launch HTTP server
    app = await make_app(db, host_cache)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, CONFIG["http_listen"][0], CONFIG["http_listen"][1])
    await site.start()
    print(f"HTTP status server listening on http://{CONFIG['http_listen'][0]}:{CONFIG['http_listen'][1]}/status")

    try:
        while True:
            start = time.time()
            print(f"[{ts_human()}] Starting discovery sweep...")
            discovered = await discover_ips(CONFIG["ip_ranges"], sem_scan, timeout=1.2)
            # Update host states & run port scans for up hosts
            ups = [ip for ip, ok in discovered.items() if ok]
            downs = [ip for ip, ok in discovered.items() if not ok]
            # update DB for each
            for ip in ups:
                await update_host_state(db, ip, True)
            for ip in downs:
                await update_host_state(db, ip, False)

            # Port scans (concurrent)
            print(f"Found {len(ups)} hosts up. Scanning ports for top services...")
            port_tasks = []
            for ip in ups:
                task = asyncio.create_task(scan_ports_for_host(ip, CONFIG["port_check"], CONFIG["port_timeout"], sem_ports))
                port_tasks.append((ip, task))
            # gather results and log if interesting
            for ip, task in port_tasks:
                try:
                    ports_res = await task
                except Exception as e:
                    ports_res = {}
                # store event if port unusual? For now, record an event for closed all ports or open uncommon
                open_ports = [p for p, ok in ports_res.items() if ok]
                detail_text = f"open_ports={open_ports}"
                await db.execute("INSERT INTO events (ts, ip, type, detail) VALUES (?, ?, ?, ?)",
                                 (now_ts(), ip, "port-scan", detail_text))
            await db.commit()

            # Bandwidth monitoring quick check (sample twice to compute rates)
            # We'll sample every X seconds for a single interval to compute B/s
            bw_samples = []
            interval = CONFIG["bandwidth_check_interval"]
            bw_samples.append(bw_monitor.sample())
            await asyncio.sleep(interval)
            bw_samples.append(bw_monitor.sample())
            # use second sample as current rates
            cur_bw = bw_samples[-1]
            host_cache["bw"] = cur_bw

            # Alert if any iface exceeds threshold
            for nic, vals in cur_bw.items():
                tx = vals["tx_B_s"]
                rx = vals["rx_B_s"]
                top = max(tx, rx)
                if top >= CONFIG["bandwidth_threshold_bytes_s"]:
                    text = f"[{CONFIG['hostname']}] High bandwidth on {nic}: tx={tx} B/s rx={rx} B/s at {ts_human()}"
                    telegram_send(text)
                    await db.execute("INSERT INTO events (ts, ip, type, detail) VALUES (?, ?, ?, ?)",
                                     (now_ts(), nic, "bw-alert", text))
                    await db.commit()

            elapsed = time.time() - start
            wait = max(1, CONFIG["scan_interval"] - elapsed)
            print(f"[{ts_human()}] Sweep done in {elapsed:.1f}s. sleeping {wait:.1f}s.")
            await asyncio.sleep(wait)
    finally:
        await db.close()
        await runner.cleanup()

# Entrypoint
def main():
    print("Raspberry Pi Network Monitoring System Indonesia")
    print("Config summary:")
    print("  IP ranges:", CONFIG["ip_ranges"])
    print("  Scan interval (s):", CONFIG["scan_interval"])
    print("  Ports to check:", CONFIG["port_check"])
    if CONFIG["telegram_bot_token"] and CONFIG["telegram_chat_id"]:
        print("  Telegram alerts enabled")
    else:
        print("  Telegram alerts disabled (configure token/chat id to enable)")

    try:
        asyncio.run(monitor_loop())
    except KeyboardInterrupt:
        print("Exiting...")

if __name__ == "__main__":
    main()