oauth2session dynamics365：不断收到错误401未经授权

问题描述

用例

我正在尝试连接到Microsoft Dynamics 365-现场服务。

我正在使用Python，Falsk和OAuth2Session执行Oauth2身份验证

我已经在Azure上设置了Azure应用。

错误消息

我一直收到HTTP错误401

谁能帮助我？

代码：config.py

"""Configuration settings for running the Python auth samples locally.

In a production deployment,this information should be saved in a database or
other secure storage mechanism.
"""

import os
from dotenv import load_dotenv

load_dotenv()

CLIENT_ID = os.environ["dynamics365_field_service_application_client_id"]
CLIENT_SECRET = os.environ["dynamics365_field_service_client_secrets"]
REDIRECT_URI = os.environ["dynamics365_field_service_redirect_path"]

# AUTHORITY_URL ending determines type of account that can be authenticated:
# /organizations = organizational accounts only
# /consumers = MSAs only (Microsoft Accounts - Live.com,Hotmail.com,etc.)
# /common = allow both types of accounts

AUTHORITY_URL = os.environ["dynamics365_field_service_authority"]
AUTHORIZATION_BASE_URL = os.environ["dynamics365_field_service_authorization_base_url"]
TOKEN_URL = os.environ["dynamics365_field_service_token_url"]

AUTH_ENDPOINT = "/oauth2/v2.0/authorize"


RESOURCE = "https://graph.microsoft.com/"
API_VERSION = os.environ["dynamics365_field_service_version"]
SCOPES = [
    "https://admin.services.crm.dynamics.com/user_impersonation"
]




#     "https://dynamics.microsoft.com/business-central/overview/user_impersonation",#     "https://graph.microsoft.com/email",#     "https://graph.microsoft.com/offline_access",#     "https://graph.microsoft.com/openid",#     "https://graph.microsoft.com/profile",#     "https://graph.microsoft.com/User.Read",#     "https://graph.microsoft.com/User.ReadBasic.All"
# ]  # Add other scopes/permissions as needed.

代码：dynamics365_flask_oauth2.py

# *-* coding:utf-8 *-*

# See https://requests-oauthlib.readthedocs.io/en/latest/index.html

from requests_oauthlib import OAuth2Session
from flask import Flask,request,redirect,session,url_for
from flask.json import jsonify
import os

import flaskr.library.dynamics365.field_service.config as config

app = Flask(__name__)
app.secret_key = os.urandom(24)

# This information is obtained upon registration of a new dynamics
# client_id = "<your client key>"
# client_secret = "<your client secret>"
# authorization_base_url = 'https://dynamics.com/login/oauth/authorize'
# token_url = 'https://dynamics.com/login/oauth/access_token'


@app.route("/")
def index():
    """Step 1: User Authorization.

    Redirect the user/resource owner to the OAuth provider (i.e. dynamics)
    using an URL with a few key OAuth parameters.
    """
    dynamics = OAuth2Session(
        config.CLIENT_ID,scope=config.SCOPES,redirect_uri=config.REDIRECT_URI
    )
    authorization_url,state = dynamics.authorization_url(config.AUTHORIZATION_BASE_URL)

    # State is used to prevent CSRF,keep this for later.
    session["oauth_state"] = state

    print(f"Please go here and authorize : {authorization_url}")
    return redirect(authorization_url)


# Step 2: User authorization,this happens on the provider.


@app.route("/login/authorized",methods=["GET"])  # callback
def callback():
    """ Step 3: Retrieving an access token.

    The user has been redirected back from the provider to your registered
    callback URL. With this redirection comes an authorization code included
    in the redirect URL. We will use that to obtain an access token.
    """

    if session.get("oauth_state") is None:
        return redirect(url_for(".index"))

    dynamics = OAuth2Session(
        config.CLIENT_ID,state=session["oauth_state"],redirect_uri=config.REDIRECT_URI
    )

    token = dynamics.fetch_token(
        token_url=config.TOKEN_URL,client_secret=config.CLIENT_SECRET,authorization_response=request.url,)
    print(f"token: {token}")
    # At this point you can fetch protected resources but lets save
    # the token and show how this is done from a persisted token
    # in /profile.
    session["oauth_token"] = token

    # return redirect(url_for(".dynamics_get_accounts_postman"))

    return redirect(url_for(".dynamics_get_accounts"))


@app.route("/profile",methods=["GET"])
def profile():
    """Fetching a protected resource using an OAuth 2 token.
    """
    dynamics = OAuth2Session(config.CLIENT_ID,token=session["oauth_token"])
    return jsonify(dynamics.get("https://api.dynamics.com/user").json())


@app.route("/get_accounts")
def dynamics_get_accounts():

    if session.get("oauth_token") is None:
        return redirect(url_for(".index"))

    dynamics = OAuth2Session(
        client_id=config.CLIENT_ID,# token="Bearer " + session["oauth_token"]["access_token"]
        token=session["oauth_token"],)

    result = dynamics.get("https://{env_name}.{region}.dynamics.com/api/data/v9.0")

    if result.status_code != 200:
        result = {"status code": result.status_code,"reason": result.reason}
    else:
        result = result.json()

    result = jsonify(result)
    return result


import requests


@app.route("/dynamics_get_accounts_postman")
def dynamics_get_accounts_postman():

    if session.get("oauth_token") is None:

        return redirect(url_for(".index"))

    url = "https://{env_name}.{region}.dynamics.com/api/data/v9.0/accounts"

    payload = {}
    headers = {
        "Accept": "application/json","OData-MaxVersion": "4.0","OData-Version": "4.0","If-None-Match": "null","Authorization": f'Bearer {session["oauth_token"]["access_token"]}',}

    response = requests.request("GET",url,headers=headers,data=payload)

    result = response.text.encode("utf8")
    print(f"result : {result}")

    return jsonify(result)


if __name__ == "__main__":
    # This allows us to use a plain HTTP callback
    os.environ["DEBUG"] = "1"

    app.secret_key = os.urandom(24)
    app.run(debug=True)

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）