Created
June 30, 2022 12:38
-
-
Save GJCav/9f5fca89ad6e5d7785ca9c7e1469a96c to your computer and use it in GitHub Desktop.
DDNS script for cloudflare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import socket as S | |
| import requests | |
| ZONE_ID = "zone_id_here" | |
| TOKEN = "token_here" | |
| DNS_NAME = "name.domain.ltd" | |
| ################################################################## | |
| # | |
| # A bunch of helper functions | |
| # | |
| ################################################################## | |
| def list_all_ip(): | |
| l = S.getaddrinfo(S.gethostname(), None, S.AF_INET) | |
| return [e[4][0] for e in l] | |
| def list_all_ipv6(): | |
| l = S.getaddrinfo(S.gethostname(), None, S.AF_INET6) | |
| return [e[4][0] for e in l] | |
| def get_inet_ip(): | |
| """ | |
| Get ip that connect to the internet. This function may still return | |
| local ip behind NAT. | |
| """ | |
| with S.socket(S.AF_INET, S.SOCK_DGRAM) as s: | |
| s.connect(("8.8.8.8", 80)) | |
| return s.getsockname()[0] | |
| def get_inet_ipv6(): | |
| """ | |
| Get IPv6 ip that connect to the internet. | |
| """ | |
| with S.socket(S.AF_INET6, S.SOCK_DGRAM) as s: | |
| s.connect(("2001:4860:4860::8888", 80)) | |
| return s.getsockname()[0] | |
| def get_inet_ip_nat(use_ipv6 = False): | |
| """ | |
| Considering NAT, use some public service to ident your public IP out of NAT. | |
| Change the host if it becomes unavailable. | |
| """ | |
| host = "ident.me" | |
| with S.socket(S.AF_INET6 if use_ipv6 else S.AF_INET, S.SOCK_STREAM) as s: | |
| s.connect((host, 80)) | |
| s.send(b"GET / HTTP/1.1\r\n") | |
| s.send(b"Host: ident.me\r\n") | |
| s.send(b"User-Agent: Python Raw Socket\r\n") | |
| s.send(b"Connection: close\r\n") | |
| s.send(b"\r\n") | |
| res = (s.recv(2048) or b'').decode("utf-8") | |
| lines = res.split("\r\n") | |
| if lines[0] and lines[0].split(" ")[-1] == 'OK': | |
| return lines[-1] | |
| return None | |
| ################################################################## | |
| # | |
| # rewrite ip_provider bellow according to your case | |
| # | |
| ################################################################## | |
| def ip_provider(): | |
| return get_inet_ipv6() | |
| ################################################################## | |
| # | |
| # connect to cloudflare api, no need to touch this part | |
| # | |
| ################################################################## | |
| def list_records(name, type="A"): | |
| params = {"type": type} | |
| if name: params["name"] = name | |
| res = requests.get( | |
| f"https://api.cloudflare.com/client/v4/zones/{ZONE_ID}/dns_records", | |
| params = params, | |
| headers={ | |
| "Authorization": "Bearer " + TOKEN | |
| } | |
| ) | |
| if res.status_code == 200: | |
| json = res.json() | |
| if json["success"] == False: | |
| raise Exception(f"CF_API: list_records API error, errors: {str(json['errors'])}") | |
| return json["result"] | |
| else: | |
| raise Exception(f"CF_API: list_records HTTP error, status_code = {res.status_code}") | |
| def create_record(type, name, content, ttl = 60): | |
| """ | |
| repeat creation leads to HTTP error 400 | |
| """ | |
| res = requests.post( | |
| f"https://api.cloudflare.com/client/v4/zones/{ZONE_ID}/dns_records", | |
| headers={"Authorization": "Bearer " + TOKEN}, | |
| json={ | |
| "type": type, | |
| "name": name, | |
| "content": content, | |
| "ttl": ttl | |
| } | |
| ) | |
| if res.status_code != 200: | |
| raise Exception(f"CF_API: create_record HTTP error, status_code = {res.status_code}") | |
| json = res.json() | |
| if json["success"] == False: | |
| raise Exception(f"CF_API: create_record API error, errors: {str(json['errors'])}") | |
| return json["result"] | |
| def update_record(record_id, type, name, content, ttl=60): | |
| res = requests.put( | |
| f"https://api.cloudflare.com/client/v4/zones/{ZONE_ID}/dns_records/{record_id}", | |
| headers={"Authorization": "Bearer " + TOKEN}, | |
| json={ | |
| "type": type, | |
| "name": name, | |
| "content": content, | |
| "ttl": ttl | |
| } | |
| ) | |
| if res.status_code != 200: | |
| raise Exception(f"CF_API: update_record HTTP error, status_code = {res.status_code}") | |
| json = res.json() | |
| if json["success"] == False: | |
| raise Exception(f"CF_API: update_record API error, errors: {str(json['errors'])}") | |
| return json["result"] | |
| if __name__ == "__main__": | |
| addr = ip_provider() | |
| name = DNS_NAME | |
| type = "A" if '.' in addr else "AAAA" | |
| try: | |
| records = list_records(name, type) | |
| if len(records) == 0: | |
| print(f"create records: {type} {name} {addr}") | |
| create_record(type, name, addr) | |
| else: | |
| record = records[0] | |
| rid = record["id"] | |
| print(f"update records: {type} {name} {addr}") | |
| update_record(rid, type, name, addr) | |
| except Exception as e: | |
| print("ERROR: " + str(e)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment