MISP integration with Wazuh
92 views
Skip to first unread message
Muhammad Ali Khan
Feb 21, 2025, 2:01:56 AM2/21/25
to Wazuh | Mailing List
Dear All
i install MISP 2.4 on my ubuntu machine 22.04 , all done and misp is working fine , but when i run script for integration , error is appear index out of range in alert part , which is attached : kindly check and give your guidance
using this script
#!/var/ossec/framework/python/bin/python3
## MISP API Integration
import sys
import json
import traceback
import os
from socket import socket, AF_UNIX, SOCK_DGRAM
import json
import ipaddress
import requests
from requests.exceptions import ConnectionError
import re
# Ensure the script gets the correct input file
pwd = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
socket_addr = f"{pwd}/queue/sockets/queue"
def send_event(msg, agent=None):
if agent and "id" in agent and "name" in agent:
if agent["id"] == "000":
string = f"1:misp:{json.dumps(msg)}"
else:
string = f"1:[{agent['id']}] ({agent['name']}) {agent.get('ip', 'any')}->misp:{json.dumps(msg)}"
else:
string = f"1:misp:{json.dumps(msg)}"
sock = socket(AF_UNIX, SOCK_DGRAM)
sock.connect(socket_addr)
sock.send(string.encode())
sock.close()
# Read configuration parameters
with open(sys.argv[1]) as alert_file:
alert = [json.loads(line) for line in alert_file if line.strip()]
# Check if alerts are loaded correctly
if not alert:
print("Error: No valid alerts found in the JSON file.")
sys.exit(1)
# Process each alert and check for 'rule' and 'groups'
# Process each alert
for alert in alert: # Make sure you're iterating through 'alerts' (plural), not 'alert'
try:
# ===== SAFE ACCESS PATTERN STARTS HERE =====
# 1. Get groups with nested dictionary safety
groups = alert.get("rule", {}).get("groups", [])
# 2. Validate groups is a list
if not isinstance(groups, list):
print(f"Invalid groups type: {type(groups)}. Expected list.")
continue
# 3. Safe index access with fallback values
event_source = groups[0] if len(groups) > 0 else "unknown"
event_type = groups[1] if len(groups) > 1 else "unknown"
event_category = groups[2] if len(groups) > 2 else "none"
# ===== SAFE ACCESS PATTERN ENDS HERE =====
# Debug print (keep this temporarily)
print(f"Processed: {event_source} | {event_type} | {event_category}")
# Continue with your MISP logic below...
# ... rest of your code ...
except Exception as e:
print(f"Error processing alert: {str(e)}")
traceback.print_exc()
# MISP API Configuration
misp_base_url = "https://172.16.1.237/attributes/restSearch/"
misp_api_auth_key = "Izy9OCQ8xdfggvuy5eyhYSsdvzUAC5kjhjG"
misp_apicall_headers = {
"Content-Type": "application/json",
"Authorization": misp_api_auth_key,
"Accept": "application/json",
}
# Extract Sysmon event details
event_source = alert["rule"]["groups"][0]
event_type = alert["rule"]["groups"][2]
regex_file_hash = re.compile(r"\b[a-fA-F0-9]{64}\b")
wazuh_event_param = None
if event_source == "windows":
if event_type in ["sysmon_event1", "sysmon_event6", "sysmon_event7", "sysmon_event_15", "sysmon_event_23", "sysmon_event_24", "sysmon_event_25"]:
if "data" in alert and "win" in alert["data"] and "eventdata" in alert["data"]["win"] and "hashes" in alert["data"]["win"]["eventdata"]:
match = regex_file_hash.search(alert["data"]["win"]["eventdata"]["hashes"])
if match:
wazuh_event_param = match.group(0)
elif event_type == "sysmon_event3" and alert["data"]["win"]["eventdata"].get("destinationIsIpv6") == "false":
dst_ip = alert["data"]["win"]["eventdata"].get("destinationIp")
if dst_ip and ipaddress.ip_address(dst_ip).is_global:
wazuh_event_param = dst_ip
elif event_type == "sysmon_event_22":
wazuh_event_param = alert["data"]["win"]["eventdata"].get("queryName")
elif event_source == "linux" and event_type == "sysmon_event3" and alert["data"]["eventdata"].get("destinationIsIpv6") == "false":
dst_ip = alert["data"]["eventdata"].get("DestinationIp")
if dst_ip and ipaddress.ip_address(dst_ip).is_global:
wazuh_event_param = dst_ip
elif event_source == "ossec" and event_type == "syscheck_entry_added":
wazuh_event_param = alert.get("syscheck", {}).get("sha256_after")
if wazuh_event_param:
misp_search_value = f"value:{wazuh_event_param}"
misp_search_url = f"{misp_base_url}{misp_search_value}"
try:
misp_api_response = requests.get(misp_search_url, headers=misp_apicall_headers, verify=False)
misp_api_response = misp_api_response.json()
except ConnectionError:
alert_output = {"misp": {"error": "Connection Error to MISP API"}, "integration": "misp"}
send_event(alert_output, alert.get("agent"))
sys.exit()
if "response" in misp_api_response and "Attribute" in misp_api_response["response"] and misp_api_response["response"]["Attribute"]:
attribute = misp_api_response["response"]["Attribute"][0]
alert_output = {
"misp": {
"event_id": attribute.get("event_id"),
"category": attribute.get("category"),
"value": attribute.get("value"),
"type": attribute.get("type"),
"source": {"description": alert["rule"].get("description", "")},
},
"integration": "misp",
}
send_event(alert_output, alert.get("agent"))
else:
sys.exit()
i install MISP 2.4 on my ubuntu machine 22.04 , all done and misp is working fine , but when i run script for integration , error is appear index out of range in alert part , which is attached : kindly check and give your guidance
using this script
#!/var/ossec/framework/python/bin/python3
## MISP API Integration
import sys
import json
import traceback
import os
from socket import socket, AF_UNIX, SOCK_DGRAM
import json
import ipaddress
import requests
from requests.exceptions import ConnectionError
import re
# Ensure the script gets the correct input file
pwd = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
socket_addr = f"{pwd}/queue/sockets/queue"
def send_event(msg, agent=None):
if agent and "id" in agent and "name" in agent:
if agent["id"] == "000":
string = f"1:misp:{json.dumps(msg)}"
else:
string = f"1:[{agent['id']}] ({agent['name']}) {agent.get('ip', 'any')}->misp:{json.dumps(msg)}"
else:
string = f"1:misp:{json.dumps(msg)}"
sock = socket(AF_UNIX, SOCK_DGRAM)
sock.connect(socket_addr)
sock.send(string.encode())
sock.close()
# Read configuration parameters
with open(sys.argv[1]) as alert_file:
alert = [json.loads(line) for line in alert_file if line.strip()]
# Check if alerts are loaded correctly
if not alert:
print("Error: No valid alerts found in the JSON file.")
sys.exit(1)
# Process each alert and check for 'rule' and 'groups'
# Process each alert
for alert in alert: # Make sure you're iterating through 'alerts' (plural), not 'alert'
try:
# ===== SAFE ACCESS PATTERN STARTS HERE =====
# 1. Get groups with nested dictionary safety
groups = alert.get("rule", {}).get("groups", [])
# 2. Validate groups is a list
if not isinstance(groups, list):
print(f"Invalid groups type: {type(groups)}. Expected list.")
continue
# 3. Safe index access with fallback values
event_source = groups[0] if len(groups) > 0 else "unknown"
event_type = groups[1] if len(groups) > 1 else "unknown"
event_category = groups[2] if len(groups) > 2 else "none"
# ===== SAFE ACCESS PATTERN ENDS HERE =====
# Debug print (keep this temporarily)
print(f"Processed: {event_source} | {event_type} | {event_category}")
# Continue with your MISP logic below...
# ... rest of your code ...
except Exception as e:
print(f"Error processing alert: {str(e)}")
traceback.print_exc()
# MISP API Configuration
misp_base_url = "https://172.16.1.237/attributes/restSearch/"
misp_api_auth_key = "Izy9OCQ8xdfggvuy5eyhYSsdvzUAC5kjhjG"
misp_apicall_headers = {
"Content-Type": "application/json",
"Authorization": misp_api_auth_key,
"Accept": "application/json",
}
# Extract Sysmon event details
event_source = alert["rule"]["groups"][0]
event_type = alert["rule"]["groups"][2]
regex_file_hash = re.compile(r"\b[a-fA-F0-9]{64}\b")
wazuh_event_param = None
if event_source == "windows":
if event_type in ["sysmon_event1", "sysmon_event6", "sysmon_event7", "sysmon_event_15", "sysmon_event_23", "sysmon_event_24", "sysmon_event_25"]:
if "data" in alert and "win" in alert["data"] and "eventdata" in alert["data"]["win"] and "hashes" in alert["data"]["win"]["eventdata"]:
match = regex_file_hash.search(alert["data"]["win"]["eventdata"]["hashes"])
if match:
wazuh_event_param = match.group(0)
elif event_type == "sysmon_event3" and alert["data"]["win"]["eventdata"].get("destinationIsIpv6") == "false":
dst_ip = alert["data"]["win"]["eventdata"].get("destinationIp")
if dst_ip and ipaddress.ip_address(dst_ip).is_global:
wazuh_event_param = dst_ip
elif event_type == "sysmon_event_22":
wazuh_event_param = alert["data"]["win"]["eventdata"].get("queryName")
elif event_source == "linux" and event_type == "sysmon_event3" and alert["data"]["eventdata"].get("destinationIsIpv6") == "false":
dst_ip = alert["data"]["eventdata"].get("DestinationIp")
if dst_ip and ipaddress.ip_address(dst_ip).is_global:
wazuh_event_param = dst_ip
elif event_source == "ossec" and event_type == "syscheck_entry_added":
wazuh_event_param = alert.get("syscheck", {}).get("sha256_after")
if wazuh_event_param:
misp_search_value = f"value:{wazuh_event_param}"
misp_search_url = f"{misp_base_url}{misp_search_value}"
try:
misp_api_response = requests.get(misp_search_url, headers=misp_apicall_headers, verify=False)
misp_api_response = misp_api_response.json()
except ConnectionError:
alert_output = {"misp": {"error": "Connection Error to MISP API"}, "integration": "misp"}
send_event(alert_output, alert.get("agent"))
sys.exit()
if "response" in misp_api_response and "Attribute" in misp_api_response["response"] and misp_api_response["response"]["Attribute"]:
attribute = misp_api_response["response"]["Attribute"][0]
alert_output = {
"misp": {
"event_id": attribute.get("event_id"),
"category": attribute.get("category"),
"value": attribute.get("value"),
"type": attribute.get("type"),
"source": {"description": alert["rule"].get("description", "")},
},
"integration": "misp",
}
send_event(alert_output, alert.get("agent"))
else:
sys.exit()
Olusegun Adenrele Oyebo
Feb 21, 2025, 5:30:48 AM2/21/25
to Wazuh | Mailing List
Hello Muhammad,
What guide or documentation are you using for this?
I was able to successfully integrate Wazuh with MISP in my lab using the below blog as a guide. My test scenario was around pinging a malicious domain (screenshots attached):
What guide or documentation are you using for this?
I was able to successfully integrate Wazuh with MISP in my lab using the below blog as a guide. My test scenario was around pinging a malicious domain (screenshots attached):
Apart from using this blog, you can also make use of the YouTube video as a guide which will also give you more explanation on the configuration:
The sysmon rules I used for my integration were the below:
You can also check the below useful link on installing sysmon on your Windows system:
Let me know if you still need any other thing.
The sysmon rules I used for my integration were the below:
You can also check the below useful link on installing sysmon on your Windows system:
Let me know if you still need any other thing.
Best regards.
Reply all
Reply to author
Forward
0 new messages