116 lines
4.2 KiB
Python
116 lines
4.2 KiB
Python
#--depends-on commands
|
|
|
|
import re, socket, typing
|
|
from src import ModuleManager, utils
|
|
import dns.resolver
|
|
|
|
URL_GEOIP = "http://ip-api.com/json/%s"
|
|
REGEX_IPv6 = r"(?:(?:[a-f0-9]{1,4}:){2,}|[a-f0-9:]*::)[a-f0-9:]*"
|
|
REGEX_IPv4 = r"(?:\d{1,3}\.){3}\d{1,3}"
|
|
REGEX_IP = re.compile("(%s)|(%s)" % (REGEX_IPv4, REGEX_IPv6), re.I)
|
|
|
|
class DnsSetting(utils.Setting):
|
|
def parse(self, value: str) -> typing.Any:
|
|
if utils.is_ip(value):
|
|
return value
|
|
return None
|
|
|
|
@utils.export("serverset", DnsSetting("dns-nameserver",
|
|
"Set DNS nameserver", example="8.8.8.8"))
|
|
class Module(ModuleManager.BaseModule):
|
|
@utils.hook("received.command.dns", min_args=1)
|
|
def dns(self, event):
|
|
"""
|
|
:help: Get all addresses for a given hostname (IPv4/IPv6)
|
|
:usage: <hostname> [type [type ...]]
|
|
:prefix: DNS
|
|
"""
|
|
hostname = event["args_split"][0]
|
|
|
|
nameserver = event["server"].get_setting("dns-nameserver", None)
|
|
|
|
record_types = []
|
|
if len(event["args_split"]) > 1:
|
|
for arg in event["args_split"][1:]:
|
|
if arg.startswith("@"):
|
|
nameserver = arg[1:]
|
|
else:
|
|
record_types.append(arg.upper())
|
|
|
|
if not record_types:
|
|
record_types = ["A?", "AAAA?"]
|
|
|
|
if not nameserver == None:
|
|
resolver = dns.resolver.Resolver(configure=False)
|
|
resolver.nameservers = [nameserver]
|
|
else:
|
|
resolver = dns.resolver
|
|
|
|
results = []
|
|
|
|
for record_type in record_types:
|
|
try:
|
|
record_type_strip = record_type.rstrip("?")
|
|
query_result = resolver.query(hostname, record_type_strip)
|
|
query_results = [q.to_text() for q in query_result]
|
|
results.append([record_type_strip, query_results])
|
|
except dns.resolver.NXDOMAIN:
|
|
raise utils.EventError("Domain not found")
|
|
except dns.resolver.NoAnswer:
|
|
if not record_type.endswith("?"):
|
|
raise utils.EventError("Domain does not have a '%s' record"
|
|
% record_type_strip)
|
|
except dns.rdatatype.UnknownRdatatype:
|
|
raise utils.EventError("Unknown record type '%s'"
|
|
% record_type_strip)
|
|
|
|
results_str = ["%s: %s" % (t, ", ".join(r)) for t, r in results]
|
|
event["stdout"].write("(%s) %s" % (hostname, " | ".join(results_str)))
|
|
|
|
@utils.hook("received.command.geoip", min_args=1)
|
|
def geoip(self, event):
|
|
"""
|
|
:help: Get geoip data on a given IPv4/IPv6 address
|
|
:usage: <IP>
|
|
:prefix: GeoIP
|
|
"""
|
|
page = utils.http.request(URL_GEOIP % event["args_split"][0],
|
|
json=True)
|
|
if page:
|
|
if page.data["status"] == "success":
|
|
data = page.data["query"]
|
|
data += " | Organisation: %s" % page.data["org"]
|
|
data += " | City: %s" % page.data["city"]
|
|
data += " | Region: %s (%s)" % (page.data["regionName"],
|
|
page.data["countryCode"])
|
|
data += " | ISP: %s" % page.data["isp"]
|
|
data += " | Lon/Lat: %s/%s" % (page.data["lon"],
|
|
page.data["lat"])
|
|
data += " | Timezone: %s" % page.data["timezone"]
|
|
event["stdout"].write(data)
|
|
else:
|
|
event["stderr"].write("No geoip data found")
|
|
else:
|
|
raise utils.EventsResultsError()
|
|
|
|
@utils.hook("received.command.rdns")
|
|
def rdns(self, event):
|
|
"""
|
|
:help: Do a reverse-DNS look up on an IPv4/IPv6 address
|
|
:usage: <IP>
|
|
:prefix: rDNS
|
|
"""
|
|
ip = event["args_split"][0] if event["args"] else ""
|
|
if not ip:
|
|
line = event["target"].buffer.find(REGEX_IP)
|
|
if line:
|
|
match = REGEX_IP.search(line.message)
|
|
ip = match.group(1) or match.group(2)
|
|
if not ip:
|
|
raise utils.EventError("No IP provided")
|
|
|
|
try:
|
|
hostname, alias, ips = socket.gethostbyaddr(ip)
|
|
except (socket.herror, socket.gaierror) as e:
|
|
raise utils.EventError(e.strerror)
|
|
event["stdout"].write("(%s) %s" % (ips[0], hostname))
|