use generic provider for ipfy + wtfismyip

pddnsc/base.py

@@ -1,7 +1,8 @@
 import httpx
 import asyncio
 from abc import ABC, abstractmethod
-from typing import NamedTuple
+from typing import NamedTuple, Optional
+from netaddr import valid_ipv4, valid_ipv6
 
 
 class IPAddreses(NamedTuple):
@@ -154,3 +155,11 @@ class BaseFilterProvider(ABC):
 
     @abstractmethod
     async def check_imp(self, source_provider, addr_v4, addr_v6): ...
+
+
+def filter_ipv4(value: str) -> Optional[str]:
+    return value and valid_ipv4(value) and value or None
+
+
+def filter_ipv6(value: str) -> Optional[str]:
+    return value and valid_ipv6(value) and value or None

pddnsc/cli.py

@@ -1,6 +1,5 @@
 import httpx
 import asyncio
-from abc import ABC, abstractmethod
 import toml
 from .base import BaseFilterProvider, BaseSourceProvider, BaseOutputProvider, IPAddreses
 from .plugins import use_plugins

pddnsc/outputs/vscale.py

@@ -22,7 +22,7 @@ class VscaleDomains(BaseOutputProvider):
 
     async def find_domain_id(self, client) -> Optional[int]:
         response = await client.get("/domains/")
-        if httpx.codes.is_success(response.status_code):
+        if response.is_success:
             data = response.json()
             if isinstance(data, list):
                 for entry in data:
@@ -37,7 +37,7 @@ class VscaleDomains(BaseOutputProvider):
         response = await client.get(
             f"/domains/{domain_id}/records/",
         )
-        if httpx.codes.is_success(response.status_code):
+        if response.is_success:
             data = response.json()
             if isinstance(data, list):
                 for entry in data:
@@ -50,7 +50,7 @@ class VscaleDomains(BaseOutputProvider):
 
     async def get_record_value(self, client, domain_id, record_id) -> str:
         response = await client.get(f"/domains/{domain_id}/records/{record_id}")
-        if httpx.codes.is_success(response.status_code):
+        if response.is_success:
             data = response.json()
             if isinstance(data, dict):
                 return data["content"]
@@ -68,7 +68,7 @@ class VscaleDomains(BaseOutputProvider):
             f"/domains/{domain_id}/records/{record_id}",
             json=data,
         )
-        if not httpx.codes.is_success(response.status_code):
+        if not response.is_success:
             raise RuntimeError(
                 f"failed to change record: {self.target=},{domain_id=}, {record_id=}, {record_type=}, {value=}"
             )
@@ -84,7 +84,7 @@ class VscaleDomains(BaseOutputProvider):
             f"/domains/{domain_id}/records/",
             json=data,
         )
-        if not httpx.codes.is_success(response.status_code):
+        if not response.is_success:
             raise RuntimeError(
                 f"failed to create record: {self.target=},{domain_id=}, {record_type=}, {value=}, {response.status_code=}"
             )

pddnsc/sources/http.py

