Python script to block malicious IPs (CYBERDUDEBIVASH EDITION)

CYBERDUDEBIVASH

 Daily Threat Intel by CyberDudeBivash
Zero-days, exploit breakdowns, IOCs, detection rules & mitigation playbooks.

Follow on LinkedInApps & Security ToolsCYBERDUDEBIVASH PVT LTD WWW.CYBERDUDEBIVASH.COM

#!/usr/bin/env python3

“””

CYBERDUDEBIVASH-Shield: Malicious IP Blocker (Defensive)

– Windows: Adds Windows Firewall inbound+outbound block rules via PowerShell

– Linux: Uses UFW if available, else iptables

– Safety: allowlist, private ranges protected, validation, dry-runaudit logs

Usage:

  python block_ips.py –ips 1.2.3.4 5.6.7.0/24 –name “CYBERDUDEBIVASH-Block” –dry-run

  python block_ips.py –file malicious_ips.txt

“””

import argparse

import datetime as dt

import ipaddress

import os

import platform

import shutil

import subprocess

import sys

from typing import List, Tuple, Set

DEFAULT_RULE_PREFIX = “CYBERDUDEBIVASH-Block”

DEFAULT_LOG = “cyberdudebivash_ipblock_audit.log”

# Protect these ranges by default (never block)

PROTECTED_RANGES = [

    “127.0.0.0/8”,     # loopback

    “10.0.0.0/8”,      # private

    “172.16.0.0/12”,   # private

    “192.168.0.0/16”,  # private

    “169.254.0.0/16”,  # link-local

    “::1/128”,         # IPv6 loopback

    “fe80::/10”,       # IPv6 link-local

    “fc00::/7”,        # IPv6 unique local

]

def log_line(path: str, line: str) -> None:

    ts = dt.datetime.utcnow().strftime(“%Y-%m-%dT%H:%M:%SZ”)

    with open(path, “a”, encoding=”utf-8″) as f:

        f.write(f”{ts} {line}\n”)

def run(cmd: List[str], dry_run: bool, log_path: str) -> Tuple[int, str]:

    pretty = ” “.join(cmd)

    log_line(log_path, f”CMD {pretty}”)

    if dry_run:

        return 0, f”[DRY-RUN] {pretty}”

    try:

        cp = subprocess.run(cmd, capture_output=True, text=True, check=False)

        out = (cp.stdout or “”) + (cp.stderr or “”)

        out = out.strip()

        log_line(log_path, f”RC {cp.returncode} OUT {out[:500]}”)

        return cp.returncode, out

    except Exception as e:

        log_line(log_path, f”ERR {pretty} EXC {repr(e)}”)

        return 1, str(e)

def is_admin_windows() -> bool:

    if platform.system().lower() != “windows”:

        return False

    try:

        import ctypes  # type: ignore

        return bool(ctypes.windll.shell32.IsUserAnAdmin())

    except Exception:

        return False

def is_root_linux() -> bool:

    return hasattr(os, “geteuid”) and os.geteuid() == 0

