agwandClaude Opus 4.8
7cf79fa901
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
664 lines
26 KiB
Python
664 lines
26 KiB
Python
"""Weircon Portal — forside (SSO-dashboard) + udvidbar admin.
|
|
|
|
- Login via Authentik (OIDC). Dashboard viser de apps brugeren har adgang til.
|
|
- Admin (kun for medlemmer af ADMIN_GROUP): IP→label-mapping, cookie-tekster, statistik.
|
|
- Admin-moduler er registreret i ADMIN_MODULES — tilføj nye ved at registrere en post.
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from authlib.integrations.starlette_client import OAuth
|
|
from fastapi import FastAPI, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
BASE = Path(__file__).resolve().parent
|
|
DATA = Path(os.environ.get("PORTAL_DATA", "/opt/portal/data"))
|
|
DATA.mkdir(parents=True, exist_ok=True)
|
|
|
|
AUTHENTIK = os.environ["AUTHENTIK_BASE"].rstrip("/")
|
|
CLIENT_ID = os.environ["OIDC_CLIENT_ID"]
|
|
CLIENT_SECRET = os.environ["OIDC_CLIENT_SECRET"]
|
|
API_TOKEN = os.environ["AUTHENTIK_API_TOKEN"]
|
|
SECRET_KEY = os.environ["SESSION_SECRET"]
|
|
REDIRECT_URI = os.environ.get("OIDC_REDIRECT_URI", "https://weircon.dk/auth/callback")
|
|
ADMIN_GROUP = os.environ.get("ADMIN_GROUP", "authentik Admins")
|
|
MATOMO_BASE = os.environ.get("MATOMO_BASE", "https://matomo.weircon.dk").rstrip("/")
|
|
MATOMO_TOKEN = os.environ.get("MATOMO_TOKEN", "")
|
|
MATOMO_SITE = os.environ.get("MATOMO_SITE", "2")
|
|
# Sync af IP-labels til NPMplus (CT101) efter hver gemning
|
|
IP_LABELS_LOCAL = DATA / "ip_labels.json"
|
|
IP_LABELS_REMOTE = os.environ.get("IP_LABELS_REMOTE", "") # fx root@192.168.10.201:/opt/npmplus/custom_nginx/ip_labels.json
|
|
SETTINGS_PATH = DATA / "settings.json"
|
|
RP_BASE_URL = os.environ.get("RP_BASE_URL", "https://rp.weircon.dk").rstrip("/")
|
|
RP_API_KEY = os.environ.get("RP_API_KEY", "")
|
|
ASSETS_VERSION_URL = os.environ.get("ASSETS_VERSION_URL", "https://assets.weircon.dk/VERSION")
|
|
GEOIP_DB = os.environ.get("GEOIP_DB", str(DATA / "dbip-country.mmdb"))
|
|
_geoip_reader = None
|
|
|
|
app = FastAPI(title="Weircon Portal")
|
|
app.add_middleware(SessionMiddleware, secret_key=SECRET_KEY, https_only=True, same_site="lax", max_age=60 * 60 * 12)
|
|
app.mount("/static", StaticFiles(directory=BASE / "static"), name="static")
|
|
templates = Jinja2Templates(directory=str(BASE / "templates"))
|
|
|
|
oauth = OAuth()
|
|
oauth.register(
|
|
name="authentik",
|
|
server_metadata_url=f"{AUTHENTIK}/application/o/weircon-portal/.well-known/openid-configuration",
|
|
client_id=CLIENT_ID,
|
|
client_secret=CLIENT_SECRET,
|
|
client_kwargs={"scope": "openid email profile groups"},
|
|
)
|
|
|
|
# ---- udvidbare admin-moduler: tilføj en post for at få en ny admin-side ----
|
|
ADMIN_MODULES = [
|
|
{"key": "ip-labels", "title": "IP → label & land", "icon": "🌍",
|
|
"desc": "Mærk kendte IP-adresser (netværk/land) i statistikken."},
|
|
{"key": "cookies", "title": "Cookie-tekster", "icon": "🍪",
|
|
"desc": "Teksten på cookie-banneret og /cookies-siden."},
|
|
{"key": "stats", "title": "Statistik", "icon": "📊",
|
|
"desc": "Hurtigt overblik fra Matomo."},
|
|
{"key": "apps", "title": "App-beskrivelser & grupper", "icon": "🧩",
|
|
"desc": "Rediger beskrivelse og gruppe på dine Authentik-apps."},
|
|
{"key": "ops", "title": "Drift / egress", "icon": "🛰️",
|
|
"desc": "Static-assets-version + random-proxy egress, rollover & inflight."},
|
|
{"key": "users", "title": "Brugere", "icon": "👤",
|
|
"desc": "Opret og administrér brugere + adgang (via grupper)."},
|
|
{"key": "groups", "title": "Grupper", "icon": "👥",
|
|
"desc": "Grupper og medlemmer — styrer hvilke services brugere har adgang til."},
|
|
]
|
|
|
|
|
|
def current_user(request: Request):
|
|
return request.session.get("user")
|
|
|
|
|
|
def is_admin(user) -> bool:
|
|
return bool(user) and ADMIN_GROUP in (user.get("groups") or [])
|
|
|
|
|
|
def load_json(path: Path, default):
|
|
try:
|
|
return json.loads(path.read_text("utf-8"))
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def save_ip_labels(data: dict) -> str:
|
|
IP_LABELS_LOCAL.write_text(json.dumps(data, ensure_ascii=False, indent=2), "utf-8")
|
|
if IP_LABELS_REMOTE:
|
|
try:
|
|
subprocess.run(
|
|
["scp", "-o", "StrictHostKeyChecking=accept-new", "-o", "ConnectTimeout=8",
|
|
str(IP_LABELS_LOCAL), IP_LABELS_REMOTE],
|
|
check=True, capture_output=True, timeout=20)
|
|
return "gemt + synkroniseret til NPMplus"
|
|
except Exception as e:
|
|
return f"gemt lokalt, men sync til NPMplus fejlede: {e}"
|
|
return "gemt lokalt"
|
|
|
|
|
|
async def fetch_user_apps(username: str):
|
|
"""Apps brugeren har adgang til, via Authentik-API (service-account-token)."""
|
|
headers = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=headers, timeout=10) as c:
|
|
u = await c.get("/api/v3/core/users/", params={"username": username})
|
|
results = u.json().get("results", [])
|
|
if not results:
|
|
return []
|
|
pk = results[0]["pk"]
|
|
a = await c.get("/api/v3/core/applications/",
|
|
params={"for_user": pk, "superuser_full_list": "false", "page_size": 100})
|
|
apps = a.json().get("results", [])
|
|
out = []
|
|
for app_ in apps:
|
|
if app_.get("slug") == "weircon-portal":
|
|
continue
|
|
url = app_.get("launch_url") or app_.get("meta_launch_url")
|
|
if not url:
|
|
continue # kun apps helt uden launch-URL (authentik-cli, ldap-auth) — vises heller ikke i Authentik
|
|
host = url.split("//", 1)[-1].split("/", 1)[0]
|
|
out.append({
|
|
"slug": app_.get("slug"),
|
|
"name": app_.get("name"),
|
|
"url": url,
|
|
"host": host,
|
|
"group": (app_.get("group") or "").strip(),
|
|
"desc": (app_.get("meta_description") or "").strip(),
|
|
"icon": app_.get("meta_icon") or "",
|
|
})
|
|
out.sort(key=lambda x: (x["name"] or "").lower())
|
|
return out
|
|
|
|
|
|
GROUP_ORDER = {n: i for i, n in enumerate(
|
|
["Home", "Familie", "Medie", "Social", "AI", "Office", "Finans", "dev", "game", "Admin"])}
|
|
|
|
|
|
USERPREFS = DATA / "userprefs.json"
|
|
|
|
|
|
def get_prefs(username):
|
|
p = (load_json(USERPREFS, {}) or {}).get(username) or {}
|
|
return {"favorites": p.get("favorites") or [], "view": p.get("view") or "grouped"}
|
|
|
|
|
|
def set_prefs(username, **kw):
|
|
allp = load_json(USERPREFS, {}) or {}
|
|
cur = allp.get(username) or {}
|
|
cur.update(kw)
|
|
allp[username] = cur
|
|
USERPREFS.write_text(json.dumps(allp, ensure_ascii=False, indent=2), "utf-8")
|
|
|
|
|
|
def group_apps(apps):
|
|
"""Grupperer apps efter Authentiks group-felt, i en fornuftig rækkefølge."""
|
|
groups = {}
|
|
for a in apps:
|
|
g = a.get("group") or "Øvrige"
|
|
groups.setdefault(g, []).append(a)
|
|
|
|
def gkey(name):
|
|
if name == "Øvrige":
|
|
return (999, "")
|
|
return (GROUP_ORDER.get(name, 50), name.lower())
|
|
|
|
return [{"name": g, "apps": sorted(groups[g], key=lambda x: (x["name"] or "").lower())}
|
|
for g in sorted(groups, key=gkey)]
|
|
|
|
|
|
async def fetch_all_apps_admin():
|
|
"""Alle apps (admin-syn) med slug/navn/gruppe/beskrivelse."""
|
|
headers = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=headers, timeout=15) as c:
|
|
r = await c.get("/api/v3/core/applications/",
|
|
params={"page_size": 100, "superuser_full_list": "true"})
|
|
apps = r.json().get("results", [])
|
|
out = [{"slug": a["slug"], "name": a.get("name"),
|
|
"group": a.get("group") or "", "desc": (a.get("meta_description") or "").strip()}
|
|
for a in apps if a.get("slug") != "weircon-portal"]
|
|
out.sort(key=lambda x: ((x["group"] or "~~").lower(), (x["name"] or "").lower()))
|
|
return out
|
|
|
|
|
|
async def patch_apps(rows):
|
|
headers = {"Authorization": f"Bearer {API_TOKEN}"}
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=headers, timeout=20) as c:
|
|
async def one(r):
|
|
try:
|
|
resp = await c.patch(f"/api/v3/core/applications/{r['slug']}/",
|
|
json={"meta_description": r["desc"], "group": r["group"]})
|
|
return resp.status_code < 300
|
|
except Exception:
|
|
return False
|
|
results = await asyncio.gather(*[one(r) for r in rows])
|
|
return sum(1 for ok in results if ok)
|
|
|
|
|
|
def geoip_country(ip):
|
|
global _geoip_reader
|
|
try:
|
|
if _geoip_reader is None:
|
|
import maxminddb
|
|
_geoip_reader = maxminddb.open_database(GEOIP_DB)
|
|
rec = _geoip_reader.get(ip) or {}
|
|
c = rec.get("country") or {}
|
|
iso = c.get("iso_code") or ""
|
|
name = (c.get("names") or {}).get("en") or ""
|
|
flag = "".join(chr(0x1F1E6 + ord(ch) - 65) for ch in iso) if len(iso) == 2 and iso.isalpha() else ""
|
|
return {"iso": iso, "name": name, "flag": flag}
|
|
except Exception:
|
|
return {"iso": "", "name": "", "flag": ""}
|
|
|
|
|
|
async def assets_version():
|
|
try:
|
|
async with httpx.AsyncClient(timeout=6) as c:
|
|
r = await c.get(ASSETS_VERSION_URL)
|
|
return r.text.strip()[:40]
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
async def rp_status():
|
|
if not RP_API_KEY:
|
|
return {"error": "ingen RP_API_KEY konfigureret"}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=6) as c:
|
|
r = await c.get(RP_BASE_URL + "/status", headers={"X-WEIRCON-RANDOM-IP": RP_API_KEY})
|
|
d = r.json()
|
|
for p in d.get("proxies", []):
|
|
p["geo"] = geoip_country(p.get("egress_ip") or "")
|
|
rot = d.get("rotation") or {}
|
|
if rot.get("next_egress_ip"):
|
|
rot["geo"] = geoip_country(rot["next_egress_ip"])
|
|
d["inflight_total"] = sum(int(p.get("inflight") or 0) for p in d.get("proxies", []))
|
|
return d
|
|
except Exception as exc:
|
|
return {"error": str(exc)[:200]}
|
|
|
|
|
|
def _ak_headers():
|
|
return {"Authorization": f"Bearer {API_TOKEN}"}
|
|
|
|
|
|
async def ak_list_groups():
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=15) as c:
|
|
gs = (await c.get("/api/v3/core/groups/", params={"page_size": 100, "ordering": "name"})).json().get("results", [])
|
|
return [{"pk": g["pk"], "name": g["name"], "members": len(g.get("users") or [])} for g in gs]
|
|
|
|
|
|
async def ak_list_users():
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=20) as c:
|
|
us = (await c.get("/api/v3/core/users/", params={"page_size": 300, "ordering": "username"})).json().get("results", [])
|
|
out = []
|
|
for u in us:
|
|
out.append({"pk": u["pk"], "username": u["username"], "name": u.get("name"),
|
|
"email": u.get("email"), "active": u.get("is_active"),
|
|
"groups": [str(x) for x in (u.get("groups") or [])],
|
|
"group_names": [g.get("name") for g in (u.get("groups_obj") or [])],
|
|
"type": u.get("type")})
|
|
return out
|
|
|
|
|
|
async def ak_get_user(pk):
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=15) as c:
|
|
u = (await c.get(f"/api/v3/core/users/{pk}/")).json()
|
|
return {"pk": u["pk"], "username": u["username"], "name": u.get("name"), "email": u.get("email"),
|
|
"active": u.get("is_active"), "groups": [str(x) for x in (u.get("groups") or [])]}
|
|
|
|
|
|
async def ak_create_user(username, name, email, password, group_pks):
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=20) as c:
|
|
r = await c.post("/api/v3/core/users/", json={
|
|
"username": username, "name": name or username, "email": email or "",
|
|
"is_active": True, "type": "internal", "path": "users", "groups": group_pks})
|
|
if r.status_code >= 300:
|
|
return False, r.text[:300]
|
|
pk = r.json()["pk"]
|
|
if password:
|
|
await c.post(f"/api/v3/core/users/{pk}/set_password/", json={"password": password})
|
|
return True, pk
|
|
|
|
|
|
async def ak_update_user(pk, group_pks, is_active, password=""):
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=20) as c:
|
|
r = await c.patch(f"/api/v3/core/users/{pk}/", json={"groups": group_pks, "is_active": is_active})
|
|
if password:
|
|
await c.post(f"/api/v3/core/users/{pk}/set_password/", json={"password": password})
|
|
return r.status_code < 300
|
|
|
|
|
|
async def ak_create_group(name):
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=15) as c:
|
|
r = await c.post("/api/v3/core/groups/", json={"name": name})
|
|
return r.status_code < 300
|
|
|
|
|
|
async def ak_group_members(gpk):
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=15) as c:
|
|
g = (await c.get(f"/api/v3/core/groups/{gpk}/")).json()
|
|
return {"pk": g["pk"], "name": g["name"], "users": [str(x) for x in (g.get("users") or [])]}
|
|
|
|
|
|
async def ak_set_group_members(gpk, user_pks):
|
|
desired = set(str(x) for x in user_pks)
|
|
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=_ak_headers(), timeout=25) as c:
|
|
g = (await c.get(f"/api/v3/core/groups/{gpk}/")).json()
|
|
cur = set(str(x) for x in (g.get("users") or []))
|
|
adds = [c.post(f"/api/v3/core/groups/{gpk}/add_user/", json={"pk": int(p)}) for p in desired - cur]
|
|
rems = [c.post(f"/api/v3/core/groups/{gpk}/remove_user/", json={"pk": int(p)}) for p in cur - desired]
|
|
await asyncio.gather(*adds, *rems)
|
|
return True
|
|
|
|
|
|
async def matomo_summary():
|
|
if not MATOMO_TOKEN:
|
|
return None
|
|
params = {
|
|
"module": "API", "method": "VisitsSummary.get", "idSite": MATOMO_SITE,
|
|
"period": "day", "date": "last7", "format": "JSON", "token_auth": MATOMO_TOKEN,
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as c:
|
|
r = await c.get(f"{MATOMO_BASE}/index.php", params=params)
|
|
data = r.json()
|
|
days = list(data.values()) if isinstance(data, dict) else []
|
|
visits = sum(int(d.get("nb_visits", 0) or 0) for d in days if isinstance(d, dict))
|
|
actions = sum(int(d.get("nb_actions", 0) or 0) for d in days if isinstance(d, dict))
|
|
return {"visits_7d": visits, "actions_7d": actions}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ---------------- ruter ----------------
|
|
@app.get("/auth/login")
|
|
async def login(request: Request):
|
|
return await oauth.authentik.authorize_redirect(request, REDIRECT_URI)
|
|
|
|
|
|
@app.get("/auth/callback")
|
|
async def callback(request: Request):
|
|
try:
|
|
token = await oauth.authentik.authorize_access_token(request)
|
|
except Exception as e:
|
|
return HTMLResponse(f"Login fejlede: {e}", status_code=400)
|
|
info = token.get("userinfo") or {}
|
|
request.session["user"] = {
|
|
"name": info.get("name") or info.get("preferred_username"),
|
|
"username": info.get("preferred_username"),
|
|
"email": info.get("email"),
|
|
"groups": info.get("groups") or [],
|
|
}
|
|
return RedirectResponse("/", status_code=302)
|
|
|
|
|
|
@app.get("/auth/logout")
|
|
async def logout(request: Request):
|
|
request.session.clear()
|
|
return RedirectResponse("/", status_code=302)
|
|
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
async def home(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not user:
|
|
return templates.TemplateResponse(request, "login.html", {"request": request})
|
|
apps, error = [], None
|
|
try:
|
|
apps = await fetch_user_apps(user["username"])
|
|
except Exception as e:
|
|
error = f"Kunne ikke hente apps fra Authentik: {e}"
|
|
prefs = get_prefs(user["username"])
|
|
fav_set = set(prefs["favorites"])
|
|
by_name = sorted(apps, key=lambda x: (x["name"] or "").lower())
|
|
return templates.TemplateResponse(request, "dashboard.html", {
|
|
"request": request, "user": user, "error": error, "is_admin": is_admin(user),
|
|
"groups": group_apps(apps), "apps_sorted": by_name,
|
|
"fav_apps": [a for a in by_name if a["slug"] in fav_set],
|
|
"favorites": list(fav_set), "view": prefs["view"], "saved": saved,
|
|
})
|
|
|
|
|
|
@app.get("/admin", response_class=HTMLResponse)
|
|
async def admin_home(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
return templates.TemplateResponse(request, "admin.html", {
|
|
"request": request, "user": user, "modules": ADMIN_MODULES, "is_admin": True,
|
|
})
|
|
|
|
|
|
@app.get("/admin/ip-labels", response_class=HTMLResponse)
|
|
async def admin_iplabels(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
raw = IP_LABELS_LOCAL.read_text("utf-8") if IP_LABELS_LOCAL.exists() else "{}"
|
|
data = load_json(IP_LABELS_LOCAL, {})
|
|
rows = [{"ip": k, "label": (v or {}).get("label", ""), "country": (v or {}).get("country", "")}
|
|
for k, v in data.items() if isinstance(v, dict)]
|
|
rows.sort(key=lambda r: r["ip"])
|
|
return templates.TemplateResponse(request, "admin_iplabels.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"rows": rows, "raw": raw, "saved": saved,
|
|
})
|
|
|
|
|
|
@app.post("/admin/ip-labels")
|
|
async def admin_iplabels_save(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
form = await request.form()
|
|
ips = form.getlist("ip")
|
|
labels = form.getlist("label")
|
|
countries = form.getlist("country")
|
|
data = {}
|
|
for ip, label, country in zip(ips, labels, countries):
|
|
ip = (ip or "").strip()
|
|
if not ip:
|
|
continue
|
|
entry = {}
|
|
if (label or "").strip():
|
|
entry["label"] = label.strip()
|
|
if (country or "").strip():
|
|
entry["country"] = country.strip().lower()
|
|
if entry:
|
|
data[ip] = entry
|
|
msg = save_ip_labels(data)
|
|
return RedirectResponse(f"/admin/ip-labels?saved={msg}", status_code=303)
|
|
|
|
|
|
@app.get("/admin/cookies", response_class=HTMLResponse)
|
|
async def admin_cookies(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
s = load_json(SETTINGS_PATH, {})
|
|
return templates.TemplateResponse(request, "admin_cookies.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"banner": s.get("cookie_banner", "Denne side bruger cookies til at forbedre din oplevelse og lave anonym statistik over besøg. Ved at bruge siden accepterer du det."),
|
|
"page": s.get("cookie_page", "Vi bruger Matomo (self-hosted) til anonym statistik og et par funktionelle cookies. Intet deles med tredjeparter."),
|
|
"saved": saved,
|
|
})
|
|
|
|
|
|
@app.post("/admin/cookies")
|
|
async def admin_cookies_save(request: Request, cookie_banner: str = Form(""), cookie_page: str = Form("")):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
s = load_json(SETTINGS_PATH, {})
|
|
s["cookie_banner"] = cookie_banner
|
|
s["cookie_page"] = cookie_page
|
|
SETTINGS_PATH.write_text(json.dumps(s, ensure_ascii=False, indent=2), "utf-8")
|
|
return RedirectResponse("/admin/cookies?saved=Gemt", status_code=303)
|
|
|
|
|
|
@app.get("/admin/ops", response_class=HTMLResponse)
|
|
async def admin_ops(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
version = await assets_version()
|
|
rp = await rp_status()
|
|
return templates.TemplateResponse(request, "admin_ops.html", {
|
|
"request": request, "user": user, "is_admin": True, "version": version, "rp": rp,
|
|
})
|
|
|
|
|
|
@app.get("/admin/stats", response_class=HTMLResponse)
|
|
async def admin_stats(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
stats = await matomo_summary()
|
|
return templates.TemplateResponse(request, "admin_stats.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"stats": stats, "matomo": MATOMO_BASE,
|
|
})
|
|
|
|
|
|
@app.get("/admin/apps", response_class=HTMLResponse)
|
|
async def admin_apps(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
rows = await fetch_all_apps_admin()
|
|
return templates.TemplateResponse(request, "admin_apps.html", {
|
|
"request": request, "user": user, "is_admin": True, "rows": rows, "saved": saved,
|
|
})
|
|
|
|
|
|
@app.post("/admin/apps")
|
|
async def admin_apps_save(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
form = await request.form()
|
|
slugs = form.getlist("slug")
|
|
descs = form.getlist("desc")
|
|
groups = form.getlist("group")
|
|
rows = [{"slug": sl, "desc": (d or "").strip(), "group": (g or "").strip()}
|
|
for sl, d, g in zip(slugs, descs, groups) if sl]
|
|
n = await patch_apps(rows)
|
|
return RedirectResponse(f"/?saved=Gemt {n} apps", status_code=303)
|
|
|
|
|
|
@app.get("/admin/users", response_class=HTMLResponse)
|
|
async def admin_users(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403, "Kun for administratorer")
|
|
return templates.TemplateResponse(request, "admin_users.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"users": await ak_list_users(), "groups": await ak_list_groups(), "saved": saved,
|
|
})
|
|
|
|
|
|
@app.post("/admin/users")
|
|
async def admin_users_create(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
f = await request.form()
|
|
username = (f.get("username") or "").strip()
|
|
if not username:
|
|
return RedirectResponse("/admin/users?saved=Brugernavn mangler", status_code=303)
|
|
ok, res = await ak_create_user(username, (f.get("name") or "").strip(),
|
|
(f.get("email") or "").strip(), f.get("password") or "",
|
|
f.getlist("groups"))
|
|
msg = f"Bruger '{username}' oprettet" if ok else f"Fejl: {res}"
|
|
return RedirectResponse(f"/admin/users?saved={msg}", status_code=303)
|
|
|
|
|
|
@app.get("/admin/users/{pk}", response_class=HTMLResponse)
|
|
async def admin_user_edit(request: Request, pk: int):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
return templates.TemplateResponse(request, "admin_user_edit.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"u": await ak_get_user(pk), "groups": await ak_list_groups(),
|
|
})
|
|
|
|
|
|
@app.post("/admin/users/{pk}")
|
|
async def admin_user_save(request: Request, pk: int):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
f = await request.form()
|
|
await ak_update_user(pk, f.getlist("groups"), f.get("is_active") == "on", f.get("password") or "")
|
|
return RedirectResponse("/admin/users?saved=Bruger gemt", status_code=303)
|
|
|
|
|
|
@app.get("/admin/groups", response_class=HTMLResponse)
|
|
async def admin_groups(request: Request, saved: str = ""):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
return templates.TemplateResponse(request, "admin_groups.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"groups": await ak_list_groups(), "saved": saved,
|
|
})
|
|
|
|
|
|
@app.post("/admin/groups")
|
|
async def admin_groups_create(request: Request):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
f = await request.form()
|
|
name = (f.get("name") or "").strip()
|
|
if name:
|
|
await ak_create_group(name)
|
|
return RedirectResponse(f"/admin/groups?saved=Gruppe '{name}' oprettet", status_code=303)
|
|
|
|
|
|
@app.get("/admin/groups/{gpk}", response_class=HTMLResponse)
|
|
async def admin_group_edit(request: Request, gpk: str):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
return templates.TemplateResponse(request, "admin_group_edit.html", {
|
|
"request": request, "user": user, "is_admin": True,
|
|
"g": await ak_group_members(gpk), "users": await ak_list_users(),
|
|
})
|
|
|
|
|
|
@app.post("/admin/groups/{gpk}")
|
|
async def admin_group_save(request: Request, gpk: str):
|
|
user = current_user(request)
|
|
if not is_admin(user):
|
|
raise HTTPException(403)
|
|
f = await request.form()
|
|
await ak_set_group_members(gpk, f.getlist("members"))
|
|
return RedirectResponse("/admin/groups?saved=Medlemmer gemt", status_code=303)
|
|
|
|
|
|
# Offentligt JSON-endpoint så frontend/andre kan læse cookie-teksterne (valgfrit)
|
|
@app.get("/api/settings")
|
|
async def api_settings():
|
|
return JSONResponse(load_json(SETTINGS_PATH, {}))
|
|
|
|
|
|
@app.post("/api/favorite")
|
|
async def api_favorite(request: Request):
|
|
user = current_user(request)
|
|
if not user:
|
|
raise HTTPException(401)
|
|
body = await request.json()
|
|
slug = str(body.get("slug") or "")
|
|
favs = set(get_prefs(user["username"])["favorites"])
|
|
favs.add(slug) if body.get("on") else favs.discard(slug)
|
|
set_prefs(user["username"], favorites=sorted(favs))
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/view")
|
|
async def api_view(request: Request):
|
|
user = current_user(request)
|
|
if not user:
|
|
raise HTTPException(401)
|
|
v = (await request.json()).get("view")
|
|
if v in ("grouped", "alpha"):
|
|
set_prefs(user["username"], view=v)
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/search-index")
|
|
async def search_index(request: Request):
|
|
"""Søgeindeks til ⌘K-paletten: brugerens apps + sider."""
|
|
user = current_user(request)
|
|
if not user:
|
|
return JSONResponse([])
|
|
items = [{"type": "page", "title": "Forside", "url": "/"}]
|
|
if is_admin(user):
|
|
items += [
|
|
{"type": "page", "title": "Admin-panel", "url": "/admin", "extra": "admin indstillinger"},
|
|
{"type": "page", "title": "IP → label & land", "url": "/admin/ip-labels", "extra": "ip land netværk geo"},
|
|
{"type": "page", "title": "Cookie-tekster", "url": "/admin/cookies", "extra": "cookies samtykke"},
|
|
{"type": "page", "title": "Statistik", "url": "/admin/stats", "extra": "matomo besøg"},
|
|
]
|
|
try:
|
|
for a in await fetch_user_apps(user["username"]):
|
|
items.append({"type": "app", "title": a["name"], "url": a["url"],
|
|
"extra": f'{a["desc"]} {a["host"]}'.strip()})
|
|
except Exception:
|
|
pass
|
|
return JSONResponse(items)
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz():
|
|
return {"ok": True, "configured": bool(CLIENT_ID and API_TOKEN)}
|