Add files via upload

Author: zebbern

Sniff.py

@@ -0,0 +1,117 @@
+import time, sys, signal, psutil, pyperclip
+from scapy.all import sniff, UDP, IP
+from collections import defaultdict
+from colorama import init, Fore, Style
+
+init(autoreset=True)
+
+INTERVAL = 1
+THRESHOLD = 250
+LOG = 'udp_monitor_log.txt'
+TARGET_PROCS = {'cod.exe'}
+
+def cprint(msg, color='green'):
+    colors = {'green': Fore.GREEN, 'cyan': Fore.CYAN, 'yellow': Fore.YELLOW, 'red': Fore.RED}
+    print(f"{colors.get(color, Fore.WHITE)}{msg}{Style.RESET_ALL}")
+
+def log_line(txt):
+    with open(LOG, 'a', encoding='utf-8') as f: 
+        f.write(txt + "\n")
+
+def run_start():
+    t = time.strftime("%Y-%m-%d %H:%M:%S")
+    log_line("\n=====================================")
+    log_line(f"=== RUN START: {t} ===")
+    log_line("=====================================")
+
+def run_end():
+    t = time.strftime("%Y-%m-%d %H:%M:%S")
+    log_line("=====================================")
+    log_line(f"=== RUN END: {t} ===")
+    log_line("=====================================")
+
+def sigint_handler(sig, frame):
+    run_end()
+    cprint("\nMonitoring canceled by user.\n", 'red')
+    sys.exit(0)
+
+signal.signal(signal.SIGINT, sigint_handler)
+
+def get_ports(proc_names):
+    ports = set()
+    pnames = {n.lower() for n in proc_names}
+    for p in psutil.process_iter(['name']):
+        try:
+            if p.info['name'] and p.info['name'].lower() in pnames:
+                for c in p.net_connections(kind='udp'):
+                    if c.laddr and c.laddr.port:
+                        ports.add(c.laddr.port)
+        except:
+            pass
+    return ports
+
+def get_interface():
+    ifaces = psutil.net_io_counters(pernic=True)
+    if not ifaces: 
+        cprint("No network interfaces found.", 'red')
+        sys.exit(1)
+    return max(ifaces, key=lambda i: ifaces[i].bytes_sent + ifaces[i].bytes_recv)
+
+def sniff_udp(tports, iface, dur):
+    data = defaultdict(int)
+    def cb(pkt):
+        if UDP in pkt and IP in pkt:
+            l = len(pkt)
+            s_ip, d_ip = pkt[IP].src, pkt[IP].dst
+            s_pt, d_pt = pkt[UDP].sport, pkt[UDP].dport
+            if s_pt in tports: data[(d_ip, d_pt)] += l
+            elif d_pt in tports: data[(s_ip, s_pt)] += l
+    flt = " or ".join(f"udp port {p}" for p in tports) if tports else "udp port 0"
+    sniff(filter=flt, prn=cb, timeout=dur, iface=iface, store=False)
+    return data
+
+def report(d, seen, sec, thr):
+    for (ip, port), bcount in d.items():
+        rate = bcount / sec
+        if rate >= thr and (ip, port) not in seen:
+            t = time.strftime("%Y-%m-%d %H:%M:%S")
+            
+            # Copy to clipboard, e.g.,: udp 12.34.56.78 120 dport=1234
+            clip_text = f"udp {ip} 120 dport={port}"
+            pyperclip.copy(clip_text)
+
+            if port == 44998:
+                cprint("=====================================", 'red')
+                cprint("Rotating IP found", 'red')
+                cprint(f"IP Address   : {ip}", 'yellow')
+                cprint(f"Port         : {port}", 'yellow')
+                cprint(f"Data Rate    : {rate:.2f} B/s", 'green')
+                cprint("=====================================", 'red')
+                log_line(f"{t} | Rotating IP found | IP={ip}, Port={port}, Rate={rate:.2f} B/s")
+            else:
+                cprint("=====================================", 'green')
+                cprint("=== Live Game Session Found ===", 'cyan')
+                cprint(f"IP Address   : {ip}", 'yellow')
+                cprint(f"Port         : {port}", 'yellow')
+                cprint(f"Data Rate    : {rate:.2f} B/s", 'green')
+                cprint("=====================================", 'green')
+                log_line(f"{t} | Live Game Session Found | IP={ip}, Port={port}, Rate={rate:.2f} B/s")
+
+            seen.add((ip, port))
+
+def main():
+    cprint("Starting UDP Traffic Monitor...", 'green')
+    run_start()
+    iface = get_interface()
+    seen = set()
+    while True:
+        tports = get_ports(TARGET_PROCS)
+        if not tports:
+            cprint("No target UDP ports found for specified processes.", 'yellow')
+            time.sleep(INTERVAL)
+            continue
+        data = sniff_udp(tports, iface, INTERVAL)
+        report(data, seen, INTERVAL, THRESHOLD)
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,4 @@
+scapy
+psutil
+colorama
+pyperclip

run.bat

@@ -0,0 +1,2 @@
+@echo off
+python Sniff.py