Add the main application code.

.flake8

@@ -1,5 +1,5 @@
 [flake8]
-max-line-length=110
+max-line-length=120
 docstring-convention=all
 import-order-style=pycharm
 exclude=constants.py,__pycache__,.cache,.git,.md,.svg,.png,tests,venv,.venv
@@ -18,3 +18,5 @@ ignore=
     ANN002,ANN003,ANN101,ANN102,ANN204,ANN206
     # Whitespace Before
     E203
+    # F-strings
+    F541

Dockerfile

@@ -2,7 +2,7 @@ FROM python:3.9-slim
 
 WORKDIR /app
 
-ENTRYPOINT ["python3", "-m", "cloudflare-ddns"]
+ENTRYPOINT ["python3", "-m", "cloudflare_ddns"]
 CMD []
 
 # Install requirements in a separate step to not rebuild everything when the requirements are updated.

cloudflare_ddns/__main__.py

@@ -0,0 +1,61 @@
+import logging
+from os import environ
+from typing import Tuple
+
+import click
+from cloudflare_ddns.app import ApplicationJob
+from cloudflare_ddns.constants import BASE_ENV_VAR, DEFAULT_DELAY, DOMAINS_ENV_VAR
+
+log = logging.getLogger("ddns")
+
+
+@click.command()
+@click.option(
+    '--delay', '-d',
+    default=DEFAULT_DELAY,
+    show_default=True,
+    show_envvar=True,
+    help="Time to wait between each update."
+)
+@click.option(
+    '--token', '-k',
+    prompt="Enter your Cloudflare Token",
+    hide_input=True,
+    show_envvar=True,
+    help="Your Cloudflare Bearer token."
+)
+@click.option('-v', '--verbose', is_flag=True, default=False, help="Show debug logging.")
+@click.argument("domains", nargs=-1)
+def start(delay: str, token: str, verbose: int, domains: Tuple[str]) -> None:
+    """
+    Update Cloudflare DNS RECORDS to your current IP every <delay> (default: 5 minutes).
+
+    The domains can be passed either as command line options, or as a space separated CF_DDNS_DOMAINS environment
+    variable.
+    Each domain can be preceded by the record type, either A or AAAA followed by a colon.
+
+    \b
+    The duration supports the following symbols for each unit of time:
+        - days: `d`, `D`, `day`, `days`
+        - hours: `H`, `h`, `hour`, `hours`
+        - minutes: `M`, `m`, `minute`, `minutes`, `min`
+        - seconds: `S`, `s`, `second`, `seconds`, `sec`
+    The units need to be provided in descending order of magnitude.
+    """
+    # Configure logging.
+    logging.basicConfig(
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        level=logging.DEBUG if verbose else logging.INFO
+    )
+
+    # Get domains from the environment variable.
+    if domain_var := environ.get(DOMAINS_ENV_VAR, None):
+        domains = list(domains)
+        domains.extend(domain_var.split(" "))
+
+    ApplicationJob(delay, token, domains).launch()
+
+
+# Main entrypoint
+if __name__ == "__main__":
+    start(auto_envvar_prefix=BASE_ENV_VAR)

cloudflare_ddns/app.py

