autopeering/web/backend/main.py

#! /usr/bin/env python3

# flask
from flask import Flask, Response, redirect, render_template, request, session, abort
import werkzeug.exceptions as werkzeug_exceptions
# other utils
import json
import os
import base64
# potential errors on base64 decode: binascii.Error
import binascii 
import logging
import random
from functools import wraps
from ipaddress import ip_address, ip_network, IPv4Network, IPv6Network
# self written modules
import kioubit_verify
from peering_manager import PeeringManager

app = Flask(__name__)


class Config (dict):
    def __init__(self, configfile: str = None):
        if configfile:
            self.configfile = configfile
        else:
            if os.path.exists("./config.json"):
                self.configfile = "./config.json"
            elif os.path.exists("/etc/dn42-autopeer/config.json"):
                self.configfile = "/etc/dn42-autopeer/config.json"
            else:
                raise FileNotFoundError(
                    "no config file found in ./config.json or /etc/dn42-autopeer/config.json")
        self._load_config()
        self.keys = self._config.keys
        #self.__getitem__ = self._config.__getitem__
        super().__init__(self)

    def __contains__(self, o):
        return self._config.__contains__(o)

    def __delitem__(self, v):
        raise NotImplementedError()
        super().__delitem__(self, v)

    def __getitem__(self, k):
        return self._config[k]

    def _load_config(self):
        with open(self.configfile) as cf:
            try:
                self._config = json.load(cf)
            except json.decoder.JSONDecodeError:
                raise SyntaxError(f"no valid JSON found in '{cf.name}'")

        if "flask-template-dir" not in self._config:
            self._config["flask-template-dir"] = "../frontend"

        if "debug-mode" not in self._config:
            self._config["debug-mode"] = False
        if "base-dir" not in self._config:
            self._config["base-dir"] = "/"

        if "peerings-data" not in self._config:
            self._config["peering-data"] = "./peerings"
        
        if "nodes" not in self._config:
            self,_config = {}
        for node in self._config["nodes"]:
            if "capacity" not in self._config["nodes"][node]:
                self._config["nodes"][node]["capacity"] = -1
        
        logging.info(self._config)



config = Config()
peerings = PeeringManager(config)
kverifyer = kioubit_verify.AuthVerifyer(config["domain"])


