LR-546 PII code implementation for data-products

ansible/inventory/env/group_vars/all.yml

@@ -101,3 +101,6 @@ s3_path_style_access: false
 s3_https_only: false
 s3_default_bucket_location: ""
 s3_storage_container: ""
+
+org_search_service_private_endpoint: "{{sunbird_learner_service_url}}/private/v2/org/search"
+tenant_preferance_read_private_service_endpoint: "{{sunbird_learner_service_url}}/private/v2/org/preferences/read"

ansible/roles/lern-data-products-deploy/templates/common.conf.j2

@@ -73,7 +73,6 @@ azure_token_client_secret="{{ media_service_azure_token_client_secret }}"
 elasticsearch.service.endpoint="http://{{groups['composite-search-cluster'][0]}}:9200"
 elasticsearch.index.compositesearch.name="{{ es_search_index }}"
 
-org.search.api.url="{{ channelSearchServiceEndpoint }}"
 org.search.api.key="{{ searchServiceAuthorizationToken }}"
 
 hierarchy.search.api.url="{{ hierarchySearchServiceUrl }}"
@@ -311,3 +310,6 @@ sunbird.course.optionalnodes="optionalnodes"
 sunbird.course.redis.host={{ groups['redisall'][0] }}
 sunbird.course.redis.port=6379
 sunbird.course.redis.relationCache.id=5
+
+org.search.private.api.url="{{ org_search_service_private_endpoint }}"
+tenant.pref.read.private.api.url="{{ tenant_preferance_read_private_service_endpoint }}"

ansible/roles/lern-data-products-deploy/templates/lern-run-job.j2

@@ -9,6 +9,7 @@ source lern-model-config.sh
 today=$(date "+%Y-%m-%d")
 
 libs_path="{{ analytics.home }}/models-{{ model_version }}/lern-data-products-1.0"
