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)