在 Python Flask App 中使用 Authlib 从 Okta 实施自省验证

Implementing Introspection Validation from Okta using Authlib in Python Flask App

我正在尝试在客户端实施自省,将 Okta 作为我的授权服务器,但不断出现错误 {"error": "missing_authorization", "error_description": "Missing \"Authorization\" in headers."}

我的实现


    class MyIntrospectTokenValidator(IntrospectTokenValidator):
        def introspect_token(self, token_string):
            print(f"Introspecting token {token_string}")
            url = f'{okta_keys.get("base_url")}/v1/introspect'
            data = {'token': token_string, 'token_type_hint': 'access_token'}
            auth = (okta_keys.get('client_id'), okta_keys.get('client_secret'))
            resp = requests.post(url, headers=headers, data=data, auth=auth)
            resp.raise_for_status()
            return resp.json()
    
    
    require_oauth = ResourceProtector()
    require_oauth.register_token_validator(MyIntrospectTokenValidator())
    
    okta = oauth.register(
        name='okta',
        client_id=secrets["internal_client_id"],
        client_secret=secrets["internal_client_secret"],
        access_token_url=f'{okta_keys.get("base_url")}/v1/token',
        authorize_url=f'{okta_keys.get("base_url")}/v1/authorize',
        api_base_url=f'{okta_keys.get("base_url")}',
        introspect=f'{okta_keys.get("base_url")}/v1/introspect',
        jwks_uri=f'{okta_keys.get("base_url")}/v1/keys',
        userinfo_endpoint=f'{okta_keys.get("base_url")}/v1/userinfo',
        client_kwargs={'scope': 'openid email profile'},
    )
    
    @app.route('/authorize', methods=["GET", "POST"])
    def authorize():
        _okta = oauth.create_client('okta')  # create the google oauth client
        token = _okta.authorize_access_token()  # Access token from google (needed to get user info)
        session.permanent = True  # make the session permanant so it keeps existing after broweser gets closed
        headers = {'Authorization': f'Bearer {token.get("access_token")}'}
        print(f"\n\n{headers}\n\n")
        return redirect(url_for('index', _external=True))
    
    @app.route('/oauth/hello-world-api', methods=["GET", "POST"])
    @require_oauth(['openid', 'email', 'profile'])
    def hello_world():
        return str('Hello World')

我已经尝试解决这个问题一段时间了,但未能成功

我发现了代码的问题,我只需要手动向我的 api

提供授权

这是代码

@app.route('/authorize', methods=["GET", "POST"])
def authorize():
    _okta = oauth.create_client('okta')  # create the google oauth client
    token = _okta.authorize_access_token()  # Access token from google (needed to get user info)
    session.permanent = True  # make the session permanant so it keeps existing after broweser gets closed
    headers = {'Authorization': f'Bearer {token.get("access_token")}'}
    url = url_for('hello_world', _external=True)
    r = requests.get(url, headers=headers)
    return redirect(url_for('index', _external=True))

之后,我从 Postman.

执行了 post 请求