109 lines
3.5 KiB
Python
Executable file
109 lines
3.5 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
"""A script which sends a Dynamic DNS update to the Ionos API."""
|
|
|
|
from pathlib import Path
|
|
import requests
|
|
from yaml import load
|
|
try:
|
|
from yaml import CLoader as Loader
|
|
except ImportError:
|
|
from yaml import Loader
|
|
|
|
|
|
_API_ENDPOINT_URL = "https://api.hosting.ionos.com/dns/v1/dyndns"
|
|
|
|
|
|
class InvalidConfigurationError(RuntimeError):
|
|
"""Raised when the configuration file's contents are invalid."""
|
|
|
|
|
|
def main():
|
|
path_to_config_file = Path("config.yml").resolve()
|
|
|
|
# Check if config file exists
|
|
if not path_to_config_file.is_file():
|
|
raise FileNotFoundError(
|
|
f"No such file: '{path_to_config_file}'."
|
|
)
|
|
|
|
# Load config file
|
|
with open(path_to_config_file, 'r') as config_file:
|
|
config = load(config_file, Loader=Loader)
|
|
|
|
# Validate config file
|
|
for top_level_key in ["apikey", "domains"]:
|
|
if not top_level_key in config:
|
|
raise InvalidConfigurationError(
|
|
f"Failed to locate key '{top_level_key}' in config."
|
|
)
|
|
# Validate 'apikey' values
|
|
for key in ["prefix", "secret"]:
|
|
if not key in config["apikey"]:
|
|
raise InvalidConfigurationError(
|
|
f"Expected key 'apikey.{key}' in config:"
|
|
)
|
|
if not isinstance(config["apikey"][key], str):
|
|
raise InvalidConfigurationError(
|
|
f"Expected type string as the value for 'apikey.{key}'."
|
|
)
|
|
# Validate 'domains' values
|
|
if not isinstance(config["domains"], list):
|
|
raise InvalidConfigurationError(
|
|
"Expected type list as the value for 'domains'."
|
|
)
|
|
for value in config["domains"]:
|
|
if not isinstance(value, str):
|
|
raise InvalidConfigurationError(
|
|
f"Expected only strings in the 'domains' list."
|
|
)
|
|
# Validate 'description' value
|
|
if "description" in config:
|
|
if not isinstance(config["description"], str):
|
|
raise InvalidConfigurationError(
|
|
"Expected type string as value for 'description'."
|
|
)
|
|
|
|
# Parse config
|
|
apikey_prefix = config["apikey"]["prefix"]
|
|
apikey_secret = config["apikey"]["secret"]
|
|
apikey = f"{apikey_prefix}.{apikey_secret}"
|
|
domains = config["domains"]
|
|
description = config["description"] if "description" in config else "Dynamic DNS update."
|
|
|
|
# Make an API call to retrieve the DNS update URL
|
|
response = requests.post(
|
|
_API_ENDPOINT_URL,
|
|
headers={"X-API-Key": apikey},
|
|
data={
|
|
"domains": domains,
|
|
"description": description,
|
|
}
|
|
)
|
|
|
|
if not response.status_code == 200:
|
|
try:
|
|
error_message = f"{response.status_code}: {response.json()}"
|
|
except requests.exceptions.JSONDecodeError:
|
|
error_message = f"{response.status_code}"
|
|
raise RuntimeError(
|
|
f"API request failed and returned: {error_message}"
|
|
)
|
|
|
|
update_url = response.json()["updateUrl"]
|
|
print(update_url) # DEBUG
|
|
|
|
# Send a dynamic DNS update using the retrieved update URL
|
|
response = requests.get(update_url)
|
|
if not response.status_code == 200:
|
|
try:
|
|
error_message = f"{response.status_code}: {response.json()}"
|
|
except requests.exceptions.JSONDecodeError:
|
|
error_message = f"{response.status_code}"
|
|
raise RuntimeError(
|
|
f"DNS update request failed and returned: {error_message}"
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|