239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
"""Server-rendered pages and the Settings form actions."""
|
|
|
|
from flask import Blueprint, current_app, flash, g, redirect, render_template, request, url_for
|
|
|
|
from . import auth, daemon, db, mailer, security, validate
|
|
from .auth import login_required, sysop_required
|
|
|
|
bp = Blueprint("views", __name__)
|
|
|
|
|
|
def conf():
|
|
return current_app.config["PB_CONF"]
|
|
|
|
|
|
@bp.get("/")
|
|
@login_required
|
|
def index():
|
|
return redirect(url_for("views.patch"))
|
|
|
|
|
|
@bp.get("/patch")
|
|
@login_required
|
|
def patch():
|
|
return render_template("patch.html")
|
|
|
|
|
|
@bp.get("/help")
|
|
@login_required
|
|
def help_page():
|
|
return render_template("help.html")
|
|
|
|
|
|
@bp.get("/services")
|
|
@login_required
|
|
def services():
|
|
conn = db.get()
|
|
hosts = [{"id": 0, "name": "Target (this server)", "online": True}]
|
|
status = daemon.request("STATUS")
|
|
online = status.get("clients", {})
|
|
for r in conn.execute("SELECT id, name, hostname FROM clients ORDER BY name"):
|
|
label = r["name"] + (f" ({r['hostname']})" if r["hostname"] else "")
|
|
hosts.append({"id": r["id"], "name": label, "online": str(r["id"]) in online})
|
|
rows = conn.execute("SELECT client_id, proto, addr, port, pid, process, updated FROM services "
|
|
"ORDER BY client_id, proto, port").fetchall()
|
|
ifaces = conn.execute("SELECT client_id, name, addr FROM interfaces ORDER BY name").fetchall()
|
|
for h in hosts:
|
|
h["services"] = [r for r in rows if r["client_id"] == h["id"]]
|
|
h["ifaces"] = [r for r in ifaces if r["client_id"] == h["id"]]
|
|
h["updated"] = max((r["updated"] for r in h["services"]), default=0)
|
|
return render_template("services.html", hosts=hosts, daemon_ok=status.get("ok", False))
|
|
|
|
|
|
@bp.get("/stats")
|
|
@login_required
|
|
def stats():
|
|
conn = db.get()
|
|
sinks = conn.execute(
|
|
"SELECT s.id, s.type, s.host, s.port, s.proto, s.label, s.iface, c.name AS client "
|
|
"FROM nodes s LEFT JOIN clients c ON c.id = s.client_id "
|
|
"WHERE s.type IN ('public_sink', 'client_sink', 'tunnel_sink') ORDER BY s.id").fetchall()
|
|
return render_template("stats.html", sinks=sinks)
|
|
|
|
|
|
@bp.get("/settings")
|
|
@login_required
|
|
def settings():
|
|
conn = db.get()
|
|
users = conn.execute("SELECT id, username, email, created FROM users ORDER BY username").fetchall()
|
|
clients = conn.execute("SELECT id, name, pubkey, added_by, created, hostname, last_seen, last_addr "
|
|
"FROM clients ORDER BY name").fetchall()
|
|
retention = {k: db.setting(k, "") for k in ("stats_minute_hours", "stats_hour_days", "stats_day_days")}
|
|
return render_template("settings.html", users=users, clients=clients, retention=retention, conf=conf())
|
|
|
|
|
|
def _done(msg):
|
|
flash(msg, "ok")
|
|
return redirect(url_for("views.settings") + request.form.get("anchor", ""))
|
|
|
|
|
|
def _fail(msg):
|
|
flash(msg, "error")
|
|
return redirect(url_for("views.settings") + request.form.get("anchor", ""))
|
|
|
|
|
|
@bp.post("/settings/account")
|
|
@login_required
|
|
def settings_account():
|
|
f = request.form
|
|
if not security.verify_password(f.get("current", ""), g.user["pwhash"]):
|
|
return _fail("Current password is wrong.")
|
|
try:
|
|
email = validate.email(f.get("email", "").strip())
|
|
new = f.get("new", "")
|
|
if new:
|
|
validate.password(new)
|
|
if new != f.get("confirm", ""):
|
|
raise validate.Invalid("new passwords do not match")
|
|
except validate.Invalid as e:
|
|
return _fail(str(e).capitalize() + ".")
|
|
|
|
if g.user["sysop"]:
|
|
changes = {"SysopEmail": email}
|
|
if new:
|
|
changes["SysopPassword"] = security.hash_password(new)
|
|
conf().update(changes)
|
|
else:
|
|
conn = db.get()
|
|
conn.execute("UPDATE users SET email = ? WHERE username = ?", (email, g.user["username"]))
|
|
if new:
|
|
conn.execute("UPDATE users SET pwhash = ? WHERE username = ?",
|
|
(security.hash_password(new), g.user["username"]))
|
|
auth.refresh_pwtag()
|
|
return _done("Account updated.")
|
|
|
|
|
|
@bp.post("/settings/users/add")
|
|
@sysop_required
|
|
def users_add():
|
|
f = request.form
|
|
try:
|
|
username = validate.name(f.get("username", "").strip(), "username")
|
|
email = validate.email(f.get("email", "").strip())
|
|
pw = validate.password(f.get("password", ""))
|
|
except validate.Invalid as e:
|
|
return _fail(str(e).capitalize() + ".")
|
|
if username == conf().get("sysopuser"):
|
|
return _fail("That name belongs to the sysop.")
|
|
conn = db.get()
|
|
if conn.execute("SELECT 1 FROM users WHERE username = ?", (username,)).fetchone():
|
|
return _fail("User already exists.")
|
|
conn.execute("INSERT INTO users (username, email, pwhash, created) VALUES (?, ?, ?, ?)",
|
|
(username, email, security.hash_password(pw), db.now()))
|
|
return _done(f"User {username} created.")
|
|
|
|
|
|
@bp.post("/settings/users/<int:uid>/edit")
|
|
@sysop_required
|
|
def users_edit(uid):
|
|
f = request.form
|
|
conn = db.get()
|
|
try:
|
|
email = validate.email(f.get("email", "").strip())
|
|
conn.execute("UPDATE users SET email = ? WHERE id = ?", (email, uid))
|
|
if f.get("password"):
|
|
conn.execute("UPDATE users SET pwhash = ? WHERE id = ?",
|
|
(security.hash_password(validate.password(f["password"])), uid))
|
|
except validate.Invalid as e:
|
|
return _fail(str(e).capitalize() + ".")
|
|
return _done("User updated.")
|
|
|
|
|
|
@bp.post("/settings/users/<int:uid>/delete")
|
|
@sysop_required
|
|
def users_delete(uid):
|
|
db.get().execute("DELETE FROM users WHERE id = ?", (uid,))
|
|
return _done("User deleted.")
|
|
|
|
|
|
@bp.post("/settings/clients/add")
|
|
@login_required
|
|
def clients_add():
|
|
f = request.form
|
|
try:
|
|
name = validate.name(f.get("name", "").strip(), "client name")
|
|
key = validate.pubkey(f.get("pubkey", ""))
|
|
except validate.Invalid as e:
|
|
return _fail(str(e).capitalize() + ".")
|
|
conn = db.get()
|
|
if conn.execute("SELECT 1 FROM clients WHERE name = ? OR pubkey = ?", (name, key)).fetchone():
|
|
return _fail("A client with that name or key already exists.")
|
|
conn.execute("INSERT INTO clients (name, pubkey, added_by, created) VALUES (?, ?, ?, ?)",
|
|
(name, key, g.user["username"], db.now()))
|
|
res = daemon.reload()
|
|
if not res.get("ok"):
|
|
flash(f"Client saved, but the daemon did not reload: {res.get('error')}", "error")
|
|
return _done(f"Client {name} added.")
|
|
|
|
|
|
@bp.post("/settings/clients/<int:cid>/delete")
|
|
@login_required
|
|
def clients_delete(cid):
|
|
conn = db.get()
|
|
conn.execute("DELETE FROM services WHERE client_id = ?", (cid,))
|
|
conn.execute("DELETE FROM clients WHERE id = ?", (cid,))
|
|
daemon.reload()
|
|
return _done("Client removed.")
|
|
|
|
|
|
@bp.post("/settings/mail")
|
|
@sysop_required
|
|
def settings_mail():
|
|
f = request.form
|
|
security_mode = f.get("security", "starttls")
|
|
if security_mode not in ("ssl", "starttls", "none"):
|
|
return _fail("Invalid security mode.")
|
|
try:
|
|
port = int(f.get("port", "587"))
|
|
except ValueError:
|
|
return _fail("Port must be a number.")
|
|
changes = {
|
|
"MailHost": f.get("host", "").strip(),
|
|
"MailPort": port,
|
|
"MailSecurity": security_mode,
|
|
"MailUser": f.get("user", "").strip(),
|
|
"MailFrom": f.get("from", "").strip(),
|
|
}
|
|
if f.get("password"):
|
|
changes["MailPassword"] = f["password"]
|
|
try:
|
|
conf().update(changes)
|
|
except ValueError as e:
|
|
return _fail(str(e))
|
|
return _done("Mail settings saved.")
|
|
|
|
|
|
@bp.post("/settings/mail/test")
|
|
@sysop_required
|
|
def settings_mail_test():
|
|
try:
|
|
mailer.send(conf(), g.user["email"], "PatchBay test mail",
|
|
"This is a test message from your PatchBay server.\n")
|
|
except mailer.MailError as e:
|
|
return _fail(f"Sending failed: {e}")
|
|
return _done(f"Test mail sent to {g.user['email']}.")
|
|
|
|
|
|
@bp.post("/settings/retention")
|
|
@sysop_required
|
|
def settings_retention():
|
|
try:
|
|
vals = {k: int(request.form.get(k, "0")) for k in ("stats_minute_hours", "stats_hour_days", "stats_day_days")}
|
|
except ValueError:
|
|
return _fail("Retention values must be whole numbers.")
|
|
if any(v < 0 for v in vals.values()) or vals["stats_minute_hours"] == 0 or vals["stats_hour_days"] == 0:
|
|
return _fail("Minute and hour retention must be positive; day retention 0 means forever.")
|
|
for k, v in vals.items():
|
|
db.set_setting(k, v)
|
|
return _done("Retention saved.")
|