[Python] Variables losing values in Flask application

I've been working on a Flask backend for quite some time and as this is my first attempt, I've been running into a lot of issues.

I have a linux server running the flask backend and I have a templates and static folder for my html and css.

The website functions as a way for people to log in via SSO, connect to a Cisco switch, and update the speed and duplex settings of the device.

I'm using Netmiko to establish a connection and retrieving user input through forms on the HTML pages. The problem I'm having is, it seems like my variables for the user inputed port and host devices don't retain their values when stored in the session.

Further, it appears my connector (net_connector) loses its value at random intervals. Please note, I'm aware of the spaghetti code in the duplex and speed settings. I had functions created for them but removed them temporarily

from flask import Flask, render_template, request, redirect, url_for, Response, session
from netmiko import ConnectHandler
from cryptography.fernet import Fernet
from datetime import timedelta
import re
import socket
import ipaddress
import os
import requests

net_connect = None

app = Flask(__name__)
app.config.update(
SESSION_COOKIE_SECURE=True,
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SAMESITE='Lax'
)

app.secret_key = os.urandom(24)
app.permanent_session_lifetime = timedelta(minutes=30)

CLIENT_ID = 'removed for example'
CLIENT_SECRET = 'removed for example'
AUTHORIZATION_BASE_URL = 'removed for example'
TOKEN_URL = 'removed for example'
REDIRECT_URI = 'removed for example'

#encrypt using functions

blocked_subnets = {
Removed from example
}

def decrypt():
with open('encryptedPWD.txt') as f:
encpwd = ''.join(f.readlines())
encpwdbyt = bytes(encpwd, 'utf-8')

with open('refKey.txt') as f:
refKey = ''.join(f.readlines())
refKeybyt = bytes(refKey, 'utf-8')

keytouse = Fernet(refKeybyt)
myPass = keytouse.decrypt(encpwdbyt).decode('utf-8').strip()
return myPass

def Connect(hostn):

global net_connect
myPass = decrypt()

try:
net_connect = ConnectHandler(
device_type='cisco_ios',
host=hostn,
username='REMOVED FOR EXAMPLE',
password=myPass
)
return "Connection to {} successful!".format(hostn)
except Exception as e:
print("Error connecting to {}: {}".format(hostn, str(e)))
return render_template('denied.html')
# return "Connection to {} failed: {}".format(hostn, str(e))

def dns_query(host):
result = socket.getaddrinfo(host, None, socket.AF_INET)
ip_address = result[0][4][0]
return ip_address

def dns_query_ip(host):
try:
result = socket.getaddrinfo(host, None, socket.AF_INET)
ip_address = result[0][4][0]
print(ip_address)
return ip_address
except socket.herror:
return None

def duplex(interface):
if net_connect is None:
return "Connection to device is not established."
output = net_connect.send_command(f'show interface {interface}')
return output

@app.errorhandler(Exception)
def handle_error(exception):
# Log the exception
app.logger.error(exception)

return render_template('error.html', error_message="An unexpected error occurred. If this error was thrown while accessing a device using a hostname, try using the IP address instead", exception=exception)


#SSO related Routes
@app.route('/sso')
def sso():
code = request.args.get('code')
if not code:
return 'Authorization code not found in request.'


response = requests.post(TOKEN_URL, data={
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': REDIRECT_URI,
'client_id': CLIENT_ID,
'client_secret': CLIENT_SECRET,
'scope': 'openid profile'
})
response_data = response.json()
access_token = response_data.get('access_token')

#
session['access_token'] = access_token

return redirect(url_for('index'))

@app.route('/auth_login')
def auth_login():
authorization_url = (
f"{AUTHORIZATION_BASE_URL}?response_type=code"
f"&client_id={CLIENT_ID}"
f"&redirect_uri={REDIRECT_URI}"
f"&response_mode=query"
f"&scope=openid profile"
)
return redirect(authorization_url)

#End of SSO Related Routes

@app.route('/', methods=['GET', 'POST'])
def initial():
if 'access_token' in session:
if request.method == 'POST':
return render_template('index.html')
else:
return redirect(url_for('auth_login'))

@app.route('/index')
def index():
if 'access_token' in session:
return render_template('index.html') # if state
else:
return redirect(url_for('auth_login'))

@app.route('/submit', methods=['POST'])
def submit():
if request.method == 'POST':
host_input = request.form['hostInput']
port_input = request.form['portInput']

