Merge pull request #22 from mietzen/check-if-ip-is-unspecified

Check if public IP is unspecified, fixes #21

porkbun_ddns/helpers.py

@@ -1,14 +1,35 @@
 import urllib.request
 import xml.etree.ElementTree as ET
+import socket
+import logging
 
+logger = logging.getLogger()
 
-def get_ips_from_fritzbox(fritzbox_ip, ipv4=True):
+def check_ipv6_connectivity() -> bool:
+    """Check IPv6 connectivity
+    Source: Pandu Poluan, https://stackoverflow.com/a/66249915
+    
+    Returns:
+        bool: IPv6 connectivity
+    """
+    if socket.has_ipv6:
+        sock = None
+        try:
+            sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+            sock.bind(('v6.ident.me', 0))
+            return True
+        except OSError:
+            logger.warning('No IPv6 connectivity!')
+        finally:
+            if sock:
+                sock.close()
+    return False
+
+def get_ips_from_fritzbox(fritzbox_ip):
     """Retrieves the IP address of the Fritzbox router's external network interface.
 
     Args:
         fritzbox_ip (str): The IP address of the Fritzbox router.
-        ipv4 (bool, optional): A boolean flag that specifies whether to retrieve the IPv4 or IPv6 address.
-            Defaults to True, which retrieves the IPv4 address.
 
     Returns:
         str: The IP address of the Fritzbox router's external network interface.
@@ -20,12 +41,9 @@ def get_ips_from_fritzbox(fritzbox_ip, ipv4=True):
 
         AttributeError: If the requested field is not found in the XML response.
     """
-    if ipv4:
-        schema = 'GetExternalIPAddress'
-        field = 'NewExternalIPAddress'
-    else:
-        schema = 'X_AVM_DE_GetExternalIPv6Address'
-        field = 'NewExternalIPv6Address'
+
+    schema = 'GetExternalIPAddress'
+    field = 'NewExternalIPAddress'
 
     req = urllib.request.Request(
         'http://' + fritzbox_ip + ':49000/igdupnp/control/WANIPConn1')

porkbun_ddns/porkbun_ddns.py

@@ -4,7 +4,9 @@
 import json
 import ipaddress
 import urllib.request
+from urllib.error import HTTPError
 from porkbun_ddns.helpers import get_ips_from_fritzbox
+from porkbun_ddns.helpers import check_ipv6_connectivity
 
 logger = logging.getLogger('porkbun_ddns')
 
@@ -80,32 +82,36 @@ def get_public_ips(self) -> list:
             if self.fritzbox_ip:
                 if self.ipv4:
                     public_ips.append(
-                        get_ips_from_fritzbox(self.fritzbox_ip, ipv4=True))
-                if self.ipv6:
-                    public_ips.append(get_ips_from_fritzbox(
-                        self.fritzbox_ip, ipv4=False))
+                        get_ips_from_fritzbox(self.fritzbox_ip))
             else:
                 if self.ipv4:
                     public_ips.append(urllib.request.urlopen(
                         'https://v4.ident.me').read().decode('utf8'))
                 if self.ipv6:
-                    public_ips.append(urllib.request.urlopen(
-                        'https://v6.ident.me').read().decode('utf8'))
+                    if check_ipv6_connectivity():
+                        public_ips.append(urllib.request.urlopen(
+                            'https://v6.ident.me').read().decode('utf8'))
             public_ips = set(public_ips)
 
         if not public_ips:
             raise PorkbunDDNS_Error('Failed to obtain IP Addresses!')
 
-        return [ipaddress.ip_address(x) for x in public_ips]
-
+        return [ipaddress.ip_address(x) for x in public_ips if not ipaddress.ip_address(x).is_unspecified]
+        
     def _api(self, target: str, data: dict = None) -> dict:
         """Send an API request to a specified target.
         """
 
         data = data or self.config
         req = urllib.request.Request(self.config['endpoint'] + target)
         req.data = json.dumps(data).encode('utf8')
-        response = urllib.request.urlopen(req).read()
+        try:
+            response = urllib.request.urlopen(req).read()
+        except HTTPError as err:
+            if err.code == 400:
+                raise PorkbunDDNS_Error('Invalid API Keys!')
+            else:
+                raise err
         return json.loads(response.decode('utf-8'))
 
     def get_records(self) -> dict: