[auth] Validate next page urls (#14776)

## Change Description

Updates the handling of the `next` query parameters on various auth URLs
to only accept absolute paths within the same domain as the auth service
or its equivalent batch service. Prevents a potential class of attacks
exploiting unvalidated redirects.

## Security Assessment

Delete all except the correct answer:
- This change has a medium security impact

### Impact Description

For medium/high impact: provide a description of the impact and the
mitigations in place.
Changes how an auth API works, but in a way that reduces its overall
functional surface. Defends against inappropriate redirections following
login.

(Reviewers: please confirm the security impact before approving)

auth/auth/auth.py

@@ -6,6 +6,7 @@
 import typing
 from contextlib import AsyncExitStack
 from typing import List, NoReturn, Optional
+from urllib.parse import urlparse
 
 import aiohttp_session
 import kubernetes_asyncio.client
@@ -192,6 +193,19 @@ def _delete(key):
     _delete('flow')
 
 
+def validate_next_page_url(next_page):
+    if not next_page:
+        raise web.HTTPBadRequest(text='Invalid next page: empty')
+    valid_next_services = ['batch', 'auth']
+    valid_next_domains = [urlparse(deploy_config.external_url(s, '/')).netloc for s in valid_next_services]
+    actual_next_page_domain = urlparse(next_page).netloc
+
+    if actual_next_page_domain not in valid_next_domains:
+        raise web.HTTPBadRequest(
+            text=f'Invalid next page: \'{next_page}\'. Domain \'{actual_next_page_domain}\' not in {valid_next_domains}'
+        )
+
+
 @routes.get('/healthcheck')
 async def get_healthcheck(_) -> web.Response:
     return web.Response()
@@ -215,6 +229,7 @@ async def creating_account(request: web.Request, userdata: Optional[UserData]) -
 
         next_url = deploy_config.external_url('auth', '/user')
         next_page = session.pop('next', next_url)
+        validate_next_page_url(next_page)
 
         cleanup_session(session)
 
@@ -285,6 +300,7 @@ async def _wait_websocket(request, login_id):
 @routes.get('/signup')
 async def signup(request) -> NoReturn:
     next_page = request.query.get('next', deploy_config.external_url('auth', '/user'))
+    validate_next_page_url(next_page)
 
     flow_data = request.app[AppKeys.FLOW_CLIENT].initiate_flow(deploy_config.external_url('auth', '/oauth2callback'))
 
@@ -300,6 +316,7 @@ async def signup(request) -> NoReturn:
 @routes.get('/login')
 async def login(request) -> NoReturn:
     next_page = request.query.get('next', deploy_config.external_url('auth', '/user'))
+    validate_next_page_url(next_page)
 
     flow_data = request.app[AppKeys.FLOW_CLIENT].initiate_flow(deploy_config.external_url('auth', '/oauth2callback'))
 
@@ -323,6 +340,7 @@ async def callback(request) -> web.Response:
 
     caller = session['caller']
     next_page = session.pop('next', next_url)
+    validate_next_page_url(next_page)
     flow_dict = session['flow']
     flow_dict['callback_uri'] = deploy_config.external_url('auth', '/oauth2callback')
     cleanup_session(session)