def check_peering_data(form):
    new_peering = {}
    # errors = 0

    # check if all required (and enabled) options are specified
    try:
        # force "peer-asn" to be the same as in the login data except if login is admin (admin get the peer-asn field enabled in the form)
        new_peering["peer-asn"] = session["user-data"]["asn"] if  "peer-asn" not in request.form or session["login"] != config["MNT"] else form["peer-asn"]
        new_peering["peer-wgkey"] = form["peer-wgkey"]
        if "peer-endpoint-enabled" in form and form["peer-endpoint-enabled"] == "on":
            new_peering["peer-endpoint"] = form["peer-endpoint"]
            if new_peering["peer-endpoint"] == "":
                raise ValueError("peer-endpoint")
        else:
            new_peering["peer-endpoint"] = None

        if "peer-v6ll-enabled" in form and form["peer-v6ll-enabled"] == "on":
            new_peering["peer-v6ll"] = form["peer-v6ll"]
            if new_peering["peer-v6ll"] == "":
                raise ValueError("peer-v6ll")
        else:
            new_peering["peer-v6ll"] = None
        if "peer-v4-enabled" in form and form["peer-v4-enabled"] == "on":
            new_peering["peer-v4"] = form["peer-v4"]
            if new_peering["peer-v4"] == "":
                raise ValueError("peer-v4")
        else:
            new_peering["peer-v4"] = None
        if "peer-v6-enabled" in form and form["peer-v6-enabled"] == "on":
            new_peering["peer-v6"] = form["peer-v6"]
            if new_peering["peer-v6"] == "":
                raise ValueError("peer-v6")
        else:
            new_peering["peer-v6"] = None
        new_peering["bgp-mp"] = form["bgp-multi-protocol"] if "bgp-multi-protocol" in form else "off"
        new_peering["bgp-mp"] = True if new_peering["bgp-mp"] == "on" else False
        new_peering["bgp-enh"] = form["bgp-extended-next-hop"] if "bgp-extended-next-hop" in form else "off"
        new_peering["bgp-enh"] = True if new_peering["bgp-enh"] == "on" else False
        #new_peering[""] = form["peer-wgkey"]
    except ValueError as e:
        print(f"error: {e.args}")
        return False, "at least one of the required/enabled fields was not filled out"

    print(new_peering)

    # check wireguard key
    wg_key_invalid = False
    if len(new_peering["peer-wgkey"]) != 44:
        wg_key_invalid = True
    try:
        base64.b64decode(new_peering["peer-wgkey"])
    except binascii.Error:
        wg_key_invalid = True
    if wg_key_invalid:
        return False, "invalid wireguard Key"

    # check endpoint
    if new_peering["peer-endpoint"]:
        if not new_peering["peer-endpoint"].split(":")[-1].isnumeric():
            return False, "no port number in endpoint"
        elif len(new_peering["peer-endpoint"].split(":")) < 2 and "." not in new_peering["peer-endpoint"]:
            return False, "endpoint doesn't look like a ip address or fqdn"

    # check if at least one ip is specified/enabled
    try:
        if not (new_peering["peer-v6ll"] or new_peering["peer-v4"] or new_peering["peer-v6"]):
            return False, "at least one of the ip addresses must be enabled and specified"
    except KeyError:
        return False, "one of the values isn't valid"

    # check if supplied ip addresses are valid
    try:
        if new_peering["peer-v6ll"]:
            ipv6ll = ip_address(new_peering["peer-v6ll"])
            if ipv6ll.version != 6:
                raise ValueError()
            if not ipv6ll.is_link_local:
                raise ValueError()
        if new_peering["peer-v4"]:
            ipv4 = ip_address(new_peering["peer-v4"])
            if ipv4.version != 4:
                raise ValueError()
            if ipv4.is_link_local:
                pass
            elif ipv4.is_private:
                if not (ipv4.compressed.startswith("172.2") or ipv4.compressed.startswith("10.")):
                    raise ValueError()
                is_in_allowed = False
                if session["user-data"]["allowed4"]:
                    if not isinstance(session["user-data"]["allowed4"],list):
                        allowed4 = session["user-data"]["allowed4"]
                        if ipv4 in ip_network(allowed4):
                            is_in_allowed = True
                    else:
                        for allowed4 in session["user-data"]["allowed4"]:
                            if ipv4 in ip_network(allowed4):
                                is_in_allowed = True
                if not is_in_allowed:
                    return False, "supplied ipv4 addr not in allowed ip range"
            else:
                raise ValueError()
        if new_peering["peer-v6"]:
            ipv6 = ip_address(new_peering["peer-v6"])
            if ipv6.version != 6:
                raise ValueError()
            if not ipv6.is_private:
                raise ValueError()
            if ipv6.is_link_local:
                raise ValueError()
            is_in_allowed = False
            if session["user-data"]["allowed6"]:
                if not isinstance(session["user-data"]["allowed6"],list):
                    allowed6 = session["user-data"]["allowed6"]
                    if ipv6 in ip_network(allowed6):
                        is_in_allowed = True
                else:
                    for allowed6 in session["user-data"]["allowed6"]:
                        if ipv6 in ip_network(allowed6):
                            is_in_allowed = True
            if not is_in_allowed:
                return False, "supplied ipv6 addr not in allowed ip range"
    except ValueError as e:
        print(e)
        return False, "invalid ip address(es) supplied"

    # check bgp options
    try:
        if new_peering["bgp-mp"] is False and new_peering["bgp-enh"] is True:
            return False, "extended next hop requires multiprotocol bgp"
        if new_peering["bgp-mp"] is False:
            if not (new_peering["peer-v4"] and (new_peering["peer-v6"] or new_peering["peer-v6ll"])):
                return False, "ipv4 and ipv6 addresses required when not having MP-BGP"
    except ValueError:
        pass
    return True, new_peering


