transip-client/transip_client/adapters.py
Sonny Bakker 84558826fb
All checks were successful
ci/woodpecker/push/tests Pipeline was successful
Adapter refactor
Allows using different ways to retrieve hosts IP address
2025-05-03 13:51:49 +02:00

86 lines
2.1 KiB
Python

import logging
import subprocess
import requests
logger = logging.getLogger(__name__)
class Adapter:
def get_ip(self) -> str:
raise NotImplementedError
class HTTPAdapter(Adapter):
service: str
def handle_response(self, response: requests.Response) -> str:
raise NotImplementedError
def get_params(self) -> dict:
raise NotImplementedError
def get_ip(self) -> str:
try:
response = requests.get(self.service, params=self.get_params(), timeout=5)
response.raise_for_status()
except requests.RequestException as e:
raise OSError(f"Unable to retrieve current IP from {self.service}") from e
ip = self.handle_response(response)
if not ip:
raise OSError(f"Unable to determine IP from response from {self.service}")
return ip
class IpifyAdapter(HTTPAdapter):
service: str = "https://api.ipify.org"
def get_params(self) -> dict:
return {"format": "text"}
def handle_response(self, response: requests.Response) -> str:
return response.text
class DNSAdapter(Adapter):
resolvers: list[str]
dns: str
def try_resolver(self, resolver: str) -> str:
output = subprocess.check_output(
["dig", "+short", self.dns, resolver], stderr=subprocess.STDOUT, text=True
)
return output.strip()
def get_ip(self) -> str:
for resolver in self.resolvers:
try:
ip = self.try_resolver(resolver)
except subprocess.CalledProcessError as e:
if e.returncode == 9:
continue
raise OSError("Unable to retrieve current IP") from e
if not ip:
logger.warning(f"No IP returned from {resolver}")
continue
return ip
raise OSError("Exhausted all known IP resolvers, unable to retrieve IP")
class OpenDNSAdapter(DNSAdapter):
dns = "myip.opendns.com"
resolvers = [
"@resolver1.opendns.com",
"@resolver2.opendns.com",
"@resolver3.opendns.com",
"@resolver4.opendns.com",
]