Taguette does not safeguard against Open Redirect

@@ -13,6 +13,7 @@
 import os
 from prometheus_async.aio import time as prom_async_time
 import prometheus_client
+import re
 import redis.asyncio as aioredis
 import signal
 import smtplib
@@ -342,6 +343,13 @@ def send_error(self, status_code=500, **kwargs):
         super(HandleStreamClosed, self).send_error(status_code, **kwargs)
 
 
+def is_next_url_safe(url, base_path):
+    return url and re.match(
+        re.escape(base_path) + r'/[a-z]([a-z0-9/_-]|\.[^.])+(?:\?.*)?$',
+        url,
+    )
+
+
 class BaseHandler(HandleStreamClosed, RequestHandler):
     """Base class for all request handlers.
     """
@@ -462,6 +470,12 @@ def logout(self):
         logger.info("Logged out")
         self.clear_cookie('user')
 
+    def sanitize_next_url(self, url):
+        if is_next_url_safe(url, self.application.config['BASE_PATH']):
+            return url
+        else:
+            return self.reverse_url('index')
+
     def render_string(self, template_name, **kwargs):
         extra_footer = self.application.config['EXTRA_FOOTER']
         if not extra_footer:

@@ -79,9 +79,7 @@ def get(self):
     @PROM_REQUESTS.sync('cookies_prompt')
     def post(self):
         self.set_cookie('cookies_accepted', 'yes', dont_check=True)
-        next_ = self.get_argument('next', '')
-        if not next_:
-            next_ = self.reverse_url('index')
+        next_ = self.sanitize_next_url(self.get_argument('next', ''))
         return self.redirect(next_)
 
     def check_xsrf_cookie(self):
@@ -138,9 +136,7 @@ async def post(self):
         )
 
     def _go_to_next(self):
-        next_ = self.get_argument('next', '')
-        if not next_:
-            next_ = self.reverse_url('index')
+        next_ = self.sanitize_next_url(self.get_argument('next', ''))
         return self.redirect(next_)
 
 

@@ -26,6 +26,7 @@
 from taguette import exact_version
 from taguette import convert, database, extract, import_codebook, main, \
     validate, web
+from taguette.web.base import is_next_url_safe
 from taguette.utils import sanitize_filename
 
 
@@ -379,6 +380,20 @@ def test_export_filename(self):
             "Rmis project",
         )
 
+    def test_next_url(self):
+        self.assertFalse(is_next_url_safe('https://google.com/', ''))
+        self.assertFalse(is_next_url_safe('https://goo.gl/', '/goo.gl'))
+        self.assertTrue(is_next_url_safe('/project/1', ''))
+        self.assertTrue(is_next_url_safe('/project?a=1&b=2', ''))
+        self.assertTrue(is_next_url_safe('/project/export.csv', ''))
+        self.assertFalse(is_next_url_safe('/a/../b', ''))
+        self.assertFalse(is_next_url_safe('/a', '/base'))
+        self.assertFalse(is_next_url_safe('/base/../a', '/base'))
+        self.assertTrue(is_next_url_safe('/base/project/1', '/base'))
+        self.assertTrue(is_next_url_safe('/b.se/project/1', '/b.se'))
+        self.assertFalse(is_next_url_safe('/bose/project/1', '/b.se'))
+        self.assertTrue(is_next_url_safe('/base/project/export.csv', '/base'))
+
 
 class TestReadCodebook(unittest.TestCase):
     def test_valid_csv(self):
@@ -661,6 +676,26 @@ async def test_login(self):
             self.assertEqual(response.status, 303)
             self.assertEqual(response.headers['Location'], '/project/1')
 
+        # Login page now instantly redirects
+        async with self.aget(
+            '/login?' + urlencode(dict(next='/project/1')),
+        ) as response:
+            self.assertEqual(response.status, 302)
+            self.assertEqual(response.headers['Location'], '/project/1')
+
+        # Watch out for CVE-2025-67502
+        async def test_redirects_to_index(url):
+            async with self.aget(
+                '/login?' + urlencode(dict(next=url)),
+            ) as response:
+                self.assertEqual(response.status, 302)
+                self.assertEqual(response.headers['Location'], '/')
+
+        await test_redirects_to_index('https://google.com/')
+        await test_redirects_to_index('//google.com/')
+        await test_redirects_to_index('../something')
+        await test_redirects_to_index('/')
+
         # Check redirect to account
         async with self.aget('/.well-known/change-password') as response:
             self.assertEqual(response.status, 301)