weircon-front: SSO-portal til weircon.dk (Authentik-login, app-dashboard, admin, favoritter, tema, palette)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,428 @@
|
||||
"""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 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"
|
||||
|
||||
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."},
|
||||
]
|
||||
|
||||
|
||||
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}"}
|
||||
n = 0
|
||||
async with httpx.AsyncClient(base_url=AUTHENTIK, headers=headers, timeout=20) as c:
|
||||
for r in rows:
|
||||
resp = await c.patch(f"/api/v3/core/applications/{r['slug']}/",
|
||||
json={"meta_description": r["desc"], "group": r["group"]})
|
||||
if resp.status_code < 300:
|
||||
n += 1
|
||||
return n
|
||||
|
||||
|
||||
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):
|
||||
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"],
|
||||
})
|
||||
|
||||
|
||||
@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/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"/admin/apps?saved=Gemt {n} apps", 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)}
|
||||
Reference in New Issue
Block a user