Added auto renewal for access tokens

TeamsEnum.py

@@ -66,9 +66,8 @@ def enumerate_user(enum, email, accounttype, presence, outfile):
  else:
  fd = None
 
- accounttype, bearertoken, skypetoken, teams_enrolled = teamsenum.auth.do_logon(args)
-
- enum = TeamsUserEnumerator(skypetoken, bearertoken, teams_enrolled)
+ accounttype, bearertoken, skypetoken, teams_enrolled, refresh_token, auth_app, auth_metadata = teamsenum.auth.do_logon(args)
+ enum = TeamsUserEnumerator(skypetoken, bearertoken, teams_enrolled, refresh_token, auth_app, auth_metadata)
 
  if args.email:
  emails = [args.email]

teamsenum/auth.py

@@ -215,7 +215,7 @@ def logon_with_devicecode(auth_metadata):
 
  return result, app
 
-def logon_with_accesstoken(auth_metadata, app):
+def logon_with_accesstoken(auth_metadata, app, scope_list=None):
  """
  Attempts to log in based on an access token. This step is required to acquire a X-Skypetoken using the previously acquired Bearer token
 
@@ -226,10 +226,18 @@ def logon_with_accesstoken(auth_metadata, app):
  Access token (dict): An object containing access tokens
  """
 
+ scopes = []
+ if( scope_list ):
+ scopes = scope_list
+ elif( 'scope' in auth_metadata ):
+ scopes = auth_metadata['scope'].split(',')
+ else:
+ p_warn("Either scopes or authentication metadata need to be provided to authenticate with the existing token", exit=True)
+
  try:
  # Fetches cached logins
  accounts = app.get_accounts()
- result = app.acquire_token_silent(scopes=["service::api.fl.spaces.skype.com::MBI_SSL openid profile"], account=accounts[0])
+ result = app.acquire_token_silent(scopes=scopes, account=accounts[0])
  except Exception as err:
  p_warn("Error while authenticating: %s" % (err.args[0]), exit=True)
 
@@ -262,7 +270,6 @@ def check_token_format(accesstoken, skypetoken):
  else:
  p_warn("Unknown access token format - If this error was raised by mistake, please reach out", exit=True)
 
-
 def get_authentication_metadata(account_type, username):
  """
  Based on the account type, returns a set of distinct oauth scope, client_id and tenant id
@@ -341,14 +348,16 @@ def do_logon(args):
  else:
  p_warn(result.get("error_description"), exit=True)
 
+ refresh_token = result['refresh_token'] if 'refresh_token' in result else ''
+
  # Login successful, Token is retrieved
  p_success("Successfully retrieved access token")
  teams_enrolled = account_is_teams_enrolled(result["access_token"], account_type.get('type'))
 
  skypetoken = None
  if account_type.get('type') == "personal":
- result_tokenlogin = logon_with_accesstoken(auth_metadata, app)
+ result_tokenlogin = logon_with_accesstoken(auth_metadata, app, scope_list=["service::api.fl.spaces.skype.com::MBI_SSL openid profile"])
  skypetoken = get_skype_token(result_tokenlogin["access_token"])
  p_success("Successfully retrieved skype token")
 
- return account_type.get('type'), result["access_token"], skypetoken, teams_enrolled
+ return account_type.get('type'), result["access_token"], skypetoken, teams_enrolled, refresh_token, app, auth_metadata

teamsenum/enum.py

@@ -3,11 +3,12 @@
 import requests
 import json
 from teamsenum.utils import p_success, p_err, p_warn, p_normal, p_file
+from teamsenum.auth import logon_with_accesstoken
 
 class TeamsUserEnumerator:
  """ Class that handles enumeration of users that use Microsoft Teams either from a personal, or corporate account """
 
- def __init__(self, skypetoken, bearertoken, teams_enrolled):
+ def __init__(self, skypetoken, bearertoken, teams_enrolled, refresh_token, auth_app, auth_metadata):
  """
  Constructor that accepts authentication tokens for use during enumeration
 
@@ -22,6 +23,9 @@ def __init__(self, skypetoken, bearertoken, teams_enrolled):
  self.skypetoken = skypetoken
  self.bearertoken = bearertoken
  self.teams_enrolled = teams_enrolled
+ self.refresh_token = refresh_token
+ self.auth_app = auth_app
+ self.auth_metadata = auth_metadata
 
  def check_user(self, email, type, presence=False, outfile=None):
  """
@@ -41,7 +45,7 @@ def check_user(self, email, type, presence=False, outfile=None):
  elif type == "corporate":
  self.check_teams_user(email, presence, outfile)
 
- def check_teams_user(self, email, presence=False, outfile=None):
+ def check_teams_user(self, email, presence=False, outfile=None, recursive_call=False):
  """
  Checks the existence and properties of a user, using the teams.microsoft.com endpoint
 
@@ -75,7 +79,15 @@ def check_teams_user(self, email, presence=False, outfile=None):
  return
 
  if content.status_code == 401:
- p_warn("Unable to enumerate user. Is the access token valid?", exit=True)
+ if( not recursive_call and self.refresh_token ):
+ p_warn("Unable to enumerate user. Trying to get a new access token...")
+ result = logon_with_accesstoken(self.auth_metadata, self.auth_app)
+ if( 'access_token' in result ):
+ p_warn("Got new access token. Rechecking the user...")
+ self.bearertoken = result['access_token']
+ return self.check_teams_user(email, presence=presence, outfile=outfile, recursive_call=True)
+ else:
+ p_warn("Unable to enumerate user. Is the access token valid?", exit=True)
 
  if content.status_code != 200:
  p_warn("Unable to enumerate user %s. Invalid target email address?" % (email))