548 lines
18 KiB
Python
548 lines
18 KiB
Python
import os
|
|
import re
|
|
import time
|
|
import ipaddress
|
|
from functools import wraps
|
|
|
|
import requests
|
|
import urllib3
|
|
from flask import (
|
|
Flask,
|
|
render_template,
|
|
request,
|
|
jsonify,
|
|
session,
|
|
redirect,
|
|
url_for,
|
|
)
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.getenv("FLASK_SECRET_KEY", "change-me-now")
|
|
|
|
PROXMOX_BASE_URL = os.getenv("PROXMOX_BASE_URL", "").rstrip("/")
|
|
PROXMOX_TOKEN_ID = os.getenv("PROXMOX_TOKEN_ID", "")
|
|
PROXMOX_TOKEN_SECRET = os.getenv("PROXMOX_TOKEN_SECRET", "")
|
|
PROXMOX_VERIFY_SSL = os.getenv("PROXMOX_VERIFY_SSL", "false").lower() == "true"
|
|
|
|
APP_USERNAME = os.getenv("APP_USERNAME", "admin")
|
|
APP_PASSWORD = os.getenv("APP_PASSWORD", "admin")
|
|
|
|
DEFAULT_NODE = os.getenv("DEFAULT_NODE", "")
|
|
DEFAULT_STORAGE = os.getenv("DEFAULT_STORAGE", "")
|
|
DEFAULT_BRIDGE = os.getenv("DEFAULT_BRIDGE", "vmbr0")
|
|
DEFAULT_SEARCH_DOMAIN = os.getenv("DEFAULT_SEARCH_DOMAIN", "local")
|
|
ALLOWED_SUBNET = os.getenv("ALLOWED_SUBNET", "192.168.50.0/24")
|
|
DEFAULT_GATEWAY = os.getenv("DEFAULT_GATEWAY", "")
|
|
DEFAULT_DNS = os.getenv("DEFAULT_DNS", "")
|
|
|
|
def proxmox_headers():
|
|
return {
|
|
"Authorization": f"PVEAPIToken={PROXMOX_TOKEN_ID}={PROXMOX_TOKEN_SECRET}"
|
|
}
|
|
|
|
|
|
def proxmox_get(path, params=None):
|
|
url = f"{PROXMOX_BASE_URL}/api2/json{path}"
|
|
response = requests.get(
|
|
url,
|
|
headers=proxmox_headers(),
|
|
params=params,
|
|
verify=PROXMOX_VERIFY_SSL,
|
|
timeout=60,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def proxmox_post(path, data=None):
|
|
url = f"{PROXMOX_BASE_URL}/api2/json{path}"
|
|
response = requests.post(
|
|
url,
|
|
headers=proxmox_headers(),
|
|
data=data,
|
|
verify=PROXMOX_VERIFY_SSL,
|
|
timeout=120,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def login_required(fn):
|
|
@wraps(fn)
|
|
def wrapper(*args, **kwargs):
|
|
if not session.get("logged_in"):
|
|
if request.path.startswith("/api/"):
|
|
return jsonify({"success": False, "error": "Nepřihlášený uživatel."}), 401
|
|
return redirect(url_for("login"))
|
|
return fn(*args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def validate_hostname(hostname: str):
|
|
hostname = hostname.strip().lower()
|
|
|
|
if len(hostname) < 2 or len(hostname) > 63:
|
|
raise ValueError("Hostname musí mít 2 až 63 znaků.")
|
|
|
|
if hostname.startswith("-") or hostname.endswith("-"):
|
|
raise ValueError("Hostname nesmí začínat ani končit pomlčkou.")
|
|
|
|
if not re.fullmatch(r"[a-z0-9-]+", hostname):
|
|
raise ValueError("Hostname může obsahovat jen malá písmena, čísla a pomlčku.")
|
|
|
|
return hostname
|
|
|
|
|
|
def validate_vmid(vmid):
|
|
try:
|
|
vmid = int(vmid)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("VMID musí být číslo.")
|
|
|
|
if vmid < 100 or vmid > 999999999:
|
|
raise ValueError("VMID je mimo povolený rozsah.")
|
|
|
|
return vmid
|
|
|
|
|
|
def validate_int_field(value, field_name, min_value, max_value):
|
|
try:
|
|
value = int(value)
|
|
except (TypeError, ValueError):
|
|
raise ValueError(f"{field_name} musí být číslo.")
|
|
|
|
if value < min_value or value > max_value:
|
|
raise ValueError(f"{field_name} musí být v rozsahu {min_value} až {max_value}.")
|
|
|
|
return value
|
|
|
|
|
|
def validate_vlan(vlan):
|
|
if vlan in (None, "", "null"):
|
|
return None
|
|
|
|
try:
|
|
vlan = int(vlan)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("VLAN musí být číslo.")
|
|
|
|
if vlan < 1 or vlan > 4094:
|
|
raise ValueError("VLAN musí být v rozsahu 1 až 4094.")
|
|
|
|
return vlan
|
|
|
|
|
|
def validate_ip_config(ip_mode, ip_last_octet, gateway):
|
|
network = ipaddress.ip_network(ALLOWED_SUBNET, strict=False)
|
|
|
|
if ip_mode == "dhcp":
|
|
return "dhcp", None
|
|
|
|
if ip_mode != "static":
|
|
raise ValueError("Neplatný IP režim.")
|
|
|
|
try:
|
|
last_octet = int(ip_last_octet)
|
|
except (TypeError, ValueError):
|
|
raise ValueError("Poslední oktet IP adresy musí být číslo.")
|
|
|
|
if last_octet < 1 or last_octet > 254:
|
|
raise ValueError("Poslední oktet IP adresy musí být v rozsahu 1 až 254.")
|
|
|
|
base = str(network.network_address).split(".")
|
|
ip_addr = f"{base[0]}.{base[1]}.{base[2]}.{last_octet}"
|
|
|
|
try:
|
|
iface = ipaddress.ip_interface(f"{ip_addr}/{network.prefixlen}")
|
|
except ValueError:
|
|
raise ValueError("Neplatná IP adresa.")
|
|
|
|
if iface.ip not in network:
|
|
raise ValueError("IP adresa není v povoleném subnetu.")
|
|
|
|
if gateway:
|
|
try:
|
|
gw = ipaddress.ip_address(gateway)
|
|
except ValueError:
|
|
raise ValueError("Neplatná gateway.")
|
|
|
|
if gw not in network:
|
|
raise ValueError("Gateway není v povoleném subnetu.")
|
|
else:
|
|
if DEFAULT_GATEWAY:
|
|
try:
|
|
gw = ipaddress.ip_address(DEFAULT_GATEWAY)
|
|
except ValueError:
|
|
raise ValueError("DEFAULT_GATEWAY v .env není platná IP adresa.")
|
|
|
|
if gw not in network:
|
|
raise ValueError("DEFAULT_GATEWAY není v povoleném subnetu.")
|
|
else:
|
|
gw = list(network.hosts())[0]
|
|
|
|
return f"{iface.ip}/{network.prefixlen}", str(gw)
|
|
|
|
|
|
def build_net0(bridge, ip_mode, ip_last_octet=None, gateway=None, vlan=None):
|
|
if not bridge:
|
|
raise ValueError("Bridge je povinný.")
|
|
|
|
parts = [f"name=eth0", f"bridge={bridge}"]
|
|
|
|
if vlan is not None:
|
|
parts.append(f"tag={vlan}")
|
|
|
|
ip_value, gw = validate_ip_config(ip_mode, ip_last_octet, gateway)
|
|
|
|
if ip_value == "dhcp":
|
|
parts.append("ip=dhcp")
|
|
else:
|
|
parts.append(f"ip={ip_value}")
|
|
parts.append(f"gw={gw}")
|
|
|
|
return ",".join(parts)
|
|
|
|
|
|
def parse_upid_node(upid: str):
|
|
# typický formát: UPID:node:...
|
|
parts = upid.split(":")
|
|
if len(parts) > 1:
|
|
return parts[1]
|
|
return DEFAULT_NODE
|
|
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
|
|
if username == APP_USERNAME and password == APP_PASSWORD:
|
|
session["logged_in"] = True
|
|
return redirect(url_for("index"))
|
|
|
|
return render_template("login.html", error="Neplatné přihlašovací údaje.")
|
|
|
|
return render_template("login.html")
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def index():
|
|
subnet_prefix = ".".join(str(ipaddress.ip_network(ALLOWED_SUBNET, strict=False).network_address).split(".")[:3])
|
|
return render_template(
|
|
"index.html",
|
|
default_node=DEFAULT_NODE,
|
|
default_storage=DEFAULT_STORAGE,
|
|
default_bridge=DEFAULT_BRIDGE,
|
|
default_search_domain=DEFAULT_SEARCH_DOMAIN,
|
|
subnet_prefix=f"{subnet_prefix}.",
|
|
default_subnet_prefix=f"{subnet_prefix}.", # ← přidat
|
|
allowed_subnet=ALLOWED_SUBNET,
|
|
default_gateway=DEFAULT_GATEWAY,
|
|
default_dns=DEFAULT_DNS,
|
|
)
|
|
|
|
@app.route("/api/nodes")
|
|
@login_required
|
|
def api_nodes():
|
|
try:
|
|
payload = proxmox_get("/nodes")
|
|
nodes = []
|
|
|
|
for item in payload.get("data", []):
|
|
node = item.get("node")
|
|
if node:
|
|
nodes.append(node)
|
|
|
|
return jsonify({"success": True, "nodes": nodes})
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/api/storage/<node>")
|
|
@login_required
|
|
def api_storage(node):
|
|
try:
|
|
payload = proxmox_get(f"/nodes/{node}/storage")
|
|
storages = []
|
|
|
|
for item in payload.get("data", []):
|
|
storage = item.get("storage")
|
|
if storage and item.get("enabled", 1):
|
|
storages.append(storage)
|
|
|
|
return jsonify({"success": True, "storage": storages})
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/api/templates/<node>/<storage>")
|
|
@login_required
|
|
def api_templates(node, storage):
|
|
try:
|
|
template_storage = os.getenv("LXC_TEMPLATE_STORAGE", "local").strip() or "local"
|
|
payload = proxmox_get(f"/nodes/{node}/storage/{template_storage}/content")
|
|
templates = []
|
|
|
|
for item in payload.get("data", []):
|
|
if item.get("content") == "vztmpl":
|
|
volid = item.get("volid")
|
|
if volid:
|
|
templates.append(volid)
|
|
|
|
return jsonify({"success": True, "templates": sorted(templates)})
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
@app.route("/api/next-vmid")
|
|
@login_required
|
|
def api_next_vmid():
|
|
try:
|
|
payload = proxmox_get("/cluster/nextid")
|
|
return jsonify({"success": True, "vmid": payload.get("data")})
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/api/create-lxc", methods=["POST"])
|
|
@login_required
|
|
def api_create_lxc():
|
|
try:
|
|
data = request.get_json(force=True)
|
|
|
|
node = data.get("node", "").strip()
|
|
hostname = validate_hostname(data.get("hostname", ""))
|
|
vmid = validate_vmid(data.get("vmid"))
|
|
cores = validate_int_field(data.get("cores"), "CPU", 1, 64)
|
|
memory = validate_int_field(data.get("memory"), "RAM", 128, 262144)
|
|
disk = validate_int_field(data.get("disk"), "Disk", 1, 2048)
|
|
swap = validate_int_field(data.get("swap", 512), "Swap", 0, 65536)
|
|
|
|
storage = data.get("storage", "").strip()
|
|
template = data.get("template", "").strip()
|
|
bridge = data.get("bridge", DEFAULT_BRIDGE).strip()
|
|
searchdomain = data.get("searchdomain", DEFAULT_SEARCH_DOMAIN).strip()
|
|
nameserver = data.get("nameserver", "").strip() or os.getenv("DEFAULT_DNS", "")
|
|
password = data.get("password", "")
|
|
start_on_boot = 1 if data.get("start_on_boot", True) else 0
|
|
unprivileged = 1 if data.get("unprivileged", True) else 0
|
|
vlan = validate_vlan(data.get("vlan"))
|
|
|
|
if not node:
|
|
raise ValueError("Node je povinný.")
|
|
if not storage:
|
|
raise ValueError("Storage je povinné.")
|
|
if not template:
|
|
raise ValueError("LXC template je povinná.")
|
|
if not password:
|
|
raise ValueError("Root heslo je povinné.")
|
|
|
|
ip_mode = data.get("ip_mode", "dhcp")
|
|
ip_last_octet = data.get("ip_last_octet")
|
|
gateway = data.get("gateway", "").strip() or None
|
|
|
|
net0 = build_net0(
|
|
bridge=bridge,
|
|
ip_mode=ip_mode,
|
|
ip_last_octet=ip_last_octet,
|
|
gateway=gateway,
|
|
vlan=vlan,
|
|
)
|
|
enable_nesting = bool(data.get("enable_nesting", False))
|
|
payload = {
|
|
"vmid": vmid,
|
|
"hostname": hostname,
|
|
"ostemplate": template,
|
|
"rootfs": f"{storage}:{disk}",
|
|
"cores": cores,
|
|
"memory": memory,
|
|
"swap": swap,
|
|
"password": password,
|
|
"net0": net0,
|
|
"onboot": start_on_boot,
|
|
"unprivileged": unprivileged,
|
|
}
|
|
if enable_nesting:
|
|
if not unprivileged:
|
|
raise ValueError("Nesting je povolen jen pro unprivileged kontejnery.")
|
|
payload["features"] = "nesting=1"
|
|
|
|
if searchdomain:
|
|
payload["searchdomain"] = searchdomain
|
|
if nameserver:
|
|
payload["nameserver"] = nameserver
|
|
|
|
result = proxmox_post(f"/nodes/{node}/lxc", data=payload)
|
|
upid = result.get("data")
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"upid": upid,
|
|
"node": node,
|
|
"vmid": vmid,
|
|
"hostname": hostname,
|
|
})
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except ValueError as e:
|
|
return jsonify({"success": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/api/task-status")
|
|
@login_required
|
|
def api_task_status():
|
|
try:
|
|
node = request.args.get("node", "").strip()
|
|
upid = request.args.get("upid", "").strip()
|
|
|
|
if not upid:
|
|
raise ValueError("Chybí UPID tasku.")
|
|
|
|
if not node:
|
|
node = parse_upid_node(upid)
|
|
|
|
status_payload = proxmox_get(f"/nodes/{node}/tasks/{upid}/status")
|
|
log_payload = proxmox_get(f"/nodes/{node}/tasks/{upid}/log", params={"start": 0, "limit": 50})
|
|
|
|
status_data = status_payload.get("data", {})
|
|
log_data = log_payload.get("data", [])
|
|
|
|
exitstatus = status_data.get("exitstatus")
|
|
running = status_data.get("status") == "running"
|
|
|
|
if running:
|
|
progress = 50
|
|
elif exitstatus == "OK":
|
|
progress = 100
|
|
else:
|
|
progress = 100
|
|
|
|
logs = [entry.get("t", "") for entry in log_data if entry.get("t")]
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"running": running,
|
|
"status": status_data.get("status"),
|
|
"exitstatus": exitstatus,
|
|
"starttime": status_data.get("starttime"),
|
|
"endtime": status_data.get("endtime"),
|
|
"progress": progress,
|
|
"logs": logs[-10:],
|
|
})
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except ValueError as e:
|
|
return jsonify({"success": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/api/container-details/<node>/<int:vmid>")
|
|
@login_required
|
|
def api_container_details(node, vmid):
|
|
try:
|
|
config_payload = proxmox_get(f"/nodes/{node}/lxc/{vmid}/config")
|
|
status_payload = proxmox_get(f"/nodes/{node}/lxc/{vmid}/status/current")
|
|
|
|
config = config_payload.get("data", {})
|
|
status = status_payload.get("data", {})
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"details": {
|
|
"vmid": vmid,
|
|
"node": node,
|
|
"hostname": config.get("hostname"),
|
|
"ostemplate": config.get("ostemplate"),
|
|
"cores": config.get("cores"),
|
|
"memory": config.get("memory"),
|
|
"swap": config.get("swap"),
|
|
"rootfs": config.get("rootfs"),
|
|
"net0": config.get("net0"),
|
|
"searchdomain": config.get("searchdomain"),
|
|
"nameserver": config.get("nameserver"),
|
|
"status": status.get("status"),
|
|
"uptime": status.get("uptime"),
|
|
"cpus": status.get("cpus"),
|
|
"maxmem": status.get("maxmem"),
|
|
"maxdisk": status.get("maxdisk"),
|
|
"ip_mode_subnet": ALLOWED_SUBNET,
|
|
}
|
|
})
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
@app.route("/health")
|
|
def health():
|
|
return jsonify({"success": True, "status": "ok"})
|
|
|
|
@app.route("/api/start-lxc", methods=["POST"])
|
|
@login_required
|
|
def api_start_lxc():
|
|
try:
|
|
data = request.get_json(force=True)
|
|
|
|
node = data.get("node", "").strip()
|
|
vmid = validate_vmid(data.get("vmid"))
|
|
|
|
if not node:
|
|
raise ValueError("Node je povinný.")
|
|
|
|
result = proxmox_post(f"/nodes/{node}/lxc/{vmid}/status/start", data={})
|
|
upid = result.get("data")
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"upid": upid,
|
|
"node": node,
|
|
"vmid": vmid,
|
|
})
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
body = e.response.text if e.response is not None else str(e)
|
|
return jsonify({"success": False, "error": f"Proxmox HTTP {e.response.status_code}: {body}"}), 500
|
|
except ValueError as e:
|
|
return jsonify({"success": False, "error": str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({"success": False, "error": f"Neočekávaná chyba: {str(e)}"}), 500
|
|
|
|
|
|
if __name__ == "__main__":
|
|
host = os.getenv("APP_HOST", "0.0.0.0")
|
|
port = int(os.getenv("APP_PORT", "8000"))
|
|
debug = os.getenv("FLASK_DEBUG", "false").lower() == "true"
|
|
|
|
app.run(host=host, port=port, debug=debug)
|