@@ -0,0 +1,75 @@
+import httpx
+
+from pddnsc.base import BaseSourceProvider, filter_ipv4, filter_ipv6
+
+
+class GenericHttpSource(BaseSourceProvider):
+    def post_init(self):
+        self.url_v4 = self.config.get("url_v4")
+        self.url_v6 = self.config.get("url_v6")
+        self.headers = self.config.get("headers", {})
+
+    async def fetch_v4(self) -> str:
+        if not self.url_v4:
+            return None
+
+        result = ""
+        async with httpx.AsyncClient(
+            transport=self.ipv4t, headers=self.headers
+        ) as client:
+            response = await client.get(self.url_v4)
+            if response.is_success:
+                result = response.text.strip()
+        return filter_ipv4(result)
+
+    async def fetch_v6(self) -> str:
+        if not self.url_v6:
+            return None
+
+        result = ""
+        async with httpx.AsyncClient(
+            transport=self.ipv6t, headers=self.headers
+        ) as client:
+            response = await client.get(self.url_v6)
+            if response.is_success:
+                result = response.text.strip()
+        return filter_ipv6(result)
+
+
+class GenericHttpJsonSource(BaseSourceProvider):
+    def post_init(self):
+        self.url_v4 = self.config.get("url_v4")
+        self.url_v6 = self.config.get("url_v6")
+        self.key_v4 = self.config.get("key_v4")
+        self.key_v6 = self.config.get("key_v6")
+        if self.config.get("use_accept", True):
+            headers = {"Accept": self.config.get("acept_type", "application/json")}
+        else:
+            headers = {}
+        self.headers = headers.update(self.config.get("headers", {}))
+
+    async def fetch_v4(self) -> str:
+        if not self.url_v4 or self.key_v4 is None:
+            return None
+
+        result = ""
+        async with httpx.AsyncClient(
+            transport=self.ipv4t, headers=self.headers
+        ) as client:
+            response = await client.get(self.url_v4)
+            if response.is_success:
+                result = response.json()[self.key_v4].strip()
+        return filter_ipv4(result)
+
+    async def fetch_v6(self) -> str:
+        if not self.url_v6 or self.key_v6 is None:
+            return None
+
+        result = ""
+        async with httpx.AsyncClient(
+            transport=self.ipv6t, headers=self.headers
+        ) as client:
+            response = await client.get(self.url_v6)
+            if response.is_success:
+                result = response.json()[self.key_v6].strip()
+        return filter_ipv6(result)

pddnsc/sources/ipfy.py

@@ -1,19 +1,9 @@
-import httpx
+from pddnsc.sources.http import GenericHttpSource
 
-from pddnsc.base import BaseSourceProvider
 
 # https://www.ipify.org/
-class IPIFYSource(BaseSourceProvider):
-    async def fetch_v4(self) -> str:
-        async with httpx.AsyncClient(transport=self.ipv4t) as client:
-            response = await client.get("https://api4.ipify.org")
-            if response.is_success:
-                result = response.text.strip() or None
-        return result
-
-    async def fetch_v6(self) -> str:
-        async with httpx.AsyncClient(transport=self.ipv6t) as client:
-            response = await client.get("https://api6.ipify.org")
-            if response.is_success:
-                result = response.text.strip() or None
-        return result
+class IPIFYSource(GenericHttpSource):
+    def post_init(self):
+        super().post_init()
+        self.url_v4 = "https://api4.ipify.org"
+        self.url_v6 = "https://api6.ipify.org"

pddnsc/sources/wtfismyip.py

@@ -1,20 +1,10 @@
-import httpx
+from pddnsc.sources.http import GenericHttpSource
 
-from pddnsc.base import BaseSourceProvider
 
 # https://wtfismyip.com/
 # https://gitlab.com/wtfismyip/wtfismyip
-class WTFIsMyIP(BaseSourceProvider):
-    async def fetch_v4(self) -> str:
-        async with httpx.AsyncClient(transport=self.ipv4t) as client:
-            response = await client.get("https://text.ipv4.myip.wtf")
-            if response.is_success:
-                result = response.text.strip() or None
-        return result
-
-    async def fetch_v6(self) -> str:
-        async with httpx.AsyncClient(transport=self.ipv6t) as client:
-            response = await client.get("https://text.ipv6.myip.wtf")
-            if response.is_success:
-                result = response.text.strip() or None
-        return result
+class WTFIsMyIP(GenericHttpSource):
+    def post_init(self):
+        super().post_init()
+        self.url_v4 = "https://text.ipv4.myip.wtf"
+        self.url_v6 = "https://text.ipv6.myip.wtf"

requirements.txt

@@ -2,3 +2,4 @@ httpx[http2]>=0.26,<1.0
 asyncio>=3.4.3,<4
 aiofiles>=23,<24
 toml>=0.10,<1
+netaddr>=1,<2