401 未经授权 https://api.powerbi.com/v1.0/myorg/GenerateToken

Question

我有一个在Azure中注册的flask中制作的Rest API，我的API允许通过Azure凭据登录，但是当我生成访问令牌以通过用户登录的access_token访问powerbi报告时会给出状态代码 401，具体是这个错误：

未经授权

检索嵌入令牌未经授权时出错： {"error":{"code":"PowerBINotAuthorizedException","pbi.error":{"code":"PowerBINotAuthorizedException","parameters":{},"details": [],"异常罪魁祸首":1}}}

状态代码：401

请求ID：d2d91052-9149-46e5-ba5b-551c18d82f09

我使用 msal 库。

在 API 中我有以下代码：

import json
from flask import Flask, render_template, request, Response, jsonify, redirect, url_for,flash,session,url_for, make_response, abort
from config import BROKER_URL, RESULT_BACKEND, SECRET_KEY, REDIRECT_PATH, CLIENT_ID, AUTHORITY, CLIENT_SECRET, TENANT_ID, ISSUER, SCOPE
from msal import ConfidentialClientApplication

@app.route('/login_microsft')
def login_microsoft():
    return redirect(AUTHORITY + "/oauth2/v2.0/authorize?" +
                    "response_type=code&client_id=" + CLIENT_ID +
                    "&redirect_uri=" + REDIRECT_PATH +
                    "&scope=User.Read")

class EmbedTokenRequestBody:

    datasets = None
    reports = None
    targetWorkspaces = None

    def __init__(self):
        self.datasets = []
        self.reports = []
        self.targetWorkspaces = []

# autorización Microsoft
@app.route('/authorize')
def authorize():

    code = request.args.get('code')  # El código de autorización que Microsoft envía a esta URL
    if code:
        clientapp = ConfidentialClientApplication(CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET)
        token_response = clientapp.acquire_token_by_authorization_code(code, SCOPE, redirect_uri=REDIRECT_PATH)
        auth_result =token_response
        print('auth_result:',auth_result)

        group_id = "id_del_grupo"
        report_id = "id_del_informe"
        
        cabecera = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + auth_result['access_token']}
        request_body = EmbedTokenRequestBody()

        request_body.reports.append({'id': report_id})
        request_body.targetWorkspaces.append({'id': group_id})

        embed_token_api = 'https://api.powerbi.com/v1.0/myorg/GenerateToken'
        api_response = requests.post(embed_token_api, data=json.dumps(request_body.__dict__), headers=cabecera)

        if api_response.status_code != 200:
            abort(api_response.status_code, description=f'Error while retrieving Embed token\n{api_response.reason}:\t{api_response.text}\nStatus Code:\t{api_response.status_code}\nRequestId:\t{api_response.headers.get("RequestId")}')
        

        # localiza el usuario en la base de datos
        usuarios= db['users']
        email = auth_result['id_token_claims']['preferred_username']
        user = usuarios.find_one({"email": email})
        if user:

            print("Usuario encontrado: " + str(user))  # imprimir usuario

            session['username'] = str(user["first_name"])
            session['position'] = str(user["position"])
            if(user['picture']):
                session['picture'] = user['picture']

            access_token = create_access_token(identity=str(user["_id"]))

            print('access_token:',access_token)

            response = make_response(redirect(url_for('home', _external=True)))

            # Establece las cookies de acceso
            response.set_cookie(
                'access_token',  # Nombre de la cookie
                value=access_token,  # Valor de la cookie, el token JWT
                httponly=True,  # Marcar la cookie como HTTP-Only
                secure=False,  # Marcar la cookie como segura (enviada solo a través de HTTPS)
                samesite='Lax'  # Política de SameSite permite que la cookie se envíe con solicitudes de navegación de nivel superior desde un origen externo
            )
            response.set_cookie(
                'access_token_ms',  # Nombre de la cookie
                value=auth_result['access_token'],  # Valor de la cookie
                httponly=True,  # Marcar la cookie como HTTP-Only
                secure=False,  # Marcar la cookie como segura (enviada solo a través de HTTPS)
                samesite='Lax'  # Política de SameSite permite que la cookie se envíe con solicitudes de navegación de nivel superior desde un origen externo
            )
            print('response:',response)
            print('status_code:',response.status_code)
            print('respuesta:',response.response)
            return response
        
    return 'Error: no se recibió el código de autorización'

它的作用基本上是使用 Microsoft 凭据登录，然后生成用于查看 PowerBi 报告的访问令牌。这样，当您从 powerBI 请求令牌时，即当您向 https://api.powerbi.com/v1.0/myorg/GenerateToken 发出 request.pos 时，您输入 中止。

我拥有 API 在 Azure 中拥有的权限：

如果有人知道或可以为我解决这个问题，我将不胜感激。预先感谢。

Answer 1

下面的代码 Flask 脚本旨在促进检索用于嵌入 Power BI 报告的嵌入令牌

需要许可：

• 内容.创建

  • 权限类型：委派
  • 描述：允许用户在 Power BI 服务中创建内容。
  • 已授予 Contoso：否

• 数据集.读取.全部

  • 权限类型：委派
  • 描述：授予查看所有数据集的能力。
  • 已授予 Contoso：否

• 报告。阅读全部

  • 权限类型：委派
  • 描述：启用需要所有报告读取权限的 API 调用。
  • 已授予 Contoso：否

• 工作区.阅读.全部

  • 权限类型：委派
  • 描述：提供查看所有工作区的功能。
  • 已授予 Contoso：否

from flask import Flask, request, jsonify
import json
import requests
from msal import ConfidentialClientApplication

app = Flask(__name__)

