Network Security Operations

# iptables, Linux packet filtering firewall

# ── Default policy: DENY all incoming traffic ─────────────────
sudo iptables -P INPUT DROP
sudo iptables -P FORWARD DROP
sudo iptables -P OUTPUT ACCEPT

# ── Allow established connections ────────────────────────────
sudo iptables -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT

# ── Allow loopback ────────────────────────────────────────────
sudo iptables -A INPUT -i lo -j ACCEPT

# ── Allow specific services ───────────────────────────────────
sudo iptables -A INPUT -p tcp --dport 22 -j ACCEPT    # SSH
sudo iptables -A INPUT -p tcp --dport 80 -j ACCEPT    # HTTP
sudo iptables -A INPUT -p tcp --dport 443 -j ACCEPT   # HTTPS

# ── Rate limiting, brute force protection ────────────────────
# Allow only 3 SSH connection attempts per 60 seconds per IP
sudo iptables -A INPUT -p tcp --dport 22 -m state --state NEW \
    -m recent --set --name SSH
sudo iptables -A INPUT -p tcp --dport 22 -m state --state NEW \
    -m recent --update --seconds 60 --hitcount 4 --name SSH -j DROP

# ── Log dropped packets (for monitoring) ─────────────────────
sudo iptables -A INPUT -j LOG --log-prefix "IPTABLES-DROP: "

# ── Save rules (persist across reboots) ──────────────────────
sudo iptables-save > /etc/iptables/rules.v4

# ── Python: Manage firewall rules ────────────────────────────
import subprocess

def block_ip(ip_address: str) -> bool:
    """Block an IP address via iptables. Returns True if successful."""
    try:
        result = subprocess.run(
            ["sudo", "iptables", "-A", "INPUT", "-s", ip_address, "-j", "DROP"],
            capture_output=True, text=True, timeout=10,
        )
        return result.returncode == 0
    except subprocess.TimeoutExpired:
        return False

def list_blocked_ips() -> list[str]:
    """List all IPs blocked by DROP rules."""
    result = subprocess.run(
        ["sudo", "iptables", "-L", "INPUT", "-n"],
        capture_output=True, text=True
    )
    blocked = []
    for line in result.stdout.split("\n"):
        if "DROP" in line and "all" in line:
            parts = line.split()
            if len(parts) >= 4:
                blocked.append(parts[3])   # Source IP
    return blocked

print("Firewall rules set up with default-deny policy")
print("SSH rate-limited to 3 new connections per 60 seconds")

# Snort/Suricata rule anatomy
# alert tcp any any -> $HOME_NET 22 (
#   msg:"SSH brute force attempt";    ← Alert message
#   flow:stateless;
#   flags:S,12;                        ← SYN flag (new connection)
#   threshold:type both, track by_src, ← Track per source IP
#             count 5, seconds 60;      ← 5+ in 60 seconds
#   classtype:attempted-admin;
#   sid:100001;                        ← Unique rule ID
#   rev:1;
# )

# Common Suricata rules for web security:
# alert http any any -> any any (msg:"SQL Injection"; content:"' OR '1'='1"; sid:100002;)
# alert http any any -> any any (msg:"XSS Attempt"; content:"<script>"; sid:100003;)
# alert http any any -> any any (msg:"Path Traversal"; content:"../"; sid:100004;)

# incident_response.py, Automated containment script
import subprocess
import logging
from datetime import datetime, timezone

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

def quarantine_ip(ip: str) -> bool:
    """Block an IP immediately via iptables."""
    cmd = ["sudo", "iptables", "-I", "INPUT", "1", "-s", ip, "-j", "DROP"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode == 0:
        logging.warning(f"QUARANTINE: Blocked IP {ip}")
        return True
    logging.error(f"Failed to block {ip}: {result.stderr}")
    return False

def send_alert(severity: str, title: str, details: str) -> None:
    """Send security alert to SIEM/Slack/PagerDuty."""
    # In production: call webhook, send email, or push to SIEM
    alert = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "severity": severity,
        "title": title,
        "details": details,
    }
    logging.critical(f"SECURITY ALERT [{severity}]: {title} | {details}")
    # requests.post(SLACK_WEBHOOK, json={"text": f"🚨 {title}
{details}"})

def respond_to_brute_force(ip: str, attempt_count: int) -> None:
    """Automated brute force incident response."""
    logging.info(f"Starting incident response for {ip} ({attempt_count} attempts)")

    # Step 1: Containment, block the IP
    if quarantine_ip(ip):
        send_alert("HIGH", "Brute Force Detected & Blocked",
                   f"IP {ip} blocked after {attempt_count} failed auth attempts")

    # Step 2: Collect evidence
    logging.info("Evidence collected, run: sudo lastb | grep " + ip)
    logging.info("Full packet capture: sudo tcpdump -i eth0 src " + ip + " -w incident.pcap")

    # Step 3: Document
    logging.info(f"Incident documented at {datetime.now(timezone.utc).isoformat()}")

# Simulate incident response
respond_to_brute_force("10.0.0.42", attempt_count=12)