148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
from flask import Flask, jsonify, request, render_template, send_from_directory
|
|
import requests
|
|
import os
|
|
import json
|
|
import time
|
|
import socket
|
|
import paramiko
|
|
|
|
from config import NOCODB_URL, NOCODB_TOKEN, NOCODB_BASE_ID, NOCODB_TABLE_ID, CONFIGS_DIR
|
|
|
|
app = Flask(__name__)
|
|
app.config['TEMPLATES_AUTO_RELOAD'] = True
|
|
|
|
os.makedirs(CONFIGS_DIR, exist_ok=True)
|
|
|
|
# ─── NocoDB ───────────────────────────────────────────────────────────────────
|
|
|
|
def get_switches(dept=None):
|
|
headers = {"xc-token": NOCODB_TOKEN}
|
|
params = {
|
|
"where": "(Active,eq,YES)",
|
|
"limit": 500,
|
|
"sort": "Hostname"
|
|
}
|
|
if dept:
|
|
params["where"] = f"(Active,eq,YES)~and(Dept,eq,{dept})"
|
|
|
|
url = f"{NOCODB_URL}/api/v1/db/data/noco/{NOCODB_BASE_ID}/{NOCODB_TABLE_ID}"
|
|
resp = requests.get(url, headers=headers, params=params, timeout=10)
|
|
resp.raise_for_status()
|
|
return resp.json().get("list", [])
|
|
|
|
|
|
# ─── Routes ───────────────────────────────────────────────────────────────────
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return render_template("index.html")
|
|
|
|
|
|
@app.route("/api/switches")
|
|
def api_switches():
|
|
try:
|
|
switches = get_switches()
|
|
return jsonify({"ok": True, "switches": switches})
|
|
except Exception as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/save", methods=["POST"])
|
|
def api_save():
|
|
data = request.get_json()
|
|
hostname = data.get("hostname", "").strip()
|
|
content = data.get("content", "")
|
|
|
|
if not hostname:
|
|
return jsonify({"ok": False, "error": "No hostname provided"}), 400
|
|
|
|
filename = f"{hostname}.conf"
|
|
filepath = os.path.join(CONFIGS_DIR, filename)
|
|
|
|
# Check if file already exists
|
|
exists = os.path.isfile(filepath)
|
|
if exists and not data.get("overwrite", False):
|
|
return jsonify({"ok": False, "exists": True, "filename": filename}), 409
|
|
|
|
try:
|
|
with open(filepath, "w") as f:
|
|
f.write(content)
|
|
return jsonify({"ok": True, "filename": filename, "path": filepath})
|
|
except Exception as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/configs")
|
|
def api_configs():
|
|
try:
|
|
files = sorted([
|
|
f for f in os.listdir(CONFIGS_DIR) if f.endswith(".conf")
|
|
])
|
|
return jsonify({"ok": True, "files": files})
|
|
except Exception as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 500
|
|
|
|
|
|
@app.route("/api/web-access", methods=["POST"])
|
|
def api_web_access():
|
|
data = request.get_json()
|
|
ip = data.get("ip", "").strip()
|
|
username = data.get("username", "").strip()
|
|
password = data.get("password", "").strip()
|
|
action = data.get("action", "") # "enable" or "disable"
|
|
|
|
if not ip or not username or not password:
|
|
return jsonify({"ok": False, "error": "Missing IP or credentials"}), 400
|
|
if action not in ("enable", "disable"):
|
|
return jsonify({"ok": False, "error": "Invalid action"}), 400
|
|
|
|
print(f"[web-access] {action} {ip} user={username!r} pass_len={len(password)}", flush=True)
|
|
|
|
cmd = "ip http secure-server" if action == "enable" else "no ip http secure-server"
|
|
|
|
try:
|
|
sock = socket.create_connection((ip, 22), timeout=10)
|
|
transport = paramiko.Transport(sock)
|
|
transport.start_client(timeout=10)
|
|
|
|
# Try password auth.
|
|
# BadAuthenticationType means the server disallows password auth entirely
|
|
# (some FS switches require keyboard-interactive instead).
|
|
# A plain AuthenticationException means the credentials were rejected — don't
|
|
# fall back to keyboard-interactive in that case.
|
|
try:
|
|
transport.auth_password(username, password)
|
|
except paramiko.ssh_exception.BadAuthenticationType as e:
|
|
if "keyboard-interactive" in e.allowed_types:
|
|
transport.auth_interactive_dumb(username, [password])
|
|
else:
|
|
raise paramiko.AuthenticationException(
|
|
f"Password auth not accepted; server allows: {e.allowed_types}"
|
|
)
|
|
|
|
shell = transport.open_session()
|
|
shell.get_pty()
|
|
shell.invoke_shell()
|
|
time.sleep(1)
|
|
shell.recv(65535) # drain login banner
|
|
|
|
for command in ["conf t", cmd, "end"]:
|
|
shell.send(command + "\n")
|
|
time.sleep(0.5)
|
|
|
|
# copy run start may prompt for filename — send \n to confirm default
|
|
shell.send("copy run start\n")
|
|
time.sleep(0.3)
|
|
shell.send("\n")
|
|
time.sleep(1.5)
|
|
|
|
output = shell.recv(65535).decode("utf-8", errors="ignore") if shell.recv_ready() else ""
|
|
transport.close()
|
|
return jsonify({"ok": True, "output": output})
|
|
except Exception as e:
|
|
return jsonify({"ok": False, "error": str(e)}), 500
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8003, debug=False)
|