KNOX-3402 [wip]: gRPC gateway support

.github/workflows/build/Dockerfile

@@ -43,6 +43,9 @@ ADD .github/workflows/build/conf/topologies/health.xml /knox-runtime/conf/topolo
 ADD .github/workflows/build/conf/topologies/knoxldap.xml /knox-runtime/conf/topologies/knoxldap.xml
 ADD .github/workflows/build/conf/topologies/remoteauth.xml /knox-runtime/conf/topologies/remoteauth.xml
 ADD .github/workflows/build/conf/topologies/k8sauth.xml /knox-runtime/conf/topologies/k8sauth.xml
+ADD .github/workflows/build/conf/topologies/sparkconnect.xml /knox-runtime/conf/topologies/sparkconnect.xml
+ADD .github/workflows/build/conf/topologies/sparkconnect-restricted.xml /knox-runtime/conf/topologies/sparkconnect-restricted.xml
+ADD .github/workflows/build/conf/topologies/sparkconnect-fgac.xml /knox-runtime/conf/topologies/sparkconnect-fgac.xml
 
 RUN chown -R gateway /knox-runtime/
 

.github/workflows/build/conf/topologies/sparkconnect-fgac.xml

@@ -0,0 +1,55 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<!--
+  A topology whose backend enforces fine-grained policy at the plan level. Any
+  authenticated user may reach it, but uploading code is refused: an uploaded
+  jar runs inside the Spark application with that application's credentials and
+  answers to no plan check.
+
+  Denial is by method name alone, which needs no knowledge of message contents.
+-->
+<topology>
+    <gateway>
+        <provider>
+            <role>federation</role>
+            <name>JWTProvider</name>
+            <enabled>true</enabled>
+            <param>
+                <name>knox.token.use.cookie</name>
+                <value>false</value>
+            </param>
+        </provider>
+        <provider>
+            <role>authorization</role>
+            <name>AclsAuthz</name>
+            <enabled>true</enabled>
+            <param>
+                <name>SPARKCONNECT.acl</name>
+                <value>*;*;*</value>
+            </param>
+            <param>
+                <name>SPARKCONNECT.methods.deny</name>
+                <value>AddArtifacts</value>
+            </param>
+        </provider>
+    </gateway>
+    <service>
+        <role>SPARKCONNECT</role>
+        <url>grpc://sparkconnect-mock:15002</url>
+    </service>
+</topology>

.github/workflows/build/conf/topologies/sparkconnect-restricted.xml

@@ -0,0 +1,48 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<!--
+  The same backend as sparkconnect.xml, but restricted to a user who does not
+  exist in the demo LDAP. Selecting a topology is a client-supplied value, so the
+  tests need a topology that an authenticated user can name and still be refused.
+-->
+<topology>
+    <gateway>
+        <provider>
+            <role>federation</role>
+            <name>JWTProvider</name>
+            <enabled>true</enabled>
+            <param>
+                <name>knox.token.use.cookie</name>
+                <value>false</value>
+            </param>
+        </provider>
+        <provider>
+            <role>authorization</role>
+            <name>AclsAuthz</name>
+            <enabled>true</enabled>
+            <param>
+                <name>SPARKCONNECT.acl</name>
+                <value>nobody;*;*</value>
+            </param>
+        </provider>
+    </gateway>
+    <service>
+        <role>SPARKCONNECT</role>
+        <url>grpc://sparkconnect-mock:15002</url>
+    </service>
+</topology>

.github/workflows/build/conf/topologies/sparkconnect.xml

@@ -0,0 +1,39 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+<!--
+  Reachable by any authenticated user. The gRPC listener authenticates bearer
+  tokens itself; the federation provider here is what the servlet side would use
+  and keeps the topology well-formed.
+-->
+<topology>
+    <gateway>
+        <provider>
+            <role>federation</role>
+            <name>JWTProvider</name>
+            <enabled>true</enabled>
+            <param>
+                <name>knox.token.use.cookie</name>
+                <value>false</value>
+            </param>
+        </provider>
+    </gateway>
+    <service>
+        <role>SPARKCONNECT</role>
+        <url>grpc://sparkconnect-mock:15002</url>
+    </service>
+</topology>

