I have an Rest API made in flask that is registered in Azure, My API allows logging in through Azure credentials, but when I go to generate an access token to access the powerbi reports through user's access_token logged in gives a status code 401, specifically this error:
Unauthorized
Error while retrieving Embed token Unauthorized: {"error":{"code":"PowerBINotAuthorizedException","pbi.error":{"code":"PowerBINotAuthorizedException","parameters":{},"details":[],"exceptionCulprit":1}}}
Status Code: 401
RequestId: d2d91052-9149-46e5-ba5b-551c18d82f09
I use the msal library.
In the API I have the following code:
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'
Basically what it does is log in with Microsoft credentials and then generate access token for viewing PowerBi reports. In such a way that when you request the token from powerBI, that is, when you make a request.pos to https://api.powerbi.com/v1.0/myorg/GenerateToken, you enter the abort.
I have the permissions that the API has in Azure:
If anyone knows or can solve this problem for me, I would appreciate it. Thanks in advance.
The below code flask script is designed to facilitate the retrieval of an embed token for embedding Power BI reports
Permission required :
Content.Create
Dataset.Read.All
Report.Read.All
Workspace.Read.All
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")