Redirect to login when external login initialization fails

CHANGELOG.md

@@ -3,9 +3,11 @@
 ## v26.15
 
 ### Pre-releases
+- v26.15-alpha1
 - v26.15-alpha
 
 ### Features
+- Redirect to login when external login initialization fails (#565, v26.15-alpha1)
 - Always include link in response if SMTP is not configured (#561, v26.15-alpha)
 
 ---

seacatauth/external_login/authentication/providers/saml.py

@@ -151,10 +151,16 @@ async def prepare_auth_request(self, state: dict, **kwargs) -> typing.Tuple[dict
 		if saml_client is None:
 			raise ExternalLoginError("SAML client is not properly initialized.")
 
-		saml_request_id, authn_request = saml_client.prepare_for_authenticate(
-			binding=saml2.BINDING_HTTP_REDIRECT,
-			relay_state=state["state_id"],
-		)
+		try:
+			saml_request_id, authn_request = saml_client.prepare_for_authenticate(
+				binding=saml2.BINDING_HTTP_REDIRECT,
+				relay_state=state["state_id"],
+			)
+		except Exception as e:
+			L.exception("Cannot prepare SAML authentication request.", struct_data={
+				"provider": self.Type,
+			})
+			raise ExternalLoginError("Failed to prepare SAML authentication request.") from e
 
 		headers = dict(authn_request.get("headers", []))
 		auth_uri = headers.get("Location")

seacatauth/external_login/authentication/service.py

@@ -11,7 +11,7 @@
 from ...models import Session
 from ...events import EventTypes
 from ...api import local_authz
-from .utils import AuthOperation
+from .utils import AuthOperation, ExtLoginResult, ExtLoginError
 from .providers.abc import ExternalAuthProviderABC
 from .providers import create_provider
 from ..exceptions import (
@@ -107,8 +107,21 @@ async def initialize_login_with_ext_provider(
 		provider_type: str,
 		redirect_uri: typing.Optional[str]
 	) -> aiohttp.web.Response:
-		response = await self._prepare_external_auth_request(
-			provider_type, operation=AuthOperation.LogIn, redirect_uri=redirect_uri)
+		try:
+			response = await self._prepare_external_auth_request(
+				provider_type, operation=AuthOperation.LogIn, redirect_uri=redirect_uri)
+		except ExternalLoginError as e:
+			L.log(asab.LOG_NOTICE, "Failed to initialize login with external account.", struct_data={
+				"provider": provider_type,
+				"error": str(e),
+			})
+			return await self._error_redirect_response(
+				self.LoginUri,
+				result=ExtLoginResult.LOGIN_FAILED,
+				delete_sso_cookie=True,
+				redirect_uri=redirect_uri or self.DefaultRedirectUri
+			)
+
 		L.log(asab.LOG_NOTICE, "Initialized login with external account.", struct_data={
 			"provider": provider_type})
 		return response
@@ -121,22 +134,53 @@ async def initialize_signup_with_ext_provider(
 	) -> aiohttp.web.Response:
 		if not self.can_sign_up_new_credentials(provider_type):
 			L.error("Signup with external account is not enabled.")
-			raise exceptions.RegistrationNotOpenError()
-		response = await self._prepare_external_auth_request(
-			provider_type, operation=AuthOperation.SignUp, redirect_uri=redirect_uri)
+			return await self._error_redirect_response(
+				self.LoginUri,
+				result=ExtLoginResult.SIGNUP_FAILED,
+				delete_sso_cookie=True,
+				redirect_uri=redirect_uri or self.DefaultRedirectUri,
+				ext_login_error=ExtLoginError.REGISTRATION_DISABLED,
+			)
+
+		try:
+			response = await self._prepare_external_auth_request(
+				provider_type, operation=AuthOperation.SignUp, redirect_uri=redirect_uri)
+		except ExternalLoginError as e:
+			L.log(asab.LOG_NOTICE, "Failed to initialize sign-up with external account.", struct_data={
+				"provider": provider_type,
+				"error": str(e),
+			})
+			return await self._error_redirect_response(
+				self.LoginUri,
+				result=ExtLoginResult.SIGNUP_FAILED,
+				delete_sso_cookie=True,
+				redirect_uri=redirect_uri or self.DefaultRedirectUri
+			)
+
 		L.log(asab.LOG_NOTICE, "Initialized sign-up with external account.", struct_data={
 			"provider": provider_type})
 		return response
 
 
-
 	async def initialize_pairing_with_ext_provider(
 		self,
 		provider_type: str,
 		redirect_uri: typing.Optional[str]
 	) -> aiohttp.web.Response:
-		response = await self._prepare_external_auth_request(
-			provider_type, operation=AuthOperation.PairAccount, redirect_uri=redirect_uri)
+		try:
+			response = await self._prepare_external_auth_request(
+				provider_type, operation=AuthOperation.PairAccount, redirect_uri=redirect_uri)
+		except ExternalLoginError as e:
+			L.log(asab.LOG_NOTICE, "Failed to initialize pairing external account.", struct_data={
+				"provider": provider_type,
+				"error": str(e),
+			})
+			return await self._error_redirect_response(
+				self.LoginUri,
+				result=ExtLoginResult.PAIRING_FAILED,
+				redirect_uri=redirect_uri or self.DefaultRedirectUri
+			)
+
 		L.log(asab.LOG_NOTICE, "Initialized pairing external account.", struct_data={
 			"provider": provider_type})
 		return response
@@ -226,10 +270,10 @@ async def _finalize_login_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="login_failed",
+				result=ExtLoginResult.LOGIN_FAILED,
 				delete_sso_cookie=True,
 				redirect_uri=self._get_final_redirect_uri(state),
-				ext_login_error="access_denied",
+				ext_login_error=ExtLoginError.ACCESS_DENIED,
 			)
 		except ExternalLoginError as e:
 			L.log(asab.LOG_NOTICE, "External authentication failed.", struct_data={
@@ -239,7 +283,7 @@ async def _finalize_login_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="login_failed",
+				result=ExtLoginResult.LOGIN_FAILED,
 				delete_sso_cookie=True,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
@@ -301,9 +345,9 @@ async def _finalize_login_with_ext_provider(
 			# Redirect to login page with error message, keep the original redirect uri in the query
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="login_failed",
+				result=ExtLoginResult.LOGIN_FAILED,
 				delete_sso_cookie=True,
-				ext_login_error="not_found",
+				ext_login_error=ExtLoginError.NOT_FOUND,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
 
@@ -316,7 +360,7 @@ async def _finalize_login_with_ext_provider(
 			)
 
 		return await self._success_redirect_response(
-			self._get_final_redirect_uri(state), "login_success", sso_session=new_sso_session)
+			self._get_final_redirect_uri(state), ExtLoginResult.LOGIN_SUCCESS, sso_session=new_sso_session)
 
 
 	async def _attempt_pair_credentials(self, provider_type: str, user_info: typing.Dict) -> typing.Optional[str]:
@@ -418,10 +462,10 @@ async def _finalize_signup_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="signup_failed",
+				result=ExtLoginResult.SIGNUP_FAILED,
 				delete_sso_cookie=True,
 				redirect_uri=self._get_final_redirect_uri(state),
-				ext_login_error="access_denied",
+				ext_login_error=ExtLoginError.ACCESS_DENIED,
 			)
 		except ExternalLoginError as e:
 			L.log(asab.LOG_NOTICE, "External authentication failed.", struct_data={
@@ -431,7 +475,7 @@ async def _finalize_signup_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="signup_failed",
+				result=ExtLoginResult.SIGNUP_FAILED,
 				delete_sso_cookie=True,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
@@ -440,7 +484,7 @@ async def _finalize_signup_with_ext_provider(
 			L.error("Sign-up with external account not enabled.")
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="signup_failed",
+				result=ExtLoginResult.SIGNUP_FAILED,
 				delete_sso_cookie=True,
 				ext_login_error="registration_disabled",
 				redirect_uri=self._get_final_redirect_uri(state)
@@ -455,9 +499,9 @@ async def _finalize_signup_with_ext_provider(
 				"provider": provider_type, "sub": user_info.get("sub")})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="signup_failed",
+				result=ExtLoginResult.SIGNUP_FAILED,
 				delete_sso_cookie=True,
-				ext_login_error="already_exists",
+				ext_login_error=ExtLoginError.ALREADY_EXISTS,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
 
@@ -473,7 +517,7 @@ async def _finalize_signup_with_ext_provider(
 			L.error("Sign-up with external account failed: {}".format(e))
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="signup_failed",
+				result=ExtLoginResult.SIGNUP_FAILED,
 				delete_sso_cookie=True,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
@@ -486,7 +530,7 @@ async def _finalize_signup_with_ext_provider(
 		)
 
 		return await self._success_redirect_response(
-			self._get_final_redirect_uri(state), "signup_success", sso_session=sso_session)
+			self._get_final_redirect_uri(state), ExtLoginResult.SIGNUP_SUCCESS, sso_session=sso_session)
 
 
 	async def _finalize_pairing_with_ext_provider(
@@ -519,9 +563,9 @@ async def _finalize_pairing_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="pairing_failed",
+				result=ExtLoginResult.PAIRING_FAILED,
 				redirect_uri=self._get_final_redirect_uri(state),
-				ext_login_error="access_denied",
+				ext_login_error=ExtLoginError.ACCESS_DENIED,
 			)
 		except ExternalLoginError as e:
 			L.log(asab.LOG_NOTICE, "External authentication failed.", struct_data={
@@ -531,7 +575,7 @@ async def _finalize_pairing_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self._get_final_redirect_uri(state),
-				result="pairing_failed",
+				result=ExtLoginResult.PAIRING_FAILED,
 			)
 
 		# Get current SSO session (if any) to determine if we are re-logging in or logging in anew
@@ -546,9 +590,9 @@ async def _finalize_pairing_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="pairing_failed",
+				result=ExtLoginResult.PAIRING_FAILED,
 				delete_sso_cookie=True,
-				ext_login_error="not_authenticated",
+				ext_login_error=ExtLoginError.NOT_AUTHENTICATED,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
 
@@ -560,9 +604,9 @@ async def _finalize_pairing_with_ext_provider(
 			})
 			return await self._error_redirect_response(
 				self.LoginUri,
-				result="pairing_failed",
+				result=ExtLoginResult.PAIRING_FAILED,
 				delete_sso_cookie=True,
-				ext_login_error="not_authenticated",
+				ext_login_error=ExtLoginError.NOT_AUTHENTICATED,
 				redirect_uri=self._get_final_redirect_uri(state)
 			)
 
@@ -584,12 +628,12 @@ async def _finalize_pairing_with_ext_provider(
 			)
 			return await self._error_redirect_response(
 				self._get_final_redirect_uri(state),
-				result="pairing_failed",
-				ext_login_error="already_exists"
+				result=ExtLoginResult.PAIRING_FAILED,
+				ext_login_error=ExtLoginError.ALREADY_EXISTS,
 			)
 
 		return await self._success_redirect_response(
-			self._get_final_redirect_uri(state), "pairing_success")
+			self._get_final_redirect_uri(state), ExtLoginResult.PAIRING_SUCCESS)
 
 
 	async def _login(

seacatauth/external_login/authentication/utils.py

@@ -16,3 +16,20 @@ def deserialize(cls, value: str):
 			return cls.SignUp
 		else:
 			raise KeyError(value)
+
+
+class ExtLoginResult(enum.StrEnum):
+	SIGNUP_SUCCESS = "signup_success"
+	PAIRING_SUCCESS = "pairing_success"
+	LOGIN_SUCCESS = "login_success"
+	LOGIN_FAILED = "login_failed"
+	SIGNUP_FAILED = "signup_failed"
+	PAIRING_FAILED = "pairing_failed"
+
+
+class ExtLoginError(enum.StrEnum):
+	REGISTRATION_DISABLED = "registration_disabled"
+	NOT_AUTHENTICATED = "not_authenticated"
+	ALREADY_EXISTS = "already_exists"
+	NOT_FOUND = "not_found"
+	ACCESS_DENIED = "access_denied"