censys_bulk_lookup.py

#!/usr/bin/env python3
import csv
import json
from concurrent.futures import ThreadPoolExecutor

from censys.common.exceptions import CensysException
from censys.search import CensysHosts


def lookup_ips(csv_file):
    c = CensysHosts()
    results: dict[str, dict] = {}

    ips_to_lookup = []
    with open(csv_file) as f:
        reader = csv.DictReader(f)
        for row in reader:
            ip = row["ip"]
            ips_to_lookup.append(ip)

    try:
        with ThreadPoolExecutor(max_workers=10) as executor:
            for ip in ips_to_lookup:
                executor.submit(lookup_ip, c, ip, results)
    except KeyboardInterrupt:
        print("Interrupted by user, exiting")
        return results

    return results


def lookup_ip(c, ip, results):
    try:
        print(f"Looking up {ip}")
        res = c.view(ip)
        results[ip] = res
    except CensysException as e:
        print(f"Error: {e}")
        return


def main():
    import sys

    if len(sys.argv) != 3:
        print(f"Usage: {sys.argv[0]} CSV_FILE OUTPUT_FILE")
        sys.exit(1)
    results = lookup_ips(sys.argv[1])
    output_file = sys.argv[2]
    print(f"Successfully looked up {len(results)} IPs")
    with open(output_file, "w") as f:
        json.dump(results, f)
    print(f"Done! Results written to {output_file}")


if __name__ == "__main__":
    main()