Add principal delegating wrapper proxy for REST APIs

Instantiate a delegating proxy per user authorization data for REST
operations on Kafka.

README.rst

@@ -314,6 +314,9 @@ Keys to take special care are the ones needed to configure Kafka and advertised_
      - ``localhost:9092``
      - The URI to the Kafka service where to store the schemas and to run
        coordination among the Karapace instances.
+   * - ``sasl_bootstrap_uri``
+     - ``None``
+     - The URI to the Kafka service to use with the Kafka REST API when SASL authorization with REST is used.
    * - ``client_id``
      - ``sr-1``
      - The ``client_id`` Karapace will use when coordinating with
@@ -413,6 +416,9 @@ Keys to take special care are the ones needed to configure Kafka and advertised_
      - ``/path/to/authfile.json``
      - Filename to specify users and access control rules for Karapace Schema Registry.
        If this is set, Schema Segistry requires authentication for most of the endpoints and applies per endpoint authorization rules.
+   * - ``rest_authorization``
+     - ``false``
+     - Use REST API's calling authorization credentials to invoke Kafka operations over SASL authentication of ``sasl_bootstrap_uri`` to delegate REST proxy authorization to Kafka.  If false, then use configured common credentials for all Kafka connections of REST proxy operations.
    * - ``metadata_max_age_ms``
      - ``60000``
      - Period of time in milliseconds after Kafka metadata is force refreshed.

karapace.config.json

@@ -2,6 +2,8 @@
     "access_logs_debug": false,
     "advertised_hostname": "localhost",
     "bootstrap_uri": "127.0.0.1:9092",
+    "sasl_bootstrap_uri": "127.0.0.1:9094",
+    "rest_authorization": false,
     "client_id": "sr-1",
     "compatibility": "FULL",
     "group_id": "schema-registry",

karapace/config.py

@@ -27,6 +27,7 @@
     "advertised_port": None,
     "advertised_protocol": "http",
     "bootstrap_uri": "127.0.0.1:9092",
+    "sasl_bootstrap_uri": None,
     "client_id": "sr-1",
     "compatibility": "BACKWARD",
     "connections_max_idle_ms": 15000,
@@ -45,6 +46,7 @@
     "registry_password": None,
     "registry_ca": None,
     "registry_authfile": None,
+    "rest_authorization": False,
     "log_level": "DEBUG",
     "log_format": "%(name)-20s\t%(threadName)s\t%(levelname)-8s\t%(message)s",
     "master_eligibility": True,
@@ -155,6 +157,11 @@ def validate_config(config: Config) -> None:
             f"Invalid master election strategy: {master_election_strategy}, valid values are {valid_strategies}"
         ) from None
 
+    if config["rest_authorization"] and config["sasl_bootstrap_uri"] is None:
+        raise InvalidConfiguration(
+            "Using 'rest_authorization' requires configuration value for 'sasl_bootstrap_uri' to be set"
+        )
+
 
 def write_config(config_path: Path, custom_values: Config) -> None:
     config_path.write_text(json.dumps(custom_values))
@@ -171,7 +178,7 @@ def read_config(config_handler: IO) -> Config:
 
 def create_client_ssl_context(config: Config) -> Optional[ssl.SSLContext]:
     # taken from conn.py, as it adds a lot more logic to the context configuration than the initial version
-    if config["security_protocol"] == "PLAINTEXT":
+    if config["security_protocol"] in ("PLAINTEXT", "SASL_PLAINTEXT"):
         return None
     ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS)
     ssl_context.options |= ssl.OP_NO_SSLv2