MCPcopy Create free account
hub / github.com/adobe/Marinus / insert_result

Function insert_result

python3_cron_scripts/zgrab_port_ip.py:254–402  ·  view source on GitHub ↗

Insert the matched domain into the collection of positive results.

(entry, port, ip_context, all_zones, results_collection)

Source from the content-addressed store, hash-verified

252
253
254def insert_result(entry, port, ip_context, all_zones, results_collection):
255 """
256 Insert the matched domain into the collection of positive results.
257 """
258 if "zgrab2" in global_zgrab_path:
259 temp_date = entry["data"][global_port_names[port]]["timestamp"]
260 new_date = parse(temp_date)
261 entry["timestamp"] = new_date
262 entry["data"][global_port_names[port]]["timestamp"] = new_date
263 else:
264 temp_date = entry["timestamp"]
265 new_date = parse(temp_date)
266 entry["timestamp"] = new_date
267
268 # Returns all entries in ip_context that contain the given IP
269 matches = check_ip_context(entry["ip"], ip_context)
270
271 # Grab the zones from the ip_context
272 zones = []
273 domains = []
274 if len(matches) > 0:
275 for match in matches:
276 if match["zone"] not in zones:
277 zones.append(match["zone"])
278 if match["domain"] not in domains:
279 domains.append(match["domain"])
280
281 # Append the zones from the TLS certificate
282 if port == "443":
283 # Make the timestamp an actual date instead of a string
284 entry["data"]["tls"]["timestamp"] = new_date
285
286 if "zgrab2" in global_zgrab_path:
287 cert_zones = check_in_zone(
288 entry["data"]["tls"]["result"]["handshake_log"], all_zones
289 )
290 else:
291 cert_zones = check_in_zone(entry["data"]["tls"], all_zones)
292 for zone in cert_zones:
293 if zone not in zones:
294 zones.append(zone)
295 elif port == "22":
296 if "zgrab2" in global_zgrab_path:
297 entry["data"]["ssh"]["timestamp"] = new_date
298 else:
299 entry["data"]["xssh"]["timestamp"] = new_date
300 elif port == "25":
301 if "zgrab2" in global_zgrab_path:
302 if "tls" in entry["data"]["smtp"]["result"]:
303 cert_zones = check_in_zone(
304 entry["data"]["smtp"]["result"]["tls"]["handshake_log"], all_zones
305 )
306 for zone in cert_zones:
307 if zone not in zones:
308 zones.append(zone)
309 else:
310 temp = entry["data"]
311 entry["data"] = {}

Callers 1

process_threadFunction · 0.70

Calls 2

check_ip_contextFunction · 0.70
check_in_zoneFunction · 0.70

Tested by

no test coverage detected