fix: validate url redirection from remote resources

lifemonitor/api/models/registries/forms.py

@@ -22,10 +22,12 @@
 from __future__ import annotations
 
 import logging
+from typing import List
 
 from flask_wtf import FlaskForm
+from wtforms import HiddenField, ValidationError
+
 from lifemonitor.auth import User
-from wtforms import HiddenField
 
 from .settings import RegistrySettings
 
@@ -38,10 +40,36 @@ def __get_registries__():
     return WorkflowRegistry.all()
 
 
+def __get_registries_names__() -> List[str]:
+    return [_.name for _ in __get_registries__()]
+
+
+def validate_registry(registry_name: str) -> bool:
+    '''Validate the given registry name against the available registries'''
+    return registry_name in __get_registries_names__()
+
+
+class RegistryNameValidator:
+
+    def __call__(self, form, field):
+        return self.validate(field.data)
+
+    @classmethod
+    def validate(cls, value: str) -> bool:
+        value_length = len(value) if value else 0
+        if value_length == 0:
+            return ValidationError("Registry name cannot be empty")
+        if not validate_registry(value):
+            return ValidationError("Registry not found")
+        logger.warning("Registry name '%s' is valid", value)
+        return True
+
+
 class RegistrySettingsForm(FlaskForm):
 
-    action = HiddenField("action", description="")
-    registry = HiddenField("registry", description="")
+    action = HiddenField("action", description="The action to perform")
+    registry = HiddenField("registry",
+                           description="Short name of the registry", validators=[RegistryNameValidator()])
     registries = HiddenField("registries", description="")
 
     available_registries = None

lifemonitor/auth/controllers.py

@@ -26,6 +26,7 @@
                    session, url_for)
 from flask.sessions import SecureCookieSessionInterface
 from flask_login import login_required, login_user, logout_user
