codekingpro/portable-devtools

1# It is currently shipped inside msal library.2# Pros: It is always available wherever msal is installed.3# Cons: Its 3rd-party dependencies (if any) may become msal's dependency.4"""MSAL Python Tester5 6Usage 1: Run it on the fly.7    python -m msal8    Note: We choose to not define a console script to avoid name conflict.9 10Usage 2: Build an all-in-one executable file for bug bash.11    shiv -e msal.__main__._main -o msaltest-on-os-name.pyz .12"""13import base64, getpass, json, logging, sys, os, atexit, msal14 15_token_cache_filename = "msal_cache.bin"16global_cache = msal.SerializableTokenCache()17atexit.register(lambda:18    open(_token_cache_filename, "w").write(global_cache.serialize())19    # Hint: The following optional line persists only when state changed20    if global_cache.has_state_changed else None21    )22 23_AZURE_CLI = "04b07795-8ddb-461a-bbee-02f9e1bf7b46"24_VISUAL_STUDIO = "04f0c124-f2bc-4f59-8241-bf6df9866bbd"25placeholder_auth_scheme = msal.PopAuthScheme(26    http_method=msal.PopAuthScheme.HTTP_GET,27    url="https://example.com/endpoint",28    nonce="placeholder",29    )30 31def print_json(blob):32    print(json.dumps(blob, indent=2, sort_keys=True))33 34def _input_boolean(message):35    return input(36        "{} (N/n/F/f or empty means False, otherwise it is True): ".format(message)37        ) not in ('N', 'n', 'F', 'f', '')38 39def _input(message, default=None):40    return input(message.format(default=default)).strip() or default41 42def _select_options(43        options, header="Your options:", footer="    Your choice? ", option_renderer=str,44        accept_nonempty_string=False,45        ):46    assert options, "options must not be empty"47    if header:48        print(header)49    for i, o in enumerate(options, start=1):50        print("    {}: {}".format(i, option_renderer(o)))51    if accept_nonempty_string:52        print("    Or you can just type in your input.")53    while True:54        raw_data = input(footer)55        try:56            choice = int(raw_data)57            if 1 <= choice <= len(options):58                return options[choice - 1]59        except ValueError:60            if raw_data and accept_nonempty_string:61                return raw_data62 63enable_debug_log = _input_boolean("Enable MSAL Python's DEBUG log?")64logging.basicConfig(level=logging.DEBUG if enable_debug_log else logging.INFO)65try:66    from dotenv import load_dotenv67    load_dotenv()68    logging.info("Loaded environment variables from .env file")69except ImportError:70    logging.warning(71        "python-dotenv is not installed. "72        "You may need to set environment variables manually.")73 74def _input_scopes():75    scopes = _select_options([76        "https://graph.microsoft.com/.default",77        "https://management.azure.com/.default",78        "User.Read",79        "User.ReadBasic.All",80        ],81        header="Select a scope (multiple scopes can only be input by manually typing them, delimited by space):",82        accept_nonempty_string=True,83        ).split()  # It also converts the input string(s) into a list84    if "https://pas.windows.net/CheckMyAccess/Linux/.default" in scopes:85        raise ValueError("SSH Cert scope shall be tested by its dedicated functions")86    return scopes87 88def _select_account(app):89    accounts = app.get_accounts()90    if accounts:91        return _select_options(92            accounts,93            option_renderer=lambda a: "{}, came from {}".format(a["username"], a["account_source"]),94            header="Account(s) already signed in inside MSAL Python:",95            )96    else:97        print("No account available inside MSAL Python. Use other methods to acquire token first.")98 99def _acquire_token_silent(app):100    """acquire_token_silent() - with an account already signed into MSAL Python."""101    account = _select_account(app)102    if account:103        print_json(app.acquire_token_silent_with_error(104            _input_scopes(),105            account=account,106            force_refresh=_input_boolean("Bypass MSAL Python's token cache?"),107            auth_scheme=placeholder_auth_scheme108                if app.is_pop_supported() and _input_boolean("Acquire AT POP via Broker?")109                else None,110            ))111 112def _acquire_token_interactive(app, scopes=None, data=None):113    """acquire_token_interactive() - User will be prompted if app opts to do select_account."""114    assert isinstance(app, msal.PublicClientApplication)115    scopes = scopes or _input_scopes()  # Let user input scope param before less important prompt and login_hint116    prompt = _select_options([117        {"value": None, "description": "Unspecified. Proceed silently with a default account (if any), fallback to prompt."},118        {"value": "none", "description": "none. Proceed silently with a default account (if any), or error out."},119        {"value": "select_account", "description": "select_account. Prompt with an account picker."},120        ],121        option_renderer=lambda o: o["description"],122        header="Prompt behavior?")["value"]123    if prompt == "select_account":124        login_hint = None  # login_hint is unnecessary when prompt=select_account125    else:126        raw_login_hint = _select_options(127            [None] + [a["username"] for a in app.get_accounts()],128            header="login_hint? (If you have multiple signed-in sessions in browser/broker, and you specify a login_hint to match one of them, you will bypass the account picker.)",129            accept_nonempty_string=True,130            )131        login_hint = raw_login_hint["username"] if isinstance(raw_login_hint, dict) else raw_login_hint132    result = app.acquire_token_interactive(133        scopes,134        parent_window_handle=app.CONSOLE_WINDOW_HANDLE,  # This test app is a console app135        enable_msa_passthrough=app.client_id in [  # Apps are expected to set this right136            _AZURE_CLI, _VISUAL_STUDIO,137            ],  # Here this test app mimics the setting for some known MSA-PT apps138        port=1234,  # Hard coded for testing. Real app typically uses default value.139        prompt=prompt, login_hint=login_hint, data=data or {},140        auth_scheme=placeholder_auth_scheme141            if app.is_pop_supported() and _input_boolean("Acquire AT POP via Broker?")142            else None,143        )144    if login_hint and "id_token_claims" in result:145        signed_in_user = result.get("id_token_claims", {}).get("preferred_username")146        if signed_in_user != login_hint:147            logging.warning('Signed-in user "%s" does not match login_hint', signed_in_user)148    print_json(result)149    return result150 151def _acquire_token_by_username_password(app):152    """acquire_token_by_username_password() - See constraints here: https://docs.microsoft.com/en-us/azure/active-directory/develop/msal-authentication-flows#constraints-for-ropc"""153    print_json(app.acquire_token_by_username_password(154        _input("username: "), getpass.getpass("password: "), scopes=_input_scopes()))155 156def _acquire_token_by_device_flow(app):157    """acquire_token_by_device_flow() - Note that this one does not go through broker"""158    assert isinstance(app, msal.PublicClientApplication)159    flow = app.initiate_device_flow(scopes=_input_scopes())160    print(flow["message"])161    sys.stdout.flush()  # Some terminal needs this to ensure the message is shown162    input("After you completed the step above, press ENTER in this console to continue...")163    result = app.acquire_token_by_device_flow(flow)  # By default it will block164    print_json(result)165 166_JWK1 = """{"kty":"RSA", "n":"2tNr73xwcj6lH7bqRZrFzgSLj7OeLfbn8216uOMDHuaZ6TEUBDN8Uz0ve8jAlKsP9CQFCSVoSNovdE-fs7c15MxEGHjDcNKLWonznximj8pDGZQjVdfK-7mG6P6z-lgVcLuYu5JcWU_PeEqIKg5llOaz-qeQ4LEDS4T1D2qWRGpAra4rJX1-kmrWmX_XIamq30C9EIO0gGuT4rc2hJBWQ-4-FnE1NXmy125wfT3NdotAJGq5lMIfhjfglDbJCwhc8Oe17ORjO3FsB5CLuBRpYmP7Nzn66lRY3Fe11Xz8AEBl3anKFSJcTvlMnFtu3EpD-eiaHfTgRBU7CztGQqVbiQ", "e":"AQAB"}"""167_SSH_CERT_DATA = {"token_type": "ssh-cert", "key_id": "key1", "req_cnf": _JWK1}168_SSH_CERT_SCOPE = ["https://pas.windows.net/CheckMyAccess/Linux/.default"]169 170def _acquire_ssh_cert_silently(app):171    """Acquire an SSH Cert silently- This typically only works with Azure CLI"""172    assert isinstance(app, msal.PublicClientApplication)173    account = _select_account(app)174    if account:175        result = app.acquire_token_silent(176            _SSH_CERT_SCOPE,177            account,178            data=_SSH_CERT_DATA,179            force_refresh=_input_boolean("Bypass MSAL Python's token cache?"),180            )181        print_json(result)182        if result and result.get("token_type") != "ssh-cert":183            logging.error("Unable to acquire an ssh-cert.")184 185def _acquire_ssh_cert_interactive(app):186    """Acquire an SSH Cert interactively - This typically only works with Azure CLI"""187    assert isinstance(app, msal.PublicClientApplication)188    result = _acquire_token_interactive(app, scopes=_SSH_CERT_SCOPE, data=_SSH_CERT_DATA)189    if result.get("token_type") != "ssh-cert":190        logging.error("Unable to acquire an ssh-cert")191 192_POP_KEY_ID = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA-AAAAAAAA'  # Fake key with a certain format and length193_RAW_REQ_CNF = json.dumps({"kid": _POP_KEY_ID, "xms_ksl": "sw"})194_POP_DATA = {  # Sampled from Azure CLI's plugin connectedk8s195    'token_type': 'pop',196    'key_id': _POP_KEY_ID,197    "req_cnf": base64.urlsafe_b64encode(_RAW_REQ_CNF.encode('utf-8')).decode('utf-8').rstrip('='),198        # Note: Sending _RAW_REQ_CNF without base64 encoding would result in an http 500 error199}  # See also https://github.com/Azure/azure-cli-extensions/blob/main/src/connectedk8s/azext_connectedk8s/_clientproxyutils.py#L86-L92200 201def _acquire_pop_token_interactive(app):202    """Acquire a POP token interactively - This typically only works with Azure CLI"""203    assert isinstance(app, msal.PublicClientApplication)204    POP_SCOPE = ['6256c85f-0aad-4d50-b960-e6e9b21efe35/.default']  # KAP 1P Server App Scope, obtained from https://github.com/Azure/azure-cli-extensions/pull/4468/files#diff-a47efa3186c7eb4f1176e07d0b858ead0bf4a58bfd51e448ee3607a5b4ef47f6R116205    result = _acquire_token_interactive(app, scopes=POP_SCOPE, data=_POP_DATA)206    print_json(result)207    if result.get("token_type") != "pop":208        logging.error("Unable to acquire a pop token")209 210def _remove_account(app):211    """remove_account() - Invalidate account and/or token(s) from cache, so that acquire_token_silent() would be reset"""212    account = _select_account(app)213    if account:214        app.remove_account(account)215        print('Account "{}" and/or its token(s) are signed out from MSAL Python'.format(account["username"]))216 217def _acquire_token_for_client(app):218    """CCA.acquire_token_for_client() - Rerun this will get same token from cache."""219    assert isinstance(app, msal.ConfidentialClientApplication)220    print_json(app.acquire_token_for_client(scopes=_input_scopes()))221 222def _remove_tokens_for_client(app):223    """CCA.remove_tokens_for_client() - Run this to evict tokens from cache."""224    assert isinstance(app, msal.ConfidentialClientApplication)225    app.remove_tokens_for_client()226 227def _exit(app):228    """Exit"""229    bug_link = (230        "https://identitydivision.visualstudio.com/Engineering/_queries/query/79b3a352-a775-406f-87cd-a487c382a8ed/"231        if app._enable_broker else232        "https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/new/choose"233        )234    print("Bye. If you found a bug, please report it here: {}".format(bug_link))235    sys.exit()236 237def _main():238    print("Welcome to the Msal Python {} Tester (Experimental)\n".format(msal.__version__))239    cache_choice = _select_options([240            {241                "choice": "empty",242                "desc": "Start with an empty token cache. Suitable for one-off tests.",243            },244            {245                "choice": "reuse",246                "desc": "Reuse the previous token cache {} (if any) "247                    "which was created during last test app exit. "248                    "Useful for testing acquire_token_silent() repeatedly".format(249                        _token_cache_filename),250            },251        ],252        option_renderer=lambda o: o["desc"],253        header="What token cache state do you want to begin with?",254        accept_nonempty_string=False)255    if cache_choice["choice"] == "reuse" and os.path.exists(_token_cache_filename):256        try:257            global_cache.deserialize(open(_token_cache_filename, "r").read())258        except IOError:259            pass  # Use empty token cache260    chosen_app = _select_options([261        {"client_id": _AZURE_CLI, "name": "Azure CLI (Correctly configured for MSA-PT)"},262        {"client_id": _VISUAL_STUDIO, "name": "Visual Studio (Correctly configured for MSA-PT)"},263        {"client_id": "95de633a-083e-42f5-b444-a4295d8e9314", "name": "Whiteboard Services (Non MSA-PT app. Accepts AAD & MSA accounts.)"},264        {265            "client_id": os.getenv("CLIENT_ID"),266            "client_secret": os.getenv("CLIENT_SECRET"),267            "name": "A confidential client app (CCA) whose settings are defined "268                "in environment variables CLIENT_ID and CLIENT_SECRET",269        },270        ],271        option_renderer=lambda a: a["name"],272        header="Impersonate this app "273            "(or you can type in the client_id of your own public client app)",274        accept_nonempty_string=True)275    is_cca = isinstance(chosen_app, dict) and "client_secret" in chosen_app276    if is_cca and not (chosen_app["client_id"] and chosen_app["client_secret"]):277        raise ValueError("You need to set environment variables CLIENT_ID and CLIENT_SECRET")278    enable_broker = (not is_cca) and _input_boolean("Enable broker? "279        "(It will error out later if your app has not registered some redirect URI)"280        )281    enable_pii_log = _input_boolean("Enable PII in broker's log?") if enable_broker and enable_debug_log else False282    authority = _select_options([283        "https://login.microsoftonline.com/common",284        "https://login.microsoftonline.com/organizations",285        "https://login.microsoftonline.com/microsoft.onmicrosoft.com",286        "https://login.microsoftonline.com/msidlab4.onmicrosoft.com",287        "https://login.microsoftonline.com/consumers",288        ],289        header="Input authority (Note that MSA-PT apps would NOT use the /common authority)",290        accept_nonempty_string=True,291        )292    instance_discovery = _input_boolean(293        "You input an unusual authority which might fail the Instance Discovery. "294        "Now, do you want to perform Instance Discovery on your input authority?"295        ) if authority and not authority.startswith(296            "https://login.microsoftonline.com") else None297    app = msal.PublicClientApplication(298        chosen_app["client_id"] if isinstance(chosen_app, dict) else chosen_app,299        authority=authority,300        instance_discovery=instance_discovery,301        enable_broker_on_windows=enable_broker,302        enable_pii_log=enable_pii_log,303        token_cache=global_cache,304        ) if not is_cca else msal.ConfidentialClientApplication(305        chosen_app["client_id"],306        client_credential=chosen_app["client_secret"],307        authority=authority,308        instance_discovery=instance_discovery,309        enable_pii_log=enable_pii_log,310        token_cache=global_cache,311        )312    methods_to_be_tested = [313            _acquire_token_silent,314        ] + ([315            _acquire_token_interactive,316            _acquire_token_by_device_flow,317            _acquire_ssh_cert_silently,318            _acquire_ssh_cert_interactive,319            _acquire_pop_token_interactive,320            ] if isinstance(app, msal.PublicClientApplication) else []321        ) + [322            _acquire_token_by_username_password,323            _remove_account,324        ] + ([325            _acquire_token_for_client,326            _remove_tokens_for_client,327            ] if isinstance(app, msal.ConfidentialClientApplication) else []328        )329    while True:330        func = _select_options(331            methods_to_be_tested + [_exit],332            option_renderer=lambda f: f.__doc__, header="MSAL Python APIs:")333        try:334            func(app)335        except ValueError as e:336            logging.error("Invalid input: %s", e)337        except KeyboardInterrupt:  # Useful for bailing out a stuck interactive flow338            print("Aborted")339 340if __name__ == "__main__":341    _main()342 343