import secrets from flask import current_app, flash, jsonify, redirect, render_template, request, session, url_for from flask_login import current_user, login_required, login_user, logout_user from .core import ( LicenseService, RuntimeState, User, discover_node, get_status_info, validate_login_payload, validate_node_payload, ) def get_runtime_state() -> RuntimeState: return current_app.extensions["runtime_state"] def get_license_service() -> LicenseService: return current_app.extensions["license_service"] def json_error(message: str, status_code: int = 400): return jsonify({"status": "error", "message": message}), status_code def get_csrf_token() -> str: token = session.get("csrf_token") if not token: token = secrets.token_urlsafe(32) session["csrf_token"] = token return token def validate_csrf() -> bool: expected = session.get("csrf_token") provided = request.headers.get("X-CSRF-Token") or request.form.get("csrf_token") return bool(expected and provided and secrets.compare_digest(expected, provided)) def register_routes(app): @app.route("/healthz") def healthz(): return jsonify({"status": "ok"}) @app.before_request def enforce_csrf(): if request.method in {"POST", "PUT", "PATCH", "DELETE"} and request.endpoint != "healthz": if not validate_csrf(): if request.path.startswith("/api/"): return json_error("CSRF invalido.", 403) flash("Sessao expirada. Tente novamente.") return redirect(url_for("login")) @app.route("/login", methods=["GET", "POST"]) def login(): if current_user.is_authenticated: return redirect(url_for("index")) if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") if get_runtime_state().authenticate(username, password): login_user(User(username)) session.permanent = True session["csrf_token"] = secrets.token_urlsafe(32) return redirect(url_for("index")) flash("Usuario ou senha invalidos.") return render_template("login.html", csrf_token=get_csrf_token()) @app.route("/") @login_required def index(): config, _, _, _ = get_runtime_state().snapshot() license_status = get_license_service().verify() monitor_status = get_runtime_state().get_monitor_status() return render_template( "index.html", config=config, licenciado=license_status.valid, lic_msg=license_status.message, hwid=license_status.hardware_id, dias_restantes=license_status.days_remaining, csrf_token=get_csrf_token(), monitor_status=monitor_status, ) @app.route("/api/nodes", methods=["POST"]) @login_required def api_nodes(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return json_error("JSON invalido.") try: zabbix_server = str(payload.get("zabbix_server", "")).strip() hostname_zabbix = str(payload.get("hostname_zabbix", "")).strip() nodes = [validate_node_payload(node) for node in payload.get("nodes", [])] except ValueError as exc: return json_error(str(exc)) if not zabbix_server: return json_error("Zabbix Server e obrigatorio.") if not hostname_zabbix: return json_error("Hostname Zabbix e obrigatorio.") get_runtime_state().update_config(zabbix_server, hostname_zabbix, nodes) return jsonify({"status": "success"}) @app.route("/api/rename", methods=["POST"]) @login_required def api_rename(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return json_error("JSON invalido.") address_key = str(payload.get("addr", "")).strip() label = str(payload.get("label", "")).strip() if not address_key: return json_error("Endereco do dispositivo e obrigatorio.") if not label: return json_error("Nome do dispositivo e obrigatorio.") if not get_runtime_state().rename_device(address_key, label): return json_error("Dispositivo nao encontrado.", 404) return jsonify({"status": "success"}) @app.route("/api/license", methods=["POST"]) @login_required def api_license(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return json_error("JSON invalido.") try: status = get_license_service().install(str(payload.get("serial", ""))) except ValueError as exc: return json_error(str(exc)) get_runtime_state().trigger_restart() return jsonify({"status": "success", "message": status.message, "days_remaining": status.days_remaining}) @app.route("/api/rescan_node", methods=["POST"]) @login_required def api_rescan(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return json_error("JSON invalido.") try: node = validate_node_payload(payload) except ValueError as exc: return json_error(str(exc)) _, current_map, _, _ = get_runtime_state().snapshot() discovered_map = discover_node(node, current_map) get_runtime_state().replace_node_devices(node, discovered_map) return jsonify({"status": "success", "count": len(discovered_map)}) @app.route("/api/data") @login_required def api_data(): _, device_map, realtime_data, node_statuses = get_runtime_state().snapshot() sensors = [] for key, info in device_map.items(): data = realtime_data.get( key, {"status": 0, "last": "--:--:--", "node": info.get("node_nome", "...")}, ) status_info = get_status_info(int(data["status"])) sensors.append( { "key": key, "label": info["label"], "node": data.get("node", "..."), "status_text": status_info["label"], "color": status_info["color"], "last": data["last"], } ) return jsonify({"sensores": sensors, "nodes_status": node_statuses}) @app.route("/api/system/status") @login_required def api_system_status(): config, device_map, realtime_data, node_statuses = get_runtime_state().snapshot() license_status = get_license_service().verify() monitor_status = get_runtime_state().get_monitor_status() return jsonify( { "status": "success", "license": { "valid": license_status.valid, "message": license_status.message, "hardware_id": license_status.hardware_id, "days_remaining": license_status.days_remaining, }, "monitor": monitor_status.__dict__, "counts": { "nodes": len(config.get("nodes", [])), "devices": len(device_map), "realtime_points": len(realtime_data), }, "nodes_status": node_statuses, } ) @app.route("/api/account", methods=["POST"]) @login_required def api_account(): payload = request.get_json(silent=True) if not isinstance(payload, dict): return json_error("JSON invalido.") username = str(payload.get("username", "")) current_password = str(payload.get("current_password", "")) new_password = str(payload.get("new_password", "")) if not get_runtime_state().authenticate(current_user.get_id(), current_password): return json_error("Senha atual invalida.", 403) try: validated_username, validated_password = validate_login_payload(username, new_password) except ValueError as exc: return json_error(str(exc)) get_runtime_state().update_credentials(validated_username, validated_password) logout_user() session.clear() return jsonify({"status": "success"}) @app.route("/logout") @login_required def logout(): logout_user() session.clear() return redirect(url_for("login"))