122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
|
||
|
|
import re
|
||
|
|
import json
|
||
|
|
import urllib
|
||
|
|
import tomllib
|
||
|
|
import urllib.parse
|
||
|
|
import urllib.request
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
#contents = urllib.request.urlopen("http://example.com/foo/bar").read()
|
||
|
|
|
||
|
|
# 5 curl -X POST -L https://api.porkbun.com/api/json/v3/dns/retrieveByNameType/klank.school/A/www --data '{"secretapikey":""}'
|
||
|
|
|
||
|
|
class API():
|
||
|
|
def __init__(self):
|
||
|
|
|
||
|
|
with (Path(__file__).parent / 'klankdns.secret.toml').open('rb') as f:
|
||
|
|
toml = tomllib.load(f)
|
||
|
|
|
||
|
|
self.key = toml['klankdns']['key']
|
||
|
|
self.secret = toml['klankdns']['secret']
|
||
|
|
self.base = toml['klankdns']['base']
|
||
|
|
self.entries = toml['klankdns']['entries']
|
||
|
|
|
||
|
|
self.public_ip = self.get_public_ip()
|
||
|
|
|
||
|
|
def get_public_ip(self):
|
||
|
|
req = urllib.request.Request("https://ipinfo.io/ip")
|
||
|
|
with urllib.request.urlopen(req) as response:
|
||
|
|
result = response.read().decode('utf-8')
|
||
|
|
|
||
|
|
if not re.match(r'[\d]+\.[\d]+\.[\d]+\.[\d]+', result):
|
||
|
|
raise Exception("ipinfo returned bullshit!!")
|
||
|
|
|
||
|
|
print(f"Got public IP: {result}")
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def post(self, url, data = None):
|
||
|
|
url = urllib.parse.urljoin(self.base, url)
|
||
|
|
fulldata = {
|
||
|
|
"secretapikey": self.secret,
|
||
|
|
"apikey": self.key,
|
||
|
|
**(data or {})
|
||
|
|
}
|
||
|
|
print(fulldata)
|
||
|
|
|
||
|
|
req = urllib.request.Request(url)
|
||
|
|
req.add_header('Content-Type', 'application/json')
|
||
|
|
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(req, json.dumps(fulldata).encode('utf_8')) as response:
|
||
|
|
result = response.read().decode('utf-8')
|
||
|
|
except urllib.error.HTTPError as err:
|
||
|
|
print(err.read().decode('utf-8'))
|
||
|
|
raise err
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def get_by_name(self, name):
|
||
|
|
return self.post(urllib.parse.urljoin("dns/retrieveByNameType/", name))
|
||
|
|
|
||
|
|
def check(self):
|
||
|
|
for entry in self.entries:
|
||
|
|
result = self.get_by_name(entry)
|
||
|
|
print(f"{entry} : {result}")
|
||
|
|
|
||
|
|
def get_current_ip(self, entry):
|
||
|
|
return json.loads(self.get_by_name(entry))['records'][0]['content']
|
||
|
|
|
||
|
|
def update(self):
|
||
|
|
for entry in self.entries:
|
||
|
|
dat = json.loads(self.get_by_name(entry))
|
||
|
|
if dat["status"] != "SUCCESS":
|
||
|
|
raise Exception("???")
|
||
|
|
|
||
|
|
if len(dat["records"]) > 1:
|
||
|
|
raise Exception("???")
|
||
|
|
|
||
|
|
if len(dat["records"]) == 0:
|
||
|
|
print(f"Creating {entry}")
|
||
|
|
result = self.post(
|
||
|
|
# TODO this is spaghettt
|
||
|
|
f := urllib.parse.urljoin("dns/create/", entry.split('/')[0]),
|
||
|
|
{
|
||
|
|
"name": entry.split('/')[2],
|
||
|
|
"type": entry.split('/')[1],
|
||
|
|
"content": self.public_ip,
|
||
|
|
"ttl": "600"
|
||
|
|
}
|
||
|
|
)
|
||
|
|
print(f"{entry} : {result}")
|
||
|
|
if json.loads(result)["status"] != "SUCCESS":
|
||
|
|
raise Exception("???")
|
||
|
|
|
||
|
|
if len(dat["records"]) == 1:
|
||
|
|
if dat["records"][0]["content"] == self.public_ip:
|
||
|
|
print(f"Not updating {entry}")
|
||
|
|
else:
|
||
|
|
print(f"updating {entry}")
|
||
|
|
|
||
|
|
result = self.post(
|
||
|
|
f := urllib.parse.urljoin("dns/editByNameType/", entry),
|
||
|
|
{
|
||
|
|
"content": self.public_ip,
|
||
|
|
"ttl": 600,
|
||
|
|
}
|
||
|
|
)
|
||
|
|
print(f"{entry} : {result}")
|
||
|
|
if json.loads(result)["status"] != "SUCCESS":
|
||
|
|
raise Exception("???")
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
api = API()
|
||
|
|
api.update()
|
||
|
|
#api.check()
|
||
|
|
|
||
|
|
|