diff --git a/cli/serve.go b/cli/serve.go index ffd6a5dc..d545bf71 100644 --- a/cli/serve.go +++ b/cli/serve.go @@ -132,12 +132,13 @@ func (l moduleConfigLookup) ModuleConfigs(ctx context.Context, moduleURN string) func setupRegistry(store *postgres.Store) module.Registry { supported := []module.Descriptor{ kubernetes.Module, - firehose.Module, job.Module, kafka.Module, flink.Module, - // dagger resolves ACL kafka streams by fetching their resource internally - // (no dependency), so it needs a resource getter backed by the store. + // dagger and firehose resolve ACL kafka streams by fetching their + // resource internally (no dependency), so they need a resource getter + // backed by the store. + firehose.Module(store.GetByURN), dagger.Module(store.GetByURN), } diff --git a/docs/modules/firehose.md b/docs/modules/firehose.md index b870e4df..4ca0142b 100644 --- a/docs/modules/firehose.md +++ b/docs/modules/firehose.md @@ -34,4 +34,72 @@ type moduleConfig struct { | `ChartVersion` | `string` Chart version you want to use. | | `Firehose` | `struct` Holds firehose configuration. | -Detailed JSONSchema for config can be referenced [here](https://github.com/goto/entropy/blob/main/modules/firehose/schema/config.json). \ No newline at end of file +Detailed JSONSchema for config can be referenced [here](https://github.com/goto/entropy/blob/main/modules/firehose/schema/config.json). + +## Kafka ACL (SASL/SSL) source streams + +A firehose names its kafka resource through `stream_name`, and flags a secured one with +`stream_security_enabled`. The stream's security profile — the same `security` block the +kafka module exposes on its output — is then resolved during Plan, in this order: + +1. an inline `stream_security` entry (prefetched by Dex), +2. a declared kafka dependency keyed by the stream name, +3. the kafka resource fetched internally by URN + (`orn:entropy:kafka::`), with no dependency declared. + +The flag exists so plaintext firehoses cost no lookup: the caller already knows whether a +profile exists, because Dex reads the same resource for the broker address. `stream_name` +is sent for plaintext streams too, so a stream that *loses* its ACLs gets the wiring a +previous plan left behind cleared. + +The wiring mirrors odin's firehose adapter, so a migrated firehose renders the same pod +spec it does today. From the resolved profile the module injects the +`SOURCE_KAFKA_CONSUMER_CONFIG_*` env variables — security protocol, SASL mechanism, SSL +protocol, truststore type, and for a stream carrying certs a fixed +`SSL_TRUSTSTORE_LOCATION` of `/etc/secret/kafka-cert/truststore.p12` (`.jks` for JKS) plus +the `SSL_TRUSTSTORE_FILENAME` the chart uses to select that key out of the secret. The cert +gets its own subdirectory rather than odin's bare `/etc/secret`, because the chart already +mounts its sink-credential secret there. OAUTHBEARER +additionally gets the `OAuthBearerLoginModule` JAAS string and the pod login callback +handler class. + +No secret value ever passes through entropy. The material is described as references in the +`acl` config, rendered as the `kafka_security` chart value alongside `service_account`: + +| field | what the chart does with it | +| :--- | :--- | +| `ssl_config_credential` + `truststore_filename` | mounts that existing secret at `/etc/secret/kafka-cert`, selecting the filename as both key and path | +| `truststore_password` (`secretName` + `key`) | renders `SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD` as a `secretKeyRef` env var | +| `jaas_config_credential` | mounts the PLAIN/SCRAM `jaas.conf` secret at `/etc/secret/kafka` | +| `kafka_token_enabled` | adds the projected service-account token (`audience: kafka`) at `/var/run/secrets/kafka/serviceaccount` | + +PLAIN/SCRAM streams never inline credentials or a JAAS string: they read a mounted +`jaas.conf`, and the module appends +`-Djava.security.auth.login.config=/etc/secret/kafka/jaas.conf` to `_JAVA_OPTIONS` (the +rest of that variable is left alone). The secret is the profile's `acls[team].secretName` +when set, otherwise odin's `--jaas` convention. + +Plaintext firehoses are untouched — no injected config, no `kafka_security` value, no +chart value changes. + +Naming a stream also relaxes the schema's `SOURCE_KAFKA_BROKERS` requirement: when the +stream is resolved, brokers are filled from its `url` unless the payload set them. Plan +fails if neither supplied them, rather than deploying a firehose with no brokers — note +that a stream named without the flag resolves nothing, so brokers must be explicit. Dex +sends the address itself in every case, so its value always wins. + +The callback handler class and default service account are deployment level settings under +the module's driver config: + +```json +{ + "kafka_security": { + "sasl_login_callback_handler_class": "io.gtflabs.kafka.security.oauthbearer.kubernetes.PodLoginCallbackHandler", + "service_account": "aegis-kafka" + } +} +``` + +> Requires firehose chart `0.2.1` or later (`goto/charts`, `stable/firehose`). Chart `0.2.0` +> renders none of this — it has no `serviceAccountName`, projected volume or `secretKeyRef` +> support — so an ACL firehose on it starts without its secrets and fails to authenticate. \ No newline at end of file diff --git a/modules/firehose/config.go b/modules/firehose/config.go index 5417de70..0e761cf7 100644 --- a/modules/firehose/config.go +++ b/modules/firehose/config.go @@ -8,6 +8,7 @@ import ( "github.com/goto/entropy/core/resource" "github.com/goto/entropy/modules" + kafkamod "github.com/goto/entropy/modules/kafka" "github.com/goto/entropy/pkg/errors" "github.com/goto/entropy/pkg/validator" ) @@ -67,6 +68,37 @@ type Config struct { ChartValues *ChartValues `json:"chart_values,omitempty"` InitContainer InitContainer `json:"init_container,omitempty"` Autoscaler *Autoscaler `json:"autoscaler,omitempty"` + + // Team owns the firehose. It selects the credential reference from a + // stream's PLAIN/SCRAM ACL list. + Team string `json:"team,omitempty"` + + // StreamName is the name of the kafka resource backing SOURCE_KAFKA_BROKERS. + // It is the key used to resolve the stream's security profile. Setting it + // also makes this module the owner of the SASL/SSL env variables, which are + // rebuilt from the stream on every plan. + StreamName string `json:"stream_name,omitempty"` + + // StreamSecurityEnabled says the named stream carries a security profile, so + // the driver resolves its kafka resource internally (by URN, without a + // declared dependency). The caller already knows this — Dex reads the same + // resource for the broker address — so a plaintext firehose costs no lookup. + StreamSecurityEnabled bool `json:"stream_security_enabled,omitempty"` + + // StreamSecurity holds the kafka security profile fetched by Dex, keyed by + // stream name. References only — never inline secret values. + StreamSecurity map[string]*kafkamod.SecurityProfile `json:"stream_security,omitempty"` + + // ACL describes the secret material an ACL (SASL_SSL/OAUTHBEARER, SSL, + // PLAIN/SCRAM) source stream needs mounted. It is computed by + // applyStreamSecurity from the resolved stream security profile and holds + // references only — never secret values. + ACL *ACLConfig `json:"acl,omitempty"` + + // ServiceAccount, when set, becomes the pod's service account. It is the + // OAuth identity authorized for ACL streams. Empty preserves the chart's + // default service account. + ServiceAccount string `json:"service_account,omitempty"` } type Telegraf struct { diff --git a/modules/firehose/driver.go b/modules/firehose/driver.go index 4b14da8b..62f8a144 100644 --- a/modules/firehose/driver.go +++ b/modules/firehose/driver.go @@ -90,8 +90,13 @@ type firehoseDriver struct { kubeGetPod kubeGetPodFn kubeGetDeployment kubeGetDeploymentFn consumerReset consumerResetFn + getResource ResourceGetter } +// ResourceGetter fetches a resource by URN. It lets the driver resolve an ACL +// kafka stream internally, without declaring it as a dependency. +type ResourceGetter func(ctx context.Context, urn string) (*resource.Resource, error) + type ( kubeDeployFn func(ctx context.Context, isCreate bool, conf kube.Config, hc helm.ReleaseConfig) error kubeGetPodFn func(ctx context.Context, conf kube.Config, ns string, labels map[string]string) ([]kube.Pod, error) @@ -147,6 +152,10 @@ type driverConf struct { KubeDeployTimeout int `json:"kube_deploy_timeout_seconds"` Autoscaler FirehoseAutoscaler `json:"autoscaler,omitempty"` + + // KafkaSecurity holds the deployment level settings used when wiring an ACL + // (SASL/SSL) source stream. + KafkaSecurity KafkaSecurity `json:"kafka_security,omitempty"` } type FirehoseAutoscaler struct { @@ -396,6 +405,35 @@ func (fd *firehoseDriver) getHelmRelease(res resource.Resource, conf Config, "mountSecrets": mountSecrets, } + // ACL (SASL/SSL) source support, mirroring odin's firehose manifest: the + // chart mounts the referenced secrets and renders the truststore password as + // a secretKeyRef. Both keys are omitted for plaintext firehoses, which keeps + // their rendered chart values unchanged. + if conf.ACL != nil { + aclValues := map[string]any{} + if conf.ACL.SSLConfigCredential != "" { + aclValues["ssl_config_credential"] = conf.ACL.SSLConfigCredential + aclValues["truststore_filename"] = conf.ACL.TruststoreFilename + } + if conf.ACL.TruststorePassword != nil { + aclValues["truststore_password"] = map[string]any{ + "secretName": conf.ACL.TruststorePassword.SecretName, + "key": conf.ACL.TruststorePassword.Key, + } + } + if conf.ACL.JaasConfigCredential != "" { + aclValues["jaas_config_credential"] = conf.ACL.JaasConfigCredential + } + if conf.ACL.KafkaTokenEnabled { + aclValues["kafka_token_enabled"] = true + } + rc.Values["kafka_security"] = aclValues + } + + if conf.ServiceAccount != "" { + rc.Values["service_account"] = conf.ServiceAccount + } + if conf.Autoscaler != nil { rc.Values["autoscaler"], err = conf.Autoscaler.GetHelmValues(conf) if err != nil { diff --git a/modules/firehose/driver_plan.go b/modules/firehose/driver_plan.go index 3803c0f6..4fb13a38 100644 --- a/modules/firehose/driver_plan.go +++ b/modules/firehose/driver_plan.go @@ -21,23 +21,23 @@ var suffixRegex = regexp.MustCompile(`^([A-Za-z0-9-]+)-([0-9]+)$`) var errCauseInvalidNamespaceUpdate = "cannot update kube namespace of a running firehose" -func (fd *firehoseDriver) Plan(_ context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { +func (fd *firehoseDriver) Plan(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { switch act.Name { case module.CreateAction: - return fd.planCreate(exr, act) + return fd.planCreate(ctx, exr, act) case ResetAction: - return fd.planReset(exr, act) + return fd.planReset(ctx, exr, act) case ResetV2Action: - return fd.planResetV2(exr, act) + return fd.planResetV2(ctx, exr, act) default: - return fd.planChange(exr, act) + return fd.planChange(ctx, exr, act) } } -func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { +func (fd *firehoseDriver) planChange(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { curConf, err := readConfig(exr.Resource, exr.Resource.Spec.Configs, fd.conf) if err != nil { return nil, err @@ -84,6 +84,12 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act } } + // resolve the ACL source stream (SASL/SSL consumer config + pod mounts). + // No-op for plaintext sources. + if err := fd.applyStreamSecurity(ctx, exr, newConf); err != nil { + return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error()) + } + curConf = newConf case ScaleAction: @@ -144,12 +150,18 @@ func (fd *firehoseDriver) planChange(exr module.ExpandedResource, act module.Act return &exr.Resource, nil } -func (fd *firehoseDriver) planCreate(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { +func (fd *firehoseDriver) planCreate(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { conf, err := readConfig(exr.Resource, act.Params, fd.conf) if err != nil { return nil, err } + // resolve the ACL source stream (SASL/SSL consumer config + pod mounts). + // No-op for plaintext sources. + if err := fd.applyStreamSecurity(ctx, exr, conf); err != nil { + return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error()) + } + chartVals, err := mergeChartValues(&fd.conf.ChartValues, conf.ChartValues) if err != nil { return nil, err @@ -193,7 +205,7 @@ func (fd *firehoseDriver) planCreate(exr module.ExpandedResource, act module.Act return &exr.Resource, nil } -func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { +func (fd *firehoseDriver) planResetV2(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { resetValue, err := kafka.ParseResetV2Params(act.Params) if err != nil { return nil, err @@ -206,6 +218,10 @@ func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.Ac return nil, err } + if err := fd.applyStreamSecurity(ctx, exr, curConf); err != nil { + return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error()) + } + curConf.ResetOffset = resetValue exr.Resource.Spec.Configs = modules.MustJSON(curConf) @@ -225,7 +241,7 @@ func (fd *firehoseDriver) planResetV2(exr module.ExpandedResource, act module.Ac return &exr.Resource, nil } -func (fd *firehoseDriver) planReset(exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { +func (fd *firehoseDriver) planReset(ctx context.Context, exr module.ExpandedResource, act module.ActionRequest) (*resource.Resource, error) { resetValue, err := kafka.ParseResetParams(act.Params) if err != nil { return nil, err @@ -238,6 +254,10 @@ func (fd *firehoseDriver) planReset(exr module.ExpandedResource, act module.Acti return nil, err } + if err := fd.applyStreamSecurity(ctx, exr, curConf); err != nil { + return nil, errors.ErrInvalid.WithMsgf("failed to resolve source stream").WithCausef("%s", err.Error()) + } + curConf.ResetOffset = resetValue curConf.EnvVariables[SourceKafkaConsumerAutoOffsetReset] = resetValue curConf.EnvVariables[confKeyConsumerID], err = getNewConsumerGroupID(curConf.EnvVariables[confKeyConsumerID]) diff --git a/modules/firehose/kafka_security.go b/modules/firehose/kafka_security.go new file mode 100644 index 00000000..dac58f74 --- /dev/null +++ b/modules/firehose/kafka_security.go @@ -0,0 +1,439 @@ +package firehose + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/goto/entropy/core/module" + "github.com/goto/entropy/core/resource" + kafkamod "github.com/goto/entropy/modules/kafka" +) + +// SASL/SSL consumer config keys. Firehose passes every +// SOURCE_KAFKA_CONSUMER_CONFIG_* env variable straight to the kafka consumer, +// so they are flat env variables here. The key set and the values mirror odin's +// firehose adapter (app/firehose/adapter.js on gtf-master), which is the +// behaviour being migrated onto entropy. +const ( + keyConsumerSecurityProtocol = "SOURCE_KAFKA_CONSUMER_CONFIG_SECURITY_PROTOCOL" + keyConsumerSaslMechanism = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_MECHANISM" + keyConsumerSaslJaasConfig = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_JAAS_CONFIG" + keyConsumerSaslLoginCallbackHandler = "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS" + keyConsumerSSLProtocol = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_PROTOCOL" + keyConsumerSSLTruststoreType = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_TYPE" + keyConsumerSSLTruststoreLocation = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_LOCATION" + keyConsumerSSLTruststoreFilename = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_FILENAME" + + // keyJavaOptions carries the JAAS file location for SCRAM/PLAIN streams. + // It is user-owned, so only the JAAS option itself is added or removed. + keyJavaOptions = "_JAVA_OPTIONS" +) + +// legacy keys from the config-provider approach. They are no longer emitted but +// are still swept, so resources planned by an older build do not keep a +// dangling provider reference. +const ( + keyConsumerSSLTruststorePassword = "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD" + keyConsumerConfigProviders = "SOURCE_KAFKA_CONSUMER_CONFIG_CONFIG_PROVIDERS" + keyConsumerConfigProviderClassPrefix = "SOURCE_KAFKA_CONSUMER_CONFIG_CONFIG_PROVIDERS_" + keyConsumerConfigProviderClassSuffix = "_CLASS" +) + +// managedSecurityKeys are owned by this module: they are wiped and rebuilt on +// every plan of a firehose that names a kafka stream, so a stream that loses +// its ACLs does not leave stale configuration behind. +var managedSecurityKeys = []string{ + keyConsumerSecurityProtocol, + keyConsumerSaslMechanism, + keyConsumerSaslJaasConfig, + keyConsumerSaslLoginCallbackHandler, + keyConsumerSSLProtocol, + keyConsumerSSLTruststoreType, + keyConsumerSSLTruststoreLocation, + keyConsumerSSLTruststoreFilename, + keyConsumerSSLTruststorePassword, + keyConsumerConfigProviders, +} + +const ( + oauthConsumerSaslJaasConfig = "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;" + + // the OAUTHBEARER login callback handler shipped by the platform kafka + // security library. Override per deployment when the firehose image + // packages it under a different name. + defaultOauthSaslLoginCallbackHandlerClass = "io.gtflabs.kafka.security.oauthbearer.kubernetes.PodLoginCallbackHandler" +) + +const ( + securityProtocolSASLSSL = "SASL_SSL" + securityProtocolSASLPlaintext = "SASL_PLAINTEXT" + securityProtocolSSL = "SSL" + saslMechanismOauthbearer = "OAUTHBEARER" + saslMechanismPlain = "PLAIN" + saslMechanismScram = "SCRAM-SHA-512" + truststoreTypePKCS12 = "PKCS12" +) + +// Mount layout, matching odin's firehose manifest. The chart mounts the stream's +// cert secret at /etc/secret and the JAAS secret at /etc/secret/kafka; the +// projected kafka service-account token lands at kafkaTokenMountPath. +const ( + secretMountPath = "/etc/secret" + // the cert secret gets its own subdirectory rather than odin's bare + // /etc/secret: the firehose chart already mounts its own secret there for + // sink credentials, and two volumes cannot share a mount path. + certMountPath = secretMountPath + "/kafka-cert" + jaasSecretMountPath = secretMountPath + "/kafka" + jaasConfigFileName = "jaas.conf" + jaasConfigJavaOpt = "-Djava.security.auth.login.config=" + jaasSecretMountPath + "/" + jaasConfigFileName + truststoreFileBase = "truststore" + jaasSecretSuffix = "jaas" + + kafkaTokenMountPath = "/var/run/secrets/kafka/serviceaccount" +) + +// KafkaSecurity carries the deployment level knobs for ACL streams. +type KafkaSecurity struct { + // SaslLoginCallbackHandlerClass is the OAUTHBEARER login callback handler + // that exchanges the projected service-account token for a kafka token. + SaslLoginCallbackHandlerClass string `json:"sasl_login_callback_handler_class,omitempty"` + + // ServiceAccount is the OAuth identity authorized for ACL streams. Used + // when the resource config does not set one. Empty preserves the chart's + // default service account. + ServiceAccount string `json:"service_account,omitempty"` +} + +func (k KafkaSecurity) withDefaults() KafkaSecurity { + if k.SaslLoginCallbackHandlerClass == "" { + k.SaslLoginCallbackHandlerClass = defaultOauthSaslLoginCallbackHandlerClass + } + return k +} + +func isOauthbearerStream(sp *kafkamod.SecurityProfile) bool { + return sp != nil && + sp.SecurityProtocol == securityProtocolSASLSSL && + sp.SaslMechanism == saslMechanismOauthbearer && + sp.SSLCertSecret != "" && + sp.SSLTruststorePasswordDetails != nil +} + +func isPlainOrScramStream(sp *kafkamod.SecurityProfile) bool { + if sp == nil { + return false + } + protoOK := sp.SecurityProtocol == securityProtocolSASLPlaintext || sp.SecurityProtocol == securityProtocolSASLSSL + mechOK := sp.SaslMechanism == saslMechanismPlain || sp.SaslMechanism == saslMechanismScram + return protoOK && mechOK +} + +// usesSSLMaterial reports whether the stream presents a truststore. odin keys +// this off the security protocol containing "SSL", covering both SSL and +// SASL_SSL. +func usesSSLMaterial(sp *kafkamod.SecurityProfile) bool { + return sp != nil && strings.Contains(sp.SecurityProtocol, securityProtocolSSL) +} + +// hasSecurityProfile reports whether the profile requires any SASL/SSL wiring. +func hasSecurityProfile(sp *kafkamod.SecurityProfile) bool { + return sp != nil && sp.SecurityProtocol != "" && !strings.EqualFold(sp.SecurityProtocol, "PLAINTEXT") +} + +// truststoreFileName is the file name the truststore is projected as, and also +// the key it is read from inside the cert secret. +func truststoreFileName(truststoreType string) string { + if strings.EqualFold(truststoreType, truststoreTypePKCS12) { + return truststoreFileBase + ".p12" + } + return truststoreFileBase + ".jks" +} + +// buildSecurityConfigs builds the SOURCE_KAFKA_CONSUMER_CONFIG_* env variables +// for the source stream. Returns nil for plaintext streams so env variables +// stay unchanged. +// +// No secret value is ever placed here: the truststore password reaches the +// container as a secretKeyRef env var and the SCRAM credentials as a mounted +// jaas.conf, both described by the ACLConfig chart values. +func buildSecurityConfigs(sp *kafkamod.SecurityProfile, sec KafkaSecurity) map[string]string { + if !hasSecurityProfile(sp) { + return nil + } + sec = sec.withDefaults() + + cfg := map[string]string{} + cfg[keyConsumerSecurityProtocol] = sp.SecurityProtocol + if sp.SaslMechanism != "" { + cfg[keyConsumerSaslMechanism] = sp.SaslMechanism + } + + if usesSSLMaterial(sp) { + if sp.SSLProtocol != "" { + cfg[keyConsumerSSLProtocol] = sp.SSLProtocol + } + if sp.SSLTruststoreType != "" { + cfg[keyConsumerSSLTruststoreType] = sp.SSLTruststoreType + } + if sp.SSLCertSecret != "" { + fileName := truststoreFileName(sp.SSLTruststoreType) + cfg[keyConsumerSSLTruststoreLocation] = certMountPath + "/" + fileName + // the chart selects this key out of the cert secret and projects it + // under the same name. + cfg[keyConsumerSSLTruststoreFilename] = fileName + } + } + + // OAUTHBEARER authenticates with the projected service-account token, so the + // JAAS config is a fixed module string rather than credentials. + if sp.SaslMechanism == saslMechanismOauthbearer { + cfg[keyConsumerSaslJaasConfig] = oauthConsumerSaslJaasConfig + cfg[keyConsumerSaslLoginCallbackHandler] = sec.SaslLoginCallbackHandlerClass + } + + return cfg +} + +// ACLConfig is the chart-facing description of a stream's security material. +// Every field is a reference to a secret that already exists in the target +// namespace — no secret value passes through entropy. +type ACLConfig struct { + // SSLConfigCredential is the secret holding the truststore. The chart mounts + // it at /etc/secret, selecting TruststoreFilename as both key and path. + SSLConfigCredential string `json:"ssl_config_credential,omitempty"` + TruststoreFilename string `json:"truststore_filename,omitempty"` + + // TruststorePassword is rendered by the chart as a secretKeyRef env var for + // SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD. + TruststorePassword *SecretKeyRef `json:"truststore_password,omitempty"` + + // JaasConfigCredential is the secret holding jaas.conf for PLAIN/SCRAM + // streams, mounted at /etc/secret/kafka. + JaasConfigCredential string `json:"jaas_config_credential,omitempty"` + + // KafkaTokenEnabled requests the projected kafka service-account token + // (audience "kafka") that OAUTHBEARER authenticates with. + KafkaTokenEnabled bool `json:"kafka_token_enabled,omitempty"` +} + +// SecretKeyRef references a single key inside an existing secret. +type SecretKeyRef struct { + SecretName string `json:"secretName"` + Key string `json:"key"` +} + +// buildACLConfig derives the chart values for the stream's security material. +// Returns nil when the stream needs none, so the rendered release is unchanged +// for plaintext firehoses. +func buildACLConfig(streamName string, sp *kafkamod.SecurityProfile, team string) *ACLConfig { + if !hasSecurityProfile(sp) { + return nil + } + + acl := &ACLConfig{} + + if usesSSLMaterial(sp) && sp.SSLCertSecret != "" { + acl.SSLConfigCredential = sp.SSLCertSecret + acl.TruststoreFilename = truststoreFileName(sp.SSLTruststoreType) + if sp.SSLTruststorePasswordDetails != nil && sp.SSLTruststorePasswordDetails.SecretName != "" { + acl.TruststorePassword = &SecretKeyRef{ + SecretName: sp.SSLTruststorePasswordDetails.SecretName, + Key: sp.SSLTruststorePasswordDetails.Key, + } + } + } + + if sp.SaslMechanism == saslMechanismOauthbearer { + acl.KafkaTokenEnabled = true + } + + // PLAIN/SCRAM read their credentials from a jaas.conf in a secret. The + // profile names it explicitly when known; otherwise fall back to odin's + // --jaas convention. + if isPlainOrScramStream(sp) { + if secretName := jaasSecretName(streamName, sp, team); secretName != "" { + acl.JaasConfigCredential = secretName + } + } + + if *acl == (ACLConfig{}) { + return nil + } + return acl +} + +func jaasSecretName(streamName string, sp *kafkamod.SecurityProfile, team string) string { + if cred, ok := sp.ACLs[team]; ok && cred.SecretName != "" { + return cred.SecretName + } + if team == "" || streamName == "" { + return "" + } + return strings.ReplaceAll(strings.Join([]string{team, streamName, jaasSecretSuffix}, "-"), "_", "-") +} + +// applyStreamSecurity resolves the source stream's kafka security profile, +// injects the consumer config into the env variables, and records the chart's +// ACL values on conf. It is a no-op for firehoses that do not name a kafka +// stream, and clears the wiring for streams that no longer carry a profile. +func (fd *firehoseDriver) applyStreamSecurity(ctx context.Context, exr module.ExpandedResource, conf *Config) error { + streamName := conf.StreamName + if streamName == "" { + return nil + } + + if conf.EnvVariables == nil { + conf.EnvVariables = map[string]string{} + } + clearManagedSecurityConfigs(conf.EnvVariables) + conf.ACL = nil + + security, err := fd.resolveStreamSecurity(ctx, exr, conf, streamName) + if err != nil { + return err + } + + // naming a stream relaxes the schema's SOURCE_KAFKA_BROKERS requirement, + // because the brokers are meant to come from the stream. If neither the + // payload nor the resolution supplied them, fail here rather than deploy a + // firehose that cannot reach any broker. + if conf.EnvVariables[confKeyKafkaBrokers] == "" { + return fmt.Errorf( + "%s is not set and kafka stream %q resolved to no url: set it explicitly", + confKeyKafkaBrokers, streamName) + } + + if !hasSecurityProfile(security) { + return nil + } + + for key, val := range buildSecurityConfigs(security, fd.conf.KafkaSecurity) { + conf.EnvVariables[key] = val + } + conf.ACL = buildACLConfig(streamName, security, conf.Team) + + // PLAIN/SCRAM point the JVM at the mounted jaas.conf. + if conf.ACL != nil && conf.ACL.JaasConfigCredential != "" { + conf.EnvVariables[keyJavaOptions] = withJaasJavaOption(conf.EnvVariables[keyJavaOptions]) + } + + if conf.ServiceAccount == "" { + conf.ServiceAccount = fd.conf.KafkaSecurity.ServiceAccount + } + + return nil +} + +// clearManagedSecurityConfigs drops everything a previous plan injected: this +// module owns the SASL/SSL keys for a stream-backed firehose. +func clearManagedSecurityConfigs(env map[string]string) { + for _, key := range managedSecurityKeys { + delete(env, key) + } + // provider class keys are named after the provider, so sweep by shape. + for key := range env { + if strings.HasPrefix(key, keyConsumerConfigProviderClassPrefix) && + strings.HasSuffix(key, keyConsumerConfigProviderClassSuffix) { + delete(env, key) + } + } + if opts := withoutJaasJavaOption(env[keyJavaOptions]); opts != "" { + env[keyJavaOptions] = opts + } else if _, ok := env[keyJavaOptions]; ok { + env[keyJavaOptions] = "" + } +} + +// withJaasJavaOption appends the JAAS location option, keeping the rest of the +// user's _JAVA_OPTIONS and never duplicating the option. +func withJaasJavaOption(opts string) string { + opts = withoutJaasJavaOption(opts) + if opts == "" { + return jaasConfigJavaOpt + } + return opts + " " + jaasConfigJavaOpt +} + +func withoutJaasJavaOption(opts string) string { + if !strings.Contains(opts, jaasConfigJavaOpt) { + return opts + } + fields := strings.Fields(opts) + kept := fields[:0] + for _, f := range fields { + if f != jaasConfigJavaOpt { + kept = append(kept, f) + } + } + return strings.Join(kept, " ") +} + +// resolveStreamSecurity resolves the source stream's kafka security profile. +// +// Resolution order: an inline conf.StreamSecurity entry, then a declared kafka +// dependency (raw-Entropy path), then — when the resource carries the +// stream_security_enabled flag — the kafka resource fetched internally by URN +// via fd.getResource, with no dependency declared. Streams with none of these +// are plaintext. +func (fd *firehoseDriver) resolveStreamSecurity(ctx context.Context, exr module.ExpandedResource, + conf *Config, streamName string, +) (*kafkamod.SecurityProfile, error) { + // 1. inline profile, if Dex prefetched one. + if security := conf.StreamSecurity[streamName]; security != nil { + return security, nil + } + + // 2. declared kafka dependency (raw-Entropy path): also carries the URL. + if dep, ok := exr.Dependencies[streamName]; ok && dep.Kind == kafkamod.Module.Kind { + var out kafkamod.Output + if err := json.Unmarshal(dep.Output, &out); err != nil { + return nil, fmt.Errorf("invalid kafka dependency output for stream %q: %w", streamName, err) + } + setKafkaBrokers(conf, out.URL) + return out.Security, nil + } + + // 3. the caller flagged a secured stream: fetch the kafka resource internally + // by URN and read its profile. Left unflagged the stream is plaintext, so + // nothing is fetched and the wiping above is all that happens. + if conf.StreamSecurityEnabled { + out, err := fd.fetchKafkaOutput(ctx, exr.Resource.Project, streamName) + if err != nil { + return nil, err + } + setKafkaBrokers(conf, out.URL) + return out.Security, nil + } + + return nil, nil +} + +// fetchKafkaOutput fetches the kafka stream's resource by URN and decodes its +// Output (url + security profile). streamName is the kafka resource name. +func (fd *firehoseDriver) fetchKafkaOutput(ctx context.Context, project, streamName string) (kafkamod.Output, error) { + var out kafkamod.Output + if fd.getResource == nil { + return out, fmt.Errorf("cannot resolve kafka stream %q: resource getter not configured", streamName) + } + + urn := resource.GenerateURN(kafkamod.Module.Kind, project, streamName) + res, err := fd.getResource(ctx, urn) + if err != nil { + return out, fmt.Errorf("failed to fetch kafka stream %q (%s): %w", streamName, urn, err) + } + if err := json.Unmarshal(res.State.Output, &out); err != nil { + return out, fmt.Errorf("invalid kafka output for stream %q: %w", streamName, err) + } + return out, nil +} + +// setKafkaBrokers fills SOURCE_KAFKA_BROKERS from the resolved stream URL, +// leaving an explicitly configured value untouched. +func setKafkaBrokers(conf *Config, url string) { + if url == "" || conf.EnvVariables[confKeyKafkaBrokers] != "" { + return + } + conf.EnvVariables[confKeyKafkaBrokers] = url +} diff --git a/modules/firehose/kafka_security_test.go b/modules/firehose/kafka_security_test.go new file mode 100644 index 00000000..fcf5a137 --- /dev/null +++ b/modules/firehose/kafka_security_test.go @@ -0,0 +1,323 @@ +package firehose + +import ( + "context" + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/goto/entropy/core/module" + "github.com/goto/entropy/core/resource" + kafkamod "github.com/goto/entropy/modules/kafka" +) + +const pocStream = "al-gp-id-s-central-kf" + +func oauthbearerProfile() *kafkamod.SecurityProfile { + return &kafkamod.SecurityProfile{ + SecurityProtocol: "SASL_SSL", + SaslMechanism: "OAUTHBEARER", + SSLProtocol: "SSL", + SSLTruststoreType: "PKCS12", + SSLCertSecret: "kafka-central-cert", + SSLTruststorePasswordDetails: &kafkamod.SecretKeyRef{ + SecretName: "scp-kafka-ssl-secrets", + Key: "truststore_password", + }, + } +} + +// the injected consumer config matches odin's: fixed /etc/secret truststore +// location plus the filename the chart selects out of the cert secret, and no +// password (it arrives as a secretKeyRef env var). +func TestBuildSecurityConfigs_OAUTHBEARER_MatchesOdin(t *testing.T) { + got := buildSecurityConfigs(oauthbearerProfile(), KafkaSecurity{}) + + want := map[string]string{ + "SOURCE_KAFKA_CONSUMER_CONFIG_SECURITY_PROTOCOL": "SASL_SSL", + "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_MECHANISM": "OAUTHBEARER", + "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_JAAS_CONFIG": "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;", + "SOURCE_KAFKA_CONSUMER_CONFIG_SASL_LOGIN_CALLBACK_HANDLER_CLASS": defaultOauthSaslLoginCallbackHandlerClass, + "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_PROTOCOL": "SSL", + "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_TYPE": "PKCS12", + "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_LOCATION": "/etc/secret/kafka-cert/truststore.p12", + "SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_FILENAME": "truststore.p12", + } + + assert.Equal(t, want, got) +} + +// no secret value is ever injected into the config. +func TestBuildSecurityConfigs_NeverInlinesSecrets(t *testing.T) { + got := buildSecurityConfigs(oauthbearerProfile(), KafkaSecurity{}) + + for key, val := range got { + assert.NotContains(t, val, "scp-kafka-ssl-secrets", "secret name leaked into %s", key) + assert.NotContains(t, val, "truststore_password", "password key leaked into %s", key) + } +} + +// JKS streams get the .jks extension in both the location and the filename. +func TestBuildSecurityConfigs_JKSTruststore(t *testing.T) { + sp := oauthbearerProfile() + sp.SSLTruststoreType = "JKS" + + got := buildSecurityConfigs(sp, KafkaSecurity{}) + + assert.Equal(t, "/etc/secret/kafka-cert/truststore.jks", got[keyConsumerSSLTruststoreLocation]) + assert.Equal(t, "truststore.jks", got[keyConsumerSSLTruststoreFilename]) +} + +func TestBuildACLConfig_OAUTHBEARER(t *testing.T) { + acl := buildACLConfig(pocStream, oauthbearerProfile(), "team-x") + + require.NotNil(t, acl) + assert.Equal(t, "kafka-central-cert", acl.SSLConfigCredential) + assert.Equal(t, "truststore.p12", acl.TruststoreFilename) + assert.Equal(t, &SecretKeyRef{SecretName: "scp-kafka-ssl-secrets", Key: "truststore_password"}, acl.TruststorePassword) + assert.True(t, acl.KafkaTokenEnabled) + assert.Empty(t, acl.JaasConfigCredential) +} + +// a plaintext stream produces no consumer config and no ACL values. +func TestPlaintextStream_NoSecurityWiring(t *testing.T) { + assert.Nil(t, buildSecurityConfigs(nil, KafkaSecurity{})) + assert.Nil(t, buildSecurityConfigs(&kafkamod.SecurityProfile{}, KafkaSecurity{})) + assert.Nil(t, buildSecurityConfigs(&kafkamod.SecurityProfile{SecurityProtocol: "PLAINTEXT"}, KafkaSecurity{})) + + assert.Nil(t, buildACLConfig(pocStream, nil, "team-x")) + assert.Nil(t, buildACLConfig(pocStream, &kafkamod.SecurityProfile{SecurityProtocol: "PLAINTEXT"}, "team-x")) +} + +// brokers are populated from the resolved stream URL when not set, and the +// consumer config is injected into the env variables. +func TestApplyStreamSecurity_PopulatesBrokersAndConfig(t *testing.T) { + out := kafkamod.Output{URL: "broker-1:9098,broker-2:9098", Security: oauthbearerProfile()} + outJSON, err := json.Marshal(out) + require.NoError(t, err) + + exr := module.ExpandedResource{ + Dependencies: map[string]module.ResolvedDependency{ + pocStream: {Kind: kafkamod.Module.Kind, Output: outJSON}, + }, + } + conf := &Config{ + Team: "team-x", + StreamName: pocStream, + EnvVariables: map[string]string{}, + } + + require.NoError(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), exr, conf)) + + assert.Equal(t, "broker-1:9098,broker-2:9098", conf.EnvVariables[confKeyKafkaBrokers]) + assert.Equal(t, "SASL_SSL", conf.EnvVariables[keyConsumerSecurityProtocol]) + require.NotNil(t, conf.ACL) + assert.True(t, conf.ACL.KafkaTokenEnabled) +} + +// product (Dex) path: the security profile is inlined on conf.StreamSecurity +// with NO kafka dependency present, and the ACL wiring still fires. +func TestApplyStreamSecurity_InlineProfile_NoDependency(t *testing.T) { + conf := &Config{ + Team: "team-x", + StreamName: pocStream, + // an inline profile carries no url, so brokers come from the payload. + EnvVariables: map[string]string{confKeyKafkaBrokers: "broker-1:9098"}, + StreamSecurity: map[string]*kafkamod.SecurityProfile{ + pocStream: oauthbearerProfile(), + }, + } + + require.NoError(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), module.ExpandedResource{}, conf)) + + assert.Equal(t, "SASL_SSL", conf.EnvVariables[keyConsumerSecurityProtocol]) + assert.Equal(t, "OAUTHBEARER", conf.EnvVariables[keyConsumerSaslMechanism]) + require.NotNil(t, conf.ACL) + assert.Equal(t, "kafka-central-cert", conf.ACL.SSLConfigCredential) +} + +// the flag makes the driver fetch the kafka resource by URN, with no dependency +// declared — the Dex product path. +func TestApplyStreamSecurity_FlagFetchesInternally(t *testing.T) { + out := kafkamod.Output{URL: "127.0.0.1:9098", Security: oauthbearerProfile()} + outJSON, err := json.Marshal(out) + require.NoError(t, err) + + var gotURN string + fd := &firehoseDriver{ + getResource: func(_ context.Context, urn string) (*resource.Resource, error) { + gotURN = urn + return &resource.Resource{State: resource.State{Output: outJSON}}, nil + }, + conf: driverConf{KafkaSecurity: KafkaSecurity{ServiceAccount: "aegis-kafka"}}, + } + + exr := module.ExpandedResource{Resource: resource.Resource{Project: "al-dp-id-s"}} + conf := &Config{ + Team: "team-x", + StreamName: pocStream, + StreamSecurityEnabled: true, + EnvVariables: map[string]string{}, + } + + require.NoError(t, fd.applyStreamSecurity(context.Background(), exr, conf)) + + assert.Equal(t, resource.GenerateURN(kafkamod.Module.Kind, "al-dp-id-s", pocStream), gotURN) + assert.Equal(t, "127.0.0.1:9098", conf.EnvVariables[confKeyKafkaBrokers]) + assert.Equal(t, "SASL_SSL", conf.EnvVariables[keyConsumerSecurityProtocol]) + assert.Equal(t, "aegis-kafka", conf.ServiceAccount) +} + +// a plaintext firehose (no stream name, no stream_security, no dependency) is +// left untouched — env variables, ACL values and service account unchanged. +func TestApplyStreamSecurity_PlaintextFirehose_NoWiring(t *testing.T) { + conf := &Config{ + Team: "team-x", + EnvVariables: map[string]string{ + confKeyKafkaBrokers: "localhost:9092", + confKeyKafkaTopic: "foo-log", + }, + } + + require.NoError(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), module.ExpandedResource{}, conf)) + + assert.Equal(t, map[string]string{ + confKeyKafkaBrokers: "localhost:9092", + confKeyKafkaTopic: "foo-log", + }, conf.EnvVariables) + assert.Nil(t, conf.ACL) + assert.Empty(t, conf.ServiceAccount) +} + +// naming a stream relaxes the schema's brokers requirement, so a stream that +// resolves to nothing must fail the plan rather than deploy without brokers. +func TestApplyStreamSecurity_MissingBrokersFails(t *testing.T) { + // the stream resolves, but carries no url and the payload has no brokers. + outJSON, err := json.Marshal(kafkamod.Output{}) + require.NoError(t, err) + fd := &firehoseDriver{ + getResource: func(context.Context, string) (*resource.Resource, error) { + return &resource.Resource{State: resource.State{Output: outJSON}}, nil + }, + } + conf := &Config{ + StreamName: pocStream, + StreamSecurityEnabled: true, + EnvVariables: map[string]string{confKeyKafkaTopic: "foo-log"}, + } + + err = fd.applyStreamSecurity(context.Background(), module.ExpandedResource{}, conf) + require.Error(t, err) + assert.Contains(t, err.Error(), confKeyKafkaBrokers) + + // same through a declared dependency whose output has no url. + depJSON, err := json.Marshal(kafkamod.Output{Security: oauthbearerProfile()}) + require.NoError(t, err) + exr := module.ExpandedResource{ + Dependencies: map[string]module.ResolvedDependency{ + pocStream: {Kind: kafkamod.Module.Kind, Output: depJSON}, + }, + } + conf = &Config{StreamName: pocStream, EnvVariables: map[string]string{}} + + require.Error(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), exr, conf)) +} + +// an explicit brokers value is not overwritten by the resolved stream URL. +func TestApplyStreamSecurity_KeepsExplicitBrokers(t *testing.T) { + out := kafkamod.Output{URL: "resolved:9098"} + outJSON, err := json.Marshal(out) + require.NoError(t, err) + + exr := module.ExpandedResource{ + Dependencies: map[string]module.ResolvedDependency{ + pocStream: {Kind: kafkamod.Module.Kind, Output: outJSON}, + }, + } + conf := &Config{ + StreamName: pocStream, + EnvVariables: map[string]string{confKeyKafkaBrokers: "explicit:9092"}, + } + + require.NoError(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), exr, conf)) + assert.Equal(t, "explicit:9092", conf.EnvVariables[confKeyKafkaBrokers]) +} + +// a stream that loses its security profile has the previously injected keys, +// ACL values and JAAS java option cleared instead of left behind — including +// the config-provider keys written by an older build. +func TestApplyStreamSecurity_ClearsStaleWiring(t *testing.T) { + conf := &Config{ + StreamName: pocStream, + StreamSecurityEnabled: true, + EnvVariables: map[string]string{ + keyConsumerSecurityProtocol: "SASL_SSL", + keyConsumerSaslMechanism: "SCRAM-SHA-512", + keyConsumerConfigProviders: "literalfile", + "SOURCE_KAFKA_CONSUMER_CONFIG_CONFIG_PROVIDERS_LITERALFILE_CLASS": "com.gtf.dagger.kafka.configproviders.LiteralFileConfigProvider", + keyJavaOptions: "-Xmx1250m " + jaasConfigJavaOpt, + confKeyKafkaTopic: "foo-log", + confKeyKafkaBrokers: "broker-1:9092", + }, + ACL: &ACLConfig{SSLConfigCredential: "stale"}, + } + + // the stream now resolves without a security profile. + outJSON, err := json.Marshal(kafkamod.Output{URL: "broker-1:9092"}) + require.NoError(t, err) + fd := &firehoseDriver{ + getResource: func(context.Context, string) (*resource.Resource, error) { + return &resource.Resource{State: resource.State{Output: outJSON}}, nil + }, + } + + require.NoError(t, fd.applyStreamSecurity(context.Background(), module.ExpandedResource{}, conf)) + + assert.Equal(t, map[string]string{ + keyJavaOptions: "-Xmx1250m", + confKeyKafkaTopic: "foo-log", + confKeyKafkaBrokers: "broker-1:9092", + }, conf.EnvVariables) + assert.Nil(t, conf.ACL) +} + +// SCRAM streams read credentials from a mounted jaas.conf: no JAAS config env +// variable, a jaas secret to mount, and the JVM option pointing at it. +func TestApplyStreamSecurity_ScramUsesJaasFile(t *testing.T) { + sp := &kafkamod.SecurityProfile{ + SecurityProtocol: "SASL_PLAINTEXT", + SaslMechanism: "SCRAM-SHA-512", + ACLs: map[string]kafkamod.ACLCredentialRef{ + "team-x": {SecretName: "team-x-creds", UsernameKey: "username", PasswordKey: "password"}, + }, + } + conf := &Config{ + Team: "team-x", + StreamName: pocStream, + EnvVariables: map[string]string{ + keyJavaOptions: "-Xmx1250m", + confKeyKafkaBrokers: "broker-1:9092", + }, + StreamSecurity: map[string]*kafkamod.SecurityProfile{pocStream: sp}, + } + + require.NoError(t, (&firehoseDriver{}).applyStreamSecurity(context.Background(), module.ExpandedResource{}, conf)) + + assert.NotContains(t, conf.EnvVariables, keyConsumerSaslJaasConfig) + assert.Equal(t, "-Xmx1250m "+jaasConfigJavaOpt, conf.EnvVariables[keyJavaOptions]) + require.NotNil(t, conf.ACL) + assert.Equal(t, "team-x-creds", conf.ACL.JaasConfigCredential) + assert.False(t, conf.ACL.KafkaTokenEnabled) +} + +// without an explicit credential secret, the jaas secret falls back to odin's +// --jaas convention, with underscores normalised to dashes. +func TestJaasSecretName_OdinConvention(t *testing.T) { + sp := &kafkamod.SecurityProfile{SecurityProtocol: "SASL_PLAINTEXT", SaslMechanism: "SCRAM-SHA-512"} + + assert.Equal(t, "team-x-al-gp-id-s-central-kf-jaas", jaasSecretName(pocStream, sp, "team-x")) + assert.Equal(t, "team-x-my-stream-jaas", jaasSecretName("my_stream", sp, "team_x")) + assert.Empty(t, jaasSecretName(pocStream, sp, "")) +} diff --git a/modules/firehose/module.go b/modules/firehose/module.go index e01a11d8..7169ee4e 100644 --- a/modules/firehose/module.go +++ b/modules/firehose/module.go @@ -33,100 +33,106 @@ const ( var mu sync.Mutex -var Module = module.Descriptor{ - Kind: "firehose", - Dependencies: map[string]string{ - keyKubeDependency: kubernetes.Module.Kind, - }, - Actions: []module.ActionDesc{ - { - Name: module.CreateAction, - Description: "Creates a new firehose", +// Module builds the firehose module descriptor. getResource lets the driver +// fetch a referenced kafka stream's resource internally (by URN) to resolve its +// security profile, without declaring it as a dependency. +func Module(getResource ResourceGetter) module.Descriptor { + return module.Descriptor{ + Kind: "firehose", + Dependencies: map[string]string{ + keyKubeDependency: kubernetes.Module.Kind, }, - { - Name: module.UpdateAction, - Description: "Update all configurations of firehose", - }, - { - Name: ResetAction, - Description: "Stop firehose, reset consumer group, restart", - }, - { - Name: ResetV2Action, - Description: "Stop firehose, reset consumer group, restart with datetime option", - }, - { - Name: StopAction, - Description: "Stop all replicas of this firehose.", - }, - { - Name: StartAction, - Description: "Start the firehose if it is currently stopped.", - }, - { - Name: ScaleAction, - Description: "Scale the number of replicas to given number.", - }, - { - Name: UpgradeAction, - Description: "Upgrade firehose version", + Actions: []module.ActionDesc{ + { + Name: module.CreateAction, + Description: "Creates a new firehose", + }, + { + Name: module.UpdateAction, + Description: "Update all configurations of firehose", + }, + { + Name: ResetAction, + Description: "Stop firehose, reset consumer group, restart", + }, + { + Name: ResetV2Action, + Description: "Stop firehose, reset consumer group, restart with datetime option", + }, + { + Name: StopAction, + Description: "Stop all replicas of this firehose.", + }, + { + Name: StartAction, + Description: "Start the firehose if it is currently stopped.", + }, + { + Name: ScaleAction, + Description: "Scale the number of replicas to given number.", + }, + { + Name: UpgradeAction, + Description: "Upgrade firehose version", + }, }, - }, - DriverFactory: func(confJSON json.RawMessage) (module.Driver, error) { - mu.Lock() - defer mu.Unlock() - - conf := defaultDriverConf // clone the default value - if err := json.Unmarshal(confJSON, &conf); err != nil { - return nil, err - } else if err := validator.TaggedStruct(conf); err != nil { - return nil, err - } + DriverFactory: func(confJSON json.RawMessage) (module.Driver, error) { + mu.Lock() + defer mu.Unlock() + + conf := defaultDriverConf // clone the default value + if err := json.Unmarshal(confJSON, &conf); err != nil { + return nil, err + } else if err := validator.TaggedStruct(conf); err != nil { + return nil, err + } - return &firehoseDriver{ - conf: conf, - timeNow: time.Now, - kubeDeploy: func(_ context.Context, isCreate bool, kubeConf kube.Config, hc helm.ReleaseConfig) error { - canUpdate := func(rel *release.Release) bool { - curLabels, ok := rel.Config[labelsConfKey].(map[string]any) - if !ok { - return false - } - newLabels, ok := hc.Values[labelsConfKey].(map[string]string) - if !ok { - return false + return &firehoseDriver{ + conf: conf, + timeNow: time.Now, + kubeDeploy: func(_ context.Context, isCreate bool, kubeConf kube.Config, hc helm.ReleaseConfig) error { + canUpdate := func(rel *release.Release) bool { + curLabels, ok := rel.Config[labelsConfKey].(map[string]any) + if !ok { + return false + } + newLabels, ok := hc.Values[labelsConfKey].(map[string]string) + if !ok { + return false + } + + isManagedByEntropy := curLabels[labelOrchestrator] == orchestratorLabelValue + isSameDeployment := curLabels[labelDeployment] == newLabels[labelDeployment] + + return isManagedByEntropy && isSameDeployment } - isManagedByEntropy := curLabels[labelOrchestrator] == orchestratorLabelValue - isSameDeployment := curLabels[labelDeployment] == newLabels[labelDeployment] - - return isManagedByEntropy && isSameDeployment - } - - helmCl := helm.NewClient(&helm.Config{Kubernetes: kubeConf}) - _, errHelm := helmCl.Upsert(&hc, canUpdate) - return errHelm - }, - kubeGetPod: func(ctx context.Context, conf kube.Config, ns string, labels map[string]string) ([]kube.Pod, error) { - kubeCl, err := kube.NewClient(ctx, conf) - if err != nil { - return nil, errors.ErrInternal.WithMsgf("failed to create new kube client on firehose driver kube get pod").WithCausef("%s", err.Error()) - } - return kubeCl.GetPodDetails(ctx, ns, labels, func(pod v1.Pod) bool { - // allow pods that are in running state and are not marked for deletion - return pod.Status.Phase == v1.PodRunning && pod.DeletionTimestamp == nil - }) - }, - kubeGetDeployment: func(ctx context.Context, conf kube.Config, ns, name string) (kube.Deployment, error) { - kubeCl, err := kube.NewClient(ctx, conf) - if err != nil { - return kube.Deployment{}, errors.ErrInternal.WithMsgf("failed to create new kube client on firehose driver kube get deployment").WithCausef("%s", err.Error()) - } - return kubeCl.GetDeploymentDetails(ctx, ns, name) - }, - consumerReset: consumerReset, - }, nil - }, + helmCl := helm.NewClient(&helm.Config{Kubernetes: kubeConf}) + _, errHelm := helmCl.Upsert(&hc, canUpdate) + return errHelm + }, + kubeGetPod: func(ctx context.Context, conf kube.Config, ns string, labels map[string]string) ([]kube.Pod, error) { + kubeCl, err := kube.NewClient(ctx, conf) + if err != nil { + return nil, errors.ErrInternal.WithMsgf("failed to create new kube client on firehose driver kube get pod").WithCausef("%s", err.Error()) + } + return kubeCl.GetPodDetails(ctx, ns, labels, func(pod v1.Pod) bool { + // allow pods that are in running state and are not marked for deletion + return pod.Status.Phase == v1.PodRunning && pod.DeletionTimestamp == nil + }) + }, + kubeGetDeployment: func(ctx context.Context, conf kube.Config, ns, name string) (kube.Deployment, error) { + kubeCl, err := kube.NewClient(ctx, conf) + if err != nil { + return kube.Deployment{}, errors.ErrInternal.WithMsgf("failed to create new kube client on firehose driver kube get deployment").WithCausef("%s", err.Error()) + } + return kubeCl.GetDeploymentDetails(ctx, ns, name) + }, + consumerReset: consumerReset, + getResource: getResource, + }, nil + }, + } } func consumerReset(ctx context.Context, conf Config, out kubernetes.Output, resetTo string, offsetResetDelaySeconds int) error { diff --git a/modules/firehose/module_test.go b/modules/firehose/module_test.go index 9e57a670..96c16206 100644 --- a/modules/firehose/module_test.go +++ b/modules/firehose/module_test.go @@ -8,6 +8,8 @@ import ( ) func BenchmarkDriverFactory(b *testing.B) { + desc := Module(nil) + b.SetParallelism(10000) b.RunParallel(func(pb *testing.PB) { for pb.Next() { @@ -18,7 +20,7 @@ func BenchmarkDriverFactory(b *testing.B) { config := json.RawMessage(configFile) - _, _ = Module.DriverFactory(config) + _, _ = desc.DriverFactory(config) } }) } diff --git a/modules/firehose/schema/config.json b/modules/firehose/schema/config.json index 8ce5cdfd..82e53233 100644 --- a/modules/firehose/schema/config.json +++ b/modules/firehose/schema/config.json @@ -6,6 +6,23 @@ "replicas", "env_variables" ], + "anyOf": [ + { + "properties": { + "env_variables": { + "required": [ + "SOURCE_KAFKA_BROKERS" + ] + } + } + }, + { + "$comment": "brokers may be omitted when the kafka stream is named: they are resolved from the stream's output during plan.", + "required": [ + "stream_name" + ] + } + ], "properties": { "stop_time": { "type": "string", @@ -19,13 +36,58 @@ "deployment_id": { "type": "string" }, + "team": { + "type": "string" + }, + "stream_name": { + "type": "string" + }, + "stream_security_enabled": { + "type": "boolean" + }, + "service_account": { + "type": "string" + }, + "stream_security": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": true, + "properties": { + "security_protocol": { "type": "string" }, + "sasl_mechanism": { "type": "string" }, + "ssl_protocol": { "type": "string" }, + "ssl_truststore_type": { "type": "string" }, + "ssl_cert_secret": { "type": "string" }, + "ssl_truststore_password_details": { + "type": "object", + "additionalProperties": true, + "properties": { + "secretName": { "type": "string" }, + "key": { "type": "string" } + } + }, + "acls": { + "type": "object", + "additionalProperties": { + "type": "object", + "additionalProperties": true, + "properties": { + "secretName": { "type": "string" }, + "usernameKey": { "type": "string" }, + "passwordKey": { "type": "string" } + } + } + } + } + } + }, "env_variables": { "type": "object", "additionalProperties": true, "required": [ "SINK_TYPE", "INPUT_SCHEMA_PROTO_CLASS", - "SOURCE_KAFKA_BROKERS", "SOURCE_KAFKA_TOPIC" ], "properties": { diff --git a/test/testbench/test_data/resource/firehose_acl_resource.json b/test/testbench/test_data/resource/firehose_acl_resource.json new file mode 100644 index 00000000..bd86ae89 --- /dev/null +++ b/test/testbench/test_data/resource/firehose_acl_resource.json @@ -0,0 +1,32 @@ +{ + "kind": "firehose", + "name": "test-firehose-acl", + "project": "test-project-0", + "labels": { + "description": "test firehose consuming a SASL_SSL kafka stream" + }, + "spec": { + "configs": { + "chart_values": { + "image_pull_policy": "IfNotPresent" + }, + "env_variables": { + "SINK_TYPE": "LOG", + "INPUT_SCHEMA_PROTO_CLASS": "com.tests.TestMessage", + "SOURCE_KAFKA_TOPIC": "test-topic" + }, + "replicas": 1, + "namespace": "default", + "team": "test-team", + "stream_name": "test-kafka-acl", + "stream_security_enabled": true, + "service_account": "aegis-kafka" + }, + "dependencies": [ + { + "key": "kube_cluster", + "value": "orn:entropy:kubernetes:test-project-0:test-kube" + } + ] + } +} diff --git a/test/testbench/test_data/resource/kafka_acl_resource.json b/test/testbench/test_data/resource/kafka_acl_resource.json new file mode 100644 index 00000000..7405e8e8 --- /dev/null +++ b/test/testbench/test_data/resource/kafka_acl_resource.json @@ -0,0 +1,36 @@ +{ + "kind": "kafka", + "name": "test-kafka-acl", + "project": "test-project-0", + "labels": { + "description": "test kafka stream secured with SASL_SSL / OAUTHBEARER" + }, + "spec": { + "configs": { + "advertise_mode": { + "host": "9098", + "address": "9098" + }, + "brokers": [ + { + "name": "test-project-test-kafka-acl-01", + "host": "test-project-test-kafka-acl-01", + "address": "127.0.0.1" + } + ], + "type": "source", + "security": { + "security_protocol": "SASL_SSL", + "sasl_mechanism": "OAUTHBEARER", + "ssl_protocol": "SSL", + "ssl_truststore_type": "PKCS12", + "ssl_cert_secret": "kafka-central-cert", + "ssl_truststore_password_details": { + "secretName": "scp-kafka-ssl-secrets", + "key": "truststore_password" + } + } + }, + "dependencies": [] + } +}