updated script for python

.gitignore

@@ -0,0 +1 @@
+.mypy_cache

README.md

@@ -1,21 +1,29 @@
 # Gandi-dyndns
 
-A simple bash script to 
+A simple script to update Gandi DNS records.
 
 # Requirements
-Probably already installed on the machine:
-- bash
-- sed
-- curl
-
-To install:
-- [jq](https://stedolan.github.io/jq/)
+You will need the following python packages:
+- pyyaml, to parse the configuration file
+- requests, to perform the DNS record change
 
 # Getting started
-1. Copy the [bash file](./update.sh) somewhere
-2. Allow script execution :
+1. Copy the [script](./src/update_dns.py) somewhere
+2. Allow script execution:
+```bash
+$ chmod u+x ./update_dns.py
+```
+3. Create a configuration file (see [the example](./example_config.yml))
+4. Launch the script:
+```bash
+$ ./update_dns.py /path/to/config.yml
 ```
-$ chmod u+x ./update.sh
+5. Install the systemd [service](./systemd/update_dns.service) and [timer](./systemd/update_dns.timer) for automatic DNS update.
+```bash
+$ vim ./systemd/update_dns.service # update parameters, config file, etc.
+$ vim ./systemd/update_dns.timer   # update interval between DNS updates, if needed
+$ sudo cp ./systemd/update_dns.* /etc/systemd/system # copy files to correct systemd directory
+$ sudo systemctl daemon-reload              # reload daemon to find the timer
+$ sudo systemctl enable update_dns.timer
+$ sudo systemctl start update_dns.timer
 ```
-3. Change settings in the script
-4. Launch the script (and possibly add a CRON entry or a systemd timer)

example_config.yml

@@ -0,0 +1,18 @@
+---
+
+api_key: "MyApiKey"
+
+ip_lookup:
+  v4: "https://homnomnom.fr/ip/"
+  v6: "https://homnomnom.fr/ip/"
+
+hosts:
+  - name: "example.com"
+    ttl: 300
+    domains:
+      v4:
+        - "subdomain1"
+        - "subdomain2"
+      v6:
+        - "subdomain3"
+        - "subdomain4"

src/update_dns.py

@@ -0,0 +1,246 @@
+#!/bin/python3
+
+from typing import (
+    Optional,
+    Tuple
+)
+
+import yaml
+import requests
+import socket
+import requests.packages.urllib3.util.connection as urllib3_cn
+
+class GandiDNS:
+    api_endpoint = "https://dns.api.gandi.net/api/v5"
+    api_key = None
+
+
+    def __init__(self, key: str):
+        """
+        Constructor
+        """
+        self.api_key = key
+
+
+    def _get_api(self, endpoint: str) -> dict:
+        """
+        Calls the API with a GET request on a given endpoint
+        """
+        r = requests.get(
+            f"{ self.api_endpoint }/{ endpoint }",
+            headers={
+                "Content-Type": "application/json",
+                "X-Api-Key": self.api_key
+            }
+        )
+        r.raise_for_status()
+        return r.json()
+
+
+    def _post_api(self, endpoint: str, payload: Optional[dict] = None) -> dict:
+        """
+        Calls the API with a POST request on a given endpoint
+
+        Arguments:
+            endpoint: api endpoint to call
+            payload: a dict that will be transformed in JSON
+        """
+
+
+    def get_domain_uuid(self, domain: str) -> str:
+        """
+        Get the Zone UUID for a specific domain name
+        """
+        return self._get_api( f"domains/{ domain }" )["zone_uuid"]
+
+
+    def _replace_ip_record(self, uuid: str, subdomain: str, record: str, ip: str, ttl: int) -> Optional[str]:
+        """
+        Replaces the IP in a DNS record 
+
+        Arguments:
+            uuid: Zone UUID of the domain to change (see `get_domain_uuid()`)
+            subdomain: name of the subdomain to change
+            record: either 'A' for IPv4, or 'AAAA' for IPv6
+            ip: IP address to use
+            ttl: time to live of that record
+
+        Returns:
+            None if the IP address was already correct,
+            the previous IP address otherwise
+        """
+        endpoint = f"zones/{uuid}/records/{subdomain}/{record}"
+
+        previous_ip = self._get_api(endpoint)["rrset_values"][0]
+        if previous_ip == ip:
+            return None
+
+        self._post_api(endpoint, {
+            "rrset_ttl": ttl,
+            "rrset_values": [ip]
+        })
+
+        return previous_ip
+
+    def replace_ipv4_record(self, uuid: str, subdomain: str, ip: str, ttl: int) -> Optional[str]:
+        """
+        Replaces an IPv4 in a DNS record.
+        See _replace_ip_record() for arguments and return values
+        """
+        return self._replace_ip_record(uuid, subdomain, 'A', ip, ttl)
+
+
+    def replace_ipv6_record(self, uuid: str, subdomain: str, ip: str, ttl: int) -> Optional[str]:
+        """
+        Replaces an IPv6 in a DNS record.
+        See _replace_ip_record() for arguments and return values
+        """
+        return self._replace_ip_record(uuid, subdomain, 'AAAA', ip, ttl)
+
+
+#############
+# Functions #
+#############
+
+def force_ipv4():
+    """
+    Function to be used as allowed_gai_family, but forces IPv4
+    """
+    return socket.AF_INET
+
+
+def force_ipv6():
+    """
+    Function to be used as allowed_gai_family, but forces IPv6
+    """
+    return socket.AF_INET6
+
+
+def get_ip( lookup_v4: str, lookup_v6: Optional[str], *, use_ipv6 = True ) -> Tuple[ str, Optional[str] ]:
+    """
+    Gets both IPv4 and IPv6 (if possible) of the current computer
+
+    This uses an external server to find the actual IP.
+    This server should return nothing but the IP.
+    One such server is https://homnomnom.fr/ip/
+
+    Arguments:
+        lookup_v4: URL to use for looking up the IPv4 address
+        lookup_v6: URL to use for looking up the IPv6 address.
+                   If None, will use lookup_v4 for this
+        use_ipv6: set to False to avoid looking up IPv6
+
+    Return:
+        A tuple [ IPv4, IPv6 ] of the addresses.
+        IPv6 may be None if IPv6 is not available, or if use_ipv6 is False
+    """
+    gai_family = urllib3_cn.allowed_gai_family
+
+    urllib3_cn.allowed_gai_family = force_ipv4
+    r = requests.get(lookup_v4)
+    r.raise_for_status()
+    ipv4 = r.text
+
+    ipv6 = None
+    if urllib3_cn.HAS_IPV6 and use_ipv6:
+        if lookup_v6 is None:
+            lookup_v6 = lookup_v4
+        
+        urllib3_cn.allowed_gai_family = force_ipv6
+        try:
+            r = requests.get(lookup_v6)
+            r.raise_for_status()
+            ipv6 = r.text
+        except OSError:
+            ipv6 = None
+
+    # stop forcing IPv4 / IPv6
+    urllib3_cn.allowed_gai_family = gai_family
+    return ( ipv4, ipv6 )
+
+
+def parse_script_arguments():
+    """
+    Parses the arguments
+    """
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        description="Gandi dynamic DNS"
+    )
+
+    parser.add_argument("config", type=str,
+        help="The config file to use"
+    )
+
+    parser.add_argument("--key", type=str,
+        help="API key to use"
+    )
+
+    return parser.parse_args()
+
+
+def display_ip_change( subdomain: str, host: str, old_ip: Optional[str], new_ip: Optional[str]):
+    """
+    Displays if the old IP was OK or if it was changed
+    """
+    header = f" - {subdomain}.{host}"
+    if old_ip is None:
+        print( f"{header}: IP already OK")
+    else:
+        print( f"{header}: replaced {old_ip} with {new_ip}")
+
+###############
+# Main script #
+###############
+
+if __name__ == "__main__":
+    args = parse_script_arguments()
+    with open(args.config, 'r') as file:
+        params = yaml.safe_load(file)
+
+    if args.key is not None:
+        params["api_key"] = args.key
+
+    gandi = GandiDNS( params["api_key"] )
+
+    ipv4, ipv6 = get_ip(
+        params["ip_lookup"]["v4"],
+        params["ip_lookup"]["v6"]
+    )
+
+    print( f"IPv4 detected: {ipv4}")
+    if ipv6 is not None:
+        print( f"IPv6 detected: {ipv6}")
+    else:
+        print( "IPv6 not supported")
+
+    for host in params["hosts"]:
+        assert "name" in host
+        assert "ttl" in host
+        assert "domains" in host
+
+        hostname = host["name"]
+        print( f"{hostname}:" )
+
+        uuid = gandi.get_domain_uuid(hostname)
+        print( f" - UUID: {uuid}" )
+
+        domains = host["domains"]
+        if "v4" in domains:
+            for domain in domains["v4"]:
+                display_ip_change(
+                    domain,
+                    hostname,
+                    gandi.replace_ipv4_record(uuid, domain, ipv4, host["ttl"]),
+                    ipv4
+                ) 
+
+        if "v6" in domains and ipv6 is not None:
+            for domain in domains["v6"]:
+                display_ip_change(
+                    domain,
+                    hostname,
+                    gandi.replace_ipv6_record(uuid, domain, ipv6, host["ttl"]),
+                    ipv6
+                ) 