.github/workflows/build/gateway-site.xml

@@ -212,4 +212,32 @@ limitations under the License.
         <value>member</value>
     </property>
 
+    <!-- gRPC listener, configured for Spark Connect and exercised by
+         test_spark_connect.py. Everything Spark-specific is a value here: the
+         gateway itself compiles against no Spark protos. -->
+    <property>
+        <name>gateway.grpc.enabled</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>gateway.grpc.service.role</name>
+        <value>SPARKCONNECT</value>
+    </property>
+    <property>
+        <name>gateway.grpc.proto.services</name>
+        <value>spark.connect.SparkConnectService</value>
+    </property>
+    <!-- Spark Connect: user_context = 2, holding user_id = 1 and user_name = 2.
+         Both carry the authenticated principal; the display name is descriptive
+         on the server, but leaving a self-asserted value there would put an
+         unverified name alongside a verified id. -->
+    <property>
+        <name>gateway.grpc.identity.rules</name>
+        <value>2.1=principal,2.2=principal</value>
+    </property>
+    <!-- No default topology on purpose: the tests cover the case where a client
+         sends no knox-topology parameter, which must be refused rather than
+         silently routed somewhere. Method denial is left to the per-topology
+         rule in sparkconnect-fgac.xml. -->
+
 </configuration>

.github/workflows/compose/docker-compose.yml

@@ -85,6 +85,47 @@ services:
     depends_on:
       - k3s
 
