Skip to content
This repository has been archived by the owner on Feb 12, 2024. It is now read-only.

Commit

Permalink
Merge pull request #316 from CybercentreCanada/update/no-dns
Browse files Browse the repository at this point in the history 
Handling DNS requests with no answers
  • Loading branch information
@cccs-kevin
cccs-kevin authored Mar 21, 2023
2 parents 869e0bb + 4980d6f commit d474a9b
Show file tree
Hide file tree
Showing 2 changed files with 99 additions and 41 deletions.
111 changes: 70 additions & 41 deletions cuckoo/cuckoo_result.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -647,6 +647,8 @@ def process_network(network: Dict[str, Any], parent_result_section: ResultSectio
del network_flow["timestamp"]

for answer, request in resolved_ips.items():
if answer.isdigit():
continue
nd = so.create_network_dns(
domain=request["domain"], resolved_ips=[answer], lookup_type=request["type"]
)
Expand Down Expand Up @@ -800,6 +802,7 @@ def _get_dns_sec(resolved_ips: Dict[str, Dict[str, Any]],
:param safelist: A dictionary containing matches and regexes for use in safelisting values
:return: the result section containing details that we care about
"""
answer_exists = False
if len(resolved_ips.keys()) == 0:
return None
dns_res_sec = ResultTableSection("Protocol: DNS")
Expand All @@ -810,13 +813,25 @@ def _get_dns_sec(resolved_ips: Dict[str, Dict[str, Any]],
request = request_dict["domain"]
_ = add_tag(dns_res_sec, "network.dynamic.ip", answer, safelist)
if add_tag(dns_res_sec, "network.dynamic.domain", request, safelist):
# If there is only UDP and no TCP traffic, then we need to tag the domains here:
dns_request = {
"domain": request,
"ip": answer,
}
if answer.isdigit():
dns_request = {
"domain": request,
}
else:
# If there is only UDP and no TCP traffic, then we need to tag the domains here:
dns_request = {
"domain": request,
"ip": answer,
}
answer_exists = True
dns_body.append(dns_request)
[dns_res_sec.add_row(TableRow(**dns)) for dns in dns_body]
if not answer_exists:
_ = ResultTextSection(
title_text="DNS services are down!",
body="Contact the CAPE administrator for details.",
parent=dns_res_sec
)
return dns_res_sec


Expand All @@ -831,55 +846,69 @@ def _get_dns_map(dns_calls: List[Dict[str, Any]], process_map: Dict[int, Dict[st
:return: the mapping of resolved IPs and their corresponding domains
"""
resolved_ips: Dict[str, Dict[str, Any]] = {}
no_answer_count = 0
for dns_call in dns_calls:
if len(dns_call["answers"]) > 0:
answer = dns_call["answers"][0]["data"]
request = dns_call["request"]
dns_type = dns_call["type"]
else:
# We still want these DNS calls in the resolved_ips map, so use int as unique ID
answer = str(no_answer_count)
no_answer_count += 1

# If the method of routing is INetSim or a variation of INetSim, then we will not use PTR records. The reason being that there is
# always a chance for collision between IPs and hostnames due to the DNS cache, and that chance increases
# the smaller the size of the random network space
if routing.lower() in [INETSIM.lower(), "none"] and dns_type == "PTR":
continue
request = dns_call.get("request")
if not request:
continue
dns_type = dns_call["type"]

# A DNS pointer record (PTR for short) provides the domain name associated with an IP address.
if dns_type == "PTR" and "in-addr.arpa" in request:
# Determine the ip from the ARPA request by extracting and reversing the IP from the "ip"
request = request.replace(".in-addr.arpa", "")
split_ip = request.split(".")
request = f"{split_ip[3]}.{split_ip[2]}.{split_ip[1]}.{split_ip[0]}"
# If the method of routing is INetSim or a variation of INetSim, then we will not use PTR records.
# The reason being that there is always a chance for collision between IPs and hostnames due to the
# DNS cache, and that chance increases the smaller the size of the random network space
if routing.lower() in [INETSIM.lower(), "none"] and dns_type == "PTR":
continue

# If PTR and A request for the same ip-domain pair, we choose the A
if request in resolved_ips:
continue
# A DNS pointer record (PTR for short) provides the domain name associated with an IP address.
if dns_type == "PTR" and "in-addr.arpa" in request:
# Determine the ip from the ARPA request by extracting and reversing the IP from the "ip"
request = request.replace(".in-addr.arpa", "")
split_ip = request.split(".")
request = f"{split_ip[3]}.{split_ip[2]}.{split_ip[1]}.{split_ip[0]}"

resolved_ips[request] = {
"domain": answer
}
elif dns_type == "PTR" and "ip6.arpa" in request:
# Drop it
continue
# Some Windows nonsense
elif answer in dns_servers:
# If PTR and A request for the same ip-domain pair, we choose the A
if request in resolved_ips:
continue
# An 'A' record provides the IP address associated with a domain name.
else:
resolved_ips[answer] = {
"domain": request,
"process_id": dns_call.get("pid"),
"process_name": dns_call.get("image"),
"time": dns_call.get("time"),
"guid": dns_call.get("guid"),
"type": dns_type,
}

resolved_ips[request] = {
"domain": answer
}
elif dns_type == "PTR" and "ip6.arpa" in request:
# Drop it
continue
# Some Windows nonsense
elif answer in dns_servers:
continue
# An 'A' record provides the IP address associated with a domain name.
else:
resolved_ips[answer] = {
"domain": request,
"process_id": dns_call.get("pid"),
"process_name": dns_call.get("image"),
"time": dns_call.get("time"),
"guid": dns_call.get("guid"),
"type": dns_type,
}
# now map process_name to the dns_call
for process, process_details in process_map.items():
for network_call in process_details["network_calls"]:
dns = next((network_call[api_call] for api_call in DNS_API_CALLS if api_call in network_call), {})
if dns != {} and dns.get("hostname"):
ip_mapped_to_host = next((ip for ip, details in resolved_ips.items()
if details["domain"] == dns["hostname"]), None)
ip_mapped_to_host = next(
(
ip
for ip, details in resolved_ips.items()
if details["domain"] == dns["hostname"] and not ip.isdigit()
),
None
)
if not ip_mapped_to_host:
continue
if not resolved_ips[ip_mapped_to_host].get("process_name"):
Expand Down
29 changes: 29 additions & 0 deletions tests/test_cuckoo_result.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -418,6 +418,21 @@ def test_get_dns_sec():
actual_res_sec = _get_dns_sec(resolved_ips, safelist)
assert check_section_equality(actual_res_sec, expected_res_sec)

resolved_ips = {"0": {"domain": "blah.com"}}
expected_res_sec = ResultSection(
"Protocol: DNS", body_format=BODY_FORMAT.TABLE, body=dumps([{"domain": "blah.com"}])
)
expected_res_sec.set_heuristic(1000)
expected_res_sec.add_tag("network.protocol", "dns")
expected_res_sec.add_tag("network.dynamic.domain", "blah.com")
expected_res_sec.add_subsection(ResultSection(
title_text="DNS services are down!",
body="Contact the CAPE administrator for details.",
))
actual_res_sec = _get_dns_sec(resolved_ips, safelist)
assert check_section_equality(actual_res_sec, expected_res_sec)


@staticmethod
@pytest.mark.parametrize(
"dns_calls, process_map, routing, expected_return",
Expand Down Expand Up @@ -473,6 +488,20 @@ def test_get_dns_sec():
([{"answers": [{"data": "1.1.1.1"}],
"request": "request", "type": "dns_type"}],
{1: {"network_calls": [{"blah": {"hostname": "blah"}}]}}, "", {}),
([{"answers": [],
"request": "request", "type": "dns_type"}],
{},
"",
{
'0': {
'domain': 'request',
'guid': None,
'process_id': None,
'process_name': None,
'time': None,
'type': 'dns_type'
}
}),
]
)
def test_get_dns_map(dns_calls, process_map, routing, expected_return):
Expand Down

0 comments on commit d474a9b

Please sign in to comment.