feat: enforce some props to be secret on RisingWave Cloud

[tabVersion]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

Highest Release Version: 2.3

What's changed and what's your intention?

as title, relate https://linear.app/risingwave-labs/issue/CLOUD-3897/%5Bdiscussion%5D-enforce-secret-for-sensitive-fields-on-cloud

Enforce sensitive fields to be SECRET when create SOURCE/SINK(in #21269)/CONNECTION

For cluster matching BOTH condition:

• hosted on cloud: by setting ENV RISINGWAVE_CLOUD (removed in feat: check SecretManagement available when enforce secret #21407)
• system config: enforce_secret_on_cloud set to true
  • the config will be set to true for new clusters
• license check: have enough paid tier for secret management (added in feat: check SecretManagement available when enforce secret #21407)

Then the kernel will do such enforcement.

manually test loacally

set_env RISINGWAVE_CLOUD "1"

dev=> create table t1( a int, b varchar )  with ( connector = 'kafka', properties.bootstrap.server = 'localhost:9092',
  topic = 'test2', properties.sasl.password = 'demo') format plain encode json ;
ERROR:  Failed to run the query

Caused by these errors (recent errors listed first):
  1: connector error
  2: Enforce Secret Error
  3: properties.sasl.password is enforced to be a SECRET on RisingWave Cloud, please use `CREATE SECRET` first

resolve #20946

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• My PR contains critical fixes that are necessary to be merged into the latest release.

Documentation

• My PR needs documentation updates.

Release note

Introduce new system setting enforce_secret_on_cloud (default to false for open source, on-prem and byoc, default to true for cloud-hosted instance)

---

detect ENV RISINGWAVE_CLOUD as the signal for the instance is running on RisingWave Cloud

• enforces some props have to be SECRET when creating source/table.

enforce SECRET props:

### AWS (mainly for loading schema from aws s3)
- access_key
- aws.credentials.access_key_id
- s3.access.key
- secret_key
- aws.credentials.secret_access_key
- s3.secret.key
- session_token
- aws.credentials.session_token

### Kafka
- properties.ssl.key.pem
- properties.ssl.key.password
- properties.sasl.password

### Pulsar
- pulsar.auth.token

### Kinesis
- kinesis.credentials.access
- kinesis.credentials.secret
- kinesis.credentials.session_token

### NATS
- password
- jwt
- nkey

### Iceberg
- s3.access.key
- s3.secret.key
- gcs.credential
- catalog.credential
- catalog.token
- catalog.oauth2_server_uri

### MQTT
- tls.client_cert
- tls.client_key
- password

### Filesystem (S3)
- s3.credentials.access
- s3.credentials.secret

### Filesystem (GCS)
- gcs.credential
- gcs.service_account

### Filesystem (Azblob)
- azblob.credentials.account_key
- azblob.credentials.account_name

### Google Pubsub
- pubsub.credentials

### Iceberg (Catalog)
- catalog.jdbc.password

### TestSource
- No specific secrets enforced (placeholder)

### DummyProperties
- No specific secrets enforced (placeholder)

comes from #21352

enforce secrets for new create CONNECTION

### Kafka: the same as Kafka source

### IcebergConnection: 
- s3.access.key
- s3.secret.key
- gcs.credential
- catalog.token

### ConfluentSchemaRegistryConnection
- schema.registry.password

### ElasticsearchConnection:
- elasticsearch.password

[Copilot]

Copilot reviewed 27 out of 27 changed files in this pull request and generated 1 comment.

[Copilot]

In the enforce_secret_on_cloud implementation for IcebergProperties, secret enforcement is performed twice (once via IcebergCommon::enforce_one and again via Self::enforce_one). Consider consolidating these calls to avoid redundant validations.

Suggested change

IcebergCommon::enforce_one(prop)?;

[kwannoel]

cool

[kwannoel]

Is there anyway to disable this on cloud, if some unforseen circumstance is encountered, and for already created source, will this affect backwards compatbility?

[tabVersion]

Is there anyway to disable this on cloud, if some unforseen circumstance is encountered, and for already created source, will this affect backwards compatbility?

I am considering following the issue suggested alter system.
The check only happens in create object stage so it should not have backwards compatibility issue.

[tabVersion]

also do we want to enforce secret when CREATE CONNECTION? If we just want to reuse some sensitive fields, we can directly ask for the privilege of CONNECTION, no bother enforcing anything.
cc @lmatz @kwannoel

[kwannoel]

also do we want to enforce secret when CREATE CONNECTION? If we just want to reuse some sensitive fields, we can directly ask for the privilege of CONNECTION, no bother enforcing anything. cc @lmatz @kwannoel

Does this mean that in RWC, they must always create connection for source and sinks, and cannot fill in the fields inline?

[tabVersion]

also do we want to enforce secret when CREATE CONNECTION? If we just want to reuse some sensitive fields, we can directly ask for the privilege of CONNECTION, no bother enforcing anything. cc @lmatz @kwannoel

Does this mean that in RWC, they must always create connection for source and sinks, and cannot fill in the fields inline?

I don't mean that. Using CONNECTION and filling inline are two alternatives. Users can grant us using CONNECTION, it does not expose the credentials in plaintext.
As long as do reduction well for CONNECTION, we do not need to enforce secret when creating CONNECTION or enforce using connection in source/sink creation. In this pr, we will enforce secrets for CREATE SOURCE/SINK/CONNECTION.

[lmatz]

I support enforcing secret also when creating CONNECTION.
The motivation is to eliminate the possibility of leaking users' credentials as much as possible.
Because once the leak happens, it becomes a mess, and it is hard to explain.

[Copilot]

Copilot reviewed 32 out of 32 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (2)

src/connector/src/source/iceberg/mod.rs:77

• [nitpick] The enforcement loop calls both IcebergCommon::enforce_one(prop) and Self::enforce_one(prop) for every property. Consider reviewing if both checks are necessary or if the logic can be consolidated to avoid potential redundancy.

for prop in prop_iter { IcebergCommon::enforce_one(prop)?; Self::enforce_one(prop)?; }

src/connector/src/source/cdc/mod.rs:150

• [nitpick] The empty implementation for EnforceSecretOnCloud in CdcProperties may bypass secret enforcement. Consider either implementing the intended checks or documenting why enforcement is intentionally omitted.

impl<T: CdcSourceTypeTrait> EnforceSecretOnCloud for CdcProperties<T> {} // todo: enforce jdbc like properties

[xiangjinwu]

For Cargo.lock changes