b4tman/pddnsc
1
0
Fork 0
Code Issues 8 Pull Requests Actions Packages Projects 1 Releases Activity

Compare commits

base: b4tman:c0d1d4b38d5b03b4f0afd70997106705b237aa3c
Branches Tags
b4tman:master
..
compare: b4tman:3e9142cd5408ebe226d3b6197a4c37b0b9737454
Branches Tags
b4tman:master

No commits in common. "c0d1d4b38d5b03b4f0afd70997106705b237aa3c" and "3e9142cd5408ebe226d3b6197a4c37b0b9737454" have entirely different histories.
c0d1d4b38d ... 3e9142cd54

7 changed files with 39 additions and 103 deletions
Show Stats Expand all files Collapse all files

11
pddnsc/base.py
View File

@ -1,8 +1,7 @@
import httpx
import asyncio
from abc import ABC, abstractmethod
from typing import NamedTuple, Optional
from netaddr import valid_ipv4, valid_ipv6
from typing import NamedTuple
class IPAddreses(NamedTuple):
@ -155,11 +154,3 @@ 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

1
pddnsc/cli.py
View File

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

10
pddnsc/outputs/vscale.py
View File

@ -22,7 +22,7 @@ class VscaleDomains(BaseOutputProvider):
async def find_domain_id(self, client) -> Optional[int]:
response = await client.get("/domains/")
if response.is_success:
if httpx.codes.is_success(response.status_code):
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 response.is_success:
if httpx.codes.is_success(response.status_code):
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 response.is_success:
if httpx.codes.is_success(response.status_code):
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 response.is_success:
if not httpx.codes.is_success(response.status_code):
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 response.is_success:
if not httpx.codes.is_success(response.status_code):
raise RuntimeError(
f"failed to create record: {self.target=},{domain_id=}, {record_type=}, {value=}, {response.status_code=}"
)

75
pddnsc/sources/http.py
View File

@ -1,75 +0,0 @@
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)

22
pddnsc/sources/ipfy.py
View File

@ -1,9 +1,19 @@
from pddnsc.sources.http import GenericHttpSource
import httpx
from pddnsc.base import BaseSourceProvider
# https://www.ipify.org/
class IPIFYSource(GenericHttpSource):
def post_init(self):
super().post_init()
self.url_v4 = "https://api4.ipify.org"
self.url_v6 = "https://api6.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

22
pddnsc/sources/wtfismyip.py
View File

@ -1,10 +1,20 @@
from pddnsc.sources.http import GenericHttpSource
import httpx
from pddnsc.base import BaseSourceProvider
# https://wtfismyip.com/
# https://gitlab.com/wtfismyip/wtfismyip
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"
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

1
requirements.txt
View File

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