timatlee/cloudflare-ddns-docker-updated
2
0
Fork 0
forked from Mirrors/cloudflare-ddns-docker
Code Issues Pull Requests 3 Actions Packages Releases Activity

Add domain update

Browse Source
This commit is contained in:
Matteo Bertucci
2021-01-14 21:34:14 +01:00
parent ad68d13078
commit 7202473af0
5 changed files with 62 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.flake8
View File

@ -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

5
cloudflare_ddns/__main__.py
View File

@ -13,16 +13,15 @@ log = logging.getLogger("ddns")
@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:
def start(delay: str, token: str, verbose: int, domain: Tuple[str]) -> 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()
ApplicationJob(delay, token, domain).launch()
# Main entrypoint

70
cloudflare_ddns/app.py
View File

@ -1,12 +1,11 @@
import logging
import threading
from dataclasses import dataclass
from typing import Tuple, List, Dict, NoReturn
from typing import Dict, List, Tuple
import requests
from cloudflare_ddns.constants import VERIFY_TOKEN, LIST_ZONES, LIST_DNS, ACCEPTED_RECORDS
from cloudflare_ddns.utils import parse_duration, BearerAuth
from cloudflare_ddns.constants import ACCEPTED_RECORDS, LIST_DNS, LIST_ZONES, VERIFY_TOKEN, PATCH_DNS
from cloudflare_ddns.utils import BearerAuth, parse_duration, get_ip
log = logging.getLogger("ddns")
@ -20,17 +19,16 @@ class Domain:
class ApplicationJob(threading.Thread):
def __init__(self, raw_delay: str, token: str, raw_domains: Tuple[str], default_ipv6: bool):
def __init__(self, raw_delay: str, token: str, raw_domains: Tuple[str]):
super().__init__()
self.stop_signal = threading.Event()
self.delay = None
self.domains: List[Domain] = []
self.found_domains: Dict[str, Domain] = {}
self.current_ip = None
self.auth = BearerAuth(token)
self.default_ipv6 = default_ipv6
self.raw_domains = raw_domains
self.raw_delay = raw_delay
@ -45,17 +43,33 @@ class ApplicationJob(threading.Thread):
self.parse_domains()
log.debug(f"Using domains: {', '.join(f'{domain.record_type}:{domain.domain}' for domain in self.domains)}")
log.info("Starting app.")
log.info(f"Starting app. Records will be updated every {self.delay} seconds.")
try:
self.update_records()
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):
...
def update_records(self) -> None:
log.info("Starting record update.")
for record in self.domains:
log.debug(f"Updating record for {record.domain}.")
requests.patch(
PATCH_DNS.format(zone_identifier=record.zone, identifier=record.id),
json={"content": get_ip(record.record_type == 'AAAA')},
auth=self.auth
).raise_for_status()
log.info("Successfully updated records.")
def parse_domains(self):
type_1 = "A" if not self.default_ipv6 else "AAAA"
type_2 = "A" if self.default_ipv6 else "AAAA"
def parse_domains(self) -> None:
found_domains = {}
for zone_json in requests.get(LIST_ZONES, auth=self.auth).json()["result"]:
for record_json in requests.get(
@ -69,11 +83,11 @@ class ApplicationJob(threading.Thread):
record_json["zone_id"],
record_json["id"]
)
self.found_domains[f'{record_json["name"]}-{record_json["type"]}'] = domain
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 self.found_domains.values())}"
f"{', '.join(f'{domain.record_type}:{domain.domain}' for domain in found_domains.values())}"
)
for domain in self.raw_domains:
if ':' in domain:
@ -84,19 +98,28 @@ class ApplicationJob(threading.Thread):
log.info(f"Exiting with code 65.")
exit(65)
if f"{domain}-{type_}" not in self.found_domains:
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:
if f"{domain}-{type_1}" in self.found_domains:
type_ = type_1
elif f"{domain}-{type_2}" in self.found_domains:
type_ = type_2
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?"
@ -104,11 +127,6 @@ class ApplicationJob(threading.Thread):
log.info(f"Exiting with code 65.")
exit(65)
self.domains.append(self.found_domains[f"{domain}-{type_}"])
def validate_arguments(self):
failed = False

4
cloudflare_ddns/constants.py
View File

@ -8,6 +8,10 @@ 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/"

8
cloudflare_ddns/utils.py
View File

@ -4,7 +4,7 @@ import requests
from requests import Request
from requests.auth import AuthBase
from .constants import VERIFY_TOKEN
from cloudflare_ddns.constants import IP_API_URL_IPV4, IP_API_URL_IPV6
DURATION_REGEX = re.compile(
r"((?P<days>\d+?) ?(days|day|D|d) ?)?"
@ -53,3 +53,9 @@ class BearerAuth(AuthBase):
"""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 = requests.get(IP_API_URL_IPV4 if not ipv6 else IP_API_URL_IPV6)
return r.text