|
25 | 25 | # logging.basicConfig(level=logging.DEBUG) # Enable DEBUG log for entire script
|
26 | 26 | # logging.getLogger("msal").setLevel(logging.INFO) # Optionally disable MSAL DEBUG logs
|
27 | 27 |
|
| 28 | +def get_preexisting_rt_and_their_scopes_from_elsewhere(): |
| 29 | + # Maybe you have an ADAL-powered app like this |
| 30 | + # https://github.com/AzureAD/azure-activedirectory-library-for-python/blob/1.2.3/sample/device_code_sample.py#L72 |
| 31 | + # which uses a resource rather than a scope, |
| 32 | + # you need to convert your v1 resource into v2 scopes |
| 33 | + # See https://docs.microsoft.com/azure/active-directory/develop/azure-ad-endpoint-comparison#scopes-not-resources |
| 34 | + # You may be able to append "/.default" to your v1 resource to form a scope |
| 35 | + # See https://docs.microsoft.com/azure/active-directory/develop/v2-permissions-and-consent#the-default-scope |
| 36 | + |
| 37 | + # Or maybe you have an app already talking to Microsoft identity platform v2, |
| 38 | + # powered by some 3rd-party auth library, and persist its tokens somehow. |
| 39 | + |
| 40 | + # Either way, you need to extract RTs from there, and return them like this. |
| 41 | + return [ |
| 42 | + ("old_rt_1", ["scope1", "scope2"]), |
| 43 | + ("old_rt_2", ["scope3", "scope4"]), |
| 44 | + ] |
| 45 | + |
| 46 | + |
| 47 | +# We will migrate all the old RTs into a new app powered by MSAL |
28 | 48 | config = json.load(open(sys.argv[1]))
|
29 |
| - |
30 |
| -def get_rt_via_old_app(): |
31 |
| - # Let's pretend this is an old app powered by ADAL |
32 |
| - app = msal.PublicClientApplication( |
33 |
| - config["client_id"], authority=config["authority"]) |
34 |
| - flow = app.initiate_device_flow(scopes=config["scope"]) |
35 |
| - if "user_code" not in flow: |
36 |
| - raise ValueError( |
37 |
| - "Fail to create device flow. Err: %s" % json.dumps(flow, indent=4)) |
38 |
| - print(flow["message"]) |
39 |
| - sys.stdout.flush() # Some terminal needs this to ensure the message is shown |
40 |
| - |
41 |
| - # Ideally you should wait here, in order to save some unnecessary polling |
42 |
| - # input("Press Enter after signing in from another device to proceed, CTRL+C to abort.") |
43 |
| - |
44 |
| - result = app.acquire_token_by_device_flow(flow) # By default it will block |
45 |
| - assert "refresh_token" in result, "We should have a successful result" |
46 |
| - return result["refresh_token"] |
47 |
| - |
48 |
| -try: # For easier testing, we try to reload a RT from previous run |
49 |
| - old_rt = json.load(open("rt.json"))[0] |
50 |
| -except: # If that is not possible, we acquire a RT |
51 |
| - old_rt = get_rt_via_old_app() |
52 |
| - json.dump([old_rt], open("rt.json", "w")) |
53 |
| - |
54 |
| -# Now we will try to migrate this old_rt into a new app powered by MSAL |
55 |
| - |
56 |
| -token_cache = msal.SerializableTokenCache() |
57 |
| -assert token_cache.serialize() == '{}', "Token cache is initially empty" |
58 | 49 | app = msal.PublicClientApplication(
|
59 |
| - config["client_id"], authority=config["authority"], token_cache=token_cache) |
60 |
| -result = app.import_refresh_token(old_rt, config["scope"]) |
61 |
| -if "error" in result: |
62 |
| - print("Migration unsuccessful. Error: ", json.dumps(result, indent=2)) |
63 |
| -else: |
64 |
| - print("Migration is successful") |
65 |
| - logging.debug("Token cache contains: %s", token_cache.serialize()) |
66 |
| - |
67 |
| -# From now on, the RT is saved inside MSAL's cache, |
68 |
| -# and becomes available in normal MSAL coding pattern. For example: |
69 |
| -accounts = app.get_accounts() |
70 |
| -if accounts: |
71 |
| - account = accounts[0] # Assuming end user pick this account |
72 |
| - result = app.acquire_token_silent(config["scope"], account) |
73 |
| - if "access_token" in result: |
74 |
| - print("RT is available in MSAL's cache, and can be used to acquire new AT") |
75 |
| - |
| 50 | + config["client_id"], authority=config["authority"], |
| 51 | + # token_cache=... # Default cache is in memory only. |
| 52 | + # You can learn how to use SerializableTokenCache from |
| 53 | + # https://msal-python.rtfd.io/en/latest/#msal.SerializableTokenCache |
| 54 | + ) |
| 55 | + |
| 56 | +# We choose a migration strategy of migrating all RTs in one loop |
| 57 | +for old_rt, scopes in get_preexisting_rt_and_their_scopes_from_elsewhere(): |
| 58 | + result = app.acquire_token_by_refresh_token(old_rt, scopes) |
| 59 | + if "error" in result: |
| 60 | + print("Discarding unsuccessful RT. Error: ", json.dumps(result, indent=2)) |
| 61 | + |
| 62 | +print("Migration completed") |
| 63 | + |
| 64 | +# From now on, those successfully-migrated RTs are saved inside MSAL's cache, |
| 65 | +# and becomes available in normal MSAL coding pattern, which is NOT part of migration. |
| 66 | +# You can refer to: |
| 67 | +# https://github.com/AzureAD/microsoft-authentication-library-for-python/blob/1.2.0/sample/device_flow_sample.py#L42-L60 |
0 commit comments