izbi-dns/app/main.py

from flask import Flask, jsonify, request, render_template, redirect, session, flash
import auth
import db
import re
from os import environ, urandom
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, EmailField, HiddenField
from wtforms.validators import DataRequired
from flask_wtf.csrf import CSRFProtect

HOSTS_PATH = ( environ["HOSTS"] if "HOSTS" in environ else"/app/data/hosts.d/" ) + "/izbi"
DB_PATH = environ["DB_PATH"] if "DB_PATH" in environ else "/app/data/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"] = environ["SECRET_KEY"] if "SECRET_KEY" in environ else urandom(32).hex()

csrf = CSRFProtect(app)

hosts = {}

# Update hosts file
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


# check if string is a valid IP
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)


class LoginForm(FlaskForm):
    username = StringField('username', validators=[DataRequired()])
    password = PasswordField('password', validators=[DataRequired()])

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

    login_form = LoginForm()

    if login_form.validate_on_submit():
        username = login_form.username.data
        password = login_form.password.data

        userid = auth.user_login(DB_PATH, username, password)

        if userid is None:
            flash("Invalid user or password")
            return redirect("/")

        session["USERID"] = userid

        return redirect("/dashboard")

    return render_template("index.html", login_form=login_form)


class LogoutForm(FlaskForm):
    pass

class GenerateTokenForm(FlaskForm):
    token_name = StringField('token-name', validators=[DataRequired()])

class RevokeTokenForm(FlaskForm):
    token_name = StringField('token-name', validators=[DataRequired()])

class ChangePasswordForm(FlaskForm):
    password = PasswordField('password', validators=[DataRequired()])
    new_password = PasswordField('new-password', validators=[DataRequired()])
    repeat_password = PasswordField('repeat-password', validators=[DataRequired()])

class CreateUserForm(FlaskForm):
    username = StringField('username', validators=[DataRequired()])
    domains = StringField('domains')
    password = PasswordField('password', validators=[DataRequired()])
    is_admin = BooleanField('is-admin')
    email = EmailField('email')

class UpdateUserForm(FlaskForm): #TODO
    pass

class DeleteUserForm(FlaskForm):
    userid = HiddenField('userid', validators=[DataRequired()])

# Display dashboard for logged in users
@app.route('/dashboard', methods=["GET"])
def dashboard():
    if session.get("USERID") is None:
        return redirect("/")

    logout_form = LogoutForm()
    generate_token_form = GenerateTokenForm()
    revoke_token_form = RevokeTokenForm()
    change_password_form = ChangePasswordForm()

    userid = session.get("USERID")

    username = db.User.get_username(DB_PATH, userid)
    domains = db.User.get_domains(DB_PATH, userid)
    tokens = db.User.get_token_names(DB_PATH, userid)

    # If user is admin, display additional options
    if db.User.is_admin(DB_PATH, userid):
        create_user_form = CreateUserForm()
        update_user_form = UpdateUserForm()
        delete_user_form = DeleteUserForm()

        users = db.list_users(DB_PATH)

        return render_template("dashboard.html",
            logout_form=logout_form, generate_token_form=generate_token_form, revoke_token_form=revoke_token_form, change_password_form=change_password_form, create_user_form=create_user_form, update_user_form=update_user_form, delete_user_form=delete_user_form,
            username=username, domains=domains, tokens=tokens, is_admin=True, users=users
        )

    return render_template("dashboard.html",
        logout_form=logout_form, generate_token_form=generate_token_form, revoke_token_form=revoke_token_form, change_password_form=change_password_form,
        username=username, domains=domains, tokens=tokens, is_admin=False
    )


