Skip to content

Instantly share code, notes, and snippets.

@ColeMundus
Last active July 22, 2018 23:20
Show Gist options
  • Save ColeMundus/5ba67aca689f7e2d0bd143908553d152 to your computer and use it in GitHub Desktop.
Save ColeMundus/5ba67aca689f7e2d0bd143908553d152 to your computer and use it in GitHub Desktop.
Acme remote DNS api thingy
from flask import Flask, request, abort, g
from functools import wraps
from uuid import uuid4
import json
import db
app = Flask(__name__)
def require_appkey(view_function):
@wraps(view_function)
def decorated_function(*args, **kwargs):
if request.args.get('key') and db.validate(request.args.get('key')):
return view_function(*args, **kwargs)
else:
abort(401)
return decorated_function
def test_domain(domain):
if domain in db.search_domain(domain):
return 'Already Exists.'
elif domain in ['mydomain.local', 'cool.local', 'lol.local', 'hmm.ya', 'yes']:
return False
else:
return 'Invalid Domain.'
def create_key(domain):
key = str(uuid4().hex)
db.create(domain, key, False)
return key
@app.route('/')
def index():
return """lol wtf it works"""
@app.route('/register')
def register():
domains = request.args.getlist('domain')
ret = {'valid': [], 'error': []}
for domain in domains:
check = test_domain(domain)
if check:
ret['error'].append({'domain': domain, 'reason': check})
else:
ret['valid'].append({'domain': domain, 'key': create_key(domain)})
if len(ret['error']) == 0:
return json.dumps({'valid': ret['valid']}), 200
else:
return json.dumps(ret), 406
@app.route('/add')
@require_appkey
def add():
s = db.search_key(request.args.get("key", type=str))
if not eval(s[2]):
abort(401)
record = request.args.get("record", type=str)
content = request.args.get("content", type=str)
# This is where you would add code to add the DNS record
return str(s), 200
@app.route('/del')
@require_appkey
def delete():
s = db.search_key(request.args.get("key", type=str))
if not eval(s[2]):
abort(401)
record = request.args.get("record", type=str)
# This is where you would add code to delete the DNS record
return str(s), 200
if __name__ == '__main__':
app.run(use_reloader=True, host='0.0.0.0')
import sqlite3
database = 'database.db'
db = sqlite3.connect(database, check_same_thread=False)
c = db.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS acme (domain text, key text, authorized text)''')
def create(domain, key, authorized):
c.execute("INSERT INTO acme VALUES (?, ?, ?)", (domain, key, authorized))
db.commit()
def validate(key):
return len([r for r in c.execute("SELECT * FROM acme WHERE key=?", (key,))])
def search_key(key):
result = c.execute("SELECT * FROM acme WHERE key=?", (key,))
return [r for r in result][0]
def search_domain(domain):
result = c.execute("SELECT * FROM acme WHERE domain=?", (domain,))
return [r[0] for r in result]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment