From d44c8d0a71ece79eb097edefbcf981abdd05dc08 Mon Sep 17 00:00:00 2001 From: Matteo Bertucci Date: Thu, 14 Jan 2021 16:39:26 +0100 Subject: [PATCH] Switching to an ApplicationJob based app --- Dockerfile | 2 +- cloudflare-ddns/__main__.py | 43 ------------ .../__init__.py | 0 cloudflare_ddns/__main__.py | 30 +++++++++ cloudflare_ddns/app.py | 67 +++++++++++++++++++ .../constants.py | 0 {cloudflare-ddns => cloudflare_ddns}/utils.py | 10 --- 7 files changed, 98 insertions(+), 54 deletions(-) delete mode 100644 cloudflare-ddns/__main__.py rename {cloudflare-ddns => cloudflare_ddns}/__init__.py (100%) create mode 100644 cloudflare_ddns/__main__.py create mode 100644 cloudflare_ddns/app.py rename {cloudflare-ddns => cloudflare_ddns}/constants.py (100%) rename {cloudflare-ddns => cloudflare_ddns}/utils.py (81%) diff --git a/Dockerfile b/Dockerfile index 47c2486..136079d 100644 --- a/Dockerfile +++ b/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. diff --git a/cloudflare-ddns/__main__.py b/cloudflare-ddns/__main__.py deleted file mode 100644 index ab4033a..0000000 --- a/cloudflare-ddns/__main__.py +++ /dev/null @@ -1,43 +0,0 @@ -import logging - -import click - -from .constants import DEFAULT_DELAY -from .utils import parse_duration, validate_bearer - -log = logging.getLogger("ddns") - - -@click.command() -@click.option('--delay', '-d', default=DEFAULT_DELAY, show_default=True) -@click.option('--token', '-k', prompt="Enter your Cloudflare Token", hide_input=True, show_envvar=True) -@click.option('-v', '--verbose', is_flag=True, default=False) -def start(delay: str, token: str, verbose: int) -> None: - """Main application entrypoint.""" - logging.basicConfig( - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - level=logging.DEBUG if verbose else logging.INFO - ) - - try: - duration = parse_duration(delay) - except ValueError as e: - log.error(f"Failed to parse delay: {e}") - log.error("Exiting with code 64.") - exit(64) - - try: - log.debug("Validating bearer token.") - - validate_bearer(token) - except ValueError as e: - log.error(f"Failed to valid bearer token: {e}") - log.error("Exiting with code 64.") - exit(64) - else: - log.info("Successfully validated the bearer token.") - - -# Main entrypoint -if __name__ == "__main__": - start(auto_envvar_prefix="CF_DDNS") diff --git a/cloudflare-ddns/__init__.py b/cloudflare_ddns/__init__.py similarity index 100% rename from cloudflare-ddns/__init__.py rename to cloudflare_ddns/__init__.py diff --git a/cloudflare_ddns/__main__.py b/cloudflare_ddns/__main__.py new file mode 100644 index 0000000..f113635 --- /dev/null +++ b/cloudflare_ddns/__main__.py @@ -0,0 +1,30 @@ +import logging + +import click +from typing import Tuple + +from cloudflare_ddns.app import ApplicationJob +from cloudflare_ddns.constants import DEFAULT_DELAY + +log = logging.getLogger("ddns") + + +@click.command() +@click.option('--delay', '-d', default=DEFAULT_DELAY, show_default=True) +@click.option('--token', '-k', prompt="Enter your Cloudflare Token", hide_input=True, show_envvar=True) +@click.option('-v', '--verbose', is_flag=True, default=False) +@click.option('--ipv6/--ipv4', '-6/-4') +@click.argument("domain", nargs=-1) +def start(delay: str, token: str, verbose: int, domain: Tuple[str], ipv6: bool) -> None: + """Main application entrypoint.""" + logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.DEBUG if verbose else logging.INFO + ) + + ApplicationJob(delay, token, domain, ipv6).launch() + + +# Main entrypoint +if __name__ == "__main__": + start(auto_envvar_prefix="CF_DDNS") diff --git a/cloudflare_ddns/app.py b/cloudflare_ddns/app.py new file mode 100644 index 0000000..32af2ab --- /dev/null +++ b/cloudflare_ddns/app.py @@ -0,0 +1,67 @@ +import logging +import threading +from dataclasses import dataclass +from typing import Tuple + +import requests + +from cloudflare_ddns.constants import VERIFY_TOKEN +from cloudflare_ddns.utils import parse_duration, BearerAuth + +log = logging.getLogger("ddns") + + +@dataclass +class Domain: + domain: str + record_type: str + + +class ApplicationJob(threading.Thread): + def __init__(self, delay: str, token: str, raw_domains: Tuple[str], default_ipv6: bool): + super().__init__() + + self.stop_signal = threading.Event() + + self.delay = delay + self.auth = BearerAuth(token) + self.raw_domains = raw_domains + self.default_ipv6 = default_ipv6 + + def launch(self) -> None: + self.validate_arguments() + + def validate_arguments(self): + failed = False + + if not self.raw_domains: + log.error("Please provide at least one domain.") + failed = True + + try: + self.delay = parse_duration(self.delay) + except ValueError as e: + log.error(f"Failed to parse delay: {e}") + failed = True + + 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) diff --git a/cloudflare-ddns/constants.py b/cloudflare_ddns/constants.py similarity index 100% rename from cloudflare-ddns/constants.py rename to cloudflare_ddns/constants.py diff --git a/cloudflare-ddns/utils.py b/cloudflare_ddns/utils.py similarity index 81% rename from cloudflare-ddns/utils.py rename to cloudflare_ddns/utils.py index 487f48e..49b8d2a 100644 --- a/cloudflare-ddns/utils.py +++ b/cloudflare_ddns/utils.py @@ -53,13 +53,3 @@ class BearerAuth(AuthBase): """Attach the Authorization header to the request.""" r.headers["Authorization"] = f"Bearer {self.token}" return r - - -def validate_bearer(token: str) -> None: - """Utility method to validate a CF bearer token.""" - bearer = BearerAuth(token) - r = requests.get(VERIFY_TOKEN, auth=bearer) - - if not r.json()["success"]: - error_message = ' / '.join(error["message"] for error in r.json()["errors"]) - raise ValueError(error_message)