app.logger.info(f"Setting session values: host_input={host_input}, port_input={port_input}")

session['host_input'] = request.form['hostInput']
session['port_input'] = request.form['portInput']
return redirect(url_for('check')) # send to submit

@app.route('/check', methods=['GET', 'POST'])
def check():
host_input = session.get('host_input')
port_input = session.get('port_input')


if host_input is None or port_input is None:
return render_template('error.html',
error_message="Host or port input is missing.")

result_messege = Connect(host_input)
output = net_connect.send_command(f'show run interface {port_input}')
if re.match("^[a-zA-Z]", host_input):
host_input = dns_query_ip(host_input)
for subnet in blocked_subnets:
if host_input.startswith(subnet):
error_message = f'Device {host_input} is a restricted device'
return render_template('denied.html', error_message=error_message)
if 'trunk'in output:
net_connect.disconnect()
error_message = f"Port {port_input} is a trunk port and cannot be modified."
return render_template('denied.html', error_message=error_message)
else:
return redirect(url_for('result', result_messege=result_messege, host_input=host_input, port_input=port_input))

@app.route('/result', methods=['GET', 'POST'])
def result():
host_input = session.get('host_input')
port_input = session.get('port_input')

port_messege = duplex(port_input)
status_messege = net_connect.send_command(f'show interface {port_input} status')

return render_template('result.html', port_messege=port_messege, status_messege=status_messege)

@app.route('/updatedDuplex', methods=['GET', 'POST'])
def updateduplex():
host_input = session.get('host_input')
port_input = session.get('port_input')

if net_connect is None:
return render_template('error.html',
error_message=f"No active connection. Host: {host_input} Port: {port_input}")

if request.method == 'POST':
chosen_option = request.form.get('duplex')
if chosen_option == 'auto':
option = net_connect.send_config_set([f'interface {port_input}', 'duplex auto'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
elif chosen_option == 'full':
option = net_connect.send_config_set([f'interface {port_input}', 'duplex full'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
elif chosen_option == 'half':
option = net_connect.send_config_set([f'interface {port_input}', 'duplex half'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
return render_template('result.html', port_messege=port_messege, status_messege=status_messege)

@app.route('/updatedSpeed', methods=['GET', 'POST'])
def updatedspeed():
host_input = session.get('host_input')
port_input = session.get('port_input')


if request.method == 'POST':
chosen_option = request.form.get('speed')
#port_messege = request.form.get('port_messege') # Retrieve port_input from query parameters
if chosen_option == 'auto':
net_connect(host_input)
option = net_connect.send_config_set([f'interface {port_input}', 'speed auto'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
net_connect.disconnect()
elif chosen_option == '10':
option = net_connect.send_config_set([f'interface {port_input}', 'speed 10'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
elif chosen_option == '100':
option = net_connect.send_config_set([f'interface {port_input}', 'speed 100'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
elif chosen_option == '1000':
option = net_connect.send_config_set([f'interface {port_input}', 'speed 1000'])
port_messege = net_connect.send_command(f'show interface {port_input}')
status_messege = net_connect.send_command(f'show interface {port_input} status')
return render_template('result.html', option=option, port_messege=port_messege, status_messege=status_messege)

@app.route('/disconnect')
def disconnect():
if net_connect is not None:
#net_connect.send_command('copy ru st') Disabled until deployment. Discuss further will pete.
net_connect.disconnect()
print("Disconnected from the device")
else:
print("No active connection to disconnect from")
return redirect('index')

@app.route('/refdisc', methods=['POST'])
def refdisc():
net_connect.disconnect()
return Response("Disconnected from the device", status=200)


@app.route('/denied')
def denied():
return render_template('denied.html')

@app.route('/incorrect')
def incorrect():
return render_template('incorrect.html')


@app.route('/logout')
def logout():
session.pop('access_token', None)
return redirect(url_for('auth_login'))


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


The main issues I appear to be getting are on the results page. I'll see "'NoneType' object has no attribute 'send_command'". Occasionally, I'll get a message about port and host inputs not having a value.

This is what I see in the browsers search bar: https://IPADDRESS/result?result_messege=Connection+to+1.1.1.1+successful!&host_input=1.1.1.1&port_input=GigabitEthernet1/0/7

Again, I know the code isn't great and I apologize. This was all new to me and has been a work in progress.

Continue reading...