| 1 | #!/usr/bin/python3 |
| 2 | # Copyright (C) 2025, SOCFortress LLC. |
| 3 | # All rights reserved. |
| 4 | |
| 5 | # This program is free software; you can redistribute it |
| 6 | # and/or modify it under the terms of the GNU General Public |
| 7 | # License (version 2) as published by the FSF - Free Software |
| 8 | # Foundation. |
| 9 | |
| 10 | import datetime |
| 11 | import json |
| 12 | import os |
| 13 | import re |
| 14 | import subprocess |
| 15 | import sys |
| 16 | |
| 17 | LOG_FILE = "/var/ossec/logs/active-responses.log" |
| 18 | HOSTS_FILE = "/etc/hosts" |
| 19 | SINKHOLE_IP = "127.0.0.1" # Loopback address for sinkholing |
| 20 | |
| 21 | COMMANDS = {"add": 0, "delete": 1, "continue": 2, "abort": 3} |
| 22 | |
| 23 | OS_SUCCESS = 0 |
| 24 | OS_INVALID = -1 |
| 25 | |
| 26 | |
| 27 | class Message: |
| 28 | def __init__(self, alert="", command=0): |
| 29 | self.alert = alert |
| 30 | self.command = command |
| 31 | |
| 32 | |
| 33 | def write_debug_file(ar_name, msg): |
| 34 | """Writes a debug message to the log file.""" |
| 35 | with open(LOG_FILE, mode="a") as log_file: |
| 36 | log_msg = { |
| 37 | "timestamp": datetime.datetime.now().strftime("%Y/%m/%d %H:%M:%S"), |
| 38 | "active_response": "domain_sinkhole", |
| 39 | "message": json.loads(msg) if isinstance(msg, str) and msg.strip().startswith("{") else msg, |
| 40 | } |
| 41 | log_file.write(json.dumps(log_msg) + "\n") |
| 42 | |
| 43 | |
| 44 | def setup_and_check_message(argv): |
| 45 | """Reads and validates the input message.""" |
| 46 | input_str = next(sys.stdin, "") |
| 47 | write_debug_file(argv[0], input_str) |
| 48 | try: |
| 49 | data = json.loads(input_str) |
| 50 | except ValueError: |
| 51 | write_debug_file(argv[0], "Decoding JSON has failed, invalid input format") |
| 52 | return Message(command=OS_INVALID) |
| 53 | command = COMMANDS.get(data.get("command"), OS_INVALID) |
| 54 | if command == OS_INVALID: |
| 55 | write_debug_file(argv[0], "Not valid command: " + data.get("command")) |
| 56 | return Message(alert=data, command=command) |
| 57 | |
| 58 | |
| 59 | def is_valid_domain(domain): |
| 60 | """Checks if a domain name is valid.""" |
| 61 | # Basic domain validation regex |
| 62 | pattern = r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)+([A-Za-z]|[A-Za-z][A-Za-z0-9\-]*[A-Za-z0-9])$" |
| 63 | return bool(re.match(pattern, domain)) |
| 64 | |
| 65 | |
| 66 | def sinkhole_domain(domain): |
| 67 | """Adds a domain to /etc/hosts pointing to loopback.""" |
| 68 | try: |
| 69 | # Check if domain already exists in hosts file |
| 70 | with open(HOSTS_FILE, "r") as f: |
| 71 | hosts_content = f.read() |
| 72 | |
| 73 | if f"{SINKHOLE_IP} {domain}" in hosts_content: |
| 74 | return f"Domain {domain} is already sinkholed" |
| 75 | |
| 76 | # Add the domain to hosts file |
| 77 | with open(HOSTS_FILE, "a") as f: |
| 78 | f.write(f"\n{SINKHOLE_IP} {domain} # Added by SOCFortress sinkhole\n") |
| 79 | |
| 80 | # Flush DNS cache if dnsmasq is running |
| 81 | try: |
| 82 | subprocess.run(["systemctl", "restart", "dnsmasq"], check=False) |
| 83 | except (subprocess.SubprocessError, FileNotFoundError): |
| 84 | # More specific exceptions for when systemctl doesn't exist or fails |
| 85 | pass # It's okay if dnsmasq isn't installed |
| 86 | |
| 87 | return f"Sinkholed domain {domain} to {SINKHOLE_IP}" |
| 88 | except Exception as e: |
| 89 | return f"Failed to sinkhole domain {domain}: {str(e)}" |
| 90 | |
| 91 | |
| 92 | def remove_sinkholed_domain(domain): |
| 93 | """Removes a domain from the /etc/hosts file.""" |
| 94 | try: |
| 95 | # Read hosts file |
| 96 | with open(HOSTS_FILE, "r") as f: |
| 97 | hosts_lines = f.readlines() |
| 98 | |
| 99 | # Filter out the domain entry |
| 100 | new_hosts = [line for line in hosts_lines if not (domain in line and "Added by SOCFortress sinkhole" in line)] |
| 101 | |
| 102 | # Write back the file without the domain |
| 103 | with open(HOSTS_FILE, "w") as f: |
| 104 | f.writelines(new_hosts) |
| 105 | |
| 106 | # Flush DNS cache if dnsmasq is running |
| 107 | try: |
| 108 | subprocess.run(["systemctl", "restart", "dnsmasq"], check=False) |
| 109 | except (subprocess.SubprocessError, FileNotFoundError): |
| 110 | # More specific exceptions for when systemctl doesn't exist or fails |
| 111 | pass # It's okay if dnsmasq isn't installed |
| 112 | |
| 113 | return f"Removed sinkholed domain {domain}" |
| 114 | except Exception as e: |
| 115 | return f"Failed to remove sinkholed domain {domain}: {str(e)}" |
| 116 | |
| 117 | |
| 118 | def extract_alert_info(msg, argv): |
| 119 | """Extracts the action and domain from the alert message.""" |
| 120 | try: |
| 121 | alert = msg.alert["parameters"]["alert"] |
| 122 | action = alert.get("action", "sinkhole") # Default to block if not specified |
| 123 | domain = alert.get("value") |
| 124 | |
| 125 | if not domain: |
| 126 | write_debug_file(argv[0], "No domain specified in alert") |
| 127 | sys.exit(OS_INVALID) |
| 128 | except KeyError as e: |
| 129 | write_debug_file(argv[0], f"Missing key in alert message: {str(e)}") |
| 130 | sys.exit(OS_INVALID) |
| 131 | return action, domain |
| 132 | |
| 133 | |
| 134 | def main(argv): |
| 135 | write_debug_file(argv[0], {"status": "Started"}) |
| 136 | |
| 137 | # Check if running as root (required to modify /etc/hosts) |
| 138 | if os.geteuid() != 0: |
| 139 | write_debug_file(argv[0], {"status": "failed", "message": "This script must run as root to modify /etc/hosts"}) |
| 140 | sys.exit(OS_INVALID) |
| 141 | |
| 142 | msg = setup_and_check_message(argv) |
| 143 | if msg.command < 0: |
| 144 | sys.exit(OS_INVALID) |
| 145 | |
| 146 | if msg.command == COMMANDS["add"]: |
| 147 | action, domain = extract_alert_info(msg, argv) |
| 148 | |
| 149 | if not is_valid_domain(domain): |
| 150 | write_debug_file(argv[0], {"status": "failed", "message": f"Invalid domain name: {domain}"}) |
| 151 | sys.exit(OS_INVALID) |
| 152 | |
| 153 | if action == "sinkhole": # Handle both "block" and "sinkhole" actions |
| 154 | result = sinkhole_domain(domain) |
| 155 | write_debug_file(argv[0], result) |
| 156 | elif action == "remove_sinkhole": |
| 157 | result = remove_sinkholed_domain(domain) |
| 158 | write_debug_file(argv[0], result) |
| 159 | else: |
| 160 | write_debug_file(argv[0], f"Unknown action: {action}") |
| 161 | |
| 162 | elif msg.command == COMMANDS["delete"]: |
| 163 | # The delete command could be used to clean up any persistent changes |
| 164 | # For this implementation, we don't need any cleanup |
| 165 | pass |
| 166 | |
| 167 | else: |
| 168 | write_debug_file(argv[0], "Invalid command") |
| 169 | |
| 170 | write_debug_file(argv[0], "Ended") |
| 171 | sys.exit(OS_SUCCESS) |
| 172 | |
| 173 | |
| 174 | if __name__ == "__main__": |
| 175 | main(sys.argv) |