izbi-dns/app/main.py

from flask import Flask, jsonify, request, render_template, redirect, session, flash
import auth
import re



HOSTS_PATH = "/app/hosts.d/izbi"
DB_PATH = "/app/sqlite.db"



app = Flask(__name__)
app.config["SESSION_PERMANENT"] = True
app.config["SESSION_TYPE"] = "memcached"
app.config["PERMANENT_SESSION_LIFETIME"] = 3600

app.config["SECRET_KEY"] = "dsadasda"

hosts = {}



def update_hosts(path: str, hosts: dict = {}) -> bool:
    try:
        hosts_str = ""
        for domain, ip in hosts.items():
            hosts_str += f"{ip}\t{domain}\n"
        with open(path, "w") as f:
            f.write(hosts_str)
        print(f"ddns - updated {path}", flush=True)
        return True
    except Exception as e:
        print(e, flush=True)
        return False



def validate_ip(ip: str) -> bool:
    exp = r"^((25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])$"
    return re.search(exp, ip)



@app.route('/', methods=["GET"])
def homepage():
    if session.get("USERID") is not None:
        return redirect("/dashboard")
    return render_template("index.html")



@app.route('/login', methods=["POST"])
def login():
    if "user" not in request.form or "pass" not in request.form:
        return redirect("/"), 400

    if request.form["user"].strip() == "" or request.form["pass"].strip() == "":
        return redirect("/"), 400

    userid = auth.user_login(DB_PATH, request.form["user"], request.form["pass"])
    if userid is None:
        return redirect("/")

    session["USERID"] = userid

    return redirect("/dashboard", code=302)

@app.route('/logout', methods=["GET"])
def logout():
    session.clear()
    return redirect("/")



@app.route('/dashboard', methods=["GET"])
def dashboard():
    if session.get("USERID") is None:
        return redirect("/")
    userid = session.get("USERID")
    username = auth.get_username(DB_PATH, userid)
    domains = auth.get_user_domains(DB_PATH, userid)
    tokens = auth.get_user_tokens(DB_PATH, userid)

    if auth.is_admin(DB_PATH, userid):
        users = auth.get_users(DB_PATH)
        return render_template("dashboard.html", username=username, domains=domains, tokens=tokens, is_admin=True, users=users)
    return render_template("dashboard.html", username=username, domains=domains, tokens=tokens, is_admin=False)



@app.route('/generate-token', methods=["POST"])
def generate_token():
    if session.get("USERID") is None:
        return redirect("/")

    if "token" not in request.form:
        return redirect("/dashboard")
    if request.form["token"].strip() == "":
        return redirect("/dashboard")

    token = auth.generate_user_token(DB_PATH, session.get("USERID"), request.form["token"])

    if token is not None:
        flash(token)

    return redirect("/dashboard")


@app.route('/revoke-token', methods=["POST"])
def revoke_token():
    if session.get("USERID") is None:
        return redirect("/")

    if "token" not in request.form:
        return redirect("/dashboard")
    if request.form["token"].strip() == "":
        return redirect("/dashboard")

    auth.revoke_user_token(DB_PATH, session.get("USERID"), request.form["token"])

    return redirect("/dashboard")


@app.route('/change-password', methods=["POST"])
def change_password():
    if session.get("USERID") is None:
        return redirect("/")

    if "pass" not in request.form or "pass-new" not in request.form or "pass-rep" not in request.form:
        flash("password change failed")
        return reidrect("/dashboard")
    if request.form["pass"] == "" or request.form["pass-new"] == "" or request.form["pass-rep"] == "":
        flash("password change failed")
        return redirect("/dashboard")

    oldpass = request.form["pass"]
    newpass = request.form["pass-new"]

    if newpass != request.form["pass-rep"]:
        flash("passwords do not match")
        return redirect("/dashboard")

    if not auth.change_user_password(DB_PATH, session.get("USERID"), oldpass, newpass):
        flash("password change failed")
        return redirect("/dashboard")

    flash("password changed successfully")
    return redirect("/dashboard")

@app.route('/health', methods=["GET"])
def health():
    return jsonify({"status": "healthy", "comment": ""})



@app.route('/update', methods=["GET"])
def update_addr():
    try:
        if 'domain' not in request.args:
            return jsonify({"status": "400", "code": "domain-error", "comment": "Missing or invalid domain"}), 400
        domain = request.args['domain']

        if 'ip' not in request.args:
            ip = request.remote_addr
        elif not validate_ip(request.args['ip']):
            return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400
        else:
            ip = request.args['ip']

        if 'token' not in request.args:
            return jsonify({"status": "400", "code": "token-error", "comment": "Missing or invalid token"}), 400
        token = request.args['token']

        if not auth.check_token_domain(DB_PATH, token, domain):
            print(f"ddns - unauthorized update attempt from {request.remote_addr} of {domain}", flush=True)
            return jsonify({"status": "401", "code": "unauthorized", "comment": "This token is not allowed to update this domain"}), 401

        success = True

        if domain not in hosts or ip != hosts[domain]:
            hosts[domain] = ip
            success = update_hosts(HOSTS_PATH, hosts)

        if not success:
            return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500

        return jsonify({"status": "200", "code": "ok", "comment": "Updated successfully"})
    except Exception as e:
        print(e, flush=True)
        return jsonify({"status": "500", "code": "internal-server-error", "comment": "An error occurred while processing request"}), 500



if __name__ == '__main__':
    auth.init_db(DB_PATH)
    app.run(host="0.0.0.0", port=8080, debug=True)