WhoisXML API is a domain data provider that offers a range of domain-related services such as domain research, monitoring, IP geolocation, and threat intelligence.
By integrating WhoisXML API with Wazuh, you can enhance your security monitoring capabilities and protect your organization against cyber threats. With WhoisXML API, you can enrich Wazuh's security alerts with domain-related data, such as domain ownership, availability, and historical records. By combining this data with Wazuh's threat detection and response capabilities, you can quickly identify and respond to security incidents before they can cause any damage.
Whether you're in the cybersecurity, financial, or e-commerce industry, integrating WhoisXML API with Wazuh can help you improve your security posture and stay ahead of emerging threats.
This article discusses how to set up Wazuh to connect with the WhoisXML API through the use of an integrator. The following topics are covered:
Configuring the integrator tool for custom integration.
Developing a Python script that can analyze alerts and use the WhoisXML API to verify the IP address provided in the log.
Setting up the Rule
In this case, we aim to produce alerts that include WhoisXML information whenever a public IP address attempts SSH authentication against a network endpoint.
To achieve this we create rules within the /var/ossec/etc/rules/local_rules.xml file in our Wazuh Manager. These rules will trigger an alert in case of a failed SSH authentication attempt from a public IP.
<!-- User Failed Authentication from Public IPv4 -->
<group name="local,syslog,sshd,">
<rule id="100004" level="10">
<if_sid>5760</if_sid>
<match type="pcre2">\b(?!(10)|192\.168|172\.(2[0-9]|1[6-9]|3[0-1])|(25[6-9]|2[6-9][0-9]|[3-9][0-9][0-9]|99[1-9]))[0-9]{1,3}\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)</match>
<description>sshd: Authentication failed from a public IP address > $(srcip).</description>
<group>authentication_failed,pci_dss_10.2.4,pci_dss_10.2.5,</group>
</rule>
</group>
When an alert is triggered by rule ID 100004, the Wazuh integration script will initiate a request to the WhoisXML API endpoint to retrieve details regarding the IP address specified in the alert.
Setting up a custom integration
To configure the integration with WhoisXML, you need to modify the ossec.conf configuration file of the Wazuh manager by adding the integration block with the following content:
<integration>
<name>custom-whois.py</name>
<hook_url>https://www.whoisxmlapi.com/whoisserver/WhoisService</hook_url>
<api_key>YOUR-OWN-API-KEY</api_key>
<level>10</level>
<rule_id>100004</rule_id>
<alert_format>json</alert_format>
</integration>
Here are the integration block parameters along with their descriptions:
name: The name of the custom script used for integration. All custom scripts should start with the prefix
custom-hook_url: Includes the API URL provided by WhoisXML
api_key: Includes the key of the API that enables its use
rule_id: Sets the rules that trigger the integration
alert_format: Indicates the format in which the script receives the alerts (JSON format is recommended). If this parameter is not set, the script will receive the alerts in full_log format
Adding the integration Script
We will create a file named custom-whois.py in the /var/ossec/integrations/ directory of the Wazuh server.
Here's the full Python script:
#!/var/ossec/framework/python/bin/python3
# Copyright (C) 2015-2022, Wazuh Inc.
# Template by WhatDoesKmean
import json
import sys
import time
import os
from socket import socket, AF_UNIX, SOCK_DGRAM
try:
import requests
from requests.auth import HTTPBasicAuth
except Exception as e:
print("No module 'requests' found. Install: pip install requests")
sys.exit(1)
# Global vars
debug_enabled = False
pwd = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
print(pwd)
#exit()
json_alert = {}
now = time.strftime("%a %b %d %H:%M:%S %Z %Y")
# Set paths
log_file = '{0}/logs/integrations.log'.format(pwd)
socket_addr = '{0}/queue/sockets/queue'.format(pwd)
def main(args):
debug("# Starting")
# Read args
alert_file_location = args[1]
apikey = args[2]
debug("# API Key")
debug(apikey)
debug("# File location")
debug(alert_file_location)
# Load alert. Parse JSON object.
with open(alert_file_location) as alert_file:
json_alert = json.load(alert_file)
debug("# Processing alert")
debug(json_alert)
# Request whois info
msg = request_whois_info(json_alert,apikey)
# If positive match, send event to Wazuh Manager
if msg:
send_event(msg, json_alert["agent"])
def debug(msg):
if debug_enabled:
msg = "{0}: {1}\n".format(now, msg)
print(msg)
f = open(log_file,"a")
f.write(str(msg))
f.close()
def collect(data):
domainName = data['domainName']
registrarName = data['registrarName']
contactEmail = data['contactEmail']
domainNameExt = data['domainNameExt']
estimatedDomainAge = data['estimatedDomainAge']
registryData = data['registryData']['strippedText']
return domainName,registrarName,contactEmail,domainNameExt,estimatedDomainAge,registryData
def in_database(data, srcip):
result = data['domainName']
if result == 0:
return False
return True
def query_api(domainName, apikey):
params = {"domainName": domainName, "apiKey": apikey, "outputFormat":"JSON"}
url = "https://www.whoisxmlapi.com/whoisserver/WhoisService"
response = requests.get(url, params=params)
if response.status_code == 200:
json_response = response.json()["WhoisRecord"]
if "estimatedDomainAge" not in json_response:
json_response["estimatedDomainAge"] = "NotFound"
data = json_response
return data
else:
alert_output = {}
alert_output["whois"] = {}
alert_output["integration"] = "custom-whois"
json_response = response.json()
debug("# Error: The whois encountered an error")
alert_output["whois"]["error"] = response.status_code
alert_output["whois"]["description"] = json_response["errors"][0]["detail"]
send_event(alert_output)
exit(0)
def request_whois_info(alert, apikey):
alert_output = {}
# If there is no source ip address present in the alert. Exit.
if not "srcip" in alert["data"]:
return(0)
# Request info using whois API
data = query_api(alert["data"]["srcip"], apikey)
# Create alert
alert_output["whois"] = {}
alert_output["integration"] = "custom-whois"
alert_output["whois"]["found"] = 0
alert_output["whois"]["source"] = {}
alert_output["whois"]["source"]["alert_id"] = alert["id"]
alert_output["whois"]["source"]["rule"] = alert["rule"]["id"]
alert_output["whois"]["source"]["description"] = alert["rule"]["description"]
alert_output["whois"]["source"]["full_log"] = alert["full_log"]
alert_output["whois"]["source"]["srcip"] = alert["data"]["srcip"]
srcip = alert["data"]["srcip"]
# Check if whois has any info about the srcip
if in_database(data, srcip):
alert_output["whois"]["found"] = 1
# Info about the IP found in whois
if alert_output["whois"]["found"] == 1:
domainName,registrarName,contactEmail,domainNameExt,estimatedDomainAge,registryData = collect(data)
# Populate JSON Output object with whois request
alert_output["whois"]["domainName"] = domainName
alert_output["whois"]["registrarName"] = registrarName
alert_output["whois"]["contactEmail"] = contactEmail
alert_output["whois"]["domainNameExt"] = domainNameExt
alert_output["whois"]["estimatedDomainAge"] = estimatedDomainAge
alert_output["whois"]["registryData"] = registryData
debug(alert_output)
return(alert_output)
def send_event(msg, agent = None):
if not agent or agent["id"] == "000":
string = '1:whois:{0}'.format(json.dumps(msg))
else:
string = '1:[{0}] ({1}) {2}->whois:{3}'.format(agent["id"], agent["name"], agent["ip"] if "ip" in agent else "any", json.dumps(msg))
debug(string)
sock = socket(AF_UNIX, SOCK_DGRAM)
sock.connect(socket_addr)
sock.send(string.encode())
sock.close()
if __name__ == "__main__":
try:
# Read arguments
bad_arguments = False
if len(sys.argv) >= 4:
msg = '{0} {1} {2} {3} {4}'.format(now, sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4] if len(sys.argv) > 4 else '')
debug_enabled = (len(sys.argv) > 4 and sys.argv[4] == 'debug')
else:
msg = '{0} Wrong arguments'.format(now)
bad_arguments = True
# Logging the call
f = open(log_file, 'a')
f.write(str(msg) + '\n')
f.close()
if bad_arguments:
debug("# Exiting: Bad arguments.")
sys.exit(1)
# Main function
main(sys.argv)
except Exception as e:
debug(str(e))
raise
The above code retrieves the source IP from an alert and utilizes the WhoisXML API to retrieve the IP's WhoIs data.
Now we need to modify the file's ownership and group to root:wazuh and grant execution permissions:chmod 750 /var/ossec/integrations/custom-whois.pychown root:wazuh /var/ossec/integrations/custom-whois.py
To apply the changes, proceed with restarting the Wazuh manager:systemctl restart wazuh-manager
or/var/ossec/bin/wazuh-control restart
Adding the custom Rule
Now we can generate alerts for a public IP address that attempted a failed SSH authentication.
To achieve this, we need to add a custom rule to the file /var/ossec/etc/rules/local_rules.xml and restart the manager for the changes to take effect.
<group name="local,syslog,sshd,">
<rule id="100005" level="10">
<field name="whois.domainName">\.+</field>
<description>IP address $(whois.domainName) trying to connect to your network.</description>
<group>authentication_failed,pci_dss_10.2.4,pci_dss_10.2.5,</group>
</rule>
</group>
To test the integration, the logs sample logs are:May 3 15:46:00 TEST_HOST sshd[321]: Failed password for whatdoeskmean from 116.237.130.82 port 8000 ssh2
May 3 16:58:00 TEST_HOST sshd[123]: Failed password for whatdoeskmean from 5.255.99.205 port 9997 ssh2
To test these rules, the logs can be injected into an endpoint that is registered with the Wazuh manager.
To inject logs on a Linux endpoint:echo "May 03 15:46:00 TEST_HOST sshd[321]: Failed password for whatdoeskmean from 116.237.130.82 port 8000 ssh2" >> /var/log/secure
After the logs are injected, we can observe the outcome of the log test on the Wazuh dashboard:
WhoIsXML integration alert (triggered by rule 100004):
Now, you know! 😉