GraphicLG/web.py

from flask import Flask, render_template, request
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

from paramiko import SSHClient, AutoAddPolicy
from graphviz import Digraph
from slugify import slugify

import time, re, diagram, base64, sys

#Try to load config 
try:
    import config 
except Exception as e:
    sys.exit(f"Can't load config dot py. Exiting GraphicLG...\n{e}")

app = Flask(__name__)

#limit req to prevent f*cking spam
limiter = Limiter(
    app,
    key_func=get_remote_address
)

#Home page
@app.route('/', methods=['GET'])
def home():
    return render_template('index.html', config = config, routers = config.routers.items(), actiondata={}, graph={}, cmd_res="")

@app.route('/<path:ip>', methods=['GET'])
def my_view_func(ip):
    return show(short_url = True, ipaddr=ip)

@app.route('/show', methods=['POST'])
@limiter.limit("20 per hour")
def show(short_url = False, ipaddr = ""):

    #Get data from form or URL 
    if(not short_url):
        actiondata = {
            "router_id": request.form.get('router_id'),
            "ip_addr": request.form.get('ip_addr'),
            "action_type": request.form.get('action_type'),
        }
    else: 
        actiondata = {
            "router_id": config.default_router,
            "ip_addr": ipaddr,
            "action_type": '1',
        }
    
    #Check for valid IPv4 / IPv6 with or without CIDR 
    #Check for valid v6
    if(not re.match('^(([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))(\/((1(1[0-9]|2[0-8]))|([0-9][0-9])|([0-9])))?$', actiondata['ip_addr'])):
        if(not config.lg_v6only): #if not v6 check for v4 as well 
            if(not re.match('^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)(/(3[0-2]|2[0-9]|1[0-9]|[0-9]))?$', actiondata['ip_addr'])):
                error_msg = "Please enter a valid IPv4 or IPv6 with or without his CIDR."
                return render_template('index.html', config = config, actiondata=actiondata, graph={}, cmd_res=error_msg)
        else:    
            error_msg = "Please enter a valid IPv6 with or without his CIDR."
            return render_template('index.html', config = config, actiondata=actiondata, graph={}, cmd_res=error_msg)

    #Init ssh client 
    client = SSHClient()
    client.set_missing_host_key_policy(AutoAddPolicy())

    #Get router from form data
    rtr = config.routers[f"{actiondata['router_id']}"]

    #Router connexion based on auth type
    try:
        if(rtr["auth-type"] == "ssh-key"):
            client.connect(rtr["ip"], username=rtr["username"], key_filename=rtr["key"], port=rtr["port"])
        else:
            client.connect(rtr["ip"], username=rtr["username"], password=rtr["password"], port=rtr["port"])
    except Exception as e: 
        error_msg = f"Can't connect to the router. {e}. Please contact the LG administrator"
        return render_template('index.html', config = config, actiondata=actiondata, graph={}, cmd_res=error_msg)
    
    #Show bgp IP_ADDR 
    if(actiondata['action_type'] == "1"):
        if(rtr["bgpcapable"]):
            #Launching CMD 
            stdin, stdout, stderr = client.exec_command(f'vtysh -c "sh bgp {actiondata["ip_addr"]}"')
            time.sleep(0.1)

            conresult = stdout.read().decode("utf8") 
            
            print(conresult)
            
            #Parsing as_list (FRRouting) via a REGEX 
            as_list = re.findall(r"  ([0-9][0-9 ]+)[\n]", conresult)
            as_list1 = re.findall(r"  ([0-9][0-9 ]+),", conresult)
            as_list = as_list + as_list1
            #Closing session
            stdin.close()
            stdout.close()
            stderr.close()
            client.close()
            
            #Graph gen
            filename = slugify(actiondata['ip_addr'])
            result = diagram.gen_diagram(as_list, config.lg_asn, filename)
            graph_details = { 
                "filename": filename + f".svg?{time.time()}",
                "success": result
            }

            if(not graph_details['success']):
                if(re.findall(r"  Local", conresult)): #If local IPs 
                    result = diagram.gen_diagram([], config.lg_asn, filename, True)
                    graph_details["success"] = result

            return render_template('index.html', config = config, actiondata=actiondata, graph=graph_details, cmd_res=conresult)
        else:
            #If the router isn't capable of bgp 
            return render_template('index.html', config = config, actiondata=actiondata, graph={}, cmd_res="'show bgp' can't be executed since this router don't do BGP")

    #Ping
    if(actiondata['action_type'] == "2"):
        #Launching CMD 
        stdin, stdout, stderr = client.exec_command(f'ping -c3 {actiondata["ip_addr"]}')

    #Traceroute
    if(actiondata['action_type'] == "3"): 
        #Launching CMD 
        stdin, stdout, stderr = client.exec_command(f'traceroute {actiondata["ip_addr"]}')

    time.sleep(0.1)
    conresult = stdout.read().decode("utf8")
    errResult = stderr.read().decode("utf8")

    if(errResult != ""): 
        conresult = "Error : " + errResult
    
    #Closing session
    stdin.close()
    stdout.close()
    stderr.close()
    client.close()
    return render_template('index.html', config = config, actiondata=actiondata, graph={}, cmd_res=conresult)

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=True)