#!/opt/imunify360/venv/bin/python3 -u """Notify the Go resident agent to synchronize ipsets. Used as a CSF post-hook script: after CSF restarts, ipsets need to be restored. Sends IPSET_SYNC RPC to the resident agent which handles the full synchronization (restore outdated ipsets, consistency check, rule updates). """ import json import socket import sys SOCKET_PATH = "/var/run/defence360agent/generic_sensor.sock.2" TIMEOUT = 30 def main() -> None: with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(TIMEOUT) sock.connect(SOCKET_PATH) msg = json.dumps({"method": "IPSET_SYNC"}).encode() + b"\n" sock.sendall(msg) if __name__ == "__main__": try: main() except (ConnectionRefusedError, FileNotFoundError): pass except socket.timeout: pass except Exception as exc: print(f"ipset_sync: {exc}", file=sys.stderr)