Search PassiveTotal for records associated with the provided email address.
(logger, email, pt, zi, jobs_manager)
| 68 | |
| 69 | |
| 70 | def search_pt_email(logger, email, pt, zi, jobs_manager): |
| 71 | """ |
| 72 | Search PassiveTotal for records associated with the provided email address. |
| 73 | """ |
| 74 | logger.info("Searching: " + email) |
| 75 | results = pt.get_email(email) |
| 76 | |
| 77 | if results is None: |
| 78 | logger.error("Error querying email: " + email) |
| 79 | jobs_manager.record_job_error() |
| 80 | exit(0) |
| 81 | |
| 82 | logger.info("Results for " + email + ": " + str(len(results["results"]))) |
| 83 | |
| 84 | for j in range(0, len(results["results"])): |
| 85 | domain = results["results"][j]["domain"].encode("utf-8").decode("utf8") |
| 86 | logger.debug("Checking domain: " + domain) |
| 87 | |
| 88 | if re.match(r"^([0-9]{1,3}\.){3}[0-9]{1,3}\/\d\d$", domain): |
| 89 | logger.debug("Matched IP address. Skipping...") |
| 90 | continue |
| 91 | |
| 92 | zi.add_zone(domain, "PassiveTotal") |
| 93 | |
| 94 | |
| 95 | def search_pt_org(logger, org, pt, zi, jobs_manager): |
no test coverage detected