Your Name
65455c55cf
- frps assigns random clientID without user field - Web UI matches clients by name, so all showed offline - Now render_frpc_ini() writes user = <client-name> - frps dashboard API returns user field for matching
1180 lines
44 KiB
Python
1180 lines
44 KiB
Python
#!/usr/bin/env python3
|
|
"""frps-manager — Flask + SQLite Web management UI for frp server (frps 0.70.1).
|
|
|
|
Design notes:
|
|
fatedier/frp 0.70.1 has a broken TOML auth story:
|
|
* The TOML parser does NOT accept the dot-notation "auth.token" key
|
|
(despite what the official example shows).
|
|
* The --token CLI flag sets c.Auth.Token but is then OVERWRITTEN by
|
|
LoadConfigureFromFile which always re-inits ServerConfig from scratch.
|
|
* Only the legacy INI format actually loads token correctly.
|
|
Therefore we generate frps.ini (legacy INI) — the example file warns it
|
|
is deprecated, but until 0.70+ fixes the TOML auth, INI is the only
|
|
option that actually works.
|
|
|
|
Same applies to frpc.ini: the verified working config is INI legacy.
|
|
|
|
This file does NOT include [[proxies]] blocks — frps 0.70+ has no
|
|
server-side proxy config; all proxies are pushed by frpc at connection.
|
|
"""
|
|
import configparser
|
|
import io
|
|
import os
|
|
import secrets
|
|
import string
|
|
import subprocess
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from functools import wraps
|
|
|
|
import requests
|
|
from flask import (Flask, Response, abort, flash, g, jsonify, redirect,
|
|
render_template, request, session, url_for)
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Paths & config
|
|
# ----------------------------------------------------------------------------
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
RUNTIME_DIR = os.path.join(BASE_DIR, "runtime")
|
|
CONF_DIR = os.path.join(RUNTIME_DIR, "conf")
|
|
LOG_DIR = os.path.join(RUNTIME_DIR, "logs")
|
|
BIN_DIR = os.path.join(RUNTIME_DIR, "bin")
|
|
INSTANCE_DIR = os.path.join(BASE_DIR, "instance")
|
|
SERVER_TOKEN_FILE = os.path.join(INSTANCE_DIR, "server_token")
|
|
|
|
FRPS_BIN = os.environ.get("FRPS_BIN", os.path.join(BIN_DIR, "frps"))
|
|
# frps 0.70.1's TOML parser doesn't accept auth.token, so we use legacy INI
|
|
FRPS_CONFIG = os.environ.get("FRPS_CONFIG", os.path.join(CONF_DIR, "frps.ini"))
|
|
FRPS_SERVICE = os.environ.get("FRPS_SERVICE", "frps")
|
|
FRPS_DASHBOARD_URL = os.environ.get(
|
|
"FRPS_DASHBOARD_URL", "http://127.0.0.1:7500")
|
|
FRPS_DASHBOARD_USER = os.environ.get("FRPS_DASHBOARD_USER", "admin")
|
|
FRPS_DASHBOARD_PASS = os.environ.get("FRPS_DASHBOARD_PASS", "admin123")
|
|
BIND_PORT = int(os.environ.get("FRPS_BIND_PORT", "7000"))
|
|
WEB_BIND_ADDR = os.environ.get("FRPS_WEB_ADDR", "127.0.0.1")
|
|
WEB_BIND_PORT = int(os.environ.get("FRPS_WEB_PORT", "7500"))
|
|
|
|
for d in (RUNTIME_DIR, CONF_DIR, LOG_DIR, INSTANCE_DIR):
|
|
os.makedirs(d, exist_ok=True)
|
|
|
|
DB_PATH = os.path.join(INSTANCE_DIR, "frps_manager.db")
|
|
SECRET_KEY_FILE = os.path.join(INSTANCE_DIR, "secret_key")
|
|
|
|
|
|
def _get_or_create_secret_key():
|
|
"""Persistent SECRET_KEY shared by all gunicorn workers.
|
|
|
|
CRITICAL: with >1 worker, a random per-import key breaks sessions —
|
|
login lands on worker A, the redirect on worker B, and B can't
|
|
verify A's cookie. The key must be identical across workers and
|
|
survive restarts, so we persist it to instance/secret_key.
|
|
"""
|
|
env = os.environ.get("FLASK_SECRET")
|
|
if env:
|
|
return env
|
|
if os.path.exists(SECRET_KEY_FILE):
|
|
with open(SECRET_KEY_FILE, "r", encoding="utf-8") as f:
|
|
key = f.read().strip()
|
|
if key:
|
|
return key
|
|
key = secrets.token_hex(32)
|
|
with open(SECRET_KEY_FILE, "w", encoding="utf-8") as f:
|
|
f.write(key)
|
|
os.chmod(SECRET_KEY_FILE, 0o600)
|
|
return key
|
|
|
|
|
|
app = Flask(__name__)
|
|
app.config["SECRET_KEY"] = _get_or_create_secret_key()
|
|
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{DB_PATH}"
|
|
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
|
app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(hours=12)
|
|
|
|
|
|
@app.context_processor
|
|
def inject_globals():
|
|
return {"now": datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")}
|
|
|
|
|
|
db = SQLAlchemy(app)
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Models
|
|
# ----------------------------------------------------------------------------
|
|
ROLES = ("admin", "tenant_admin", "user")
|
|
|
|
|
|
class User(db.Model):
|
|
__tablename__ = "users"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(64), unique=True, nullable=False)
|
|
password_hash = db.Column(db.String(256), nullable=False)
|
|
role = db.Column(db.String(16), nullable=False, default="user")
|
|
tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"), nullable=True)
|
|
display_name = db.Column(db.String(128))
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
last_login_at = db.Column(db.DateTime)
|
|
tenant = db.relationship("Tenant", backref="users", foreign_keys=[tenant_id])
|
|
|
|
def set_password(self, raw):
|
|
self.password_hash = generate_password_hash(raw)
|
|
|
|
def check_password(self, raw):
|
|
return check_password_hash(self.password_hash, raw)
|
|
|
|
|
|
class Tenant(db.Model):
|
|
__tablename__ = "tenants"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(128), unique=True, nullable=False)
|
|
description = db.Column(db.String(256))
|
|
bandwidth_in_mbps = db.Column(db.Integer, default=0)
|
|
bandwidth_out_mbps = db.Column(db.Integer, default=0)
|
|
max_connections = db.Column(db.Integer, default=0)
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
|
|
|
|
class Client(db.Model):
|
|
"""An frpc registration identified by its [common] user/name."""
|
|
__tablename__ = "clients"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
tenant_id = db.Column(db.Integer, db.ForeignKey("tenants.id"), nullable=False)
|
|
name = db.Column(db.String(128), nullable=False)
|
|
# Label token (NOT used by frp 0.70+; for documentation / future)
|
|
label_token = db.Column(db.String(64), nullable=False)
|
|
server_addr = db.Column(db.String(128))
|
|
description = db.Column(db.String(256))
|
|
is_active = db.Column(db.Boolean, default=True)
|
|
# When proxy config changes, we set this; the dashboard tells user
|
|
# "frpc config out of date — pull fresh one and frpc reload"
|
|
config_revision = db.Column(db.Integer, default=0)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
tenant = db.relationship("Tenant", backref="clients")
|
|
proxies = db.relationship(
|
|
"Proxy", backref="client", cascade="all, delete-orphan")
|
|
__table_args__ = (
|
|
db.UniqueConstraint("tenant_id", "name", name="uq_client_tenant_name"),
|
|
)
|
|
|
|
|
|
PROXY_TYPES = ("tcp", "udp", "http", "https", "tcpmux")
|
|
|
|
|
|
class Proxy(db.Model):
|
|
"""A proxy rule owned by a client (rendered into frpc.ini)."""
|
|
__tablename__ = "proxies"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
client_id = db.Column(db.Integer, db.ForeignKey("clients.id"), nullable=False)
|
|
name = db.Column(db.String(128), nullable=False)
|
|
type = db.Column(db.String(16), nullable=False, default="tcp")
|
|
local_ip = db.Column(db.String(64), default="127.0.0.1")
|
|
local_port = db.Column(db.Integer)
|
|
remote_port = db.Column(db.Integer)
|
|
custom_domains = db.Column(db.String(512))
|
|
subdomain = db.Column(db.String(64))
|
|
locations = db.Column(db.String(256))
|
|
use_encryption = db.Column(db.Boolean, default=False)
|
|
use_compression = db.Column(db.Boolean, default=False)
|
|
health_check_type = db.Column(db.String(16))
|
|
health_check_url = db.Column(db.String(128))
|
|
health_check_interval_s = db.Column(db.Integer, default=30)
|
|
health_check_timeout_s = db.Column(db.Integer, default=3)
|
|
note = db.Column(db.String(256))
|
|
is_enabled = db.Column(db.Boolean, default=True)
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
updated_at = db.Column(db.DateTime, default=datetime.utcnow,
|
|
onupdate=datetime.utcnow)
|
|
__table_args__ = (
|
|
db.UniqueConstraint("client_id", "name", name="uq_proxy_client_name"),
|
|
)
|
|
|
|
|
|
class AuditLog(db.Model):
|
|
__tablename__ = "audit_logs"
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
|
|
username = db.Column(db.String(64))
|
|
action = db.Column(db.String(64), nullable=False)
|
|
target_type = db.Column(db.String(32))
|
|
target_id = db.Column(db.String(64))
|
|
target_label = db.Column(db.String(256))
|
|
detail = db.Column(db.Text)
|
|
ip = db.Column(db.String(64))
|
|
created_at = db.Column(db.DateTime, default=datetime.utcnow)
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Helpers
|
|
# ----------------------------------------------------------------------------
|
|
def log_action(action, target_type=None, target_id=None, target_label=None,
|
|
detail=None):
|
|
u = g.current_user if hasattr(g, "current_user") else None
|
|
try:
|
|
entry = AuditLog(
|
|
user_id=u.id if u else None,
|
|
username=u.username if u else None,
|
|
action=action, target_type=target_type,
|
|
target_id=str(target_id) if target_id is not None else None,
|
|
target_label=target_label, detail=detail,
|
|
ip=request.remote_addr if request else None,
|
|
)
|
|
db.session.add(entry)
|
|
db.session.commit()
|
|
except Exception as e:
|
|
app.logger.warning("audit log failed: %s", e)
|
|
try:
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def random_token(n=24):
|
|
alphabet = string.ascii_letters + string.digits
|
|
return "".join(secrets.choice(alphabet) for _ in range(n))
|
|
|
|
|
|
def current_user():
|
|
if "user_id" not in session:
|
|
return None
|
|
if hasattr(g, "current_user_obj") and g.current_user_obj:
|
|
return g.current_user_obj
|
|
u = User.query.get(session["user_id"])
|
|
if not u or not u.is_active:
|
|
return None
|
|
g.current_user_obj = u
|
|
return u
|
|
|
|
|
|
def login_required(view):
|
|
@wraps(view)
|
|
def wrapped(*args, **kwargs):
|
|
u = current_user()
|
|
if not u:
|
|
return redirect(url_for("login", next=request.path))
|
|
g.current_user = u
|
|
return view(*args, **kwargs)
|
|
return wrapped
|
|
|
|
|
|
def role_required(*roles):
|
|
def deco(view):
|
|
@wraps(view)
|
|
def wrapped(*args, **kwargs):
|
|
u = current_user()
|
|
if not u:
|
|
return redirect(url_for("login"))
|
|
if u.role not in roles:
|
|
abort(403)
|
|
g.current_user = u
|
|
return view(*args, **kwargs)
|
|
return wrapped
|
|
return deco
|
|
|
|
|
|
def visible_tenants(user):
|
|
if user.role == "admin":
|
|
return Tenant.query.order_by(Tenant.name).all()
|
|
if user.tenant_id:
|
|
return Tenant.query.filter_by(id=user.tenant_id).all()
|
|
return []
|
|
|
|
|
|
def visible_clients(user, tenant_id=None):
|
|
q = Client.query
|
|
if user.role == "admin":
|
|
if tenant_id:
|
|
q = q.filter_by(tenant_id=tenant_id)
|
|
else:
|
|
if not user.tenant_id:
|
|
return []
|
|
q = q.filter_by(tenant_id=user.tenant_id)
|
|
if tenant_id and tenant_id != user.tenant_id:
|
|
return []
|
|
return q.order_by(Client.name).all()
|
|
|
|
|
|
def visible_proxies(user, client_id=None, tenant_id=None):
|
|
client_ids = [c.id for c in visible_clients(user, tenant_id=tenant_id)]
|
|
if not client_ids:
|
|
return []
|
|
q = Proxy.query.filter(Proxy.client_id.in_(client_ids))
|
|
if client_id:
|
|
q = q.filter_by(client_id=client_id)
|
|
return q.order_by(Proxy.name).all()
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Server token
|
|
# ----------------------------------------------------------------------------
|
|
def get_or_create_server_token():
|
|
if os.path.exists(SERVER_TOKEN_FILE):
|
|
with open(SERVER_TOKEN_FILE, "r") as f:
|
|
tok = f.read().strip()
|
|
if tok:
|
|
return tok
|
|
tok = random_token(32)
|
|
with open(SERVER_TOKEN_FILE, "w") as f:
|
|
f.write(tok)
|
|
os.chmod(SERVER_TOKEN_FILE, 0o600)
|
|
return tok
|
|
|
|
|
|
def rotate_server_token():
|
|
tok = random_token(32)
|
|
with open(SERVER_TOKEN_FILE, "w") as f:
|
|
f.write(tok)
|
|
os.chmod(SERVER_TOKEN_FILE, 0o600)
|
|
return tok
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# frps INI rendering (legacy INI is the only format that loads auth token
|
|
# correctly in frps 0.70.1)
|
|
# ----------------------------------------------------------------------------
|
|
def render_frps_ini():
|
|
"""Build runtime frps.ini in legacy INI format."""
|
|
token = get_or_create_server_token()
|
|
cp = configparser.RawConfigParser()
|
|
cp.optionxform = str # preserve case
|
|
cp["common"] = {
|
|
"bind_addr": "0.0.0.0",
|
|
"bind_port": str(BIND_PORT),
|
|
# Authentication
|
|
"authentication_method": "token",
|
|
"token": token,
|
|
# Dashboard
|
|
"dashboard_addr": WEB_BIND_ADDR,
|
|
"dashboard_port": str(WEB_BIND_PORT),
|
|
"dashboard_user": FRPS_DASHBOARD_USER,
|
|
"dashboard_pwd": FRPS_DASHBOARD_PASS,
|
|
# HTTP/HTTPS virtual host ports — required for http/https proxies
|
|
"vhost_http_port": str(os.environ.get("FRPS_VHOST_HTTP_PORT", "80")),
|
|
"vhost_https_port": str(os.environ.get("FRPS_VHOST_HTTPS_PORT", "443")),
|
|
# Logging
|
|
"log_file": os.path.join(LOG_DIR, "frps.log"),
|
|
"log_level": "info",
|
|
"log_max_days": "7",
|
|
}
|
|
out = io.StringIO()
|
|
cp.write(out)
|
|
return (
|
|
"# Managed by frps-manager — DO NOT EDIT BY HAND\n"
|
|
+ out.getvalue()
|
|
)
|
|
|
|
|
|
def write_frps_config():
|
|
content = render_frps_ini()
|
|
tmp = FRPS_CONFIG + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
os.replace(tmp, FRPS_CONFIG)
|
|
return content
|
|
|
|
|
|
def reload_frps(action="restart"):
|
|
"""Restart frps and wait for dashboard to come back."""
|
|
if action not in ("reload", "restart"):
|
|
action = "restart"
|
|
try:
|
|
result = subprocess.run(
|
|
["systemctl", action, FRPS_SERVICE + ".service"],
|
|
capture_output=True, text=True, timeout=20)
|
|
if result.returncode != 0:
|
|
return False, (result.stdout + result.stderr).strip()
|
|
except FileNotFoundError:
|
|
return _restart_frps_direct()
|
|
except Exception as e:
|
|
return False, str(e)
|
|
deadline = time.time() + 12
|
|
while time.time() < deadline:
|
|
try:
|
|
r = requests.get(
|
|
FRPS_DASHBOARD_URL + "/api/serverinfo",
|
|
auth=(FRPS_DASHBOARD_USER, FRPS_DASHBOARD_PASS), timeout=2)
|
|
if r.status_code == 200:
|
|
return True, "ok"
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.4)
|
|
return False, "frps did not become ready within 12s"
|
|
|
|
|
|
def _restart_frps_direct():
|
|
"""Restart frps directly (dev mode)."""
|
|
pidfile = os.path.join(LOG_DIR, "frps.pid")
|
|
if os.path.exists(pidfile):
|
|
try:
|
|
with open(pidfile) as f:
|
|
old_pid = int(f.read().strip())
|
|
os.kill(old_pid, 15)
|
|
time.sleep(0.5)
|
|
except (OSError, ValueError):
|
|
pass
|
|
log_file = open(os.path.join(LOG_DIR, "frps.log"), "a")
|
|
proc = subprocess.Popen(
|
|
[FRPS_BIN, "-c", FRPS_CONFIG],
|
|
stdout=log_file, stderr=log_file, cwd=RUNTIME_DIR)
|
|
with open(pidfile, "w") as f:
|
|
f.write(str(proc.pid))
|
|
time.sleep(1.5)
|
|
try:
|
|
r = requests.get(
|
|
FRPS_DASHBOARD_URL + "/api/serverinfo",
|
|
auth=(FRPS_DASHBOARD_USER, FRPS_DASHBOARD_PASS), timeout=2)
|
|
return (r.status_code == 200), "started pid=%d" % proc.pid
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
def get_dashboard_info():
|
|
base = FRPS_DASHBOARD_URL
|
|
auth = (FRPS_DASHBOARD_USER, FRPS_DASHBOARD_PASS)
|
|
out = {}
|
|
try:
|
|
r = requests.get(f"{base}/api/serverinfo", auth=auth, timeout=3)
|
|
r.raise_for_status()
|
|
out["server"] = r.json()
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
try:
|
|
r = requests.get(f"{base}/api/v2/clients", auth=auth, timeout=3)
|
|
r.raise_for_status()
|
|
out["clients"] = r.json().get("data", {})
|
|
except Exception as e:
|
|
out["clients"] = {"items": [], "total": 0, "error": str(e)}
|
|
try:
|
|
r = requests.get(f"{base}/api/v2/proxies", auth=auth, timeout=3)
|
|
r.raise_for_status()
|
|
out["proxies"] = r.json().get("data", {})
|
|
except Exception as e:
|
|
out["proxies"] = {"items": [], "total": 0, "error": str(e)}
|
|
return out
|
|
|
|
|
|
def enrich_clients_with_status(items, info):
|
|
client_index = {}
|
|
if info and not info.get("error"):
|
|
for c in info.get("clients", {}).get("items", []):
|
|
client_index[c.get("user", "") or c.get("clientID", "")] = c
|
|
for c in items:
|
|
d = client_index.get(c.name, {})
|
|
c._online = bool(d)
|
|
c._nat = d.get("nat", "-") if d else "-"
|
|
c._version = d.get("version", "-") if d else "-"
|
|
c._conn_count = d.get("conn_count", 0) if d else 0
|
|
|
|
|
|
def bump_client_config_revision(client_id):
|
|
"""Mark the client's config as needing reload."""
|
|
c = Client.query.get(client_id)
|
|
if c:
|
|
c.config_revision = (c.config_revision or 0) + 1
|
|
db.session.commit()
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# frpc.ini rendering (legacy INI)
|
|
# ----------------------------------------------------------------------------
|
|
def render_frpc_ini(client, server_addr, bind_port):
|
|
"""Build frpc.ini in legacy INI format for a single client.
|
|
|
|
Always includes admin_addr/admin_port (default 7400) so the user can
|
|
run `frpc reload -c /etc/frpc.ini` after pulling a new copy.
|
|
"""
|
|
token = get_or_create_server_token()
|
|
proxies = Proxy.query.filter_by(
|
|
client_id=client.id, is_enabled=True).order_by(Proxy.name).all()
|
|
lines = []
|
|
lines.append(f"# Generated by frps-manager for client: {client.name}")
|
|
lines.append("# (tenant={}, label_token={})".format(
|
|
client.tenant.name, client.label_token))
|
|
lines.append("# Server token comes from your admin (shared secret).")
|
|
lines.append("# Copy to your frpc host as /etc/frpc.ini, then:")
|
|
lines.append("# systemctl enable --now frpc")
|
|
lines.append("# frpc reload -c /etc/frpc.ini # to apply changes")
|
|
lines.append("")
|
|
lines.append("[common]")
|
|
lines.append(f"server_addr = {server_addr}")
|
|
lines.append(f"server_port = {bind_port}")
|
|
# Set user so frps identifies this client by name (for dashboard matching)
|
|
lines.append(f"user = {client.name}")
|
|
lines.append("authentication_method = token")
|
|
lines.append(f"token = {token}")
|
|
# Admin API for `frpc reload` command
|
|
lines.append(f"admin_addr = {os.environ.get('FRPC_ADMIN_ADDR', '127.0.0.1')}")
|
|
lines.append(f"admin_port = {os.environ.get('FRPC_ADMIN_PORT', '7400')}")
|
|
lines.append(f"admin_user = {os.environ.get('FRPC_ADMIN_USER', 'admin')}")
|
|
lines.append(f"admin_pwd = {os.environ.get('FRPC_ADMIN_PWD', 'admin')}")
|
|
lines.append("")
|
|
for p in proxies:
|
|
lines.append(f"[{p.name}]")
|
|
lines.append(f"type = {p.type}")
|
|
lines.append(f"local_ip = {p.local_ip}")
|
|
lines.append(f"local_port = {p.local_port}")
|
|
if p.type in ("tcp", "udp") and p.remote_port:
|
|
lines.append(f"remote_port = {p.remote_port}")
|
|
if p.type in ("http", "https"):
|
|
if p.custom_domains:
|
|
# INI supports comma-separated values
|
|
lines.append(f"custom_domains = {p.custom_domains}")
|
|
if p.subdomain:
|
|
lines.append(f"subdomain = {p.subdomain}")
|
|
if p.locations:
|
|
lines.append(f"locations = {p.locations}")
|
|
if p.use_encryption:
|
|
lines.append("use_encryption = true")
|
|
if p.use_compression:
|
|
lines.append("use_compression = true")
|
|
if p.health_check_type:
|
|
lines.append(f"health_check_type = {p.health_check_type}")
|
|
if p.health_check_type == "http" and p.health_check_url:
|
|
lines.append(f"health_check_url = {p.health_check_url}")
|
|
lines.append(f"health_check_interval_s = {p.health_check_interval_s or 30}")
|
|
lines.append(f"health_check_timeout_s = {p.health_check_timeout_s or 3}")
|
|
lines.append("")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Seed
|
|
# ----------------------------------------------------------------------------
|
|
def seed_admin():
|
|
if not User.query.filter_by(username="admin").first():
|
|
u = User(username="admin", role="admin", display_name="Administrator")
|
|
u.set_password("admin")
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
print("[seed] created default admin (admin/admin) — CHANGE PASSWORD")
|
|
if Tenant.query.count() == 0:
|
|
t = Tenant(name="default", description="Default tenant")
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
print("[seed] created default tenant")
|
|
get_or_create_server_token()
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Routes — auth
|
|
# ----------------------------------------------------------------------------
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "")
|
|
password = request.form.get("password", "")
|
|
app.logger.info(f"login attempt: user={username!r} pw_len={len(password)} "
|
|
f"ip={request.remote_addr} ua={request.user_agent!r}")
|
|
u = User.query.filter_by(username=username).first()
|
|
if u and u.is_active and u.check_password(password):
|
|
session["user_id"] = u.id
|
|
session.permanent = True
|
|
u.last_login_at = datetime.utcnow()
|
|
db.session.commit()
|
|
log_action("login", target_type="user", target_id=u.id,
|
|
target_label=u.username)
|
|
app.logger.info(f"login OK: user={username!r} user_id={u.id}")
|
|
return redirect(request.args.get("next") or url_for("dashboard"))
|
|
app.logger.warning(f"login FAIL: user={username!r} u_exists={bool(u)} "
|
|
f"active={u.is_active if u else 'N/A'} pw_match={u.check_password(password) if u else False}")
|
|
flash("Invalid credentials", "error")
|
|
return render_template("login.html")
|
|
|
|
|
|
@app.route("/logout")
|
|
def logout():
|
|
u = current_user()
|
|
if u:
|
|
log_action("logout", target_type="user", target_id=u.id,
|
|
target_label=u.username)
|
|
session.clear()
|
|
return redirect(url_for("login"))
|
|
|
|
|
|
@app.route("/change-password", methods=["GET", "POST"])
|
|
@login_required
|
|
def change_password():
|
|
u = g.current_user
|
|
if request.method == "POST":
|
|
old = request.form.get("old_password", "")
|
|
new = request.form.get("new_password", "")
|
|
if not u.check_password(old):
|
|
flash("Old password incorrect", "error")
|
|
elif len(new) < 6:
|
|
flash("New password must be at least 6 characters", "error")
|
|
else:
|
|
u.set_password(new)
|
|
db.session.commit()
|
|
log_action("change_password", target_type="user", target_id=u.id,
|
|
target_label=u.username)
|
|
flash("Password updated", "success")
|
|
return redirect(url_for("dashboard"))
|
|
return render_template("change_password.html", user=u)
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Routes — main pages
|
|
# ----------------------------------------------------------------------------
|
|
@app.route("/")
|
|
@login_required
|
|
def dashboard():
|
|
info = get_dashboard_info()
|
|
tenant_count = Tenant.query.count()
|
|
client_count = Client.query.count()
|
|
proxy_count = Proxy.query.count()
|
|
return render_template(
|
|
"dashboard.html", user=g.current_user,
|
|
info=info, tenant_count=tenant_count,
|
|
client_count=client_count, proxy_count=proxy_count,
|
|
server_token=get_or_create_server_token(),
|
|
)
|
|
|
|
|
|
# ---- Tenants ----------------------------------------------------------------
|
|
@app.route("/tenants")
|
|
@role_required("admin")
|
|
def tenants_list():
|
|
items = Tenant.query.order_by(Tenant.name).all()
|
|
return render_template("tenants.html", user=g.current_user, items=items)
|
|
|
|
|
|
@app.route("/tenants/new", methods=["GET", "POST"])
|
|
@role_required("admin")
|
|
def tenants_new():
|
|
if request.method == "POST":
|
|
name = request.form.get("name", "").strip()
|
|
if not name or Tenant.query.filter_by(name=name).first():
|
|
flash("Tenant name must be unique", "error")
|
|
return redirect(url_for("tenants_new"))
|
|
t = Tenant(
|
|
name=name,
|
|
description=request.form.get("description", ""),
|
|
bandwidth_in_mbps=int(request.form.get("bandwidth_in_mbps", 0) or 0),
|
|
bandwidth_out_mbps=int(request.form.get("bandwidth_out_mbps", 0) or 0),
|
|
max_connections=int(request.form.get("max_connections", 0) or 0),
|
|
is_active=bool(request.form.get("is_active")),
|
|
)
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
log_action("create", target_type="tenant", target_id=t.id,
|
|
target_label=t.name)
|
|
flash(f"Tenant '{t.name}' created", "success")
|
|
return redirect(url_for("tenants_list"))
|
|
return render_template("tenant_form.html", user=g.current_user, item=None)
|
|
|
|
|
|
@app.route("/tenants/<int:tid>/edit", methods=["GET", "POST"])
|
|
@role_required("admin")
|
|
def tenants_edit(tid):
|
|
t = Tenant.query.get_or_404(tid)
|
|
if request.method == "POST":
|
|
t.description = request.form.get("description", "")
|
|
t.bandwidth_in_mbps = int(request.form.get("bandwidth_in_mbps", 0) or 0)
|
|
t.bandwidth_out_mbps = int(request.form.get("bandwidth_out_mbps", 0) or 0)
|
|
t.max_connections = int(request.form.get("max_connections", 0) or 0)
|
|
t.is_active = bool(request.form.get("is_active"))
|
|
db.session.commit()
|
|
log_action("update", target_type="tenant", target_id=t.id,
|
|
target_label=t.name)
|
|
flash("Tenant updated", "success")
|
|
return redirect(url_for("tenants_list"))
|
|
return render_template("tenant_form.html", user=g.current_user, item=t)
|
|
|
|
|
|
@app.route("/tenants/<int:tid>/delete", methods=["POST"])
|
|
@role_required("admin")
|
|
def tenants_delete(tid):
|
|
t = Tenant.query.get_or_404(tid)
|
|
if t.clients:
|
|
flash(f"Cannot delete: tenant has {len(t.clients)} client(s)", "error")
|
|
return redirect(url_for("tenants_list"))
|
|
name = t.name
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
log_action("delete", target_type="tenant", target_id=tid,
|
|
target_label=name)
|
|
flash(f"Tenant '{name}' deleted", "success")
|
|
return redirect(url_for("tenants_list"))
|
|
|
|
|
|
# ---- Clients ---------------------------------------------------------------
|
|
@app.route("/clients")
|
|
@login_required
|
|
def clients_list():
|
|
tid = request.args.get("tenant_id", type=int)
|
|
tenants = visible_tenants(g.current_user)
|
|
items = visible_clients(g.current_user, tenant_id=tid)
|
|
info = get_dashboard_info()
|
|
enrich_clients_with_status(items, info)
|
|
return render_template(
|
|
"clients.html", user=g.current_user, items=items,
|
|
tenants=tenants, filter_tenant_id=tid,
|
|
)
|
|
|
|
|
|
@app.route("/clients/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def clients_new():
|
|
tenants = visible_tenants(g.current_user)
|
|
if not tenants:
|
|
flash("No tenants available — ask an admin to create one first",
|
|
"error")
|
|
return redirect(url_for("clients_list"))
|
|
if request.method == "POST":
|
|
name = request.form.get("name", "").strip()
|
|
tid = int(request.form.get("tenant_id", 0))
|
|
if not any(t.id == tid for t in tenants):
|
|
flash("Tenant not allowed", "error")
|
|
return redirect(url_for("clients_new"))
|
|
if not name or Client.query.filter_by(
|
|
tenant_id=tid, name=name).first():
|
|
flash("Client name must be unique within tenant", "error")
|
|
return redirect(url_for("clients_new"))
|
|
c = Client(
|
|
tenant_id=tid, name=name,
|
|
label_token=random_token(),
|
|
server_addr=request.form.get("server_addr", "").strip() or None,
|
|
description=request.form.get("description", ""),
|
|
is_active=bool(request.form.get("is_active", "on")),
|
|
)
|
|
db.session.add(c)
|
|
db.session.commit()
|
|
log_action("create", target_type="client", target_id=c.id,
|
|
target_label=c.name, detail=f"tenant_id={tid}")
|
|
flash(f"Client '{c.name}' created. Add proxies or download frpc.ini.",
|
|
"success")
|
|
return redirect(url_for("clients_list"))
|
|
return render_template("client_form.html", user=g.current_user,
|
|
item=None, tenants=tenants)
|
|
|
|
|
|
@app.route("/clients/<int:cid>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
def clients_edit(cid):
|
|
c = Client.query.get_or_404(cid)
|
|
tenants = visible_tenants(g.current_user)
|
|
if not any(t.id == c.tenant_id for t in tenants):
|
|
abort(403)
|
|
if request.method == "POST":
|
|
if g.current_user.role == "admin":
|
|
new_tid = int(request.form.get("tenant_id", c.tenant_id))
|
|
if any(t.id == new_tid for t in tenants):
|
|
c.tenant_id = new_tid
|
|
c.server_addr = request.form.get("server_addr", "").strip() or None
|
|
c.description = request.form.get("description", "")
|
|
c.is_active = bool(request.form.get("is_active"))
|
|
db.session.commit()
|
|
log_action("update", target_type="client", target_id=c.id,
|
|
target_label=c.name)
|
|
flash("Client updated", "success")
|
|
return redirect(url_for("clients_list"))
|
|
return render_template("client_form.html", user=g.current_user,
|
|
item=c, tenants=tenants)
|
|
|
|
|
|
@app.route("/clients/<int:cid>/delete", methods=["POST"])
|
|
@login_required
|
|
def clients_delete(cid):
|
|
c = Client.query.get_or_404(cid)
|
|
tenants = visible_tenants(g.current_user)
|
|
if not any(t.id == c.tenant_id for t in tenants):
|
|
abort(403)
|
|
name = c.name
|
|
db.session.delete(c)
|
|
db.session.commit()
|
|
log_action("delete", target_type="client", target_id=cid,
|
|
target_label=name)
|
|
flash(f"Client '{name}' and its proxies deleted.", "success")
|
|
return redirect(url_for("clients_list"))
|
|
|
|
|
|
@app.route("/clients/<int:cid>/frpc-config")
|
|
@login_required
|
|
def clients_frpc_config(cid):
|
|
c = Client.query.get_or_404(cid)
|
|
tenants = visible_tenants(g.current_user)
|
|
if not any(t.id == c.tenant_id for t in tenants):
|
|
abort(403)
|
|
server_addr = request.args.get(
|
|
"server_addr", "").strip() or c.server_addr or request.host.split(":")[0]
|
|
bind_port = request.args.get("bind_port", type=int) or BIND_PORT
|
|
body = render_frpc_ini(c, server_addr, bind_port)
|
|
return Response(body, mimetype="text/plain; charset=utf-8",
|
|
headers={"Content-Disposition":
|
|
f"attachment; filename=frpc-{c.name}.ini"})
|
|
|
|
|
|
# ---- Proxies ---------------------------------------------------------------
|
|
@app.route("/proxies")
|
|
@login_required
|
|
def proxies_list():
|
|
tid = request.args.get("tenant_id", type=int)
|
|
cid = request.args.get("client_id", type=int)
|
|
tenants = visible_tenants(g.current_user)
|
|
clients = visible_clients(g.current_user, tenant_id=tid)
|
|
items = visible_proxies(g.current_user, client_id=cid)
|
|
info = get_dashboard_info()
|
|
online_names = set()
|
|
if info and not info.get("error"):
|
|
for p in info.get("proxies", {}).get("items", []):
|
|
online_names.add(p.get("name", ""))
|
|
return render_template(
|
|
"proxies.html", user=g.current_user, items=items,
|
|
clients=clients, tenants=tenants,
|
|
filter_tenant_id=tid, filter_client_id=cid,
|
|
online_names=online_names,
|
|
)
|
|
|
|
|
|
@app.route("/proxies/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def proxies_new():
|
|
clients = visible_clients(g.current_user)
|
|
if not clients:
|
|
flash("Create a client first before adding proxies", "error")
|
|
return redirect(url_for("proxies_list"))
|
|
if request.method == "POST":
|
|
try:
|
|
cid = int(request.form.get("client_id", 0))
|
|
except ValueError:
|
|
cid = 0
|
|
if not any(c.id == cid for c in clients):
|
|
abort(403)
|
|
name = request.form.get("name", "").strip()
|
|
if not name or Proxy.query.filter_by(
|
|
client_id=cid, name=name).first():
|
|
flash("Proxy name must be unique within client", "error")
|
|
return redirect(url_for("proxies_new"))
|
|
ptype = request.form.get("type", "tcp")
|
|
if ptype not in PROXY_TYPES:
|
|
flash("Invalid proxy type", "error")
|
|
return redirect(url_for("proxies_new"))
|
|
rp = request.form.get("remote_port", "")
|
|
p = Proxy(
|
|
client_id=cid, name=name, type=ptype,
|
|
local_ip=request.form.get("local_ip", "127.0.0.1"),
|
|
local_port=int(request.form.get("local_port", 0) or 0),
|
|
remote_port=int(rp) if rp else None,
|
|
custom_domains=request.form.get("custom_domains", ""),
|
|
subdomain=request.form.get("subdomain", ""),
|
|
locations=request.form.get("locations", ""),
|
|
use_encryption=bool(request.form.get("use_encryption")),
|
|
use_compression=bool(request.form.get("use_compression")),
|
|
health_check_type=request.form.get("health_check_type", ""),
|
|
health_check_url=request.form.get("health_check_url", ""),
|
|
health_check_interval_s=int(
|
|
request.form.get("health_check_interval_s", 30) or 30),
|
|
health_check_timeout_s=int(
|
|
request.form.get("health_check_timeout_s", 3) or 3),
|
|
note=request.form.get("note", ""),
|
|
is_enabled=bool(request.form.get("is_enabled")),
|
|
)
|
|
db.session.add(p)
|
|
db.session.commit()
|
|
bump_client_config_revision(cid)
|
|
log_action("create", target_type="proxy", target_id=p.id,
|
|
target_label=p.name, detail=f"client_id={cid} type={ptype}")
|
|
flash(f"Proxy '{p.name}' created. Tell the frpc owner to pull "
|
|
"new frpc.ini and run `frpc reload`.", "success")
|
|
return redirect(url_for("proxies_list"))
|
|
prefill_client = request.args.get("client_id", type=int)
|
|
return render_template(
|
|
"proxy_form.html", user=g.current_user, item=None,
|
|
clients=clients, types=PROXY_TYPES,
|
|
prefill_client=prefill_client,
|
|
)
|
|
|
|
|
|
@app.route("/proxies/<int:pid>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
def proxies_edit(pid):
|
|
p = Proxy.query.get_or_404(pid)
|
|
clients = visible_clients(g.current_user)
|
|
if not any(c.id == p.client_id for c in clients):
|
|
abort(403)
|
|
if request.method == "POST":
|
|
p.local_ip = request.form.get("local_ip", "127.0.0.1")
|
|
try:
|
|
p.local_port = int(request.form.get("local_port", 0) or 0)
|
|
except ValueError:
|
|
p.local_port = 0
|
|
rp = request.form.get("remote_port", "")
|
|
p.remote_port = int(rp) if rp else None
|
|
p.custom_domains = request.form.get("custom_domains", "")
|
|
p.subdomain = request.form.get("subdomain", "")
|
|
p.locations = request.form.get("locations", "")
|
|
p.use_encryption = bool(request.form.get("use_encryption"))
|
|
p.use_compression = bool(request.form.get("use_compression"))
|
|
p.health_check_type = request.form.get("health_check_type", "")
|
|
p.health_check_url = request.form.get("health_check_url", "")
|
|
try:
|
|
p.health_check_interval_s = int(
|
|
request.form.get("health_check_interval_s", 30) or 30)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
p.health_check_timeout_s = int(
|
|
request.form.get("health_check_timeout_s", 3) or 3)
|
|
except ValueError:
|
|
pass
|
|
p.note = request.form.get("note", "")
|
|
p.is_enabled = bool(request.form.get("is_enabled"))
|
|
p.updated_at = datetime.utcnow()
|
|
db.session.commit()
|
|
bump_client_config_revision(p.client_id)
|
|
log_action("update", target_type="proxy", target_id=p.id,
|
|
target_label=p.name)
|
|
flash(f"Proxy '{p.name}' updated. Tell the frpc owner to pull "
|
|
"new frpc.ini and run `frpc reload`.", "success")
|
|
return redirect(url_for("proxies_list"))
|
|
return render_template(
|
|
"proxy_form.html", user=g.current_user, item=p,
|
|
clients=clients, types=PROXY_TYPES, prefill_client=p.client_id,
|
|
)
|
|
|
|
|
|
@app.route("/proxies/<int:pid>/toggle", methods=["POST"])
|
|
@login_required
|
|
def proxies_toggle(pid):
|
|
p = Proxy.query.get_or_404(pid)
|
|
clients = visible_clients(g.current_user)
|
|
if not any(c.id == p.client_id for c in clients):
|
|
abort(403)
|
|
p.is_enabled = not p.is_enabled
|
|
db.session.commit()
|
|
bump_client_config_revision(p.client_id)
|
|
log_action("toggle", target_type="proxy", target_id=p.id,
|
|
target_label=p.name, detail=f"enabled={p.is_enabled}")
|
|
flash(f"Proxy '{p.name}' {'enabled' if p.is_enabled else 'disabled'}.",
|
|
"success")
|
|
return redirect(request.referrer or url_for("proxies_list"))
|
|
|
|
|
|
@app.route("/proxies/<int:pid>/delete", methods=["POST"])
|
|
@login_required
|
|
def proxies_delete(pid):
|
|
p = Proxy.query.get_or_404(pid)
|
|
clients = visible_clients(g.current_user)
|
|
if not any(c.id == p.client_id for c in clients):
|
|
abort(403)
|
|
name = p.name
|
|
cid = p.client_id
|
|
db.session.delete(p)
|
|
db.session.commit()
|
|
bump_client_config_revision(cid)
|
|
log_action("delete", target_type="proxy", target_id=pid,
|
|
target_label=name)
|
|
flash(f"Proxy '{name}' deleted.", "success")
|
|
return redirect(url_for("proxies_list"))
|
|
|
|
|
|
# ---- Audit log -------------------------------------------------------------
|
|
@app.route("/logs")
|
|
@role_required("admin")
|
|
def logs_list():
|
|
page = max(1, request.args.get("page", 1, type=int))
|
|
per = 50
|
|
q = AuditLog.query.order_by(AuditLog.id.desc())
|
|
total = q.count()
|
|
items = q.offset((page - 1) * per).limit(per).all()
|
|
return render_template(
|
|
"logs.html", user=g.current_user, items=items,
|
|
page=page, per=per, total=total,
|
|
)
|
|
|
|
|
|
# ---- frps management -------------------------------------------------------
|
|
@app.route("/frps/status")
|
|
@login_required
|
|
def frps_status():
|
|
return jsonify(get_dashboard_info())
|
|
|
|
|
|
@app.route("/frps/config")
|
|
@role_required("admin")
|
|
def frps_config_view():
|
|
if not os.path.exists(FRPS_CONFIG):
|
|
return Response("# frps.ini not generated yet\n", mimetype="text/plain")
|
|
with open(FRPS_CONFIG, "r", encoding="utf-8") as f:
|
|
return Response(f.read(), mimetype="text/plain; charset=utf-8")
|
|
|
|
|
|
@app.route("/frps/reload", methods=["POST"])
|
|
@role_required("admin")
|
|
def frps_reload():
|
|
write_frps_config()
|
|
ok, msg = reload_frps("restart")
|
|
log_action("frps_reload", target_type="frps", detail=msg,
|
|
target_label="ok" if ok else "failed")
|
|
flash(("frps reloaded" if ok else f"frps reload failed: {msg}"),
|
|
"success" if ok else "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
@app.route("/frps/rotate-token", methods=["POST"])
|
|
@role_required("admin")
|
|
def frps_rotate_token():
|
|
rotate_server_token()
|
|
write_frps_config()
|
|
ok, msg = reload_frps("restart")
|
|
log_action("frps_rotate_token", target_type="frps", detail=msg)
|
|
flash(("Server token rotated. frps reloaded. "
|
|
"All existing frpc connections will drop."
|
|
if ok else f"rotate failed: {msg}"),
|
|
"success" if ok else "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
# ---- User management -------------------------------------------------------
|
|
@app.route("/users")
|
|
@role_required("admin")
|
|
def users_list():
|
|
items = User.query.order_by(User.username).all()
|
|
tenants = Tenant.query.order_by(Tenant.name).all()
|
|
return render_template("users.html", user=g.current_user,
|
|
items=items, tenants=tenants)
|
|
|
|
|
|
@app.route("/users/new", methods=["GET", "POST"])
|
|
@role_required("admin")
|
|
def users_new():
|
|
tenants = Tenant.query.order_by(Tenant.name).all()
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
pw = request.form.get("password", "")
|
|
if not username or not pw:
|
|
flash("Username and password required", "error")
|
|
elif User.query.filter_by(username=username).first():
|
|
flash("Username already exists", "error")
|
|
else:
|
|
u = User(
|
|
username=username,
|
|
role=request.form.get("role", "user"),
|
|
display_name=request.form.get("display_name", ""),
|
|
tenant_id=int(request.form.get("tenant_id", 0) or 0) or None,
|
|
is_active=bool(request.form.get("is_active")),
|
|
)
|
|
u.set_password(pw)
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
log_action("create", target_type="user", target_id=u.id,
|
|
target_label=u.username, detail=f"role={u.role}")
|
|
flash(f"User '{u.username}' created", "success")
|
|
return redirect(url_for("users_list"))
|
|
return render_template("user_form.html", user=g.current_user, item=None,
|
|
tenants=tenants)
|
|
|
|
|
|
@app.route("/users/<int:uid>/edit", methods=["GET", "POST"])
|
|
@role_required("admin")
|
|
def users_edit(uid):
|
|
u = User.query.get_or_404(uid)
|
|
tenants = Tenant.query.order_by(Tenant.name).all()
|
|
if request.method == "POST":
|
|
u.role = request.form.get("role", u.role)
|
|
u.display_name = request.form.get("display_name", "")
|
|
u.tenant_id = int(request.form.get("tenant_id", 0) or 0) or None
|
|
u.is_active = bool(request.form.get("is_active"))
|
|
new_pw = request.form.get("password", "")
|
|
if new_pw:
|
|
if len(new_pw) < 6:
|
|
flash("Password must be at least 6 characters", "error")
|
|
return redirect(url_for("users_edit", uid=uid))
|
|
u.set_password(new_pw)
|
|
db.session.commit()
|
|
log_action("update", target_type="user", target_id=u.id,
|
|
target_label=u.username)
|
|
flash("User updated", "success")
|
|
return redirect(url_for("users_list"))
|
|
return render_template("user_form.html", user=g.current_user, item=u,
|
|
tenants=tenants)
|
|
|
|
|
|
@app.route("/users/<int:uid>/delete", methods=["POST"])
|
|
@role_required("admin")
|
|
def users_delete(uid):
|
|
u = User.query.get_or_404(uid)
|
|
if u.id == g.current_user.id:
|
|
flash("Cannot delete yourself", "error")
|
|
return redirect(url_for("users_list"))
|
|
if u.username == "admin":
|
|
flash("Cannot delete the bootstrap admin user", "error")
|
|
return redirect(url_for("users_list"))
|
|
name = u.username
|
|
db.session.delete(u)
|
|
db.session.commit()
|
|
log_action("delete", target_type="user", target_id=uid,
|
|
target_label=name)
|
|
flash(f"User '{name}' deleted", "success")
|
|
return redirect(url_for("users_list"))
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Jinja filters
|
|
# ----------------------------------------------------------------------------
|
|
@app.template_filter("fmt_dt")
|
|
def fmt_dt(dt):
|
|
if not dt:
|
|
return "-"
|
|
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
@app.template_filter("fmt_bytes")
|
|
def fmt_bytes(n):
|
|
if n is None:
|
|
return "-"
|
|
try:
|
|
n = int(n)
|
|
except Exception:
|
|
return str(n)
|
|
for unit in ("B", "KB", "MB", "GB", "TB"):
|
|
if n < 1024:
|
|
return f"{n:.1f} {unit}" if unit != "B" else f"{n} {unit}"
|
|
n /= 1024
|
|
return f"{n:.1f} PB"
|
|
|
|
|
|
@app.template_filter("fmt_mbps")
|
|
def fmt_mbps(n):
|
|
try:
|
|
n = int(n)
|
|
except Exception:
|
|
return "-"
|
|
return f"{n} Mbps" if n else "unlimited"
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Error handlers
|
|
# ----------------------------------------------------------------------------
|
|
@app.errorhandler(403)
|
|
def err_403(e):
|
|
return render_template("error.html", user=current_user(),
|
|
code=403, message="Forbidden"), 403
|
|
|
|
|
|
@app.errorhandler(404)
|
|
def err_404(e):
|
|
return render_template("error.html", user=current_user(),
|
|
code=404, message="Not found"), 404
|
|
|
|
|
|
# ----------------------------------------------------------------------------
|
|
# Bootstrap DB on import
|
|
# ----------------------------------------------------------------------------
|
|
def init_db():
|
|
with app.app_context():
|
|
db.create_all()
|
|
seed_admin()
|
|
|
|
|
|
# Top-level import requires io
|
|
init_db()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0",
|
|
port=int(os.environ.get("PORT", "5390")),
|
|
debug=False) |