Function json_search

python3_cron_scripts/get_gcp_ranges.py:119–169

This function performs the new method of fetching the JSON file.

(google_dns, logger)

def json_search(google_dns, logger):
    """
    This function performs the new method of fetching the JSON file.
    """

    ipv4_ranges = []
    ipv6_ranges = []

    try:
        req = _requests_retry_session().get(
            "https://www.gstatic.com/ipranges/cloud.json", timeout=120
        )
    except Exception as ex:
        logger.error("Google GCP range download attempts failed!")
        logger.error(str(ex))
        return None, None

    if req.status_code != 200:
        logger.debug("HTTP Error fetching GCP ranges!")
        return None

    json_results = json.loads(req.text)

    already_seen = set()

    try:
        for entry in json_results["prefixes"]:
            if "ipv4Prefix" in entry and entry["ipv4Prefix"] not in already_seen:
                ipv4_ranges.append(
                    {
                        "ip_prefix": entry["ipv4Prefix"],
                        "service": entry["service"],
                        "scope": entry["scope"],
                    }
                )
                already_seen.add(entry["ipv4Prefix"])
            elif "ipv6Prefix" in entry and entry["ipv6Prefix"] not in already_seen:
                ipv6_ranges.append(
                    {
                        "ipv6_prefix": entry["ipv6Prefix"],
                        "service": entry["service"],
                        "scope": entry["scope"],
                    }
                )
                already_seen.add(entry["ipv6Prefix"])
    except Exception as ex:
        logger.error("Error parsing GCP ranges!")
        logger.error(str(ex))
        return None, None

    return ipv4_ranges, ipv6_ranges


def main(logger=None):