Add CORS support to webapp

operator/Makefile

@@ -1,4 +1,4 @@
-VERSION ?= 0.14.0
+VERSION ?= 0.15.0
 GIT_TAG := operator_v$(VERSION)
 KEIP_INTEGRATION_IMAGE ?= ghcr.io/codice/keip/minimal-app:latest
 

operator/controller/addons/certmanager/keip-certmanager-controller.yaml

@@ -15,4 +15,4 @@ spec:
   hooks:
     sync:
       webhook:
-        url: http://integrationroute-webhook.keip/addons/certmanager/sync
+        url: http://integrationroute-webhook.keip/webhook/addons/certmanager/sync

operator/controller/core-controller.yaml

@@ -57,7 +57,7 @@ spec:
       serviceAccountName: keip-controller-service
       containers:
         - name: webhook
-          image: ghcr.io/codice/keip/webapp:0.17.0
+          image: ghcr.io/codice/keip/webapp:0.18.0
           ports:
             - containerPort: 7080
               name: webhook-http

operator/webapp/Makefile

@@ -1,4 +1,4 @@
-VERSION ?= 0.17.0
+VERSION ?= 0.18.0
 HOST_PORT ?= 7080
 GIT_TAG := webapp_v$(VERSION)
 

operator/webapp/app.py

@@ -1,19 +1,36 @@
 import logging.config
 
 from starlette.applications import Starlette
-from starlette.routing import Route, Mount
+from starlette.middleware.cors import CORSMiddleware
 from starlette.responses import JSONResponse
+from starlette.routing import Route, Mount
 from starlette.types import ASGIApp
 
 import config as cfg
+from logconf import LOG_CONF
 from routes import webhook
 from routes.deploy import deploy_route
-from logconf import LOG_CONF
-
 
 _LOGGER = logging.getLogger(__name__)
 
 
+def _with_cors(app: Starlette, origins_env: str):
+    origins = [s for part in origins_env.split(",") if (s := part.strip())]
+    if not origins:
+        _LOGGER.warning(
+            "Failed to parse 'CORS_ALLOWED_ORIGINS' env var. CORS headers are disabled."
+        )
+        return app
+
+    _LOGGER.info("Enable CORS headers. Allowed origins: %s", origins)
+
+    return CORSMiddleware(
+        app=app,
+        allow_origins=origins,
+        allow_methods=["GET", "PUT"],
+    )
+
+
 async def status(request):
     return JSONResponse({"status": "UP"})
 
@@ -29,8 +46,12 @@ def create_app() -> ASGIApp:
         Route("/status", status, methods=["GET"]),
         Mount(path="/webhook", routes=webhook.routes),
     ]
+
     starlette_app = Starlette(debug=cfg.DEBUG, routes=routes)
 
+    if cfg.CORS_ALLOWED_ORIGINS:
+        starlette_app = _with_cors(starlette_app, cfg.CORS_ALLOWED_ORIGINS)
+
     return starlette_app
 
 

operator/webapp/config.py

@@ -5,6 +5,9 @@
 # Server
 DEBUG = cfg("DEBUG", cast=bool, default=False)
 
+# Comma-separated list of origin URLs (e.g. "http://localhost:8123,https://www.example.com")
+CORS_ALLOWED_ORIGINS = cfg("CORS_ALLOWED_ORIGINS", cast=str, default="")
+
 # Application
 INTEGRATION_CONTAINER_IMAGE = cfg(
     "INTEGRATION_IMAGE", cast=str, default="keip-integration"

operator/webapp/routes/test/test_cors.py

@@ -0,0 +1,54 @@
+import pytest
+from starlette.testclient import TestClient
+
+from app import app, _with_cors
+
+ACCESS_CONTROL_ALLOW_ORIGIN = "Access-Control-Allow-Origin"
+
+
+@pytest.mark.parametrize(
+    "input_origins, request_origin, allowed_origin_header",
+    [
+        ("http://localhost:8000", "http://localhost:8000", "http://localhost:8000"),
+        (
+            "https://www.example.com,http://localhost:8000",
+            "http://localhost:8000",
+            "http://localhost:8000",
+        ),
+        (
+            "  https://www.example.com ,  http://localhost:8000  ",
+            "https://www.example.com",
+            "https://www.example.com",
+        ),
+    ],
+)
+def test_status_endpoint_with_cors_success(
+    input_origins, request_origin, allowed_origin_header
+):
+    test_client = TestClient(_with_cors(app, input_origins))
+    response = test_client.get("/status", headers={"Origin": request_origin})
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "UP"}
+
+    assert response.headers[ACCESS_CONTROL_ALLOW_ORIGIN] == allowed_origin_header
+
+
+@pytest.mark.parametrize(
+    "input_origins, request_origin",
+    [
+        ("", "http://localhost:8000"),
+        ("http://localhost:8000", "http://localhost:3000"),
+        (",,,", "http://localhost:8000"),
+    ],
+)
+def test_status_endpoint_with_cors_not_enabled_on_bad_input(
+    input_origins, request_origin
+):
+    test_client = TestClient(_with_cors(app, input_origins))
+    response = test_client.get("/status", headers={"Origin": request_origin})
+
+    assert response.status_code == 200
+    assert response.json() == {"status": "UP"}
+
+    assert ACCESS_CONTROL_ALLOW_ORIGIN not in response.headers