+from wtforms import ValidationError
 
 from lifemonitor.cache import Timeout, cached, clear_cache
 from lifemonitor.utils import (NextRouteRegistry, next_route_aware,
@@ -173,6 +174,13 @@ def profile(form=None, passwordForm=None, currentView=None,
     logger.debug(OpenApiSpecs.get_instance().authorization_code_scopes)
     back_param = request.args.get('back', None)
     logger.debug("detected back param: %r", back_param)
+    # validate back_param if defined
+    try:
+        NextRouteRegistry.validate_next_route_url(back_param)
+    except ValidationError:
+        logger.warning("Detected an invalid back param: %r", back_param)
+        back_param = None
+    # set the back param in the session
     if not current_user.is_authenticated:
         session['lm_back_param'] = back_param
         logger.debug("Pushing back param to session")
@@ -273,7 +281,18 @@ def login():
             session.pop('_flashes', None)
             flash("You have logged in", category="success")
             return redirect(NextRouteRegistry.pop(url_for("auth.profile")))
-    flask.session['lm_back_param'] = flask.request.args.get('back', None)
+    # Configure the next route
+    back_param = session.pop('lm_back_param', None)
+    # validate back_param if defined
+    if back_param:
+        try:
+            NextRouteRegistry.validate_next_route_url(back_param)
+        except ValidationError:
+            flash('back param not valid')
+            back_param = None
+    # set the back param in the session
+    flask.session['lm_back_param'] = back_param
+    # render the login page
     return render_template("auth/login.j2", form=form,
                            providers=get_providers(), is_service_available=is_service_alive)
 
@@ -283,11 +302,21 @@ def login():
 def logout():
     logout_user()
     session.pop('_flashes', None)
-    back_param = session.pop('lm_back_param', None)
     flash("You have logged out", category="success")
+    # Configure the next route
     NextRouteRegistry.clear()
+    back_param = session.pop('lm_back_param', None)
+    # validate back_param if defined
+    if back_param:
+        try:
+            NextRouteRegistry.validate_next_route_url(back_param)
+        except ValidationError:
+            flash('back param not valid')
+            back_param = None
+    # set the next route
     next_route = request.args.get('next', '/logout' if back_param else '/')
-    logger.debug("Next route after logout: %r", next_route)
+    logger.debug("Next route after logout: %r", back_param)
+    # redirect to the next route
     return redirect(next_route)
 
 
@@ -404,8 +433,8 @@ def update_github_settings():
 def enable_registry_sync():
     logger.debug("Enabling Registry Sync")
     if request.method == "GET":
-        registry_name = request.values.get("s", None)
-        logger.debug("Enabling Registry Sync: %r", registry_name)
+        registry_name = session.pop("registry_integration_name", None)
+        logger.debug("Registry name from session: %r", registry_name)
         if not registry_name:  # or not current_user.check_secret(registry_name, secret_hash):
             flash("Invalid request: registry cannot be enabled", category="error")
             return redirect(url_for('auth.profile', currentView='registrySettingsTab'))
@@ -426,18 +455,21 @@ def enable_registry_sync():
             settings.set_token(registry.client_name, registry_user_identity.tokens['read write'])
             current_user.save()
             flash(f"Integration with registry \"{registry.name}\" enabled", category="success")
+            logger.info("Integration with registry \"%s\" enabled", registry.name)
             return redirect(url_for('auth.profile', currentView='registrySettingsTab'))
 
     elif request.method == "POST":
         from lifemonitor.api.models.registries.forms import \
             RegistrySettingsForm
         form = RegistrySettingsForm()
         if form.validate_on_submit():
-            registry_name = request.values.get("registry", None)
-            return redirect(f'/oauth2/login/{registry_name}?scope=read+write&next=/account/enable_registry_sync?s={registry_name}')
-        else:
-            logger.debug("Form validation failed")
-            flash("Invalid request", category="error")
+            registry_name = form.registry.data
+            if registry_name:
+                session['registry_integration_name'] = registry_name
+                return redirect(f'/oauth2/login/{registry_name}?scope=read+write&next=/account/enable_registry_sync')
+        # if we are here, then the form validation failed
+        logger.debug("Form validation failed")
+        flash("Invalid request", category="error")
     # set a fallback redirect
     return redirect(url_for('auth.profile', currentView='registrySettingsTab'))
 
@@ -446,16 +478,22 @@ def enable_registry_sync():
 @login_required
 def disable_registry_sync():
     from lifemonitor.api.models.registries.forms import RegistrySettingsForm
-    registry_name = request.values.get("registry", None)
-    logger.debug("Disabling sync for registry: %r", registry_name)
     form = RegistrySettingsForm()
-    settings = current_user.registry_settings
+    # extract the registry name from the form
+    # and check if it is valid
+    registry_name = form.registry.data
+    if not registry_name:
+        flash("Invalid request: registry not found", category="error")
+        return redirect(url_for('auth.profile', currentView='registrySettingsTab'))
+    # validate the form
     if form.validate_on_submit():
+        logger.debug("Disabling sync for registry: %r", registry_name)
         from lifemonitor.api.models import WorkflowRegistry
         registry = WorkflowRegistry.find_by_client_name(registry_name)
         if not registry:
             flash("Invalid request: registry not found", category="error")
             return redirect(url_for('auth.profile', currentView='registrySettingsTab'))
+        settings = current_user.registry_settings
         settings.remove_registry(registry_name)
         current_user.save()
         flash(f"Integration with registry \"{registry.name}\" disabled", category="success")

lifemonitor/auth/oauth2/client/controllers.py

@@ -27,6 +27,7 @@
 from flask import (Blueprint, abort, current_app, flash, redirect, request,
                    session, url_for)
 from flask_login import current_user, login_user
+from wtforms import ValidationError
 
 from lifemonitor import exceptions, utils
 from lifemonitor.auth.models import User
@@ -134,8 +135,15 @@ def logout(name: str):
 
         # Determine the right next hop
         next_url = NextRouteRegistry.pop()
-        return redirect(next_url, code=307) if next_url \
-            else RequestHelper.response() or redirect('/account', code=302)
+        if next_url:
+            try:
+                NextRouteRegistry.validate_next_route_url(next_url)
+                return redirect(next_url, code=307)
+            except ValidationError as e:
+                logger.error(e)
+        # redirect to the account page if the next url is not defined or invalid
+        return RequestHelper.response() or redirect('/account', code=302)
+
     return blueprint
 
 
@@ -256,12 +264,18 @@ def handle_authorize(self, provider: FlaskRemoteApp, token, user_info: OAuthUser
             db.session.commit()
             db.session.flush()
             logger.debug("Identity flushed")
+            flash(f"Logged with your <b>\"{identity.provider.name}\"</b> identity.", category="success")
 
             # Determine the right next hop
             next_url = NextRouteRegistry.pop()
-            flash(f"Logged with your <b>\"{identity.provider.name}\"</b> identity.", category="success")
-            return redirect(next_url, code=307) if next_url \
-                else RequestHelper.response() or redirect('/account', code=302)
+            if next_url:
+                try:
+                    NextRouteRegistry.validate_next_route_url(next_url)
+                    return redirect(next_url, code=307)
+                except ValidationError as e:
+                    logger.error(e)
+            # redirect to the account page if the next url is not defined or invalid
+            return RequestHelper.response() or redirect('/account', code=302)
 
     @staticmethod
     def validate_identity_token(identity: OAuthIdentity):

lifemonitor/errors.py

@@ -72,8 +72,9 @@ def handle_400(e: Exception = None, description: str = None):
 def handle_404(e: Exception = None):
     resource = request.args.get("resource", None, type=str)
     logger.debug(f"Resource not found: {resource}")
-    if not validate_url(resource):
-        return handle_400(decription="Invalid URL")
+    if resource and not validate_url(resource):
+        logger.error(f"Invalid URL: {resource}")
+        return handle_400(description="Invalid URL")
     return handle_error(
         {
             "title": "LifeMonitor: Page not found",

lifemonitor/utils.py

@@ -59,6 +59,7 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding, rsa
 from dateutil import parser
+from wtforms import ValidationError
 
 from lifemonitor.cache import cached
 
@@ -207,6 +208,10 @@ def decodeBase64(str, as_object=False, encoding='utf-8'):
     return result
 
 
+def get_netloc(url: str) -> str:
+    return urlparse(url).netloc
+
+
 def get_base_url():
     server_name = None
     try:
@@ -227,6 +232,10 @@ def get_external_server_url():
     return get_base_url() if not external_server_url else external_server_url
 
 
+def get_valid_server_domains():
+    return ['/', get_netloc(get_base_url()), get_netloc(get_external_server_url())]
+
+
 def get_validation_schema_url():
     return f"{get_external_server_url()}/integrations/github/config/schema.json"
 
@@ -1037,19 +1046,28 @@ def _save_route_registry(cls, registry):
         logger.debug("Route registry saved")
 
     @classmethod
-    def save(cls, route=None):
+    def save(cls, route=None, skipValidation: bool = False):
         route = route or flask.request.args.get('next', False)
         logger.debug("'next' route param found: %r", route)
         if route:
-            registry = cls._get_route_registry()
-            registry.append(route)
-            logger.debug("Route registry changed: %r", registry)
-            cls._save_route_registry(registry)
+            try:
+                if not skipValidation:
+                    cls.validate_next_route_url(route)
+                registry = cls._get_route_registry()
+                registry.append(route)
+                logger.debug("Route registry changed: %r", registry)
+                cls._save_route_registry(registry)
+            except ValidationError as e:
+                logger.error(e)
+                if logger.isEnabledFor(logging.DEBUG):
+                    logger.exception(e)
 
     @classmethod
-    def pop(cls, default=None):
+    def pop(cls, default=None, skipValidation=False):
+        # extract the route from the request
         route = flask.request.args.get('next', None)
         logger.debug("'next' route as param: %r", route)
+        # if the route is not defined as param, try to get it from the registry
         if route is None:
             registry = cls._get_route_registry()
             try:
@@ -1060,12 +1078,62 @@ def pop(cls, default=None):
                 logger.debug(e)
             finally:
                 cls._save_route_registry(registry)
-        return route or default
+        if skipValidation:
+            return route or default
+        # validate the actual route
+        try:
+            cls.validate_next_route_url(route)
+        except ValidationError as e:
+            logger.error(e)
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.exception(e)
+            # if the route is not valid, try to get the default route
+            try:
+                cls.validate_next_route_url(default)
+                route = default
+            except ValidationError as ex:
+                logger.error(ex)
+                if logger.isEnabledFor(logging.DEBUG):
+                    logger.exception(ex)
+                route = None
+        # return the validated route
+        return route
 
     @classmethod
     def clear(cls):
         flask.session[cls.__LM_NEXT_ROUTE_REGISTRY__] = json.dumps([])
 
+    @classmethod
+    def validate_next_route_url(cls, url: str) -> bool:
+        # check whether the URL is valid
+        url_domain = None
+        try:
+            url_domain = get_netloc(url)
+        except Exception as e:
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.exception(e)
+        # check whether a url domain has been extracted
+        if url_domain is None:
+            raise ValidationError("Invalid URL: unable to detect domain")
+        # check if the URL domain matches the main domain of the back-end app
+        valid_server_domains = get_valid_server_domains()
+        logger.debug("Valid app domains: %r", valid_server_domains)
+        if not url_domain or url_domain in valid_server_domains:
+            return True
+        # check whether the URL belong to a client
+        from lifemonitor.auth.oauth2.server.models import Client
+        for c in Client.all():
+            for c_redirect_uri in c.redirect_uris:
+                try:
+                    if url_domain == get_netloc(c_redirect_uri):
+                        logger.debug(f"Found a match for the URL url: {url}")
+                        return True
+                except Exception as e:
+                    if logger.isEnabledFor(logging.DEBUG):
+                        logger.exception(e)
+        # the URL doesn't belong to any of the client domains
+        raise ValidationError(message="URL not allowed as next route")
+
 
 class ClassManager: