Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_cache_key(entry):
return AdalTokenCacheKey(
entry.get(TokenResponseFields._AUTHORITY),
entry.get(TokenResponseFields.RESOURCE),
entry.get(TokenResponseFields._CLIENT_ID),
entry.get(TokenResponseFields.USER_ID))
# pylint: enable=protected-access
from . import log
from . import util
from .constants import OAuth2, TokenResponseFields, IdTokenFields
from .adal_error import AdalError
TOKEN_RESPONSE_MAP = {
OAuth2.ResponseParameters.TOKEN_TYPE : TokenResponseFields.TOKEN_TYPE,
OAuth2.ResponseParameters.ACCESS_TOKEN : TokenResponseFields.ACCESS_TOKEN,
OAuth2.ResponseParameters.REFRESH_TOKEN : TokenResponseFields.REFRESH_TOKEN,
OAuth2.ResponseParameters.CREATED_ON : TokenResponseFields.CREATED_ON,
OAuth2.ResponseParameters.EXPIRES_ON : TokenResponseFields.EXPIRES_ON,
OAuth2.ResponseParameters.EXPIRES_IN : TokenResponseFields.EXPIRES_IN,
OAuth2.ResponseParameters.RESOURCE : TokenResponseFields.RESOURCE,
OAuth2.ResponseParameters.ERROR : TokenResponseFields.ERROR,
OAuth2.ResponseParameters.ERROR_DESCRIPTION : TokenResponseFields.ERROR_DESCRIPTION,
}
_REQ_OPTION = {'headers' : {'content-type': 'application/x-www-form-urlencoded'}}
_ERROR_TEMPLATE = u"{} request returned http error: {}"
def map_fields(in_obj, map_to):
return dict((map_to[k], v) for k, v in in_obj.items() if k in map_to)
def _get_user_id(id_token):
user_id = None
is_displayable = False
if id_token.get('upn'):
user_id = id_token['upn']
is_displayable = True
def _get_header_from_dict(token):
return _get_header(token[TokenResponseFields.TOKEN_TYPE], token[TokenResponseFields.ACCESS_TOKEN])
from . import oauth2_client
from . import self_signed_jwt
from . import user_realm
from . import wstrust_request
from .adal_error import AdalError
from .cache_driver import CacheDriver
from .constants import WSTrustVersion
OAUTH2_PARAMETERS = constants.OAuth2.Parameters
TOKEN_RESPONSE_FIELDS = constants.TokenResponseFields
OAUTH2_GRANT_TYPE = constants.OAuth2.GrantType
OAUTH2_SCOPE = constants.OAuth2.Scope
OAUTH2_DEVICE_CODE_RESPONSE_PARAMETERS = constants.OAuth2.DeviceCodeResponseParameters
SAML = constants.Saml
ACCOUNT_TYPE = constants.UserRealm.account_type
USER_ID = constants.TokenResponseFields.USER_ID
_CLIENT_ID = constants.TokenResponseFields._CLIENT_ID #pylint: disable=protected-access
def add_parameter_if_available(parameters, key, value):
if value:
parameters[key] = value
def _get_saml_grant_type(wstrust_response):
token_type = wstrust_response.token_type
if token_type == SAML.TokenTypeV1 or token_type == SAML.OasisWssSaml11TokenProfile11:
return OAUTH2_GRANT_TYPE.SAML1
elif token_type == SAML.TokenTypeV2 or token_type == SAML.OasisWssSaml2TokenProfile2:
return OAUTH2_GRANT_TYPE.SAML2
else:
raise AdalError("RSTR returned unknown token type: {}".format(token_type))
try:
from urllib.parse import urlencode, urlparse
except ImportError:
from urllib import urlencode # pylint: disable=no-name-in-module
from urlparse import urlparse # pylint: disable=import-error,ungrouped-imports
import requests
from . import log
from . import util
from .constants import OAuth2, TokenResponseFields, IdTokenFields
from .adal_error import AdalError
TOKEN_RESPONSE_MAP = {
OAuth2.ResponseParameters.TOKEN_TYPE : TokenResponseFields.TOKEN_TYPE,
OAuth2.ResponseParameters.ACCESS_TOKEN : TokenResponseFields.ACCESS_TOKEN,
OAuth2.ResponseParameters.REFRESH_TOKEN : TokenResponseFields.REFRESH_TOKEN,
OAuth2.ResponseParameters.CREATED_ON : TokenResponseFields.CREATED_ON,
OAuth2.ResponseParameters.EXPIRES_ON : TokenResponseFields.EXPIRES_ON,
OAuth2.ResponseParameters.EXPIRES_IN : TokenResponseFields.EXPIRES_IN,
OAuth2.ResponseParameters.RESOURCE : TokenResponseFields.RESOURCE,
OAuth2.ResponseParameters.ERROR : TokenResponseFields.ERROR,
OAuth2.ResponseParameters.ERROR_DESCRIPTION : TokenResponseFields.ERROR_DESCRIPTION,
}
_REQ_OPTION = {'headers' : {'content-type': 'application/x-www-form-urlencoded'}}
_ERROR_TEMPLATE = u"{} request returned http error: {}"
def map_fields(in_obj, map_to):
return dict((map_to[k], v) for k, v in in_obj.items() if k in map_to)
def _argument_entry_with_cached_metadata(self, entry):
if _entry_has_metadata(entry):
return
if _is_mrrt(entry):
self._log.debug('Added entry is MRRT')
entry[TokenResponseFields.IS_MRRT] = True
else:
entry[TokenResponseFields.RESOURCE] = self._resource
entry[TokenResponseFields._CLIENT_ID] = self._client_id
entry[TokenResponseFields._AUTHORITY] = self._authority
def _query_cache(self, is_mrrt, user_id, client_id):
matches = []
for k in self._cache:
v = self._cache[k]
#None value will be taken as wildcard match
#pylint: disable=too-many-boolean-expressions
if ((is_mrrt is None or is_mrrt == v.get(TokenResponseFields.IS_MRRT)) and
(user_id is None or _string_cmp(user_id, v.get(TokenResponseFields.USER_ID))) and
(client_id is None or _string_cmp(client_id, v.get(TokenResponseFields._CLIENT_ID)))):
matches.append(v)
return matches
def _acquire_authorization_header(self):
if self._authentication_method is AuthenticationMethod.aad_token:
return _get_header("Bearer", self._token)
token = self._adal_context.acquire_token(self._kusto_cluster, self._username, self._client_id)
if token is not None:
expiration_date = dateutil.parser.parse(token[TokenResponseFields.EXPIRES_ON])
if expiration_date > datetime.now() + timedelta(minutes=1):
return _get_header_from_dict(token)
if TokenResponseFields.REFRESH_TOKEN in token:
token = self._adal_context.acquire_token_with_refresh_token(
token[TokenResponseFields.REFRESH_TOKEN], self._client_id, self._kusto_cluster
)
if token is not None:
return _get_header_from_dict(token)
if self._authentication_method is AuthenticationMethod.aad_username_password:
token = self._adal_context.acquire_token_with_username_password(
self._kusto_cluster, self._username, self._password, self._client_id
)
elif self._authentication_method is AuthenticationMethod.aad_application_key:
token = self._adal_context.acquire_token_with_client_credentials(
self._kusto_cluster, self._client_id, self._client_secret
)
elif self._authentication_method is AuthenticationMethod.aad_device_login:
code = self._adal_context.acquire_user_code(self._kusto_cluster, self._client_id)
print(code[OAuth2DeviceCodeResponseParameters.MESSAGE])
webbrowser.open(code[OAuth2DeviceCodeResponseParameters.VERIFICATION_URL])
def _create_entry_from_refresh(self, entry, refresh_response):
new_entry = copy.deepcopy(entry)
new_entry.update(refresh_response)
# It is possible the response payload has no 'resource' field, like in ADFS, so we manually
# fill it here. Note, 'resource' is part of the token cache key, so we have to set it to avoid
# corrupting the cache.
if 'resource' not in refresh_response:
new_entry['resource'] = self._resource
if entry[TokenResponseFields.IS_MRRT] and self._authority != entry[TokenResponseFields._AUTHORITY]:
new_entry[TokenResponseFields._AUTHORITY] = self._authority
self._log.debug('Created new cache entry from refresh response.')
return new_entry
def _get_cache_key(entry):
return AdalTokenCacheKey(
entry.get(TokenResponseFields._AUTHORITY),
entry.get(TokenResponseFields.RESOURCE),
entry.get(TokenResponseFields._CLIENT_ID),
entry.get(TokenResponseFields.USER_ID))
# pylint: enable=protected-access