Schema registry USER_INFO authentication seemed didn't work #31
Comments
|
I haven't used replicator with this, but you can see #25 for MirrorMaker2 |
|
I have had a custom build jar working by upgrading the confluent version to schema-registry-transfer-smt/pom.xml Line 70 in ddf5cd1 Because I saw there was an authentication change in this repo #19 which was last September but the schema registry client version in pom.xml didn't being changed accordingly. So, this issue should be solved as long as we upgrade the confluent version to a version released after this pull request confluentinc/schema-registry#1204 |
|
I've got two cents to offer here. PR #19 was a workaround for the fact that, as of 5.1.0, Confluent maintained a global/static cache of In order to proceed without the ability to modify the SchemaRegistryClient in a way that would remove this static cache. Without being able to change the implementation of that cache, the workaround implemented in #19 was to register two identical copies of the existing UserInfoBasicAuthenticationProvider, but with slightly modified getAlias( ) methods, such that one called itself In 5.4.1, Confluent updated its implementation of There is a test case on line #296 here that tests whether the client gets configured correctly to send basic authentication credentials if only the destination USER_INFO strategy is configured, and that test case does still seem to be passing, so I'm surprised the request you captured shows no such header. It is possible that the something changed between 5.1.0 and 5.4.1 about how Confluent uses the provider to cause basic authentication credentials to be attached, but I've no reason to believe that's the case. Your case is somewhat unique insofar as it is using SASL security to protect the Kafka client, but I don't see anything in the source that would suggest those configuration properties are in any way exposed to the Schema Registry client, therefore I would not expect them to be a part of this behavior. I am curious as to how you managed to inspect the HTTP request sent, given that the communication channel is configured for https? If you are relaying what appeared in the log files on the registry, I think it is possible that Confluent is stripping any authentication credentials from the log file to prevent the log from being a security liability. If that's so, then you should be able to send a valid registry request from any other client and also not find the authentication header in the log. Presuming the previous paragraph's assumption that the sample request came from a log file where the authentication header would have been stripped anyway as a security feature, I think the only way to observe what is really being sent from the SMT transform to the registry server would be to set up a Man-In-The-Middle TLS proxy configured to log each decoded request it receives before initiating a new TLS session with the actual registry and forwarding the original request unmodified. If that's not how the sample request was collected and you're interested in running such an experiment, I'd start by checking out the Quick Start guide at https://mitmproxy.org/. |
|
Hi @jheinnic , Thanks for your reply.
I was having the issue when I use 5.1.0, not any later release version. And I was surprised by the result as well because I saw the #19 was mean to solve this kind of problem.
In terms of how did I get the HTTP response, I updated the target schema registry server in connect configuration to a local a dummy HTTP server By the way, I think this issue can be solved when we merge #29 |
Hi there, I am currently trying to use this transformer to migrate multiple Kafka clusters and schema registries to one cluster on Confluent cloud.
But with the distributed replicator configuration below, I am keeping getting
401 unauthorizederror when the transformer tries to register the schema on the dest schema registry，the fetching schema from the src cluster part works fine as there is no authentication required for that cluster. By the way, the dest schema registry is the Confluent cloud hosted version.My connector config details are:
{ "name": "us-replication", "config": { "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector", "key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter", "value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter", "key.serializer": "org.apache.kafka.common.serialization.StringSerializer", "value.serializer": "io.confluent.kafka.serializers.KafkaAvroSerializer", "src.kafka.bootstrap.servers": "consumer-kafka:9094", "src.kafka.security.protocol": "SASL_SSL", "src.kafka.ssl.truststore.type": "jks", "src.kafka.ssl.endpoint.identification.algorithm": "https", "src.kafka.ssl.truststore.location": "/xxxx.jks", "src.kafka.ssl.truststore.password": "xxxx", "src.kafka.sasl.mechanism": "SCRAM-SHA-256", "src.kafka.sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"xxx\" password=\"xxx\";", "src.kafka.request.timeout.ms": "30000", "tasks.max": "1", "dest.kafka.bootstrap.servers": "dest-kafka:9092", "dest.kafka.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"xxxx\" password=\"xxxx\";", "dest.kafka.security.protocol": "SASL_SSL", "dest.kafka.ssl.endpoint.identification.algorithm": "https", "dest.kafka.sasl.mechanism": "PLAIN", "dest.topic.replication.factor": "3", "dest.kafka.request.timeout.ms": "30000", "dest.kafka.retry.backoff.ms": "500", "topic.rename.format": "${topic}", "topic.regex": "^(qa.).*$", "offset.topic.commit": false, "offset.translator.tasks.max": 1, "offset.timestamps.commit": false, "transforms": "AvroSchemaTransfer", "transforms.AvroSchemaTransfer.type": "cricket.jmoore.kafka.connect.transforms.SchemaRegistryTransfer", "transforms.AvroSchemaTransfer.transfer.message.keys": "false", "transforms.AvroSchemaTransfer.src.schema.registry.url": "http://consumer-registry", "transforms.AvroSchemaTransfer.dest.schema.registry.url": "https://producer-registry", "transforms.AvroSchemaTransfer.dest.basic.auth.credentials.source": "USER_INFO", "transforms.AvroSchemaTransfer.dest.basic.auth.user.info": "user:pass" } }The errors are：
And I also inspected the HTTP request of egistering a schema but didn't see any authentication headers inside, and it looks like the following:
Furthermore, I've tested both
0.2.0and the current master branch, and none of them works in my case.So, can you help to have a look with this? Thanks.
The text was updated successfully, but these errors were encountered: