# type: ignore """keycloak_test_server.""" # ./bin/start_keycloak # starts keycloak on 8080 # pip install flask_oidc # pip install itsdangerous==2.0.1 # python ./bin/keycloak_test_server.py # starts flask on 5005 import json import logging import requests from flask import Flask from flask import g from flask_oidc import OpenIDConnect logging.basicConfig(level=logging.DEBUG) app = Flask(__name__) app.config.update( { "SECRET_KEY": "SomethingNotEntirelySecret", "TESTING": True, "DEBUG": True, "OIDC_CLIENT_SECRETS": "bin/keycloak_test_secrets.json", "OIDC_ID_TOKEN_COOKIE_SECURE": False, "OIDC_REQUIRE_VERIFIED_EMAIL": False, "OIDC_USER_INFO_ENABLED": True, "OIDC_OPENID_REALM": "flask-demo", "OIDC_SCOPES": ["openid", "email", "profile"], "OIDC_INTROSPECTION_AUTH_METHOD": "client_secret_post", } ) oidc = OpenIDConnect(app) @app.route("/") def hello_world(): """Hello_world.""" if oidc.user_loggedin: return ( 'Hello, %s, See private ' 'Log out' ) % oidc.user_getfield("preferred_username") else: return 'Welcome anonymous, Log in' @app.route("/private") @oidc.require_login def hello_me(): """Example for protected endpoint that extracts private information from the OpenID Connect id_token. Uses the accompanied access_token to access a backend service. """ info = oidc.user_getinfo(["preferred_username", "email", "sub"]) username = info.get("preferred_username") email = info.get("email") user_id = info.get("sub") if user_id in oidc.credentials_store: try: from oauth2client.client import OAuth2Credentials access_token = OAuth2Credentials.from_json( oidc.credentials_store[user_id] ).access_token print("access_token=<%s>" % access_token) headers = {"Authorization": "Bearer %s" % (access_token)} # YOLO greeting = requests.get( "http://localhost:8080/greeting", headers=headers ).text except BaseException: print("Could not access greeting-service") greeting = "Hello %s" % username return """{} your email is {} and your user_id is {}!
""".format( greeting, email, user_id, ) @app.route("/api", methods=["POST"]) @oidc.accept_token(require_token=True, scopes_required=["openid"]) def hello_api(): """OAuth 2.0 protected API endpoint accessible via AccessToken.""" return json.dumps({"hello": "Welcome %s" % g.oidc_token_info["sub"]}) @app.route("/logout") def logout(): """Performs local logout by removing the session cookie.""" oidc.logout() return 'Hi, you have been logged out! Return' if __name__ == "__main__": app.run(port=5005)