NODEDC_TASKMANAGER/plane-src/apps/api/plane/authentication/middleware/nodedc_access.py

import os
import time

import requests
from django.contrib.auth import logout
from django.http import HttpResponseRedirect, JsonResponse

from plane.db.models import ExternalIdentityLink, Session


OIDC_PROVIDER = "authentik"


class NodeDCAccessMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self._enforce(request)

        if response is not None:
            return response

        return self.get_response(request)

    def _enforce(self, request):
        config = get_access_config()

        if not config["enabled"] or should_skip_path(request.path_info):
            return None

        user = getattr(request, "user", None)

        if not user or not user.is_authenticated:
            return None

        link = ExternalIdentityLink.objects.filter(
            provider=OIDC_PROVIDER,
            user=user,
            status=ExternalIdentityLink.Status.ACTIVE,
        ).first()

        if link is None:
            return deny_unlinked_user(request, config) if config["enforce_unlinked"] else None

        cached = get_cached_access_decision(request, config["cache_seconds"])

        if cached is not None:
            return None if cached else revoke_session(request, user, "nodedc_access_revoked")

        try:
            decision = check_launcher_access(config, link, user)
        except (ValueError, requests.RequestException):
            return service_unavailable(request)

        cache_access_decision(request, decision["allowed"], config["cache_seconds"])

        if decision["groups"] is not None and decision["groups"] != link.groups:
            link.groups = decision["groups"]
            link.save(update_fields=["groups", "updated_at"])

        if not decision["allowed"]:
            return revoke_session(request, user, decision["reason"])

        return None


def get_access_config():
    check_url = os.environ.get("PLANE_NODEDC_ACCESS_CHECK_URL", "").strip()
    token = (
        os.environ.get("PLANE_NODEDC_ACCESS_TOKEN", "").strip()
        or os.environ.get("NODEDC_INTERNAL_ACCESS_TOKEN", "").strip()
        or os.environ.get("PLANE_OIDC_CLIENT_SECRET", "").strip()
    )

    return {
        "enabled": is_truthy(os.environ.get("PLANE_NODEDC_ACCESS_ENFORCEMENT", "0")) and bool(check_url and token),
        "check_url": check_url,
        "token": token,
        "service_slug": os.environ.get("PLANE_NODEDC_ACCESS_SERVICE_SLUG", "task-manager").strip() or "task-manager",
        "timeout": float(os.environ.get("PLANE_NODEDC_ACCESS_TIMEOUT_SECONDS", "3") or "3"),
        "cache_seconds": max(0, int(os.environ.get("PLANE_NODEDC_ACCESS_CACHE_SECONDS", "0") or "0")),
        "enforce_unlinked": is_truthy(os.environ.get("PLANE_NODEDC_ACCESS_ENFORCE_UNLINKED", "0")),
    }


def check_launcher_access(config, link, user):
    response = requests.post(
        config["check_url"],
        json={
            "serviceSlug": config["service_slug"],
            "subject": link.subject,
            "email": link.email or user.email,
            "userId": None,
        },
        headers={
            "Authorization": f"Bearer {config['token']}",
            "Accept": "application/json",
        },
        timeout=config["timeout"],
    )
    response.raise_for_status()
    payload = response.json()

    return {
        "allowed": bool(payload.get("allowed")),
        "reason": payload.get("reason") or "nodedc_access_denied",
        "groups": payload.get("groups") if isinstance(payload.get("groups"), list) else None,
    }


def get_cached_access_decision(request, cache_seconds):
    if cache_seconds <= 0:
        return None

    checked_at = request.session.get("nodedc_access_checked_at")
    allowed = request.session.get("nodedc_access_allowed")

    if not checked_at or allowed is None:
        return None

    try:
        checked_at_value = float(checked_at)
    except (TypeError, ValueError):
        return None

    if time.time() - checked_at_value > cache_seconds:
        return None

    return bool(allowed)


def cache_access_decision(request, allowed, cache_seconds):
    if cache_seconds <= 0:
        request.session.pop("nodedc_access_checked_at", None)
        request.session.pop("nodedc_access_allowed", None)
        return

    request.session["nodedc_access_checked_at"] = time.time()
    request.session["nodedc_access_allowed"] = bool(allowed)


def revoke_session(request, user, reason):
    Session.objects.filter(user_id=str(user.id)).delete()
    logout(request)

    if is_api_request(request):
        status_code = 200 if request.path_info == "/api/users/session/" else 403
        payload = {"is_authenticated": False} if status_code == 200 else {}
        payload.update({"error": "nodedc_access_revoked", "reason": reason})
        return JsonResponse(payload, status=status_code)

    return HttpResponseRedirect(os.environ.get("PLANE_NODEDC_ACCESS_DENIED_REDIRECT_URL", "http://launcher.local.nodedc/"))


def deny_unlinked_user(request, config):
    if not config["enforce_unlinked"]:
        return None

    return revoke_session(request, request.user, "nodedc_identity_not_linked")


def service_unavailable(request):
    if is_api_request(request):
        return JsonResponse({"error": "nodedc_access_check_unavailable"}, status=503)

    return JsonResponse({"error": "nodedc_access_check_unavailable"}, status=503)


def should_skip_path(path):
    return path.startswith(
        (
            "/auth/",
            "/api/public/",
            "/api/schema/",
            "/static/",
            "/assets/",
            "/robots.txt",
        )
    )


def is_api_request(request):
    return request.path_info.startswith("/api/")


def is_truthy(value):
    return str(value).strip().lower() in {"1", "true", "yes", "on"}