systemd/update_dns.service

@@ -0,0 +1,9 @@
+[Unit]
+Description=Update the dynamic DNS record
+
+[Service]
+Type=oneshot
+ExecStart="/path/to/update_dns.py"
+
+[Install]
+WantedBy=multi-user.target

systemd/update_dns.timer

@@ -0,0 +1,9 @@
+[Unit]
+Description=Update DNS record on boot and periodically
+
+[Timer]
+OnBootSec=5min
+OnUnitActiveSec=3h
+
+[Install]
+WantedBy=timers.target

update_dns.sh

@@ -1,97 +0,0 @@
-#!/bin/bash
-
-api_key='MyApiKey'
-api_endpoint='https://dns.api.gandi.net/api/v5'
-
-ifconfig='https://homnomnom.fr/ip/'
-
-host="example.com"
-ttl='300'
-
-domainsV4=(
-    "subdomain1"
-    "subdomain2"
-)
-
-domainsV6=(
-    "subdomain3"
-    "subdomain4"
-)
-
-#script below
-function die {
-    echo -e $1 >&2
-    kill -HUP $$
-}
-
-function get_api {
-    tmpfile=$(mktemp)
-    http_code=$(curl --silent "${api_endpoint}/${1}" \
-                     --header "Content-Type: application/json" \
-                     --header "X-Api-Key: ${api_key}" \
-                     --output ${tmpfile} \
-                     --write-out '%{http_code}')
-
-    message=$(cat ${tmpfile})
-    rm ${tmpfile}
-
-    if [[ ${http_code} != "200" ]]; then
-        die "API GET call exited with code ${http_code}.\nMessage: ${message}"
-    fi
-
-    echo ${message}
-}
-
-function put_api {
-    tmpfile=$(mktemp)
-    http_code=$(curl --request PUT \
-         --header "Content-Type: application/json" \
-         --header "X-Api-Key: ${api_key}" \
-         --silent \
-         --output ${tmpfile} \
-         --write-out '%{http_code}' \
-         "${api_endpoint}/${1}" \
-         -d "${2}")
-
-    message=$(cat ${tmpfile})
-    rm ${tmpfile}
-
-    if [[ ${http_code} != "201" ]]; then
-        die "API PUT call exited with code ${http_code}.\nMessage: ${message}"
-    fi
-
-    echo ${message}
-}
-
-#find out zone UUID for domain
-uuid=$(get_api "domains/${host}")
-uuid=$(echo ${uuid} | jq '.zone_uuid' | sed 's/"//g')
-echo "UUID: ${uuid}"
-
-ipv4=$(curl -4 --silent ${ifconfig})
-echo "detected IPv4 address: ${ipv4}"
-for domain in ${domainsV4[@]}; do
-    ip=$(get_api "zones/${uuid}/records/${domain}/A" | jq '.rrset_values[0]' | sed 's/"//g')
-    if [ "${ip}" == "${ipv4}" ]; then
-        echo "- ${domain}.${host}: IP already OK"
-    else
-        payload='{"rrset_ttl": '${ttl}', "rrset_values": ["'${ipv4}'"]}'
-
-        put_api "zones/${uuid}/records/${domain}/A" "${payload}" >/dev/null
-        echo "- ${domain}.${host}: replaced ${ip} with ${ipv4}"
-    fi
-done
-
-ipv6=$(curl -6 --silent ${ifconfig})
-echo "detected IPv6 address: ${ipv6}"
-for domain in ${domainsV6[@]}; do
-    ip=$(get_api "zones/${uuid}/records/${domain}/AAAA" | jq '.rrset_values[0]' | sed 's/"//g')
-    if [ "${ip}" == "${ipv6}" ]; then
-        echo "- ${domain}.${host}: IP already OK"
-    else
-        payload='{"rrset_ttl": '${ttl}', "rrset_values": ["'${ipv6}'"]}'
-
-        put_api "zones/${uuid}/records/${domain}/AAAA" "${payload}" >/dev/null
-        echo "- ${domain}.${host}: replaced ${ip} with ${ipv6}"
-    fi
-done