From 33c22e1d7b7a9c44ac0af23915174253a723e3ed Mon Sep 17 00:00:00 2001 From: Arpit Shah Date: Thu, 27 Aug 2026 15:08:26 -0400 Subject: [PATCH 1/2] Fixed ipv4 ipv6 verification. --- scripts/verify_reverse_zones.py | 118 +++++++++++++++++--------------- 1 file changed, 61 insertions(+), 57 deletions(-) diff --git a/scripts/verify_reverse_zones.py b/scripts/verify_reverse_zones.py index e69d32d..f7c176a 100644 --- a/scripts/verify_reverse_zones.py +++ b/scripts/verify_reverse_zones.py @@ -31,15 +31,15 @@ Usage: python verify_reverse_zones.py - + Output: JSON format showing subnet, corresponding zone name, and existence status - + Example input file content: 192.168.1.0/24 2001:558:4ffe:3::/64 10.0.0.0/16 - + Example output: { "subnet": "192.168.1.0/24", @@ -86,36 +86,36 @@ def safe_get_env_vars(env_vars: List[str]) -> Dict[str, str]: def ipv4_subnet_to_reverse_zone(subnet: str) -> str: """ Convert an IPv4 subnet to its reverse DNS zone name. - + Args: subnet (str): IPv4 subnet in CIDR notation (e.g., "192.168.1.0/24") - + Returns: str: Reverse DNS zone name (e.g., "1.168.192.in-addr.arpa") """ try: network = ipaddress.IPv4Network(subnet, strict=False) - + # Get the network address and prefix length network_addr = network.network_address prefix_length = network.prefixlen - + # Convert to octets octets = str(network_addr).split('.') - + # Calculate how many octets we need based on prefix length # Each octet represents 8 bits octets_needed = prefix_length // 8 - + # Take the required octets from the beginning and reverse them relevant_octets = octets[:octets_needed] reversed_octets = '.'.join(reversed(relevant_octets)) - + # Add the in-addr.arpa suffix reverse_zone = f"{reversed_octets}.in-addr.arpa" - + return reverse_zone - + except Exception as e: logging.error(f"Error converting IPv4 subnet {subnet} to reverse zone: {e}") return "" @@ -124,38 +124,38 @@ def ipv4_subnet_to_reverse_zone(subnet: str) -> str: def ipv6_subnet_to_reverse_zone(subnet: str) -> str: """ Convert an IPv6 subnet to its reverse DNS zone name. - + Args: subnet (str): IPv6 subnet in CIDR notation (e.g., "2001:558:4ffe:3::/64") - + Returns: str: Reverse DNS zone name (e.g., "3.0.0.0.e.f.f.4.8.5.5.0.1.0.0.2.ip6.arpa") """ try: network = ipaddress.IPv6Network(subnet, strict=False) - + # Get the network address and prefix length network_addr = network.network_address prefix_length = network.prefixlen - + # Convert to full hex representation (32 hex digits) full_hex = format(int(network_addr), '032x') - + # Calculate how many nibbles (4-bit hex digits) we need based on prefix length # Each nibble represents 4 bits nibbles_needed = prefix_length // 4 - + # Take the required nibbles from the beginning relevant_hex = full_hex[:nibbles_needed] - + # Reverse the nibbles and add dots between them reversed_nibbles = '.'.join(reversed(relevant_hex)) - + # Add the ip6.arpa suffix reverse_zone = f"{reversed_nibbles}.ip6.arpa" - + return reverse_zone - + except Exception as e: logging.error(f"Error converting IPv6 subnet {subnet} to reverse zone: {e}") return "" @@ -164,16 +164,18 @@ def ipv6_subnet_to_reverse_zone(subnet: str) -> str: def subnet_to_reverse_zone(subnet: str) -> str: """ Convert an IPv4 or IPv6 subnet to its reverse DNS zone name. - + Args: subnet (str): IP subnet in CIDR notation (e.g., "192.168.1.0/24" or "2001:558:4ffe:3::/64") - + Returns: str: Reverse DNS zone name """ try: # Try to determine if it's IPv4 or IPv6 - if ':' in subnet: + value = ipaddress.ip_network(subnet, strict=False) # Validate the subnet before processing + + if value.version == 6: return ipv6_subnet_to_reverse_zone(subnet) else: return ipv4_subnet_to_reverse_zone(subnet) @@ -185,10 +187,10 @@ def subnet_to_reverse_zone(subnet: str) -> str: def read_ip_subnets(file_path: str) -> List[str]: """ Read IPv4/IPv6 subnets from a text file. - + Args: file_path (str): Path to the file containing IP subnets - + Returns: List[str]: List of IP subnets (both IPv4 and IPv6) """ @@ -200,18 +202,20 @@ def read_ip_subnets(file_path: str) -> List[str]: # Skip empty lines and table formatting if not line or line.startswith('+') or line.startswith('|') or 'rows in set' in line: continue - + # Skip header lines if line.startswith('name') or line == 'name': continue - + # Check if line contains a subnet (IPv4 or IPv6) if '/' in line: subnets.append(line) - + else: + logging.warning(f"Skipping invalid subnet line {line_num}: {line}") + logging.info(f"Read {len(subnets)} IP subnets from {file_path}") return subnets - + except FileNotFoundError: logging.error(f"File not found: {file_path}") raise @@ -223,11 +227,11 @@ def read_ip_subnets(file_path: str) -> List[str]: def check_zone_exists(client: VinylDNSClient, zone_name: str) -> Dict[str, Any]: """ Check if a zone exists in VinylDNS. - + Args: client (VinylDNSClient): Initialized VinylDNS client instance zone_name (str): Name of the DNS zone to check - + Returns: Dict[str, Any]: Dictionary with zone information or error details """ @@ -262,22 +266,22 @@ def check_zone_exists(client: VinylDNSClient, zone_name: str) -> Dict[str, Any]: def verify_ip_zones(client: VinylDNSClient, subnets: List[str]) -> List[Dict[str, Any]]: """ Verify if reverse DNS zones exist for the given IP subnets (IPv4 and IPv6). - + Args: client (VinylDNSClient): Initialized VinylDNS client instance subnets (List[str]): List of IP subnets to check - + Returns: List[Dict[str, Any]]: List of dictionaries with verification results """ results = [] - + for i, subnet in enumerate(subnets, 1): logging.info(f"Processing subnet {i}/{len(subnets)}: {subnet}") - + # Convert subnet to reverse zone name reverse_zone = subnet_to_reverse_zone(subnet) - + if not reverse_zone: result = { "subnet": subnet, @@ -298,36 +302,36 @@ def verify_ip_zones(client: VinylDNSClient, subnets: List[str]) -> List[Dict[str "status": zone_info["status"], "error": zone_info["error"] } - + results.append(result) - + return results def write_results_to_file(results: List[Dict[str, Any]], output_dir: str = "output") -> str: """ Write verification results to a timestamped JSON file. - + Args: results (List[Dict[str, Any]]): List of verification results output_dir (str, optional): Directory to save the file. Defaults to "output" - + Returns: str: Full file path of the written file """ os.makedirs(output_dir, exist_ok=True) - + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"ip_zone_verification_{timestamp}.json" filepath = os.path.join(output_dir, filename) - + try: with open(filepath, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2) - + logging.info(f"Results written to: {filepath}") return filepath - + except Exception as e: logging.error(f"Error writing results to {filepath}: {e}") raise @@ -348,43 +352,43 @@ def main() -> None: default="output", help="Directory to save output files (default: output)" ) - + args = parser.parse_args() - + try: # Validate environment variables env_vars = safe_get_env_vars(REQUIRED_ENV_VARS) - + # Initialize VinylDNS client client = VinylDNSClient( env_vars["VINYLDNS_HOST"], env_vars["VINYLDNS_ACCESS_KEY"], env_vars["VINYLDNS_SECRET_KEY"], ) - + # Read IP subnets from file subnets = read_ip_subnets(args.subnets_file) - + if not subnets: logging.error("No valid IP subnets found in the input file") sys.exit(2) - + # Verify zones results = verify_ip_zones(client, subnets) - + # Output results to stdout print(json.dumps(results, indent=2)) - + # Write results to file write_results_to_file(results, args.output_dir) - + # Print summary total = len(results) existing = sum(1 for r in results if r["exists"]) missing = total - existing - + logging.info(f"Summary: {existing}/{total} zones exist, {missing} zones missing") - + except EnvironmentError as env_err: logging.error(f"Environment error: {env_err}") sys.exit(3) From 1ceda940c393d37b462871a64a4d358b9bcd944f Mon Sep 17 00:00:00 2001 From: Arpit Shah Date: Thu, 27 Aug 2026 16:47:24 -0400 Subject: [PATCH 2/2] Added CIDR subnet lookup. --- scripts/verify_reverse_zones.py | 58 ++++++++++++++++++++++++++------- 1 file changed, 47 insertions(+), 11 deletions(-) diff --git a/scripts/verify_reverse_zones.py b/scripts/verify_reverse_zones.py index f7c176a..8026174 100644 --- a/scripts/verify_reverse_zones.py +++ b/scripts/verify_reverse_zones.py @@ -86,12 +86,13 @@ def safe_get_env_vars(env_vars: List[str]) -> Dict[str, str]: def ipv4_subnet_to_reverse_zone(subnet: str) -> str: """ Convert an IPv4 subnet to its reverse DNS zone name. + Handles both classful and CIDR subnets properly. Args: - subnet (str): IPv4 subnet in CIDR notation (e.g., "192.168.1.0/24") + subnet (str): IPv4 subnet in CIDR notation (e.g., "192.168.1.0/24", "192.168.1.128/25") Returns: - str: Reverse DNS zone name (e.g., "1.168.192.in-addr.arpa") + str: Reverse DNS zone name (e.g., "1.168.192.in-addr.arpa", "128/25.1.168.192.in-addr.arpa") """ try: network = ipaddress.IPv4Network(subnet, strict=False) @@ -103,16 +104,40 @@ def ipv4_subnet_to_reverse_zone(subnet: str) -> str: # Convert to octets octets = str(network_addr).split('.') - # Calculate how many octets we need based on prefix length - # Each octet represents 8 bits - octets_needed = prefix_length // 8 + # Handle different CIDR scenarios + if prefix_length % 8 == 0: + # Classful boundary - simple case + octets_needed = prefix_length // 8 - # Take the required octets from the beginning and reverse them - relevant_octets = octets[:octets_needed] - reversed_octets = '.'.join(reversed(relevant_octets)) + relevant_octets = octets[:octets_needed] + reversed_octets = '.'.join(reversed(relevant_octets)) - # Add the in-addr.arpa suffix - reverse_zone = f"{reversed_octets}.in-addr.arpa" + # Add the in-addr.arpa suffix + reverse_zone = f"{reversed_octets}.in-addr.arpa" + + else: + # CIDR subnet that doesn't align on octet boundary + + # Calculate the range of addresses this subnet covers + start_addr = network.network_address + end_addr = network.broadcast_address + + # Get the full octets (complete 8-bit boundaries) + full_octets = prefix_length // 8 + remaining_bits = prefix_length % 8 + + # Build the zone name + # Include the complete octets in reverse order + base_octets = octets[:full_octets] + reversed_base = '.'.join(reversed(base_octets)) + + # Add the partial octet information + partial_octet_start = int(octets[full_octets]) + partial_octet_end = int(str(end_addr).split('.')[full_octets]) + + reverse_zone = f"{partial_octet_start}/{prefix_length}.{reversed_base}.in-addr.arpa" + + logging.info(f"CIDR reverse zone for {subnet}: {reverse_zone}") return reverse_zone @@ -236,7 +261,17 @@ def check_zone_exists(client: VinylDNSClient, zone_name: str) -> Dict[str, Any]: Dict[str, Any]: Dictionary with zone information or error details """ try: - zone = client.get_zone_by_name(zone_name) + logging.info(f"Checking if zone exists for: {zone_name}") + if "/" in zone_name: + zone = client.list_zones(name_filter=zone_name) + if zone and hasattr(zone, 'zones') and zone.zones: + zone = zone.zones[0] + else: + zone = None + + else: + zone = client.get_zone_by_name(zone_name) + if zone: return { "exists": True, @@ -277,6 +312,7 @@ def verify_ip_zones(client: VinylDNSClient, subnets: List[str]) -> List[Dict[str results = [] for i, subnet in enumerate(subnets, 1): + logging.info(f"\n\n") logging.info(f"Processing subnet {i}/{len(subnets)}: {subnet}") # Convert subnet to reverse zone name