elif config.USE_HEURISTICS:
if ord(dns_data[2:3]) & 0x80: # standard response
if ord(dns_data[3:4]) == 0x80: # recursion available, no error
_ = offset + 5
try:
while _ < len(dns_data):
if ord(dns_data[_:_ + 1]) & 0xc0 != 0 and dns_data[_ + 2] == "\00" and dns_data[_ + 3] == "\x01": # Type A
break
else:
_ += 12 + struct.unpack("!H", dns_data[_ + 10: _ + 12])[0]
_ = dns_data[_ + 12:_ + 16]
if _:
answer = socket.inet_ntoa(_)
if answer in trails and not _check_domain_whitelisted(query):
_ = trails[answer]
if "sinkhole" in _[0]:
trail = "(%s).%s" % ('.'.join(parts[:-1]), '.'.join(parts[-1:]))
log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, trail, "sinkholed by %s (malware)" % _[0].split(" ")[1], "(heuristic)"), packet) # (e.g. kitro.pl, devomchart.com, jebena.ananikolic.su, vuvet.cn)
elif "parking" in _[0]:
trail = "(%s).%s" % ('.'.join(parts[:-1]), '.'.join(parts[-1:]))
log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, trail, "parked site (suspicious)", "(heuristic)"), packet)
except IndexError:
pass
elif ord(dns_data[3:4]) == 0x83: # recursion available, no such name
if '.'.join(parts[-2:]) not in _dns_exhausted_domains and not _check_domain_whitelisted(query) and not _check_domain_member(query, trails):
if parts[-1].isdigit():
return
if not (len(parts) > 4 and all(_.isdigit() and int(_) < 256 for _ in parts[:4])): # generic check for DNSBL IP lookups
if not is_local(dst_ip): # prevent FPs caused by local queries
for _ in filter(None, (query, "*.%s" % '.'.join(parts[-2:]) if query.count('.') > 1 else None)):
if _ not in NO_SUCH_NAME_COUNTERS or NO_SUCH_NAME_COUNTERS[_][0] != sec // 3600:
NO_SUCH_NAME_COUNTERS[_] = [sec // 3600, 1, set()]
else:
NO_SUCH_NAME_COUNTERS[_][1] += 1
NO_SUCH_NAME_COUNTERS[_][2].add(query)
if NO_SUCH_NAME_COUNTERS[_][1] > NO_SUCH_NAME_PER_HOUR_THRESHOLD:
if _.startswith("*."):
trail = "%s%s" % ("(%s)" % ','.join(item.replace(_[1:], "") for item in NO_SUCH_NAME_COUNTERS[_][2]), _[1:])
if not any(subdomain in trail for subdomain in LOCAL_SUBDOMAIN_LOOKUPS): # generic check for local DNS resolutions
log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, trail, "excessive no such domain (suspicious)", "(heuristic)"), packet)
for item in NO_SUCH_NAME_COUNTERS[_][2]:
try:
del NO_SUCH_NAME_COUNTERS[item]
except KeyError:
pass
else:
log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, _, "excessive no such domain (suspicious)", "(heuristic)"), packet)
try:
del NO_SUCH_NAME_COUNTERS[_]
except KeyError:
pass
break
if len(parts) == 2 and parts[0] and '-' not in parts[0]:
part = parts[0]
trail = "(%s).%s" % (parts[0], parts[1])
result = _result_cache.get(part)
if result is None:
# Reference: https://github.com/exp0se/dga_detector
probabilities = (float(part.count(c)) / len(part) for c in set(_ for _ in part))
entropy = -sum(p * math.log(p) / math.log(2.0) for p in probabilities)
if entropy > SUSPICIOUS_DOMAIN_ENTROPY_THRESHOLD:
result = "entropy threshold no such domain (suspicious)"
if not result:
if sum(_ in CONSONANTS for _ in part) > SUSPICIOUS_DOMAIN_CONSONANT_THRESHOLD:
result = "consonant threshold no such domain (suspicious)"
_result_cache[part] = result or False
if result:
log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, trail, result, "(heuristic)"), packet)
elif protocol in IPPROTO_LUT: # non-TCP/UDP (e.g. ICMP)
if protocol == socket.IPPROTO_ICMP:
if ord(ip_data[iph_length:iph_length + 1]) != 0x08: # Non-echo request
return
elif protocol == socket.IPPROTO_ICMPV6:
if ord(ip_data[iph_length:iph_length + 1]) != 0x80: # Non-echo request
return
if dst_ip in trails:
log_event((sec, usec, src_ip, '-', dst_ip, '-', IPPROTO_LUT[protocol], TRAIL.IP, dst_ip, trails[dst_ip][0], trails[dst_ip][1]), packet)
elif src_ip in trails:
log_event((sec, usec, src_ip, '-', dst_ip, '-', IPPROTO_LUT[protocol], TRAIL.IP, src_ip, trails[src_ip][0], trails[src_ip][1]), packet)
except struct.error:
pass各行代码功能及函数总体功能
在函数 `_process_packet` 中，这段代码的功能是处理UDP协议的数据包，特别是DNS响应，并进行一系列的分析和日志记录。这段代码是函数的一部分，用于处理捕获到的单个原始IP层数据包。
代码中的 `elif` 块尝试执行以下操作：
1. **使用启发式分析处理DNS响应**：
- `elif config.USE_HEURISTICS:`: 如果启用了启发式分析，则继续处理DNS响应。
- `if ord(dns_data[2:3]) & 0x80: # standard response`: 如果DNS响应是标准响应，则继续处理。
- `if ord(dns_data[3:4]) == 0x80: # recursion available, no error`: 如果DNS响应是递归可用的且没有错误，则继续处理。
- `_ = offset + 5`: 设置一个变量 `_` 为当前偏移量加5。
- `while _ < len(dns_data):`: 遍历DNS响应数据，查找A记录。
- `if ord(dns_data[_:_ + 1]) & 0xc0 != 0 and dns_data[_ + 2] == "\00" and dns_data[_ + 3] == "\x01": # Type A`: 如果找到A记录，则继续处理。
- `_ = dns_data[_ + 12:_ + 16]`: 设置一个变量 `_` 为找到A记录后的偏移量。
- `if _:`: 如果A记录存在，则继续处理。
- `answer = socket.inet_ntoa(_)`: 将A记录的IP地址转换为点分十进制格式。
- `if answer in trails and not _check_domain_whitelisted(query):`: 如果A记录的IP地址在轨迹中且查询名称不在白名单中，则记录事件。
2. **记录DNS响应事件**：
- `log_event((sec, usec, src_ip, src_port, dst_ip, dst_port, PROTO.UDP, TRAIL.DNS, trail, "sinkholed by %s (malware)" % _[0].split(" ")[1], "(heuristic)"), packet)`: 记录DNS响应事件，包括时间戳、源IP、源端口、目标IP、目标端口、协议类型、轨迹类型、轨迹描述和原始数据包。
3. **处理DNS响应中的其他记录类型**：
- `elif ord(dns_data[3:4]) == 0x83: # recursion available, no such name`: 如果DNS响应是递归可用的且没有找到名称，则继续处理。
- `if '.'.join(parts[-2:]) not in _dns_exhausted_domains and not _check_domain_whitelisted(query) and not _check_domain_member(query, trails):`: 如果查询名称不在已耗尽的域名列表中且不在白名单中且不在轨迹中，则继续处理。
- `if parts[-1].isdigit():`: 如果查询名称的最后一个部分是数字，则返回。
- `if not (len(parts) > 4 and all(_.isdigit() and int(_) < 256 for _ in parts[:4])):`: 如果查询名称的前四个部分都是数字且小于256，则返回。
- `if len(parts) == 2 and parts[0] and '-' not in parts[0]:`: 如果查询名称的第一个部分不是数字且不包含连字符，则继续处理。
- `part = parts[0]`: 设置一个变量 `part` 为查询名称的第一个部分。
- `trail = "(%s).%s" % (parts[0], parts[1])`: 创建一个轨迹描述。
- `result = _result_cache.get(part)`: 从缓存中获取结果。
- `if result is None:`: 如果结果不存在，则继续处理。
- `probabilities = (float(part.count(c)) / len(part) for c in set(_ for _ in part))`: 计算每个字符的概率。
- `entropy = -sum(