def auth_required():
    def wrapper(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            if "login" not in session:
                request_url = f"{config['base-dir']}{request.full_path}".replace("//", "/")
                return redirect(f"{config['base-dir']}login?return={request_url}")
            else:
                return f(*args, **kwargs)
        return decorated
    return wrapper


@app.route("/api/auth/kverify", methods=["GET", "POST"])
def kioubit_auth():
    try:
        params = request.args["params"]
        signature = request.args["signature"]
    except KeyError:
        return render_template("login.html", session=session, config=config, return_addr=session["return_url"], msg='"params" or "signature" missing')

    success, msg = kverifyer.verify(params, signature)
    try:
        logging.debug(base64.b64decode(params))
    except:
        logging.debug("invalid Base64 data provided")

    if success:
        session["user-data"] = msg
        session["login"] = msg['mnt']
        try:
            return redirect(session["return_url"])
        except KeyError:
            return redirect(f"{config['base-dir']}peerings")
    else:
        try:
            return render_template("login.html", session=session, config=config, return_addr=session["return_url"], msg=msg)
        except KeyError:
            return render_template("login.html", session=session, config=config, return_addr=f"{config['base-dir']}peerings", msg=msg)


@app.route("/logout")
def logout():
    session.clear()
    return redirect(config["base-dir"])


@app.route("/login", methods=["GET", "POST"])
def login():
    print(session)
    if request.method == "GET":
        session["return_url"] = request.args["return"] if "return" in request.args else ""

        return render_template("login.html", session=session, config=config, return_addr=session["return_url"])
    elif request.method == "POST" and (config["debug-mode"] or session["login"] == config["MNT"]):
        try:
            print(request.form)
            if request.form["theanswer"] != "42":
                msg = """what is the answer for everything? <a href="https://en.wikipedia.org/wiki/42_(answer)">hint</a>"""
                return render_template("login.html", session=session, config=config, return_addr=session["return_url"], msg=msg)
            mnt = request.form["mnt"]
            if not mnt.upper().endswith("-MNT"):
                raise ValueError
            asn = request.form["asn"]
            asn = asn[2:] if asn[:2].lower() == "as" else asn
            int(asn)
            if "allowed4" in request.form:
                allowed4 = request.form["allowed4"]
                v4_ranges = allowed4.split(",") if "," in allowed4 else [allowed4]
                for v4_range in v4_ranges:
                    IPv4Network(v4_range)
            else:
                allowed4 = None
            if "allowed6" in request.form:
                allowed6 = request.form["allowed6"]
                v6_ranges = allowed6.split(",") if "," in allowed6 else [allowed6]
                for v6_range in v6_ranges:
                    IPv6Network(v6_range)
            else:
                allowed6 = None
            session["user-data"] = {'asn': asn, 'allowed4': allowed4,
                                    'allowed6': allowed6, 'mnt': mnt, 'authtype': "debug"}
            session["login"] = mnt if "login" not in session else session["login"]
            return redirect(session["return_url"])
        except ValueError:
            msg = "at least one of the values provided is wrong/invalid"
            return render_template("login.html", session=session, config=config, return_addr=session["return_url"], msg=msg)
        except KeyError:
            msg = "not all required field were specified"
            return render_template("login.html", session=session, config=config, return_addr=session["return_url"], msg=msg)
    elif request.method == "POST" and not config["debug-mode"]:
        abort(405)
    return redirect(request.args["return"])


@app.route("/peerings/delete", methods=["GET", "POST", "DELETE"])
@auth_required()
def peerings_delete():

    if request.method == "GET":
        return render_template("peerings-delete.html", session=session, config=config, request_args=request.args)
        return f"{request.method} /peerings/delete?{str(request.args)}{str(request.form)}"
    elif request.method in ["POST", "DELETE"]:
        if request.form["confirm"] != "on":
            return render_template("peerings-delete.html", session=session, config=config, request_args=request.args, msg="you have to confirm the deletion first")
        # mnt=None, if admin
        if not peerings.exists(request.args["asn"], request.args["node"], mnt=session["user-data"]["mnt"] if session["login"] != config["MNT"] else None):
            return render_template("peerings-delete.html", session=session, config=config, request_args=request.args, msg="the peering you requested to delete doesn't exist (anymore) or you are not authorized to delete it")
        print(str(request))
        # mnt=None, if admin
        if not peerings.delete_peering(request.args["asn"], request.args["node"], mnt=session["user-data"]["mnt"] if session["login"] != config["MNT"] else None):
            return render_template("peerings-delete.html", session=session, config=config, request_args=request.args, msg="deletion of the peering requested failed, maybe you are not authorized or that peering doesn't exist")
        session["msg"] = {"msg": "peer-del",
                          "node": request.args["node"], "asn": request.args["asn"]}
        return redirect("../peerings")
        return f"{request.method} /peerings/delete {request.args} {request.form}"


@app.route("/peerings/edit", methods=["GET", "POST"])
@auth_required()
def peerings_edit():
    print(session)
    if request.method == "GET":
        if "node" not in request.args or not request.args["node"]:
            return render_template("peerings-edit.html",  session=session, config=config, peerings=peerings, msg="no peering selected, please click one of the buttons above")
        mnt_peerings = peerings.get_peerings_by_mnt(session["user-data"]["mnt"])
        # print(mnt_peerings)
        if "node" in request.args and request.args["node"] in config["nodes"]:
            selected_peering = None
            for p in mnt_peerings:
                if p["node"] == request.args["node"] and p["ASN"] == request.args["asn"]:
                    selected_peering = p
                    print(p)
                    break
            return render_template("peerings-edit.html", session=session, config=config, mnt_peerings=mnt_peerings, selected_peering=selected_peering, selected_node=selected_peering["node"])
        else:
            print(request.args)
            return render_template("peerings-edit.html", session=session, config=config, mnt_peerings=mnt_peerings, selected_peering=None)
    elif request.method == "POST":
        print(request.args)
        print(request.form)
        if "node" not in request.args or not request.args["node"]:
            return render_template("peerings-edit.html",  session=session, config=config, peerings=peerings, msg="no peering selected, please click one of the buttons above")

        peering_valid, peering_or_msg = check_peering_data(request.form)
        print(peering_valid)
        print(peering_or_msg)
        selected_peering = None
        mnt_peerings = peerings.get_peerings_by_mnt(session["user-data"]["mnt"])
        for p in mnt_peerings:
            if p["node"] == request.args["node"] and p["ASN"] == request.args["asn"]:
                selected_peering = p
                print(p)
                break
        if not peering_valid:
            return render_template("peerings-edit.html",  session=session, config=config, peerings=peerings, msg=peering_or_msg, selected_peering=selected_peering), 400
        if not peerings.update_peering(session["user-data"]["asn"], request.args["node"], session["user-data"]["mnt"], peering_or_msg["peer-wgkey"], peering_or_msg["peer-endpoint"], peering_or_msg["peer-v6ll"], peering_or_msg["peer-v4"], peering_or_msg["peer-v6"], peering_or_msg["bgp-mp"], peering_or_msg["bgp-enh"]):
            return render_template("peerings-edit.html",  session=session, config=config, peerings=peerings, msg="such a peering doesn't exist(yet)", selected_peering=selected_peering), 400

        return redirect(f"{config['base-dir']}peerings")
        return f"{request.method} /peerings/edit?{str(request.args)}{str(request.form)}"


@app.route("/peerings/new", methods=["GET", "POST"])
@auth_required()
def peerings_new():
    print(session)
    if request.method == "GET":
        if "node" in request.args and request.args["node"] in config["nodes"]:
            return render_template("peerings-new.html", config=config, selected_node=request.args["node"], peerings=peerings)
        else:
            return render_template("peerings-new.html",  session=session, config=config, peerings=peerings)
    elif request.method == "POST":
        print(request.args)
        print(request.form)
        if "node" not in request.args or not request.args["node"]:
            return render_template("peerings-new.html",  session=session, config=config, peerings=peerings, msg="no node specified, please click one of the buttons above")

        peering_valid, peering_or_msg = check_peering_data(request.form)

        if not peering_valid:
            return render_template("peerings-new.html",  session=session, config=config, peerings=peerings, msg=peering_or_msg), 400
        success, code = peerings.add_peering(session["user-data"]["asn"], request.args["node"], session["user-data"]["mnt"], peering_or_msg["peer-wgkey"], peering_or_msg["peer-endpoint"], peering_or_msg["peer-v6ll"], peering_or_msg["peer-v4"], peering_or_msg["peer-v6"], peering_or_msg["bgp-mp"], peering_or_msg["bgp-enh"])
        print(f"{success}, {code}")
        if not success:
            return render_template("peerings-new.html",  session=session, config=config, peerings=peerings, msg="this ASN already has a peering with the requested node or something failed in the backend, please retry later"), code

        return redirect(f"{config['base-dir']}peerings")


@app.route("/peerings", methods=["GET", "POST", "DELETE"])
@auth_required()
def peerings_view():
    print(session)
    if request.method == "GET":
        if "node" in request.args and request.args["node"] in config["nodes"]:
            return render_template("peerings.html", config=config, selected_node=request.args["node"], peerings=peerings)
        else:
            return render_template("peerings.html",  session=session, config=config, peerings=peerings)
    elif request.method == "POST":
        return peerings_new()
    elif request.method == "DELETE":
        return peerings_delete()
    else:
        # shouldn't get here
        abort(405)

@app.route("/hidden", methods=["GET", "POST", "DELETE"])
@auth_required()
def hidden_page():
    ...


@app.route("/")
def index():
    print(session)
    # print(config._config["nodes"])
    # for node in config["nodes"].values():
    #     print (node)
    return render_template("index.html",  session=session, config=config, peerings = peerings)


app.static_folder = config["flask-template-dir"]+"/static/"
app.template_folder = config["flask-template-dir"]
app.secret_key = config["flask-secret-key"]
def main():
    if "production" in config and config["production"] == False:
        logging.getLogger(__name__).setLevel(0)
        app.run(host=config["listen"], port=config["port"],
                debug=config["debug-mode"], threaded=True)
    else:
        from waitress import serve
        logging.getLogger(__name__).setLevel(logging.INFO)
        serve(app, host=config["listen"], port=config["port"])


if __name__ == "__main__":
    main()