Dodaj admin web panel u repo (admin/), sinhronizuj systemd unite
- admin/: LAN web panel (Flask + waitress + python-pam), status servisa, PAM login sa idriz korisnikom, vertikalni meni. Port 8080, samo LAN. - systemd/idriz-admin.service: korisnički servis, enabled + Restart=always + StartLimitIntervalSec=0. - systemd/idriz.service: dodan StartLimitIntervalSec=0 (repo je bio zastario). Od sada sav aplikacioni source code ide u ovaj repo. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LzKpYaEsuWDevuNnvSjcFW
This commit is contained in:
idriz
201
admin/app.py
Normal file
201
admin/app.py
Normal file
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Idriz admin — jednostavan LAN web panel pod PAM autentikacijom.
|
||||
|
||||
Namjerno minimalno: Flask + waitress + python-pam, jedan fajl, bez baze.
|
||||
Prijava ide preko PAM-a (Linux korisnik `idriz`).
|
||||
"""
|
||||
import os
|
||||
import secrets
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
from functools import wraps
|
||||
|
||||
import pam as pam_module
|
||||
from flask import (
|
||||
Flask, render_template, request, redirect, url_for, session, flash
|
||||
)
|
||||
|
||||
# --- Konfiguracija -----------------------------------------------------------
|
||||
|
||||
# Samo ovaj Linux korisnik smije da se prijavi.
|
||||
ALLOWED_USER = os.environ.get("IDRIZ_ADMIN_USER", "idriz")
|
||||
PAM_SERVICE = os.environ.get("IDRIZ_ADMIN_PAM_SERVICE", "idriz-admin")
|
||||
LISTEN_HOST = os.environ.get("IDRIZ_ADMIN_HOST", "0.0.0.0")
|
||||
LISTEN_PORT = int(os.environ.get("IDRIZ_ADMIN_PORT", "8080"))
|
||||
|
||||
# Servisi koje panel prati. scope: "system" ili "user".
|
||||
SERVICES = [
|
||||
{"unit": "idriz.service", "scope": "user", "label": "Idriz asistent", "desc": "Glasovni pomoćnik za djecu"},
|
||||
{"unit": "cups.service", "scope": "system", "label": "CUPS", "desc": "Štampanje / print server"},
|
||||
{"unit": "cups-browsed.service", "scope": "system", "label": "CUPS Browsed", "desc": "Dijeljenje štampača"},
|
||||
{"unit": "avahi-daemon.service", "scope": "system", "label": "Avahi (mDNS)", "desc": "Otkrivanje štampača na mreži"},
|
||||
{"unit": "tigervncserver@:2.service","scope": "system","label": "VNC", "desc": "Daljinski desktop :2 / 5902"},
|
||||
{"unit": "gdm3.service", "scope": "system", "label": "GDM", "desc": "Grafički login + autologin"},
|
||||
{"unit": "pipewire.service", "scope": "user", "label": "PipeWire", "desc": "Audio server"},
|
||||
{"unit": "wireplumber.service", "scope": "user", "label": "WirePlumber", "desc": "Audio session manager"},
|
||||
]
|
||||
|
||||
# --- Aplikacija --------------------------------------------------------------
|
||||
|
||||
app = Flask(__name__)
|
||||
# Tajni ključ za sesije; nov pri svakom startu (odjavi sve pri restartu — ok za nas).
|
||||
app.secret_key = secrets.token_hex(32)
|
||||
app.config.update(
|
||||
SESSION_COOKIE_HTTPONLY=True,
|
||||
SESSION_COOKIE_SAMESITE="Lax",
|
||||
PERMANENT_SESSION_LIFETIME=8 * 3600,
|
||||
)
|
||||
|
||||
|
||||
def login_required(f):
|
||||
@wraps(f)
|
||||
def wrapper(*args, **kwargs):
|
||||
if not session.get("user"):
|
||||
return redirect(url_for("login", next=request.path))
|
||||
return f(*args, **kwargs)
|
||||
return wrapper
|
||||
|
||||
|
||||
def _systemctl(scope, *args):
|
||||
cmd = ["systemctl"]
|
||||
if scope == "user":
|
||||
cmd.append("--user")
|
||||
cmd.extend(args)
|
||||
try:
|
||||
out = subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=8
|
||||
)
|
||||
return out.stdout.strip()
|
||||
except Exception as e: # noqa: BLE001
|
||||
return f"error: {e}"
|
||||
|
||||
|
||||
def get_service_status(svc):
|
||||
"""Vrati dict sa stanjem servisa (bez roota — samo čita)."""
|
||||
scope, unit = svc["scope"], svc["unit"]
|
||||
props = _systemctl(
|
||||
scope, "show", unit,
|
||||
"-p", "ActiveState", "-p", "SubState", "-p", "UnitFileState",
|
||||
"-p", "ActiveEnterTimestampMonotonic", "-p", "ActiveEnterTimestamp",
|
||||
"-p", "MainPID", "-p", "MemoryCurrent",
|
||||
)
|
||||
data = {}
|
||||
for line in props.splitlines():
|
||||
if "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
data[k] = v
|
||||
|
||||
active = data.get("ActiveState", "unknown")
|
||||
sub = data.get("SubState", "")
|
||||
enabled = data.get("UnitFileState", "unknown")
|
||||
|
||||
# uptime iz ActiveEnterTimestamp (realtime, npr. "Pet 2026-09-04 12:00:00 CEST")
|
||||
rt = _systemctl(scope, "show", unit, "--value", "-p", "ActiveEnterTimestamp")
|
||||
uptime = "—"
|
||||
if rt and rt not in ("", "n/a"):
|
||||
try:
|
||||
started = datetime.strptime(rt.rsplit(" ", 1)[0], "%a %Y-%m-%d %H:%M:%S")
|
||||
delta = datetime.now() - started
|
||||
total = int(delta.total_seconds())
|
||||
if total >= 0:
|
||||
d, rem = divmod(total, 86400)
|
||||
h, rem = divmod(rem, 3600)
|
||||
m, _ = divmod(rem, 60)
|
||||
parts = []
|
||||
if d:
|
||||
parts.append(f"{d}d")
|
||||
if h:
|
||||
parts.append(f"{h}h")
|
||||
if m or not parts:
|
||||
parts.append(f"{m}min")
|
||||
uptime = " ".join(parts)
|
||||
except (ValueError, IndexError):
|
||||
uptime = "—"
|
||||
|
||||
mem = data.get("MemoryCurrent", "")
|
||||
mem_h = "—"
|
||||
if mem.isdigit():
|
||||
n = int(mem)
|
||||
if n > 0:
|
||||
for unit_s in ("B", "KB", "MB", "GB"):
|
||||
if n < 1024:
|
||||
mem_h = f"{n:.0f} {unit_s}"
|
||||
break
|
||||
n /= 1024
|
||||
else:
|
||||
mem_h = f"{n:.1f} TB"
|
||||
|
||||
if active == "active":
|
||||
state = "ok"
|
||||
elif active in ("activating", "reloading", "deactivating"):
|
||||
state = "warn"
|
||||
else:
|
||||
state = "bad"
|
||||
|
||||
return {
|
||||
**svc,
|
||||
"active": active,
|
||||
"sub": sub,
|
||||
"enabled": enabled,
|
||||
"uptime": uptime,
|
||||
"pid": data.get("MainPID", "0"),
|
||||
"mem": mem_h,
|
||||
"state": state,
|
||||
}
|
||||
|
||||
|
||||
# --- Rute --------------------------------------------------------------------
|
||||
|
||||
@app.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if session.get("user"):
|
||||
return redirect(url_for("status"))
|
||||
if request.method == "POST":
|
||||
username = (request.form.get("username") or "").strip()
|
||||
password = request.form.get("password") or ""
|
||||
if username != ALLOWED_USER:
|
||||
flash("Pogrešno korisničko ime ili lozinka.")
|
||||
else:
|
||||
p = pam_module.pam()
|
||||
if p.authenticate(username, password, service=PAM_SERVICE):
|
||||
session["user"] = username
|
||||
session.permanent = True
|
||||
nxt = request.args.get("next") or url_for("status")
|
||||
if not nxt.startswith("/"):
|
||||
nxt = url_for("status")
|
||||
return redirect(nxt)
|
||||
flash("Pogrešno korisničko ime ili lozinka.")
|
||||
return render_template("login.html")
|
||||
|
||||
|
||||
@app.route("/logout")
|
||||
def logout():
|
||||
session.clear()
|
||||
return redirect(url_for("login"))
|
||||
|
||||
|
||||
@app.route("/")
|
||||
@login_required
|
||||
def index():
|
||||
return redirect(url_for("status"))
|
||||
|
||||
|
||||
@app.route("/status")
|
||||
@login_required
|
||||
def status():
|
||||
services = [get_service_status(s) for s in SERVICES]
|
||||
ok = sum(1 for s in services if s["state"] == "ok")
|
||||
return render_template(
|
||||
"status.html",
|
||||
services=services,
|
||||
ok=ok,
|
||||
total=len(services),
|
||||
now=datetime.now().strftime("%d.%m.%Y %H:%M:%S"),
|
||||
active_page="status",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
from waitress import serve
|
||||
print(f"Idriz admin na http://{LISTEN_HOST}:{LISTEN_PORT}")
|
||||
serve(app, host=LISTEN_HOST, port=LISTEN_PORT, threads=8)
|
||||
Reference in New Issue
Block a user