def parse_ip_targets(raw: List[str]) -> List[str]:

    “””

    Accepts single IPs or CIDRs, validates, normalizes.

    Returns normalized string representations.

    “””

    normalized = []

    for item in raw:

        item = item.strip()

        if not item or item.startswith(“#”):

            continue

        try:

            # Try network first (CIDR), else single IP

            if “/” in item:

                net = ipaddress.ip_network(item, strict=False)

                normalized.append(str(net))

            else:

                ip = ipaddress.ip_address(item)

                normalized.append(str(ip))

        except ValueError:

            raise ValueError(f”Invalid IP/CIDR: {item}”)

    return normalized

def build_protected_set() -> List[ipaddress._BaseNetwork]:  # type: ignore

    return [ipaddress.ip_network(r, strict=False) for r in PROTECTED_RANGES]

def is_protected(target: str, protected_nets: List[ipaddress._BaseNetwork]) -> bool:  # type: ignore

    if “/” in target:

        net = ipaddress.ip_network(target, strict=False)

        return any(net.subnet_of(p) or net.overlaps(p) for p in protected_nets)

    ip = ipaddress.ip_address(target)

    return any(ip in p for p in protected_nets)

def filter_targets(

    targets: List[str],

    allowlist: Set[str],

    protected_nets: List[ipaddress._BaseNetwork]  # type: ignore

) -> Tuple[List[str], List[str]]:

    “””

    Returns: (safe_to_block, skipped)

    “””

    safe = []

    skipped = []

    for t in targets:

        if t in allowlist:

            skipped.append(f”{t} (allowlisted)”)

            continue

        if is_protected(t, protected_nets):

            skipped.append(f”{t} (protected range)”)

            continue

        safe.append(t)

    return safe, skipped

# —————- Windows firewall —————-

def block_windows_firewall(targets: List[str], rule_prefix: str, dry_run: bool, log_path: str) -> None:

    if not is_admin_windows():

        raise PermissionError(“Windows firewall changes require Administrator privileges.”)

    # One rule per direction, combined RemoteAddress list (but keep reasonably sized).

    # Windows has practical limits, so chunk if large.

    chunk_size = 200  # safe chunk size for remote addresses

    def chunks(lst, n):

        for i in range(0, len(lst), n):

            yield lst[i:i+n]

    for idx, chunk in enumerate(chunks(targets, chunk_size), start=1):

        addr_list = “,”.join(chunk)

        in_name = f”{rule_prefix}-IN-{idx}”

        out_name = f”{rule_prefix}-OUT-{idx}”

        # Inbound block

        cmd_in = [

            “powershell”, “-NoProfile”, “-ExecutionPolicy”, “Bypass”,

            “-Command”,

            f”New-NetFirewallRule -DisplayName ‘{in_name}’ -Direction Inbound -Action Block -RemoteAddress ‘{addr_list}’ -Profile Any”

        ]

        rc, out = run(cmd_in, dry_run, log_path)

        if rc != 0 and “already exists” not in (out or “”).lower():

            raise RuntimeError(f”Failed to add inbound rule: {in_name} :: {out}”)

        # Outbound block

        cmd_out = [

            “powershell”, “-NoProfile”, “-ExecutionPolicy”, “Bypass”,

            “-Command”,

            f”New-NetFirewallRule -DisplayName ‘{out_name}’ -Direction Outbound -Action Block -RemoteAddress ‘{addr_list}’ -Profile Any”

        ]

        rc, out = run(cmd_out, dry_run, log_path)

        if rc != 0 and “already exists” not in (out or “”).lower():

            raise RuntimeError(f”Failed to add outbound rule: {out_name} :: {out}”)

# —————- Linux firewall —————-

def have_cmd(name: str) -> bool:

    return shutil.which(name) is not None

def block_linux_ufw(targets: List[str], dry_run: bool, log_path: str) -> None:

    if not is_root_linux():

        raise PermissionError(“Linux firewall changes require root privileges.”)

    # Ensure ufw enabled if you want persistence; we only add rules.

    for t in targets:

        # UFW supports “ufw deny from X” and “ufw deny out to X” variations.

        # We’ll add both inbound + outbound blocks for parity with Windows.

        cmd_in = [“ufw”, “deny”, “from”, t, “to”, “any”]

        rc, out = run(cmd_in, dry_run, log_path)

        if rc != 0 and “Skipping” not in (out or “”):

            raise RuntimeError(f”UFW inbound rule failed for {t}: {out}”)

        cmd_out = [“ufw”, “deny”, “out”, “to”, t]

        rc, out = run(cmd_out, dry_run, log_path)

        if rc != 0 and “Skipping” not in (out or “”):

            raise RuntimeError(f”UFW outbound rule failed for {t}: {out}”)

def block_linux_iptables(targets: List[str], dry_run: bool, log_path: str) -> None:

    if not is_root_linux():

        raise PermissionError(“Linux firewall changes require root privileges.”)

    # NOTE: iptables may not persist across reboot unless you save rules (distro-specific).

    for t in targets:

        # Input chain: drop packets from source

        cmd_in = [“iptables”, “-I”, “INPUT”, “-s”, t, “-j”, “DROP”]

        rc, out = run(cmd_in, dry_run, log_path)

        if rc != 0:

            raise RuntimeError(f”iptables INPUT rule failed for {t}: {out}”)

        # Output chain: drop packets to destination

        cmd_out = [“iptables”, “-I”, “OUTPUT”, “-d”, t, “-j”, “DROP”]

        rc, out = run(cmd_out, dry_run, log_path)

        if rc != 0:

            raise RuntimeError(f”iptables OUTPUT rule failed for {t}: {out}”)

def block_linux(targets: List[str], dry_run: bool, log_path: str) -> None:

    if have_cmd(“ufw”):

        log_line(log_path, “INFO using UFW”)

        block_linux_ufw(targets, dry_run, log_path)

    elif have_cmd(“iptables”):

        log_line(log_path, “INFO using iptables”)

        block_linux_iptables(targets, dry_run, log_path)

    else:

        raise EnvironmentError(“No supported Linux firewall tool found (ufw or iptables).”)

# —————- Main —————-

def read_file_lines(path: str) -> List[str]:

    with open(path, “r”, encoding=”utf-8″) as f:

        return [line.strip() for line in f.readlines()]

def main():

    ap = argparse.ArgumentParser(description=”CYBERDUDEBIVASH-Shield: block malicious IPs safely”)

    ap.add_argument(“–ips”, nargs=”*”, default=[], help=”IP(s) or CIDR(s) to block”)

    ap.add_argument(“–file”, default=””, help=”File containing IPs/CIDRs (one per line)”)

    ap.add_argument(“–allow”, nargs=”*”, default=[], help=”Allowlist IPs/CIDRs to NEVER block”)

    ap.add_argument(“–name”, default=DEFAULT_RULE_PREFIX, help=”Rule name prefix (Windows)”)

    ap.add_argument(“–log”, default=DEFAULT_LOG, help=”Audit log path”)

    ap.add_argument(“–dry-run”, action=”store_true”, help=”Print actions without applying”)

    args = ap.parse_args()

    raw = list(args.ips)

    if args.file:

        raw.extend(read_file_lines(args.file))

    if not raw:

        print(“No IPs provided. Use –ips or –file.”)

        sys.exit(2)

    try:

        targets = parse_ip_targets(raw)

    except ValueError as e:

        print(f”Input error: {e}”)

        sys.exit(2)

    allowlist = set(parse_ip_targets(args.allow)) if args.allow else set()

    protected_nets = build_protected_set()

    safe, skipped = filter_targets(targets, allowlist, protected_nets)

    log_line(args.log, f”INFO start platform={platform.system()} dry_run={args.dry_run} total={len(targets)} safe={len(safe)} skipped={len(skipped)}”)

    if skipped:

        print(“Skipped (safety/allowlist):”)

        for s in skipped:

            print(f” – {s}”)

    if not safe:

        print(“Nothing to block after safety filtering.”)

        sys.exit(0)

    print(f”\nWill block {len(safe)} target(s):”)

    for t in safe[:20]:

        print(f” – {t}”)

    if len(safe) > 20:

        print(f” … and {len(safe) – 20} more”)

    system = platform.system().lower()

    try:

        if system == “windows”:

            if not is_admin_windows():

                raise PermissionError(“Run PowerShell/Terminal as Administrator.”)

            block_windows_firewall(safe, args.name, args.dry_run, args.log)

        elif system == “linux”:

            if not is_root_linux():

                raise PermissionError(“Run as root (sudo).”)

            block_linux(safe, args.dry_run, args.log)

        else:

            raise NotImplementedError(f”Unsupported OS: {platform.system()}”)

    except Exception as e:

        log_line(args.log, f”FAIL {repr(e)}”)

        print(f”\nERROR: {e}”)

        print(f”Audit log: {args.log}”)

        sys.exit(1)

    log_line(args.log, “OK completed”)

    print(“\nDone. Rules applied (or simulated in dry-run).”)

    print(f”Audit log: {args.log}”)

if __name__ == “__main__”:

    main()

#cyberdudebivash #CyberDudeBivash #CYBERDUDEBIVASHShield #MaliciousIP #IPBlocking #FirewallAutomation #WindowsFirewall #LinuxFirewall #UFW #IPTables #ThreatIntel #ThreatHunting #IncidentResponse #SOC #BlueTeam #DetectionEngineering #CyberDefense #NetworkSecurity #EDR #SIEM #PythonSecurity #SecurityAutomation #DevSecOps #ZeroTrust #OWASP #DefensiveSecurity #Blocklist #IOC #CyberSecurity

Leave a comment

Design a site like this with WordPress.com
Get started