File size: 2,462 Bytes
18fa92b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
import socket
import os
import logging
import threading
from scapy.all import sniff, IP, TCP, UDP
# Configuration
rules = {"192.168.1.100", "10.0.0.200"} # IPs to block
blocked = {"example.com", "test.net"} # Domains to block
log_path = "events.log"
# Logging setup
logging.basicConfig(filename=log_path, level=logging.INFO,
https://www.google.com/search?q=grok+3&ie=UTF-8&oe=UTF-8&hl=en-ph&client=safari#scso=_BVy2Z4uXJsCovr0P4tq76Aw_111:496 format="%(asctime)s - %(levelname)s - %(message)s")
def log_action(entry):
"""Log an action and print it to the console."""
logging.info(entry)
print(f"[System] {entry}")
def check_rule(item):
"""Check if an IP is in the rules set."""
return item in rules
def check_data(data):
"""Check if any blocked domain is in the data."""
return any(item in data for item in blocked)
def resolve(item):
"""Resolve a domain name to an IP address."""
try:
return socket.gethostbyname(item)
except socket.gaierror:
return None
def analyze(packet):
"""Analyze a network packet and enforce rules."""
if IP in packet:
src = packet[IP].src
dest = packet[IP].dst
# Block traffic to/from restricted IPs
if check_rule(src) or check_rule(dest):
log_action(f"Blocked {src} -> {dest}")
return
# Check payload for blocked domains
if TCP in packet or UDP in packet:
content = bytes(packet[TCP].payload).decode(errors="ignore")
for item in blocked:
if item in content:
log_action(f"Prevented access to {item} from {src}")
return
def restrict(item):
"""Block an IP address using system commands."""
try:
if os.name == "nt":
os.system(f"netsh advfirewall firewall add rule name='Restricted' dir=in action=block remoteip={item}")
else:
os.system(f"iptables -A INPUT -s {item} -j DROP")
log_action(f"Restricted {item}")
except Exception as e:
log_action(f"Failed to restrict {item}: {e}")
def monitor():
"""Start packet sniffing."""
log_action("System initialized.")
sniff(filter="ip", prn=analyze, store=0)
if __name__ == "__main__":
# Run the monitor in a separate thread
monitor_thread = threading.Thread(target=monitor)
monitor_thread.start()
log_action("Monitoring started in a separate thread.")
|