+file_path="lern{{ env }}.conf"
 
 get_report_job_model_name(){
   case "$1" in
@@ -72,6 +73,6 @@ echo "Starting the job - $1" >> "$DP_LOGS/$today-job-execution.log"
 
 echo "Job modelName - $job_id" >> "$DP_LOGS/$today-job-execution.log"
 
-nohup $SPARK_HOME/bin/spark-submit --master local[*] --jars $(echo ${libs_path}/lib/*.jar | tr ' ' ','),$MODELS_HOME/analytics-framework-2.0.jar,$MODELS_HOME/scruid_2.12-2.5.0.jar,$MODELS_HOME/batch-models-2.0.jar --class org.ekstep.analytics.job.JobExecutor $MODELS_HOME/batch-models-2.0.jar --model "$job_id" --config "$job_config$batchIds" >> "$DP_LOGS/$today-job-execution.log" 2>&1
+nohup $SPARK_HOME/bin/spark-submit --conf spark.driver.extraJavaOptions="-Dconfig.file=$MODELS_HOME/$file_path" --conf spark.executor.extraJavaOptions="-Dconfig.file=$MODELS_HOME/$file_path" --master local[*] --jars $(echo ${libs_path}/lib/*.jar | tr ' ' ','),$MODELS_HOME/analytics-framework-2.0.jar,$MODELS_HOME/scruid_2.12-2.5.0.jar,$MODELS_HOME/batch-models-2.0.jar --class org.ekstep.analytics.job.JobExecutor $MODELS_HOME/batch-models-2.0.jar --model "$job_id" --config "$job_config$batchIds" >> "$DP_LOGS/$today-job-execution.log" 2>&1
 
-echo "Job execution completed - $1" >> "$DP_LOGS/$today-job-execution.log"
+echo "Job execution completed - $1" >> "$DP_LOGS/$today-job-execution.log"

lern-data-products/pom.xml

@@ -351,6 +351,12 @@
 			<version>0.7.1</version>
 			<scope>test</scope>
 		</dependency>
+
+		<dependency>
+			<groupId>com.moparisthebest</groupId>
+			<artifactId>junidecode</artifactId>
+			<version>0.1.1</version>
+		</dependency>
 	</dependencies>
 	<build>
 		<sourceDirectory>src/main/scala</sourceDirectory>
@@ -461,6 +467,14 @@
 					</filters>
 				</configuration>
 			</plugin>
-		</plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>11</source>
+                    <target>11</target>
+                </configuration>
+            </plugin>
+        </plugins>
 	</build>
 </project>

lern-data-products/src/main/resources/application.conf

@@ -143,8 +143,6 @@ druid.deletesegment.path="/druid/coordinator/v1/datasources/"
 druid.content.consumption.query="{\"query\":\"SELECT COUNT(*) as \\\"play_sessions_count\\\", SUM(total_time_spent) as \\\"total_time_spent\\\", dimensions_pdata_id, object_id\\nFROM \\\"summary-events\\\"\\nWHERE \\\"dimensions_mode\\\" = 'play' AND \\\"dimensions_type\\\" ='content'\\nGROUP BY object_id, dimensions_pdata_id\"}"
 
 // TPD Configurations
-org.search.api.url="https://dev.sunbirded.org/api"
-org.search.api.path="/org/v1/search"
 druid.host="http://localhost:8082/druid/v2"
 elasticsearch.index.coursebatch.name="course-batch"
 //ETB Configurations
@@ -202,4 +200,7 @@ redis.user.index.source.key="id" # this will be used as key for redis
 cassandra.read.timeoutMS="500000"
 cassandra.query.retry.count="100"
 cassandra.input.consistency.level="LOCAL_QUORUM"
-## user cache indexer job Configuration - end ##
+## user cache indexer job Configuration - end ##
+
+org.search.private.api.url="{{sunbird_learner_service_url}}/private/v2/org/search"
+tenant.pref.read.private.api.url="{{sunbird_learner_service_url}}/private/v2/org/preferences/read"

lern-data-products/src/main/scala/org/sunbird/core/exception/APIException.scala

@@ -0,0 +1,5 @@
+package org.sunbird.core.exception
+
+class APIException(message: String, cause: Throwable) extends Exception(message, cause)
+
+class ServerException(code: String, msg: String, cause: Throwable = null) extends Exception(msg, cause)

lern-data-products/src/main/scala/org/sunbird/core/exhaust/OnDemandExhaustJob.scala

@@ -1,8 +1,5 @@
 package org.sunbird.core.exhaust
 
-import net.lingala.zip4j.ZipFile
-import net.lingala.zip4j.model.ZipParameters
-import net.lingala.zip4j.model.enums.EncryptionMethod
 import org.apache.commons.lang.StringUtils
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.sql.functions._
@@ -11,9 +8,8 @@ import org.ekstep.analytics.framework.Level.INFO
 import org.ekstep.analytics.framework.conf.AppConf
 import org.ekstep.analytics.framework.util.{CommonUtil, JobLogger}
 import org.ekstep.analytics.framework.{FrameworkContext, StorageConfig}
+import org.sunbird.core.util.DataSecurityUtil.{zipAndPasswordProtect}
 
-import java.io.File
-import java.nio.file.Paths
 import java.sql.{Connection, DriverManager, PreparedStatement, Timestamp}
 import java.util.Properties
 import java.util.concurrent.CompletableFuture
@@ -22,6 +18,7 @@ import java.util.function.Supplier
 case class JobRequest(tag: String, request_id: String, job_id: String, var status: String, request_data: String, requested_by: String, requested_channel: String,
                       dt_job_submitted: Long, var download_urls: Option[List[String]], var dt_file_created: Option[Long], var dt_job_completed: Option[Long],
                       var execution_time: Option[Long], var err_message: Option[String], var iteration: Option[Int], encryption_key: Option[String], var processed_batches : Option[String] = None) {
+
     def this() = this("", "", "", "", "", "", "", 0, None, None, None, None, None, None, None, None)
 }
 case class RequestStatus(channel: String, batchLimit: Long, fileLimit: Long)
@@ -114,32 +111,35 @@ trait OnDemandExhaustJob {
 
   }
 
-  def saveRequests(storageConfig: StorageConfig, requests: Array[JobRequest])(implicit conf: Configuration, fc: FrameworkContext) = {
-    val zippedRequests = for (request <- requests) yield processRequestEncryption(storageConfig, request)
+  def saveRequests(storageConfig: StorageConfig, requests: Array[JobRequest], reqOrgAndLevelDtl: List[(String, String, String)])(implicit conf: Configuration, fc: FrameworkContext) = {
+    val zippedRequests = for (request <- requests) yield {
+      val reqOrgAndLevel = reqOrgAndLevelDtl.filter(_._1 == request.request_id).headOption
+      processRequestEncryption(storageConfig, request, reqOrgAndLevel.getOrElse("", "", ""))
+    }
     updateRequests(zippedRequests)
   }
 
-  def saveRequestAsync(storageConfig: StorageConfig, request: JobRequest)(implicit conf: Configuration, fc: FrameworkContext): CompletableFuture[JobRequest] = {
+  def saveRequestAsync(storageConfig: StorageConfig, request: JobRequest, reqOrgAndLevel: (String, String, String))(implicit conf: Configuration, fc: FrameworkContext): CompletableFuture[JobRequest] = {
 
     CompletableFuture.supplyAsync(new Supplier[JobRequest]() {
       override def get() : JobRequest =  {
-        val res = CommonUtil.time(saveRequest(storageConfig, request))
+        val res = CommonUtil.time(saveRequest(storageConfig, request, reqOrgAndLevel))
         JobLogger.log("Request is zipped", Some(Map("requestId" -> request.request_id, "timeTakenForZip" -> res._1)), INFO)
         request
       }
     })
 
   }
 
-  def saveRequest(storageConfig: StorageConfig, request: JobRequest)(implicit conf: Configuration, fc: FrameworkContext): Boolean = {
-    updateRequest(processRequestEncryption(storageConfig, request))
+  def saveRequest(storageConfig: StorageConfig, request: JobRequest, reqOrgAndLevel: (String, String, String))(implicit conf: Configuration, fc: FrameworkContext): Boolean = {
+    updateRequest(processRequestEncryption(storageConfig, request, reqOrgAndLevel))
   }
 
-  def processRequestEncryption(storageConfig: StorageConfig, request: JobRequest)(implicit conf: Configuration, fc: FrameworkContext): JobRequest = {
+  def processRequestEncryption(storageConfig: StorageConfig, request: JobRequest, reqOrgAndLevel: (String, String, String))(implicit conf: Configuration, fc: FrameworkContext): JobRequest = {
     val downloadURLs = CommonUtil.time(for (url <- request.download_urls.getOrElse(List())) yield {
       if (zipEnabled())
         try {
-          zipAndEncrypt(url, storageConfig, request)
+          zipAndPasswordProtect(url, storageConfig, request, null, reqOrgAndLevel._3)
           url.replace(".csv", ".zip")
         } catch {
           case ex: Exception => ex.printStackTrace();
@@ -161,60 +161,6 @@ trait OnDemandExhaustJob {
 
   def canZipExceptionBeIgnored(): Boolean = true
 
-  @throws(classOf[Exception])
-  private def zipAndEncrypt(url: String, storageConfig: StorageConfig, request: JobRequest)(implicit conf: Configuration, fc: FrameworkContext): String = {
-
-    val path = Paths.get(url);
-    val storageService = fc.getStorageService(storageConfig.store, storageConfig.accountKey.getOrElse(""), storageConfig.secretKey.getOrElse(""));
-    val tempDir = AppConf.getConfig("spark_output_temp_dir") + request.request_id + "/"
-    val localPath = tempDir + path.getFileName;
-    fc.getHadoopFileUtil().delete(conf, tempDir);
-    val filePrefix = storageConfig.store.toLowerCase() match {
-      // $COVERAGE-OFF$ Disabling scoverage
-      case "s3" =>
-        CommonUtil.getS3File(storageConfig.container, "")
-      case "azure" =>
-        CommonUtil.getAzureFile(storageConfig.container, "", storageConfig.accountKey.getOrElse("azure_storage_key"))
-      case "gcloud" =>
-        CommonUtil.getGCloudFile(storageConfig.container, "")
-      // $COVERAGE-ON$ for case: local
-      case _ =>
-        storageConfig.fileName
-    }
-    val objKey = url.replace(filePrefix, "");
-    if (storageConfig.store.equals("local")) {
-      fc.getHadoopFileUtil().copy(filePrefix, localPath, conf)
-    }
-    // $COVERAGE-OFF$ Disabling scoverage
-    else {
-      storageService.download(storageConfig.container, objKey, tempDir, Some(false));
-    }
-    // $COVERAGE-ON$
-    val zipPath = localPath.replace("csv", "zip")
-    val zipObjectKey = objKey.replace("csv", "zip")
-    val zipLocalObjKey = url.replace("csv", "zip")
-
-    request.encryption_key.map(key => {
-      val zipParameters = new ZipParameters();
-      zipParameters.setEncryptFiles(true);
-      zipParameters.setEncryptionMethod(EncryptionMethod.ZIP_STANDARD); // AES encryption is not supported by default with various OS.
-      val zipFile = new ZipFile(zipPath, key.toCharArray());
-      zipFile.addFile(localPath, zipParameters)
-    }).getOrElse({
-      new ZipFile(zipPath).addFile(new File(localPath));
-    })
-    val resultFile = if (storageConfig.store.equals("local")) {
-      fc.getHadoopFileUtil().copy(zipPath, zipLocalObjKey, conf)
-    }
-    // $COVERAGE-OFF$ Disabling scoverage
-    else {
-      storageService.upload(storageConfig.container, zipPath, zipObjectKey, Some(false), Some(0), Some(3), None);
-    }
-    // $COVERAGE-ON$
-    fc.getHadoopFileUtil().delete(conf, tempDir);
-    resultFile;
-  }
-
   def markRequestAsFailed(request: JobRequest, failedMsg: String, completed_Batches: Option[String] = None): JobRequest = {
     request.status = "FAILED";
     request.dt_job_completed = Option(System.currentTimeMillis());

lern-data-products/src/main/scala/org/sunbird/core/util/Constants.scala

@@ -33,9 +33,11 @@ object Constants {
     val LP_URL = AppConf.getConfig("lp.url")
     val SEARCH_SERVICE_URL = AppConf.getConfig("service.search.url")
     val COMPOSITE_SEARCH_URL = s"$SEARCH_SERVICE_URL" + AppConf.getConfig("service.search.path")
-    val ORG_SEARCH_URL: String = AppConf.getConfig("org.search.api.url") + AppConf.getConfig("org.search.api.path")
     val ORG_SEARCH_API_KEY: String = AppConf.getConfig("org.search.api.key")
     val USER_SEARCH_URL : String = AppConf.getConfig("user.search.api.url")
+    val TENANT_PREFERENCE_PRIVATE_READ_URL = AppConf.getConfig("tenant.pref.read.private.api.url")
+    val ORG_PRIVATE_SEARCH_URL: String = AppConf.getConfig("org.search.private.api.url")
+    val TEMP_DIR = AppConf.getConfig("spark_output_temp_dir")
 
     val HIERARCHY_STORE_KEY_SPACE_NAME = AppConf.getConfig("cassandra.hierarchy_store_prefix")+"hierarchy_store"
     val CONTENT_HIERARCHY_TABLE = "content_hierarchy"