Add a cors-always-allowed-origins option

edb/server/args.py

@@ -292,6 +292,8 @@ class ServerConfig(NamedTuple):
 
     admin_ui: bool
 
+    override_cors_allowed_origins: Optional[str]
+
 
 class PathPath(click.Path):
     name = 'path'
@@ -1081,6 +1083,16 @@ def resolve_envvar_value(self, ctx: click.Context):
         ),
         default='default',
         help='Enable admin UI.'),
+    click.option(
+        '--override-cors-allowed-origins',
+        envvar="GEL_SERVER_OVERRIDE_CORS_ALLOWED_ORIGINS",
+        cls=EnvvarResolver,
+        hidden=True,
+        help='A comma separated list of origins to always allow CORS requests '
+             'from regardless of the `cors_allow_orgin` config. The `*` '
+             'character can be used as a wildcard. Intended for use by cloud '
+             'to always allow the cloud UI to make requests to the instance.'
+    ),
     click.option(
         '--disable-dynamic-system-config', is_flag=True,
         envvar="GEL_SERVER_DISABLE_DYNAMIC_SYSTEM_CONFIG",

edb/server/main.py

@@ -246,6 +246,7 @@ async def _run_server(
             pidfile_dir=args.pidfile_dir,
             new_instance=new_instance,
             admin_ui=args.admin_ui,
+            override_cors_origins=args.override_cors_allowed_origins,
             disable_dynamic_system_config=args.disable_dynamic_system_config,
             compiler_state=compiler.state,
             tenant=tenant,

edb/server/multitenant.py

@@ -468,6 +468,7 @@ async def run_server(
             default_auth_method=args.default_auth_method,
             testmode=args.testmode,
             admin_ui=args.admin_ui,
+            override_cors_origins=args.override_cors_allowed_origins,
             disable_dynamic_system_config=args.disable_dynamic_system_config,
             compiler_pool_size=args.compiler_pool_size,
             compiler_pool_mode=srvargs.CompilerPoolMode.MultiTenant,

edb/server/protocol/protocol.pyx

@@ -840,13 +840,21 @@ cdef class HttpProtocol:
             config = self.tenant.get_sys_config().get('cors_allow_origins')
 
         allowed_origins = config.value if config else None
+        overrides = self.server.get_override_cors_origins()
 
-        if allowed_origins is None:
+        if allowed_origins is None and overrides == []:
             return False
 
         origin = request.origin.decode() if request.origin else None
+
         origin_allowed = origin is not None and (
-            origin in allowed_origins or '*' in allowed_origins)
+                any(
+                    override.match(origin) if isinstance(override, re.Pattern)
+                    else origin == override
+                    for override in overrides
+                )
+                or (origin in allowed_origins or '*' in allowed_origins)
+            )
 
         if origin_allowed:
             response.custom_headers['Access-Control-Allow-Origin'] = origin

edb/server/server.py

@@ -20,6 +20,7 @@
 
 
 from __future__ import annotations
+import re
 from typing import (
     Any,
     Callable,
@@ -160,6 +161,7 @@ def __init__(
         default_auth_method: srvargs.ServerAuthMethods = (
             srvargs.DEFAULT_AUTH_METHODS),
         admin_ui: bool = False,
+        override_cors_origins: Optional[str] = None,
         disable_dynamic_system_config: bool = False,
         compiler_state: edbcompiler.CompilerState,
         use_monitor_fs: bool = False,
@@ -252,6 +254,15 @@ def __init__(
 
         self._admin_ui = admin_ui
 
+        self._override_cors_origins = [
+            re.compile(
+                '^' + origin
+                    .replace('.', '\\.')
+                    .replace('*', '.*') + '$'
+            ) if '*' in origin else origin
+            for origin in override_cors_origins.split(',')
+        ] if override_cors_origins else []
+
         self._file_watch_handles = []
         self._tls_certs_reload_retry_handle: Any | asyncio.TimerHandle = None
 
@@ -308,6 +319,9 @@ def in_test_mode(self):
     def is_admin_ui_enabled(self):
         return self._admin_ui
 
+    def get_override_cors_origins(self):
+        return self._override_cors_origins
+
     def on_binary_client_created(self) -> str:
         self._binary_proto_id_counter += 1