Function main

python3_cron_scripts/get_gcp_ranges.py:172–223

This function extract the IP address ranges from the TXT records and stores them in gcp_ips collection within the database.

(logger=None)

def main(logger=None):
    """
    This function extract the IP address ranges from the TXT records
    and stores them in gcp_ips collection within the database.
    """
    if logger is None:
        logger = LoggingUtil.create_log(__name__)

    now = datetime.now()
    print("Starting: " + str(now))
    logger.info("Starting...")

    parser = argparse.ArgumentParser(
        description="Retrieve GCP IP ranges from DNS TXT records or the public JSON file."
    )
    parser.add_argument(
        "--source",
        required=False,
        default="json",
        choices=["json", "dns"],
        help="Specify whether to fetch from DNS TXT records or the public JSON file",
    )

    args = parser.parse_args()

    mongo_connector = MongoConnector.MongoConnector()
    gcp_collection = mongo_connector.get_gcp_ips_connection()
    google_dns = GoogleDNS.GoogleDNS()
    jobs_manager = JobsManager.JobsManager(mongo_connector, "get_gcp_ranges")
    jobs_manager.record_job_start()

    if args.source == "dns":
        ipv4_ranges, ipv6_ranges = dns_search(google_dns, logger)
    elif args.source == "json":
        ipv4_ranges, ipv6_ranges = json_search(google_dns, logger)
    else:
        logger.error("Invalid source: " + args.source)
        exit(1)

    new_data = {}
    new_data["prefixes"] = ipv4_ranges
    new_data["ipv6_prefixes"] = ipv6_ranges
    new_data["created"] = now

    gcp_collection.delete_many({})
    mongo_connector.perform_insert(gcp_collection, new_data)

    jobs_manager.record_job_complete()

    now = datetime.now()
    print("Ending: " + str(now))
    logger.info("Complete.")


if __name__ == "__main__":