Last active
July 22, 2018 23:20
-
-
Save ColeMundus/5ba67aca689f7e2d0bd143908553d152 to your computer and use it in GitHub Desktop.
Acme remote DNS api thingy
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
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') |
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
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