Created
October 21, 2019 11:15
-
-
Save cmouse/34fadc9a8f2f7e3dbac38adbe2ebe120 to your computer and use it in GitHub Desktop.
FastCGI script to support dyndns from EdgeOS
This file contains 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
#!/usr/bin/python3 | |
from flup.server.fcgi import WSGIServer | |
import bottle | |
from bottle_log import LoggingPlugin | |
import json | |
import logging | |
import requests | |
logging.basicConfig(format='localhost - - [%(asctime)s] %(message)s', level=logg | |
ing.DEBUG) | |
log = logging.getLogger(__name__) | |
bottle.debug(True) | |
app = bottle.Bottle() | |
@app.get("/") | |
def do_get_root(): | |
bottle.response.content_type="application/json" | |
return json.dumps({ | |
"service":"dyndns" | |
}) | |
class PowerDNSAPI(object): | |
def __init__(self, url, apikey): | |
self.api_url = url | |
self.headers = {"X-Api-Key": apikey} | |
# get server ID | |
servers = self.get("/servers") | |
self.server_id = servers[0]["id"] | |
def get(self, endpoint, *args): | |
resp = [] | |
try: | |
resp = requests.get(("%s%s" % (self.api_url, endpoint)) % args, headers=se | |
lf.headers) | |
except: | |
pass | |
if not resp.ok: | |
if resp.status_code == 404: | |
return {} | |
raise bottle.HTTPError(500, resp.text) | |
return resp.json() | |
def get_domain(self, name): | |
zones = self.get("/servers/%s/zones", self.server_id) | |
zone = [n for n in zones if n["id"] == self.to_api_name(name)] | |
if zone: | |
return zone[0] | |
return None | |
def get_record(self, domain_id, name, rectype): | |
zone = self.get("/servers/%s/zones/%s", self.server_id, domain_id) | |
recs = [rr for rr in zone["rrsets"] if rr["name"] == self.to_api_name(name) | |
and rr["type"] == rectype] | |
return recs | |
def set_record(self, domain_id, name, rectype, content): | |
rrsets = { "rrsets" : [{ | |
"name": self.to_api_name(name), | |
"type": rectype, | |
"ttl": 60, | |
"changetype": "REPLACE", | |
"records": [{ | |
"content": content, | |
"disabled": False, | |
}], | |
}]} | |
resp = requests.patch("%s/servers/%s/zones/%s" % (self.api_url, self.server_ | |
id, domain_id), | |
headers=self.headers, data=json.dumps(rrsets)) | |
if not resp.ok: | |
raise bottle.HTTPError(500, resp.text) | |
recs = self.get_record(domain_id, name, rectype) | |
return recs[0] | |
def delete_record(self, domain_id, name, rectype, content): | |
rrsets = {"rrsets" : [{ | |
"name": self.to_api_name(name), | |
"type": rectype, | |
"ttl": 60, | |
"changetype": "DELETE", | |
"records": [], | |
}]} | |
resp = requests.patch("%s/servers/%s/zones/%s" % (self.api_url, self.server_ | |
id, domain_id), | |
headers=self.headers, data=json.dumps(rrsets)) | |
if not resp.ok: | |
raise bottle.HTTPError(500, resp.text) | |
def to_api_name(self, name): | |
return "%s." % name | |
def from_api_name(self, name): | |
return name[0:-1] | |
API = PowerDNSAPI("http://localhost:8081/api/v1", "password") | |
@app.get("/nic/update") | |
def do_nic_update(): | |
zone = "zonename" | |
hostname = bottle.request.query.get("hostname") | |
content = bottle.request.query.get("myip") | |
dom = API.get_domain(zone) | |
if not dom: | |
return bottle.HTTPError(404, "Zone not found") | |
rec = API.set_record(dom["name"], hostname, 'A', content) | |
bottle.response.content_type = "application/json" | |
return json.dumps({ | |
"active":"y", | |
"address":rec["records"][0]["content"], | |
"fqdn":API.from_api_name(rec["name"]), | |
"record_type":rec["type"], | |
"zone":API.from_api_name(dom["name"]) | |
}) | |
WSGIServer(app).run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment