Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:return: response status code and message
(200, uuid) or (404, ticket)
"""
data = {
"idp": issuer,
"id": id,
"redirect_endpoint": "%s/account_linking%s" % (self.base_url, self.endpoint)
}
jws = JWS(json.dumps(data), alg=self.signing_key.alg).sign_compact([self.signing_key])
try:
request = "{}/get_id?jwt={}".format(self.api_url, jws)
response = requests.get(request)
except Exception as con_exc:
msg = "Could not connect to account linking service"
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.critical(logline)
raise SATOSAAuthenticationError(context.state, msg) from con_exc
if response.status_code not in [200, 404]:
msg = "Got status code '{}' from account linking service".format(response.status_code)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.critical(logline)
raise SATOSAAuthenticationError(context.state, msg)
return response.status_code, response.text
self._check_error_response(authn_resp, context)
access_token, id_token_claims = self._get_tokens(authn_resp, context)
if id_token_claims:
self._verify_nonce(id_token_claims["nonce"], context)
else:
id_token_claims = {}
userinfo = {}
if access_token:
# make userinfo request
userinfo = self._get_userinfo(authn_resp["state"], context)
if not id_token_claims and not userinfo:
msg = "No id_token or userinfo, nothing to do.."
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.error(logline)
raise SATOSAAuthenticationError(context.state, "No user info available.")
all_user_claims = dict(list(userinfo.items()) + list(id_token_claims.items()))
msg = "UserInfo: {}".format(all_user_claims)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
del context.state[self.name]
internal_resp = self._translate_response(all_user_claims, self.client.authorization_endpoint)
return self.auth_callback_func(context, internal_resp)
:rtype: http.cookies.SimpleCookie
:param state: The state to save
:param name: Name identifier of the cookie
:param path: Endpoint path the cookie will be associated to
:param encryption_key: Key to encrypt the state information
:return: A cookie
"""
cookie_data = "" if state.delete else state.urlstate(encryption_key)
max_age = 0 if state.delete else STATE_COOKIE_MAX_AGE
msg = "Saving state as cookie, secure: {secure}, max-age: {max_age}, path: {path}".format(
secure=STATE_COOKIE_SECURE, max_age=STATE_COOKIE_MAX_AGE, path=path
)
logline = lu.LOG_FMT.format(id=lu.get_session_id(state), message=msg)
logger.debug(logline)
cookie = SimpleCookie()
cookie[name] = cookie_data
cookie[name]["secure"] = STATE_COOKIE_SECURE
cookie[name]["path"] = path
cookie[name]["max-age"] = max_age
return cookie
def constructPrimaryIdentifier(self, data, ordered_identifier_candidates):
"""
Construct and return a primary identifier value from the
data asserted by the IdP using the ordered list of candidates
from the configuration.
"""
logprefix = PrimaryIdentifier.logprefix
context = self.context
attributes = data.attributes
msg = "{} Input attributes {}".format(logprefix, attributes)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
value = None
for candidate in ordered_identifier_candidates:
msg = "{} Considering candidate {}".format(logprefix, candidate)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
# Get the values asserted by the IdP for the configured list of attribute names for this candidate
# and substitute None if the IdP did not assert any value for a configured attribute.
values = [ attributes.get(attribute_name, [None])[0] for attribute_name in candidate['attribute_names'] ]
msg = "{} Found candidate values {}".format(logprefix, values)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
body = io.BytesIO(environ['wsgi.input'].read(content_length))
environ['wsgi.input'] = body
context.request = unpack_request(environ, content_length)
environ['wsgi.input'].seek(0)
context.cookie = environ.get("HTTP_COOKIE", "")
context.request_authorization = environ.get("HTTP_AUTHORIZATION", "")
try:
resp = self.run(context)
if isinstance(resp, Exception):
raise resp
return resp(environ, start_response)
except SATOSANoBoundEndpointError as e:
msg = str(e)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
resp = NotFound("The Service or Identity Provider you requested could not be found.")
return resp(environ, start_response)
except Exception as e:
if type(e) != UnknownSystemEntity:
logline = "{}".format(e)
logger.exception(logline)
if debug:
raise
resp = ServiceError("%s" % e)
return resp(environ, start_response)
attributes = self.converter.to_internal(
self.attribute_profile, response.ava,
)
internal_resp = InternalData(
auth_info=auth_info,
attributes=attributes,
subject_type=name_id_format,
subject_id=name_id,
)
msg = "backend received attributes:\n{}".format(
json.dumps(response.ava, indent=4)
)
logline = lu.LOG_FMT.format(id=lu.get_session_id(state), message=msg)
logger.debug(logline)
return internal_resp
if status_code == 200:
msg = "issuer/id pair is linked in AL service"
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.info(logline)
internal_response.subject_id = message
if self.id_to_attr:
internal_response.attributes[self.id_to_attr] = [message]
del context.state[self.name]
return super().process(context, internal_response)
else:
# User selected not to link their accounts, so the internal.response.subject_id is based on the
# issuers id/sub which is fine
msg = "User selected to not link their identity in AL service"
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.info(logline)
del context.state[self.name]
return super().process(context, internal_response)
if len(responses) > 0:
if len(responses) > 1:
msg = "LDAP server returned {} records using search filter"
msg = msg + " value {}"
msg = msg.format(len(responses), filter_val)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.warning(logline)
record = responses[0]
break
# Before using a found record, if any, to populate attributes
# clear any attributes incoming to this microservice if so configured.
if config["clear_input_attributes"]:
msg = "Clearing values for these input attributes: {}"
msg = msg.format(data.attributes)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
data.attributes = {}
# This adapts records with different search and connection strategy
# (sync without pool), it should be tested with anonimous bind with
# message_id.
if isinstance(results, bool) and record:
record = {
"dn": record.entry_dn if hasattr(record, "entry_dn") else "",
"attributes": (
record.entry_attributes_as_dict
if hasattr(record, "entry_attributes_as_dict")
else {}
),
}
def _check_error_response(self, response, context):
"""
Check if the response is an OAuth error response.
:param response: the OIDC response
:type response: oic.oic.message
:raise SATOSAAuthenticationError: if the response is an OAuth error response
"""
if "error" in response:
msg = "{name} error: {error} {description}".format(
name=type(response).__name__,
error=response["error"],
description=response.get("error_description", ""),
)
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.debug(logline)
raise SATOSAAuthenticationError(context.state, "Access denied")
only_idp = only_one_idp_in_metadata and idps[0]
target_entity_id = context.get_decoration(Context.KEY_TARGET_ENTITYID)
force_authn = get_force_authn(context, self.config, self.sp.config)
memorized_idp = get_memorized_idp(context, self.config, force_authn)
entity_id = only_idp or target_entity_id or memorized_idp or None
msg = {
"message": "Selected IdP",
"only_one": only_idp,
"target_entity_id": target_entity_id,
"force_authn": force_authn,
"memorized_idp": memorized_idp,
"entity_id": entity_id,
}
logline = lu.LOG_FMT.format(id=lu.get_session_id(context.state), message=msg)
logger.info(logline)
return entity_id