weircon-front/authentik.py

"""Authentik Core-API: apps, brugere, grupper. Læsninger er cachet (cache.py);
skrivninger rydder cachen via cache.clear() i kaldsstedet (main.py)."""
import asyncio

import httpx

import cache
from config import API_TOKEN, AUTHENTIK, GROUP_ORDER


def _h():
    return {"Authorization": f"Bearer {API_TOKEN}"}


def _client(timeout=15):
    return httpx.AsyncClient(base_url=AUTHENTIK, headers=_h(), timeout=timeout)


async def fetch_user_apps(username: str):
    """Apps brugeren har adgang til. Cachet 45s."""
    ck = f"apps:{username}"
    hit = cache.get(ck)
    if hit is not None:
        return hit
    async with _client(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
        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())
    cache.put(ck, out, 45)
    return out


def group_apps(apps):
    groups = {}
    for a in apps:
        groups.setdefault(a.get("group") or "Øvrige", []).append(a)

    def gkey(name):
        return (999, "") if name == "Øvrige" else (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():
    hit = cache.get("apps_admin")
    if hit is not None:
        return hit
    async with _client(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()))
    cache.put("apps_admin", out, 45)
    return out


async def patch_apps(rows):
    async with _client(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)


async def list_groups():
    hit = cache.get("groups")
    if hit is not None:
        return hit
    async with _client(15) as c:
        gs = (await c.get("/api/v3/core/groups/", params={"page_size": 100, "ordering": "name"})).json().get("results", [])
    res = [{"pk": g["pk"], "name": g["name"], "members": len(g.get("users") or [])} for g in gs]
    cache.put("groups", res, 30)
    return res


async def list_users():
    hit = cache.get("users")
    if hit is not None:
        return hit
    async with _client(20) as c:
        us = (await c.get("/api/v3/core/users/", params={"page_size": 300, "ordering": "username"})).json().get("results", [])
    out = [{"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")}
           for u in us]
    cache.put("users", out, 30)
    return out


async def get_user(pk):
    async with _client(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 create_user(username, name, email, password, group_pks):
    async with _client(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 update_user(pk, group_pks, is_active, password=""):
    async with _client(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 create_group(name):
    async with _client(15) as c:
        r = await c.post("/api/v3/core/groups/", json={"name": name})
    return r.status_code < 300


async def group_members(gpk):
    async with _client(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 set_group_members(gpk, user_pks):
    desired = set(str(x) for x in user_pks)
    async with _client(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 []))
        ops = [c.post(f"/api/v3/core/groups/{gpk}/add_user/", json={"pk": int(p)}) for p in desired - cur]
        ops += [c.post(f"/api/v3/core/groups/{gpk}/remove_user/", json={"pk": int(p)}) for p in cur - desired]
        await asyncio.gather(*ops)
    return True