CLIENT_ID =  'CLIENT_ID'
CLIENT_SECRET = 'CLIENT_SECRET'
AUTHORITY = 'https://login.microsoftonline.com/tenantiD'
REDIRECT_PATH = 'https://jwt.ms'  # Update with your redirect URI
SCOPE = ['https://analysis.windows.net/powerbi/api/.default']

class EmbedTokenRequestBody:
    datasets = None
    reports = None
    targetWorkspaces = None

    def __init__(self):
        self.datasets = []
        self.reports = []
        self.targetWorkspaces = []

AUTHORIZE_URL = f'{AUTHORITY}/oauth2/v2.0/authorize'

def get_authorization_url(client_id, redirect_uri, scope):
    return f'{AUTHORIZE_URL}?response_type=code&client_id={client_id}&redirect_uri={redirect_uri}&scope={"%20".join(scope)}'

def get_access_token(authorization_code):
    clientapp = ConfidentialClientApplication(CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET)
    token_response = clientapp.acquire_token_by_authorization_code(authorization_code, scopes=SCOPE, redirect_uri=REDIRECT_PATH)
    return token_response.get('access_token')

def generate_embed_token(access_token, group_id, report_id):
    request_body = {
        "accessLevel": "View",
        "allowSaveAs": True
    }

    headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + access_token}
    embed_token_api = f'https://api.powerbi.com/v1.0/myorg/groups/{group_id}/reports/{report_id}/GenerateToken'
    api_response = requests.post(embed_token_api, data=json.dumps(request_body), headers=headers)

    if api_response.status_code != 200:
        error_data = api_response.json()
        error_code = error_data.get('error', {}).get('code', '')
        error_message = error_data.get('error', {}).get('message', api_response.reason)
        raise Exception(f"Error while retrieving Embed token: {error_message} (Error Code: {error_code})")

    try:
        embed_token = api_response.json().get('token')
        return embed_token
    except json.JSONDecodeError:
        raise Exception("Error: Invalid JSON response received from the server")

@app.route('/')
def index():
    authorization_url = get_authorization_url(CLIENT_ID, REDIRECT_PATH, SCOPE)
    return f"<a href='{authorization_url}'>Click here to authenticate and get authorization code</a>"

@app.route('/auth_callback')
def auth_callback():
    authorization_code = request.args.get('code')
    access_token = get_access_token(authorization_code)
    if access_token:
        group_id = 'groupid'  # Replace with your group ID
        report_id = 'reportid'  # Replace with your report ID
        try:
            embed_token = generate_embed_token(access_token, group_id, report_id)
            print("\nAccess token:", access_token)  # Print access token to console
            print("Embed token:", embed_token)  # Print embed token to console
            return jsonify({"embed_token": embed_token})
        except Exception as e:
            print("Error:", e)  # Print to console
            return jsonify({"error": str(e)}), 500
    else:
        return "Error: No authorization code received", 400

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000, debug=True)

import json
import requests
from msal import ConfidentialClientApplication

CLIENT_ID = 'CLIENT_ID'
CLIENT_SECRET = 'CLIENT_SECRET'
AUTHORITY = 'https://login.microsoftonline.com/tenantiD'
REDIRECT_PATH = 'https://jwt.ms'  # Update with your redirect URI
SCOPE = ['https://analysis.windows.net/powerbi/api/.default']

class EmbedTokenRequestBody:
    datasets = None
    reports = None
    targetWorkspaces = None

    def __init__(self):
        self.datasets = []
        self.reports = []
        self.targetWorkspaces = []

AUTHORIZE_URL = f'{AUTHORITY}/oauth2/v2.0/authorize'

def get_authorization_url(client_id, redirect_uri, scope):
    return f'{AUTHORIZE_URL}?response_type=code&client_id={client_id}&redirect_uri={redirect_uri}&scope={"%20".join(scope)}'

def get_access_token(authorization_code):
    clientapp = ConfidentialClientApplication(CLIENT_ID, authority=AUTHORITY, client_credential=CLIENT_SECRET)
    token_response = clientapp.acquire_token_by_authorization_code(authorization_code, scopes=SCOPE, redirect_uri=REDIRECT_PATH)
    return token_response.get('access_token')

def generate_embed_token(access_token, group_id, report_id):
    request_body = {
        "accessLevel": "View",
        "allowSaveAs": True
    }

    headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + access_token}
    embed_token_api = f'https://api.powerbi.com/v1.0/myorg/groups/{group_id}/reports/{report_id}/GenerateToken'
    api_response = requests.post(embed_token_api, data=json.dumps(request_body), headers=headers)

    if api_response.status_code != 200:
        error_data = api_response.json()
        error_code = error_data.get('error', {}).get('code', '')
        error_message = error_data.get('error', {}).get('message', api_response.reason)
        raise Exception(f"Error while retrieving Embed token: {error_message} (Error Code: {error_code})")

    try:
        embed_token = api_response.json().get('token')
        return embed_token
    except json.JSONDecodeError:
        raise Exception("Error: Invalid JSON response received from the server")

# Example usage:
authorization_url = get_authorization_url(CLIENT_ID, REDIRECT_PATH, SCOPE)
print("Authorization URL:", authorization_url)

authorization_code = input("Enter authorization code: ")
access_token = get_access_token(authorization_code)
if access_token:
    print("\nAccess token:", access_token)
    
    group_id = 'groupid'  # Replace with your group ID
    report_id = 'reportid'  # Replace with your report ID
    
    try:
        embed_token = generate_embed_token(access_token, group_id, report_id)
        print("\nEmbed token:", embed_token)
    except Exception as e:
        print("Error:", e)
else:
    print("Error: No authorization code received")