+  # One-shot: generates Python protobuf/gRPC stubs from the same vendored
+  # spark/connect/*.proto files the gateway compiles against, into a volume
+  # shared by the mock backend and the tests. Generating rather than depending on
+  # pyspark keeps the images small and means a proto refresh that broke the wire
+  # contract would break these tests too.
+  sparkconnect-protos:
+    image: python:3.10-slim
+    entrypoint:
+      - /bin/sh
+      - -c
+    command:
+      - |
+        set -e
+        pip install --no-cache-dir --quiet grpcio-tools==1.60.0
+        python -m grpc_tools.protoc -I/protos \
+          --python_out=/out --grpc_python_out=/out \
+          /protos/spark/connect/*.proto
+        # Generated modules import each other as spark.connect.*, so the output
+        # has to be an importable package.
+        touch /out/spark/__init__.py /out/spark/connect/__init__.py
+        echo 'spark connect stubs generated'
+    volumes:
+      - ../../../gateway-service-grpc/src/test/proto:/protos:ro
+      - sparkconnect-protos:/out
+
+  # Stands in for a Spark Connect server on a private network. Plaintext, which
+  # is what Knox's grpc:// backend scheme describes.
+  sparkconnect-mock:
+    image: python:3.10-slim
+    environment:
+      - PYTHONPATH=/stubs
+    volumes:
+      - ./sparkconnect:/mock:ro
+      - sparkconnect-protos:/stubs:ro
+    command: >
+      sh -c "pip install --no-cache-dir --quiet grpcio==1.60.0 protobuf==4.25.8
+      && python /mock/mock_server.py"
+    depends_on:
+      sparkconnect-protos:
+        condition: service_completed_successfully
+
   knox:
     image: apache/knox-dev:${IMAGE_TAG:-master}
     command: /gateway.sh
@@ -101,14 +142,21 @@ services:
         condition: service_started
       k8s-bootstrap:
         condition: service_completed_successfully
+      sparkconnect-mock:
+        condition: service_started
 
   tests:
     image: python:3.10-slim
     working_dir: /tests
     volumes:
       - ../tests:/tests
+      - sparkconnect-protos:/stubs:ro
     environment:
       - KNOX_GATEWAY_URL=https://knox:8443/
+      - KNOX_SPARKCONNECT_HOST=knox
+      - KNOX_SPARKCONNECT_PORT=15002
+      # Generated Spark Connect stubs, shared with the mock backend.
+      - PYTHONPATH=/stubs
     command: >
       bash -c "pip install -r requirements.txt
       && pylint *.py
@@ -120,3 +168,4 @@ services:
 
 volumes:
   k3s-output:
+  sparkconnect-protos:

.github/workflows/compose/sparkconnect/mock_server.py

@@ -0,0 +1,113 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to you under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A stand-in Spark Connect server for the Knox integration tests.
+
+Running real Spark would add a gigabyte of image and a minute of startup to
+test a gateway, and the gateway does not care what is behind it -- only that it
+speaks `spark.connect.SparkConnectService`. So this implements just enough of
+that service to make the gateway's behavior observable.
+
+The important trick is that the RPCs echo back what the *backend* received,
+rather than returning canned data. Knox overwrites `user_context.user_id` with
+the authenticated principal on its way through, and that rewrite is invisible
+from the client side -- the client only knows what it sent. By reflecting the
+observed identity into the response, an assertion about what Spark would have
+seen becomes an ordinary assertion in the test.
+
+The stubs are generated at container start from the same vendored
+`spark/connect/*.proto` files the gateway compiles against, so a proto refresh
+that broke the wire contract would break this too.
+"""
+
+import logging
+import os
+from concurrent import futures
+
+import grpc
+
+from spark.connect import base_pb2
+from spark.connect import base_pb2_grpc
+
+LOG = logging.getLogger("mock-spark-connect")
+
+# Enough responses to prove a server stream is relayed message by message rather
+# than collapsed or truncated.
+EXECUTE_PLAN_RESPONSE_COUNT = 5
+
+
+def _observed_user(request):
+    """The user_id the backend actually received, i.e. after Knox's rewrite."""
+    return request.user_context.user_id
+
+
+class MockSparkConnectService(base_pb2_grpc.SparkConnectServiceServicer):
+    """Implements the handful of RPCs the integration tests exercise."""
+
+    def AnalyzePlan(self, request, context):  # noqa: N802 - gRPC naming
+        observed = _observed_user(request)
+        LOG.info("AnalyzePlan session=%s user_id=%s", request.session_id, observed)
+        # explain_string is a free-form string field, so it can carry the observed
+        # identity back to the test without inventing a side channel.
+        return base_pb2.AnalyzePlanResponse(
+            session_id=request.session_id,
+            explain=base_pb2.AnalyzePlanResponse.Explain(explain_string=observed),
+        )
+
+    def ExecutePlan(self, request, context):  # noqa: N802 - gRPC naming
+        observed = _observed_user(request)
+        LOG.info("ExecutePlan session=%s user_id=%s", request.session_id, observed)
+        for index in range(EXECUTE_PLAN_RESPONSE_COUNT):
+            yield base_pb2.ExecutePlanResponse(
+                session_id=request.session_id,
+                operation_id=observed,
+                response_id=f"response-{index}",
+            )
+
+    def Config(self, request, context):  # noqa: N802 - gRPC naming
+        observed = _observed_user(request)
+        LOG.info("Config session=%s user_id=%s", request.session_id, observed)
+        return base_pb2.ConfigResponse(
+            session_id=request.session_id,
+            # Echo the observed identity as a config value so the Config path can
+            # be asserted the same way as AnalyzePlan.
+            pairs=[base_pb2.KeyValue(key="knox.observed.user", value=observed)],
+        )
+
+    def AddArtifacts(self, request_iterator, context):  # noqa: N802 - gRPC naming
+        observed = ""
+        count = 0
+        for request in request_iterator:
+            observed = _observed_user(request)
+            count += 1
+        LOG.info("AddArtifacts messages=%d user_id=%s", count, observed)
+        return base_pb2.AddArtifactsResponse()
+
+
+def serve():
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    port = os.environ.get("MOCK_PORT", "15002")
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=8))
+    base_pb2_grpc.add_SparkConnectServiceServicer_to_server(MockSparkConnectService(), server)
+    # Plaintext: this stands in for a Spark Connect server on a private network,
+    # which is exactly the posture Knox's grpc:// backend scheme describes.
+    server.add_insecure_port(f"[::]:{port}")
+    server.start()
+    LOG.info("Mock Spark Connect server listening on %s", port)
+    server.wait_for_termination()
+
+
+if __name__ == "__main__":
+    serve()

.github/workflows/tests/requirements.txt

@@ -1,4 +1,6 @@
 requests==2.33.0
 pytest==9.0.3
 pylint==4.0.5
-ldap3==2.9.1
+ldap3==2.9.1
+grpcio==1.60.0
+protobuf==4.25.8