# Log out by clearing session data
@app.route('/logout', methods=["POST"])
def logout():
    logout_form = LogoutForm()

    if logout_form.validate_on_submit():
        session.clear()

    return redirect("/")


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

    generate_token_form = GenerateTokenForm()

    if generate_token_form.validate_on_submit():
        token_name = generate_token_form.token_name.data
        try:
            token = db.User.generate_token(DB_PATH, session.get("USERID"), token_name)
        except ValueError as e:
            flash(str(e))
            token = None

        if token is not None:
            flash(token)

    return redirect("/dashboard")


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

    revoke_token_form = RevokeTokenForm()

    if revoke_token_form.validate_on_submit():
        token_name = revoke_token_form.token_name.data
        try:
            db.User.revoke_token(DB_PATH, session.get("USERID"), token_name)
        except ValueError as e:
            flash(str(e))

    return redirect("/dashboard")


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

    change_password_form = ChangePasswordForm()

    if change_password_form.validate_on_submit():
        password = change_password_form.password.data
        new_password = change_password_form.new_password.data
        repeat_password = change_password_form.repeat_password.data

        if new_password != repeat_password:
            flash("Passwords do not match")
            return redirect("/dashboard")

        try:
            auth.change_user_password(DB_PATH, session.get("USERID"), password, new_password)
        except ValueError as e:
            flash(str(e))
            return redirect("/dashboard")

        flash("Password changed successfully")

    return redirect("/dashboard")


# Create new user (by admin)
@app.route('/create-user', methods=["POST"])
def create_user():
    if session.get("USERID") is None:
        return redirect("/")
    if not db.User.is_admin(DB_PATH, session.get("USERID")):
        return redirect("/dashboard")

    create_user_form = CreateUserForm()

    if create_user_form.validate_on_submit():
        username = create_user_form.username.data.strip()
        password = create_user_form.password.data

        domains = create_user_form.domains.data
        if domains is None:
            domains = []
        else:
            # Collapse whitespaces
            collapse_exp = re.compile(r'\s+')
            domains = collapse_exp.sub(' ', domains.strip())
            domains = domains.split(" ")

        is_admin = create_user_form.is_admin.data

        email = create_user_form.email.data
        if email is not None:
            if email == "":
                email = None

        try:
            db.User.new(DB_PATH, username, password, domains, is_admin, email)
        except ValueError as e:
            flash(str(e))

    return redirect("/dashboard")


# Update user's field (by admin)
@app.route('/update-user', methods=["POST"])
def update_user():
    if session.get("USERID") is None:
        return redirect("/")
    if not db.User.is_admin(DB_PATH, session.get("USERID")):
        return redirect("/dashboard")

    update_user_form = UpdateUserForm()

    if not update_user_form.validate_on_submit():
        return redirect("/dashboard")

    userid = request.form.get("userid")
    if userid is None:
        flash("UserID not supplied")
        return redirect("/dashboard")

    if not db.User.exists(DB_PATH, userid):
        flash("User does not exist")
        return redirect("/dashboard")

    domains = request.form.get('domains')
    if domains is not None:
        collapse_exp = re.compile(r'\s+')
        domains = collapse_exp.sub(' ', domains.strip())
        domains = domains.split(" ")
        try:
            db.User.set_domains(DB_PATH, userid, domains)
        except ValueError as e:
            flash(str(e))

    password = request.form.get("password")
    if password is not None:
        try:
            db.User.set_password(DB_PATH, userid, password)
        except ValueError as e:
            flash(str(e))

    is_admin = request.form.get("is_admin")
    if is_admin is not None:
        is_admin = is_admin == "true"
        db.User.set_admin(DB_PATH, userid, is_admin)

    email = request.form.get("email")
    if email is not None:
        if email == "":
            email = None
        db.User.set_email(DB_PATH, userid, email)

    return redirect("/dashboard")


# Delete user (by admin)
@app.route('/delete-user', methods=["POST"])
def delete_user():
    if session.get("USERID") is None:
        return redirect("/")
    if not db.User.is_admin(DB_PATH, session.get("USERID")):
        return redirect("/dashboard")

    delete_user_form = DeleteUserForm()

    if delete_user_form.validate_on_submit():
        userid = delete_user_form.userid.data

        if not db.User.exists(DB_PATH, userid):
            flash("User does not exst")
            return redirect("/dashboard")

        db.User.delete(DB_PATH, userid)

    return redirect("/dashboard")



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


# API for updating domain's IP
@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 validate_ip(request.args['ip']):
            ip = request.args['ip']
        else:
            return jsonify({"status": "400", "code": "ip-error", "comment": "Invalid IP"}), 400

        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

        # Update hosts file if something changed
        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__':
    db.init(DB_PATH)
    app.run(host="0.0.0.0", port=8080, debug=False)