feat(external-login): add redirect_uri validation

seacatauth/external_login/handler/public.py

@@ -68,6 +68,9 @@ async def pair_external_account(self, request):
 		"""
 		redirect_uri = request.query.get("redirect_uri")
 		provider_type = request.match_info["provider_type"]
+		client_id = self.ExternalLoginService.get_client_id(redirect_uri)
+		if not await self.ExternalLoginService.validate_client_id_and_redirect_uri(redirect_uri, client_id):
+			return aiohttp.web.HTTPBadRequest()
 		authorization_url = await self.ExternalLoginService.initialize_pairing_external_account(
 			provider_type, redirect_uri)
 		return aiohttp.web.HTTPFound(authorization_url)
@@ -91,6 +94,9 @@ async def login_with_external_account(self, request):
 		"""
 		redirect_uri = request.query.get("redirect_uri")
 		provider_type = request.match_info["provider_type"]
+		client_id = self.ExternalLoginService.get_client_id(redirect_uri)
+		if not await self.ExternalLoginService.validate_client_id_and_redirect_uri(redirect_uri, client_id):
+			return aiohttp.web.HTTPBadRequest()
 		authorization_url = await self.ExternalLoginService.initialize_login_with_external_account(
 			provider_type, redirect_uri)
 		return aiohttp.web.HTTPFound(authorization_url)
@@ -112,6 +118,9 @@ async def sign_up_with_external_account(self, request):
 		"""
 		redirect_uri = request.query.get("redirect_uri")
 		provider_type = request.match_info["provider_type"]
+		client_id = self.ExternalLoginService.get_client_id(redirect_uri)
+		if not await self.ExternalLoginService.validate_client_id_and_redirect_uri(redirect_uri, client_id):
+			return aiohttp.web.HTTPBadRequest()
 		try:
 			authorization_url = await self.ExternalLoginService.initialize_signup_with_external_account(
 				provider_type, redirect_uri)

seacatauth/external_login/service.py

@@ -5,11 +5,13 @@
 
 import asab
 import asab.web.rest
+import urllib.parse
 
 from .. import exceptions, AuditLogger
 from ..last_activity import EventCode
 from .utils import AuthOperation
 from .providers import create_provider, GenericOAuth2Login
+from seacatauth.client.service import validate_redirect_uri
 from .storage import ExternalLoginStateStorage, ExternalLoginAccountStorage
 from .exceptions import (
 	LoginWithExternalAccountError,
@@ -69,6 +71,7 @@ async def initialize(self, app):
 		self.RoleService = app.get_service("seacatauth.RoleService")
 		self.LastActivityService = app.get_service("seacatauth.LastActivityService")
 		self.CookieService = app.get_service("seacatauth.CookieService")
+		self.ClientService = app.get_service("seacatauth.ClientService")
 
 		for provider in self.Providers.values():
 			await provider.initialize(app)
@@ -553,3 +556,21 @@ def _get_final_redirect_uri(self, state: dict):
 			return state["redirect_uri"]
 		# No redirect_uri was specified; redirect to default URL
 		return self.DefaultRedirectUri
+
+	def get_client_id(self, redirect_uri: str):
+		parsed = urllib.parse.urlparse(redirect_uri)._asdict()
+		parsed_dict = dict(urllib.parse.parse_qsl(parsed["query"]))
+
+		return parsed_dict.get("client_id")
+
+	async def validate_client_id_and_redirect_uri(self, redirect_uri: str, client_id: str):		
+		try:
+			client = await self.ClientService.get(client_id)			
+		except KeyError:
+			L.error("Client not found in external login.", struct_data={"client_id": client_id})
+			return False
+		if not validate_redirect_uri(redirect_uri, client["redirect_uris"], client.get("redirect_uri_validation_method")):
+			L.error("Invalid redirect uri for external login.", struct_data={"redirect_uri": redirect_uri, "client_id": client_id})
+			return False
+
+		return True