Update app.py
This commit is contained in:
parent
2ee69a737f
commit
598ca21296
1 changed files with 31 additions and 31 deletions
58
app.py
58
app.py
|
|
@ -6,7 +6,7 @@ from flask import Flask, render_template, jsonify, request, Response
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
from modules.network_tools import execute_command_streaming
|
from modules.network_tools import execute_command_streaming
|
||||||
from modules.visual_route import generate_visual_route_graph
|
from modules.visual_route import generate_visual_route_graph, get_raw_bgp_route
|
||||||
|
|
||||||
load_dotenv()
|
load_dotenv()
|
||||||
|
|
||||||
|
|
@ -15,11 +15,8 @@ app = Flask(__name__)
|
||||||
FQDN_REGEX = re.compile(r'^(?!-)(?:[a-zA-Z0-9-]{0,62}[a-zA-Z0-9]\.){1,126}(?!-d$)[a-zA-Z0-9-]{2,63}$')
|
FQDN_REGEX = re.compile(r'^(?!-)(?:[a-zA-Z0-9-]{0,62}[a-zA-Z0-9]\.){1,126}(?!-d$)[a-zA-Z0-9-]{2,63}$')
|
||||||
MAX_TARGET_LENGTH = 255
|
MAX_TARGET_LENGTH = 255
|
||||||
|
|
||||||
def is_valid_ip_or_prefix(target: str) -> bool:
|
def is_valid_ip(target: str) -> bool:
|
||||||
try:
|
try:
|
||||||
if '/' in target:
|
|
||||||
ipaddress.ip_network(target, strict=False)
|
|
||||||
else:
|
|
||||||
ipaddress.ip_address(target)
|
ipaddress.ip_address(target)
|
||||||
return True
|
return True
|
||||||
except ValueError:
|
except ValueError:
|
||||||
|
|
@ -28,21 +25,25 @@ def is_valid_ip_or_prefix(target: str) -> bool:
|
||||||
def is_valid_hostname(hostname: str) -> bool:
|
def is_valid_hostname(hostname: str) -> bool:
|
||||||
if len(hostname) > MAX_TARGET_LENGTH:
|
if len(hostname) > MAX_TARGET_LENGTH:
|
||||||
return False
|
return False
|
||||||
if is_valid_ip_or_prefix(hostname):
|
if is_valid_ip(hostname):
|
||||||
return False
|
return False
|
||||||
return FQDN_REGEX.match(hostname) is not None
|
return FQDN_REGEX.match(hostname) is not None
|
||||||
|
|
||||||
def get_ip_version(target):
|
def get_ip_version_from_ip(target):
|
||||||
try:
|
try:
|
||||||
ip = ipaddress.ip_address(target)
|
ip = ipaddress.ip_address(target)
|
||||||
return f"ipv{ip.version}"
|
return f"ipv{ip.version}"
|
||||||
except ValueError:
|
|
||||||
try:
|
|
||||||
net = ipaddress.ip_network(target, strict=False)
|
|
||||||
return f"ipv{net.version}"
|
|
||||||
except ValueError:
|
except ValueError:
|
||||||
return 'ipv4'
|
return 'ipv4'
|
||||||
|
|
||||||
|
@app.route("/ping")
|
||||||
|
def ping():
|
||||||
|
return "pong"
|
||||||
|
|
||||||
|
@app.errorhandler(404)
|
||||||
|
def page_not_found(e):
|
||||||
|
return render_template("404.html"), 404
|
||||||
|
|
||||||
@app.route('/')
|
@app.route('/')
|
||||||
def index():
|
def index():
|
||||||
return render_template('looking_glass.html')
|
return render_template('looking_glass.html')
|
||||||
|
|
@ -78,40 +79,39 @@ def execute_command():
|
||||||
target = data.get('target', '').strip()
|
target = data.get('target', '').strip()
|
||||||
|
|
||||||
if not target:
|
if not target:
|
||||||
return Response("Error: A target is required.", status=400, mimetype='text/plain')
|
return jsonify({"error": "A target is required."})
|
||||||
|
|
||||||
if len(target) > MAX_TARGET_LENGTH:
|
if len(target) > MAX_TARGET_LENGTH:
|
||||||
return Response(f"Error: Target exceeds maximum length.", status=400, mimetype='text/plain')
|
return jsonify({"error": f"Target exceeds maximum length of {MAX_TARGET_LENGTH} characters."})
|
||||||
|
|
||||||
if target.startswith('-'):
|
if target.startswith('-'):
|
||||||
return Response("Error: Target cannot start with a hyphen.", status=400, mimetype='text/plain')
|
return jsonify({"error": "Target cannot start with a hyphen."})
|
||||||
|
|
||||||
is_ip = is_valid_ip_or_prefix(target)
|
if not (is_valid_ip(target) or is_valid_hostname(target)):
|
||||||
is_host = is_valid_hostname(target)
|
return jsonify({"error": "Invalid input. Please provide a valid IP address or a fully qualified domain name."})
|
||||||
|
|
||||||
if not (is_ip or is_host):
|
version = get_ip_version_from_ip(target)
|
||||||
return Response(
|
|
||||||
"Error: Invalid input. Please provide a valid IP, prefix, or fully qualified domain name.",
|
|
||||||
status=400,
|
|
||||||
mimetype='text/plain'
|
|
||||||
)
|
|
||||||
|
|
||||||
version = get_ip_version(target)
|
|
||||||
|
|
||||||
if version == 'ipv6' and method in ['ping', 'mtr', 'traceroute']:
|
if version == 'ipv6' and method in ['ping', 'mtr', 'traceroute']:
|
||||||
method += '6'
|
method += '6'
|
||||||
|
|
||||||
return Response(execute_command_streaming(method, target), mimetype='text/plain')
|
return Response(execute_command_streaming(method, target), mimetype='text/plain')
|
||||||
|
|
||||||
|
@app.route('/api/bgp_raw_lookup', methods=['POST'])
|
||||||
|
def bgp_raw_lookup():
|
||||||
|
target = request.json.get('target', '').strip()
|
||||||
|
data, error = get_raw_bgp_route(target)
|
||||||
|
|
||||||
|
if error:
|
||||||
|
return jsonify({"error": error})
|
||||||
|
|
||||||
|
return Response(data, mimetype='text/plain')
|
||||||
|
|
||||||
@app.route('/api/visualize', methods=['POST'])
|
@app.route('/api/visualize', methods=['POST'])
|
||||||
def visualize_route():
|
def visualize_route():
|
||||||
ip_address_str = request.json.get('ip_address')
|
ip_address_str = request.json.get('ip_address', '').strip()
|
||||||
|
|
||||||
graph_data = generate_visual_route_graph(ip_address_str)
|
graph_data = generate_visual_route_graph(ip_address_str)
|
||||||
|
|
||||||
if "error" in graph_data:
|
if "error" in graph_data:
|
||||||
return jsonify(graph_data), 400
|
return jsonify(graph_data)
|
||||||
|
|
||||||
return jsonify(graph_data)
|
return jsonify(graph_data)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue