datafilecollector/src/main/java/org/oran/datafile/configuration/AppConfig.java

   1 /*-
   2  * ============LICENSE_START======================================================================
   3  * Copyright (C) 2018, 2020-2022 Nokia. All rights reserved.
   4  * Copyright (C) 2018-2023 Nordix Foundation. All rights reserved.
   5  * ===============================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
   7  * in compliance with the License. You may obtain a copy of the License at
   8  *
   9  * http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software distributed under the License
  12  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  13  * or implied. See the License for the specific language governing permissions and limitations under
  14  * the License.
  15  * ============LICENSE_END========================================================================
  16  */
  17
  18 package org.oran.datafile.configuration;
  19
  20 import java.util.Map;
  21
  22 import lombok.Getter;
  23
  24 import lombok.Setter;
  25 import org.apache.kafka.clients.CommonClientConfigs;
  26 import org.apache.kafka.common.config.SaslConfigs;
  27 import org.apache.kafka.common.config.SslConfigs;
  28 import org.apache.kafka.common.security.auth.SecurityProtocol;
  29 import org.oran.datafile.oauth2.OAuthKafkaAuthenticateLoginCallbackHandler;
  30 import org.springframework.beans.factory.annotation.Value;
  31 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  32 import org.springframework.stereotype.Component;
  33
  34 /**
  35  * Holds all configuration for the DFC.
  36  */
  37
  38 @Component
  39 @EnableConfigurationProperties
  40 public class AppConfig {
  41
  42     @Value("${app.kafka.bootstrap-servers:}")
  43     private String kafkaBootStrapServers;
  44
  45     @Value("${app.kafka.collected-file-topic:}")
  46     @Getter
  47     private String collectedFileTopic;
  48
  49     @Value("${app.kafka.file-ready-event-topic:}")
  50     @Getter
  51     private String inputTopic;
  52
  53     @Value("${app.kafka.client-id:undefined}")
  54     @Getter
  55     private String kafkaClientId;
  56
  57     @Value("${app.collected-files-path}")
  58     @Getter
  59     @Setter
  60     private String collectedFilesPath;
  61
  62     @Value("${app.sftp.strict-host-key-checking:false}")
  63     private boolean strictHostKeyChecking;
  64
  65     @Value("${app.sftp.known-hosts-file-path:}")
  66     @Getter
  67     private String knownHostsFilePath;
  68
  69     @Value("${app.ssl.key-store-password-file}")
  70     private String clientKeyStorePassword = "";
  71
  72     @Value("${app.ssl.key-store:}")
  73     private String clientKeyStore = "";
  74
  75     @Value("${app.ssl.trust-store:}")
  76     private String clientTrustStore = "";
  77
  78     @Value("${app.ssl.trust-store-password-file:}")
  79     private String clientTrustStorePassword;
  80
  81     @Getter
  82     @Setter
  83     @Value("${app.s3.endpointOverride:}")
  84     private String s3EndpointOverride;
  85
  86     @Getter
  87     @Setter
  88     @Value("${app.s3.accessKeyId:}")
  89     private String s3AccessKeyId;
  90
  91     @Getter
  92     @Setter
  93     @Value("${app.s3.secretAccessKey:}")
  94     private String s3SecretAccessKey;
  95
  96     @Getter
  97     @Setter
  98     @Value("${app.s3.bucket:}")
  99     private String s3Bucket;
 100
 101     @Value("${app.s3.locksBucket:}")
 102     @Setter
 103     private String s3LocksBucket;
 104
 105     @Value("${app.number-of-worker-treads:200}")
 106     @Getter
 107     private int noOfWorkerThreads;
 108
 109     @Value("${app.kafka.ssl.key-store-location}")
 110     private String kafkaKeyStoreLocation;
 111
 112     @Value("${app.kafka.ssl.key-store-type}")
 113     private String kafkaKeyStoreType;
 114
 115     @Value("${app.kafka.ssl.key-store-password}")
 116     private String kafkaKeyStorePassword;
 117
 118     @Value("${app.kafka.ssl.trust-store-type}")
 119     private String kafkaTrustStoreType;
 120
 121     @Value("${app.kafka.ssl.trust-store-location}")
 122     private String kafkTrustStoreLocation;
 123
 124     @Value("${app.kafka.use-oath-token}")
 125     private boolean useOathToken;
 126
 127     public String getS3LocksBucket() {
 128         return s3LocksBucket.isEmpty() ? s3Bucket : s3LocksBucket;
 129     }
 130
 131     public boolean isS3Enabled() {
 132         return !s3EndpointOverride.isEmpty() && !s3Bucket.isEmpty();
 133     }
 134
 135     public String getKafkaBootStrapServers() {
 136         return kafkaBootStrapServers;
 137     }
 138
 139     public synchronized CertificateConfig getCertificateConfiguration() {
 140         return CertificateConfig.builder() //
 141             .trustedCa(this.clientTrustStore) //
 142             .trustedCaPasswordPath(this.clientTrustStorePassword) //
 143             .keyCert(this.clientKeyStore) //
 144             .keyPasswordPath(this.clientKeyStorePassword) //
 145             .build();
 146     }
 147
 148     public synchronized SftpConfig getSftpConfiguration() {
 149         return SftpConfig.builder() //
 150             .knownHostsFilePath(this.knownHostsFilePath) //
 151             .strictHostKeyChecking(this.strictHostKeyChecking) //
 152             .build();
 153     }
 154
 155     public void addKafkaSecurityProps(Map<String, Object> props) {
 156
 157         if (useOathToken) {
 158             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.name);
 159             props.put(SaslConfigs.SASL_MECHANISM, "OAUTHBEARER");
 160             props.put(SaslConfigs.SASL_LOGIN_CALLBACK_HANDLER_CLASS,
 161                 OAuthKafkaAuthenticateLoginCallbackHandler.class.getName());
 162             props.put(SaslConfigs.SASL_JAAS_CONFIG,
 163                 "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub=\"alice\"; ");
 164         }
 165         if (!kafkaKeyStoreLocation.isEmpty()) {
 166             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_SSL.name);
 167             // SSL
 168             props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, kafkaKeyStoreType);
 169             props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, kafkaKeyStoreLocation);
 170             if (!kafkaKeyStorePassword.isEmpty()) {
 171                 props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, kafkaKeyStorePassword);
 172             }
 173             if (!kafkTrustStoreLocation.isEmpty()) {
 174                 props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, kafkaTrustStoreType);
 175                 props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkTrustStoreLocation);
 176             }
 177         }
 178     }
 179
 180 }