AuthPowerBI endpoint with some tests

Author: szabozoltan69

api/test_views.py

@@ -1,7 +1,11 @@
+import re
 import uuid
+from unittest.mock import patch
 
 from django.contrib.auth.models import User
 from django.core.files.uploadedfile import SimpleUploadedFile
+from django.test import override_settings
+from django.urls import reverse
 
 import api.models as models
 from api.factories.event import (
@@ -18,6 +22,82 @@
 from main.test_case import APITestCase
 
 
+class AuthPowerBITest(APITestCase):
+    def setUp(self):
+        self.url = reverse("auth_power_bi")
+
+    def test_not_authenticated_returns_401(self):
+        resp = self.client.get(self.url)
+        self.assertEqual(resp.status_code, 401)
+        self.assertIn("error_code", resp.json())
+
+    def test_authenticated_returns_ok(self):
+        user = User.objects.create_user(username="alice", password="pass1234")
+        # self.client.login(username="alice", password="pass1234")  # session auth, if needed in the future
+        # Use token authentication instead of session auth
+        self.authenticate(user=user)
+        resp = self.client.get(self.url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data.get("detail"), "ok")
+        self.assertEqual(data.get("user"), user.username)
+
+    def test_authenticated_returns_mock_values_shape(self):
+        user = User.objects.create_user(username="bob", password="pass1234")
+        self.assertEqual("bob", user.username)
+        # Use token authentication instead of session auth
+        self.authenticate(user=user)
+
+        # Force mock path regardless of settings by returning no MI token
+        with patch("api.views._pbi_token_via_managed_identity", return_value=None):
+            resp = self.client.get(self.url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+
+        # embed_token: 16 hex chars
+        self.assertIsInstance(data.get("embed_token"), str)
+        self.assertTrue(re.fullmatch(r"[0-9a-f]{16}", data["embed_token"]))
+        # embed_url: 10-char random string
+        self.assertIsInstance(data.get("embed_url"), str)
+        self.assertEqual(len(data["embed_url"]), 10)
+        # report_id: positive int
+        self.assertIsInstance(data.get("report_id"), int)
+        self.assertGreater(data["report_id"], 0)
+
+    def test_authenticated_powerbi_values_when_configured(self):
+        user = User.objects.create_user(username="carol", password="pass1234")
+        self.assertEqual("carol", user.username)
+        # Use token authentication instead of session auth
+        self.authenticate(user=user)
+
+        expected = {
+            "embed_url": "https://app.powerbi.com/reportEmbed?reportId=rep-123",
+            "report_id": "rep-123",
+            "embed_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",  # dummy string
+            "expires_at": "2099-01-01T00:00:00Z",
+        }
+
+        with override_settings(POWERBI_WORKSPACE_ID="ws-abc"):
+            with (
+                patch("api.views._pbi_token_via_managed_identity", return_value="access-token") as p_token,
+                patch(
+                    "api.views._pbi_get_embed_info",
+                    return_value=(expected["embed_url"], expected["report_id"], expected["embed_token"], expected["expires_at"]),
+                ) as p_info,
+            ):
+                resp = self.client.get(self.url)
+
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data.get("embed_url"), expected["embed_url"])
+        self.assertEqual(data.get("report_id"), expected["report_id"])
+        self.assertEqual(data.get("embed_token"), expected["embed_token"])
+        self.assertEqual(data.get("user"), "carol")
+        # helpers were called
+        p_token.assert_called_once()
+        p_info.assert_called_once()
+
+
 class SecureFileFieldTest(APITestCase):
     def is_valid_uuid(self, uuid_to_test):
         try:

api/views.py

@@ -1,7 +1,12 @@
+import base64
 import json
+import os
+import secrets
 from datetime import datetime, timedelta
 from urllib.parse import urlparse
 
+import requests
+from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
 from django.conf import settings
 from django.contrib.auth import authenticate, login, logout
 from django.contrib.auth.models import User
@@ -1065,3 +1070,155 @@ def logout_user(request):
     if request.method == "POST" and request.user.is_authenticated:
         logout(request)
     return redirect(reverse(settings.LOGIN_URL))
+
+
+# Power BI embedding via managed identity
+PBI_SCOPE = "https://analysis.windows.net/powerbi/api/.default"
+PBI_BASE = "https://api.powerbi.com/v1.0/myorg"
+
+
+def _pbi_token_via_managed_identity() -> str | None:
+    """
+    Acquire an AAD access token for Power BI using the AKS managed identity.
+    If AZURE_CLIENT_ID is provided, target that user-assigned MI.
+    """
+    try:
+        client_id = getattr(settings, "AZURE_CLIENT_ID", None) or os.getenv("AZURE_CLIENT_ID")
+        if client_id:
+            cred = ManagedIdentityCredential(client_id=client_id)
+        else:
+            # Will use workload identity / MSI / Azure CLI (in dev) automatically
+            cred = DefaultAzureCredential(exclude_interactive_browser_credential=True)
+        return cred.get_token(PBI_SCOPE).token
+    except Exception as exc:
+        logger.exception("Power BI MI token acquisition failed: %s", exc)
+        return None
+
+
+def _pbi_get_embed_info(access_token: str, workspace_id: str, report_id: str | None = None, timeout: int = 10):
+    """
+    Get report embedUrl and generate an embed token (embed-for-your-organization).
+    Requires the service principal (managed identity) to be a member of the workspace.
+    """
+    headers = {"Authorization": f"Bearer {access_token}"}
+
+    # Resolve report metadata
+    if report_id:
+        r = requests.get(f"{PBI_BASE}/groups/{workspace_id}/reports/{report_id}", headers=headers, timeout=timeout)
+        r.raise_for_status()
+        rep = r.json()
+    else:
+        r = requests.get(f"{PBI_BASE}/groups/{workspace_id}/reports", headers=headers, timeout=timeout)
+        r.raise_for_status()
+        items = r.json().get("value", [])
+        if not items:
+            raise RuntimeError("No reports found in workspace.")
+        rep = items[0]
+        report_id = rep["id"]
+
+    embed_url = rep["embedUrl"]
+
+    # Generate embed token (view access)
+    payload = {"accessLevel": "View"}
+    t = requests.post(
+        f"{PBI_BASE}/groups/{workspace_id}/reports/{report_id}/GenerateToken",
+        headers={**headers, "Content-Type": "application/json"},
+        json=payload,
+        timeout=timeout,
+    )
+    t.raise_for_status()
+    token_json = t.json()
+    embed_token = token_json["token"]
+    expires = token_json.get("expiration")
+
+    return embed_url, report_id, embed_token, expires
+
+
+def _log_token_claims_safe(access_token: str) -> None:
+    """
+    Debug-lite helper: log selected AAD JWT token claims without exposing the token.
+    Logs tid (tenant), appid (client), oid (object id), aud (audience), exp (UTC).
+    """
+    try:
+        parts = (access_token or "").split(".")
+        if len(parts) < 2:
+            logger.warning("AuthPowerBI: token not in JWT format; cannot decode claims")
+            return
+        # Add base64 padding if needed
+        pad = "=" * (-len(parts[1]) % 4)
+        payload = json.loads(base64.urlsafe_b64decode(parts[1] + pad))
+        exp = payload.get("exp", 0)
+        try:
+            exp_iso = datetime.utcfromtimestamp(exp).isoformat() + "Z"
+        except Exception:
+            exp_iso = str(exp)
+        logger.info(
+            "AuthPowerBI token claims: tid=%s appid=%s oid=%s aud=%s exp=%s",
+            payload.get("tid"),
+            payload.get("appid"),
+            payload.get("oid"),
+            payload.get("aud"),
+            exp_iso,
+        )
+    except Exception as exc:
+        logger.warning("AuthPowerBI: failed to decode token claims: %s", exc)
+
+
+class AuthPowerBI(APIView):
+    authentication_classes = (authentication.TokenAuthentication,)  # later to SessionAuthentication
+    permission_classes = (permissions.IsAuthenticated,)
+
+    def get(self, request):
+        # Try real Power BI via managed identity
+        # Accept config from settings or environment for parity with diagnostics command
+        workspace_id = getattr(settings, "POWERBI_WORKSPACE_ID", None) or os.getenv("POWERBI_WORKSPACE_ID")
+        # Support both POWERBI_REPORT_ID (preferred) and legacy REPORT_ID, plus env override
+        report_id_cfg = (
+            getattr(settings, "POWERBI_REPORT_ID", None) or getattr(settings, "REPORT_ID", None) or os.getenv("POWERBI_REPORT_ID")
+        )
+        access_token = _pbi_token_via_managed_identity()
+
+        # Optional debug-lite: log selected token claims and config when requested
+        debug_flag = str(request.query_params.get("debug", "")).lower() in {"1", "true", "yes", "on"}
+        if debug_flag:
+            logger.info(
+                "AuthPowerBI debug-lite enabled: workspace_id=%s report_id_cfg=%s has_token=%s",
+                workspace_id,
+                report_id_cfg,
+                bool(access_token),
+            )
+            if access_token:
+                _log_token_claims_safe(access_token)
+
+        if access_token and workspace_id:
+            try:
+                embed_url, report_id, embed_token, expires_at = _pbi_get_embed_info(access_token, workspace_id, report_id_cfg)
+                return Response(
+                    {
+                        "detail": "ok",
+                        "embed_url": embed_url,
+                        "report_id": report_id,
+                        "embed_token": embed_token,
+                        "expires_at": expires_at,
+                        "user": request.user.username,
+                    }
+                )
+            except Exception as e:
+                logger.exception("Power BI REST call failed, falling back to mock: %s", e)
+
+        # Fallback mock if not configured or failed
+        embed_token = secrets.token_hex(8)  # 16-char hex
+        embed_url = get_random_string(10)
+        report_id = secrets.randbelow(2_147_483_647) + 1
+        expires_at = (timezone.now() + timedelta(hours=1)).isoformat()
+
+        return Response(
+            {
+                "detail": "ok",
+                "embed_url": embed_url,
+                "report_id": report_id,
+                "embed_token": embed_token,
+                "expires_at": expires_at,
+                "user": request.user.username,
+            }
+        )

main/urls.py

@@ -30,6 +30,7 @@
     AggregateByTime,
     AggregateHeaderFigures,
     AreaAggregate,
+    AuthPowerBI,
     Brief,
     DelSubscription,
     DummyExceptionError,
@@ -241,6 +242,7 @@
     url(r"^api/v2/event/(?P<slug>[-\w]+)", api_views.EventViewset.as_view({"get": "retrieve"}, lookup_field="slug")),
     url(r"^api/v2/delegation-office/(?P<pk>\d+)", DelegationOfficeDetailAPIView.as_view()),
     url(r"^api/v2/delegation-office/", DelegationOfficeListAPIView.as_view()),
+    url(r"^api/v2/auth-power-bi/", AuthPowerBI.as_view(), name="auth_power_bi"),
     url(r"^tinymce/", include("tinymce.urls")),
     url(r"^$", RedirectView.as_view(url="/admin")),
     # url(r'^', admin.site.urls),