问题描述

用例

我正在尝试连接到 Microsoft Dynamics 365 - Field Service。

我正在使用 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)

标签： python-3.xflaskazure-active-directorymicrosoft-dynamics

python-3.x - oauth2session dynamics365：不断收到错误 401 未授权

问题描述

用例

错误信息

代码：config.py

代码：dynamics365_flask_oauth2.py

解决方案

推荐阅读