@@ -0,0 +1,193 @@
+import logging
+import threading
+from dataclasses import dataclass
+from typing import List
+
+import requests
+from cloudflare_ddns.constants import ACCEPTED_RECORDS, LIST_DNS, LIST_ZONES, PATCH_DNS, VERIFY_TOKEN
+from cloudflare_ddns.utils import BearerAuth, CloudflareHTTPError, check_status, get_ip, parse_duration
+from requests import HTTPError
+
+log = logging.getLogger("ddns")
+
+
+@dataclass
+class Domain:
+    """
+    Dataclass representing one domain record to update.
+
+    Args:
+        domain: The domain name.
+        record_type: The type of the record to update.
+        zone: The ID of the Cloudflare zone it belongs to.
+        id: The Cloudflare ID of this record.
+    """
+
+    domain: str
+    record_type: str
+    zone: str
+    id: str
+
+
+class ApplicationJob(threading.Thread):
+    """Main application class."""
+
+    def __init__(self, raw_delay: str, token: str, raw_domains: List[str]):
+        super().__init__()
+
+        self.stop_signal = threading.Event()
+
+        self.delay = None
+        self.domains: List[Domain] = []
+        self.current_ip = None
+
+        self.auth = BearerAuth(token)
+
+        self.raw_domains = raw_domains
+        self.raw_delay = raw_delay
+
+    def launch(self) -> None:
+        """Launch the application by validating arguments and starting the thread."""
+        self.validate_arguments()
+        log.debug("Starting job.")
+        self.start()
+
+    def run(self) -> None:
+        """Main application function, in charge of controlling the periodic updates."""
+        log.debug("Parsing domains.")
+        self.parse_domains()
+        log.debug(f"Using domains: {', '.join(f'{domain.record_type}:{domain.domain}' for domain in self.domains)}")
+
+        log.info(f"Starting app. Records will be updated every {self.delay} seconds.")
+        try:
+            self.update_records()
+
+        except HTTPError as e:
+            log.error(
+                f"HTTP error {'from Cloudflare' if isinstance(e, CloudflareHTTPError) else ''} "
+                f"while updating records for the first time, aborting."
+            )
+            log.error(e)
+            log.info("Exiting with code 70.")
+
+        except Exception:
+            log.exception("Error while updating records for the first time, aborting.")
+            log.info("Exiting with code 70.")
+            exit(70)
+
+        while not self.stop_signal.wait(self.delay):
+            try:
+                self.update_records()
+            except Exception:
+                log.exception(f"Error while updating records. Retrying in {self.delay} seconds.")
+
+    def update_records(self) -> None:
+        """Update all the registered records."""
+        log.info("Starting record update.")
+        for record in self.domains:
+            log.debug(f"Updating record for {record.domain}.")
+
+            check_status(requests.patch(
+                PATCH_DNS.format(zone_identifier=record.zone, identifier=record.id),
+                json={"content": get_ip(record.record_type == 'AAAA')},
+                auth=self.auth
+            ))
+
+        log.info("Successfully updated records.")
+
+    def parse_domains(self) -> None:
+        """Parse the domain in `raw_domains` and populate the `domains` array with `Domain` objects."""
+        found_domains = {}
+
+        for zone_json in check_status(requests.get(LIST_ZONES, auth=self.auth)).json()["result"]:
+            for record_json in check_status(requests.get(
+                    LIST_DNS.format(zone_identifier=zone_json["id"]),
+                    auth=self.auth
+            )).json()["result"]:
+                if record_json["type"] in ACCEPTED_RECORDS:
+                    domain = Domain(
+                        record_json["name"],
+                        record_json["type"],
+                        record_json["zone_id"],
+                        record_json["id"]
+                    )
+                    found_domains[f'{record_json["name"]}-{record_json["type"]}'] = domain
+
+        log.debug(
+            f"Found domains: "
+            f"{', '.join(f'{domain.record_type}:{domain.domain}' for domain in found_domains.values())}"
+        )
+        for domain in self.raw_domains:
+            if ':' in domain:
+                type_, domain = domain.split(':', maxsplit=1)
+
+                if type_ not in ACCEPTED_RECORDS:
+                    log.error(f"Invalid record type {type_}. Must be one of {', '.join(ACCEPTED_RECORDS)}.")
+                    log.info(f"Exiting with code 65.")
+                    exit(65)
+
+                if f"{domain}-{type_}" not in found_domains:
+                    log.error(
+                        f"Cannot find an {type_} record for the domain {domain} in your Cloudflare settings. "
+                        f"Have you defined this record yet?"
+                    )
+                    log.info(f"Exiting with code 65.")
+                    exit(65)
+
+                self.domains.append(found_domains[f"{domain}-{type_}"])
+
+            else:
+                found = False
+
+                if f"{domain}-A" in found_domains:
+                    self.domains.append(found_domains[f"{domain}-A"])
+                    found = True
+
+                if f"{domain}-AAAA" in found_domains:
+                    self.domains.append(found_domains[f"{domain}-AAAA"])
+                    found = True
+
+                if not found:
+                    log.error(
+                        f"Cannot find the domain {domain} in your Cloudflare settings. "
+                        f"Have you defined this record yet?"
+                    )
+                    log.info(f"Exiting with code 65.")
+                    exit(65)
+
+    def validate_arguments(self) -> None:
+        """Validate the provided arguments."""
+        failed = False
+
+        if not self.raw_domains:
+            log.error("Please provide at least one domain.")
+            failed = True
+
+        try:
+            self.delay = parse_duration(self.raw_delay)
+        except ValueError as e:
+            log.error(f"Failed to parse delay: {e}")
+            failed = True
+
+        if not failed:
+            try:
+                log.debug("Validating bearer token.")
+
+                self.validate_bearer()
+            except ValueError as e:
+                log.error(f"Failed to validate bearer token: {e}")
+                failed = True
+            else:
+                log.info("Successfully validated the bearer token.")
+
+        if failed:
+            log.info("Exiting with code 64.")
+            exit(64)
+
+    def validate_bearer(self) -> None:
+        """Utility method to validate a CF bearer token."""
+        r = requests.get(VERIFY_TOKEN, auth=self.auth)
+
+        if not r.json()["success"]:
+            error_message = ' / '.join(error["message"] for error in r.json()["errors"])
+            raise ValueError(error_message)

cloudflare_ddns/constants.py

@@ -0,0 +1,21 @@
+# App defaults
+DEFAULT_DELAY = "5 minutes"
+
+# App constants
+BASE_ENV_VAR = "CF_DDNS"
+DOMAINS_ENV_VAR = BASE_ENV_VAR + "_DOMAINS"
+
+# Endpoints
+BASE_ENDPOINT = "https://api.cloudflare.com/client/v4/"
+
+VERIFY_TOKEN = BASE_ENDPOINT + "user/tokens/verify"
+
+LIST_ZONES = BASE_ENDPOINT + "zones"
+LIST_DNS = BASE_ENDPOINT + "zones/{zone_identifier}/dns_records"
+PATCH_DNS = BASE_ENDPOINT + "zones/{zone_identifier}/dns_records/{identifier}"
+
+# Utilities
+ACCEPTED_RECORDS = ('A', 'AAAA')
+
+IP_API_URL_IPV4 = "https://api.ipify.org/"
+IP_API_URL_IPV6 = "https://api6.ipify.org/"

cloudflare_ddns/utils.py

@@ -0,0 +1,77 @@
+import re
+
+import requests
+from cloudflare_ddns.constants import BASE_ENDPOINT, IP_API_URL_IPV4, IP_API_URL_IPV6
+from requests import HTTPError, Request, Response, codes
+from requests.auth import AuthBase
+
+DURATION_REGEX = re.compile(
+    r"((?P<days>\d+?) ?(days|day|D|d) ?)?"
+    r"((?P<hours>\d+?) ?(hours|hour|H|h) ?)?"
+    r"((?P<minutes>\d+?) ?(minutes|minute|min|M|m) ?)?"
+    r"((?P<seconds>\d+?) ?(seconds|second|sec|S|s))?"
+)
+UNIT_TO_SECONDS = {
+    "days": 86400,
+    "hours": 3600,
+    "minutes": 60,
+    "seconds": 1
+}
+
+
+def parse_duration(duration: str) -> int:
+    """
+    Parameter type for durations.
+
+    The converter supports the following symbols for each unit of time:
+        - days: `d`, `D`, `day`, `days`
+        - hours: `H`, `h`, `hour`, `hours`
+        - minutes: `M`, `m`, `minute`, `minutes`, `min`
+        - seconds: `S`, `s`, `second`, `seconds`, `sec`
+        The units need to be provided in descending order of magnitude.
+    """
+    match = DURATION_REGEX.fullmatch(duration)
+    if not match:
+        raise ValueError(f"{duration} isn't a valid duration.")
+
+    duration = 0
+    for unit, time_value in match.groupdict().items():
+        if time_value:
+            duration += int(time_value) * UNIT_TO_SECONDS[unit]
+
+    return duration
+
+
+class BearerAuth(AuthBase):
+    """Bearer based authentication."""
+
+    def __init__(self, token: str) -> None:
+        self.token = token
+
+    def __call__(self, r: Request) -> Request:
+        """Attach the Authorization header to the request."""
+        r.headers["Authorization"] = f"Bearer {self.token}"
+        return r
+
+
+def get_ip(ipv6: bool) -> str:
+    """Return the host public IP as detected by ipify.org."""
+    r = check_status(requests.get(IP_API_URL_IPV4 if not ipv6 else IP_API_URL_IPV6))
+    return r.text
+
+
+class CloudflareHTTPError(HTTPError):
+    """HTTPError coming from a Cloudflare endpoint."""
+
+
+def check_status(r: Response) -> Response:
+    """Check the status code of a response and return it."""
+    if not r.status_code == codes.ok:
+        if r.url.startswith(BASE_ENDPOINT) and not r.json()["success"]:
+            errors = "\n".join(f"{err['code']}: {err['message']}" for err in r.json()["errors"])
+
+            raise CloudflareHTTPError(f"{r.status_code} {r.reason} while querying {r.url}: {errors}")
+        else:
+            r.raise_for_status()
+
+    return r