LR-588 checking PII fields with respect to security levels

ansible/roles/lern-data-products-deploy/templates/lern-model-config.j2

@@ -55,16 +55,16 @@ config() {
     echo '{"search":{"type":"none"},"model":"org.sunbird.lms.audit.CollectionSummaryJobV2","modelParams":{"storageKeyConfig":"druid_storage_account_key","storageSecretConfig":"druid_storage_account_secret","batchSize":50,"generateForAllBatches":true,"contentFields":["identifier","name","organisation","channel","status","keywords","createdFor","medium","subject"],"contentStatus":["Live","Unlisted","Retired"],"store":"azure","specPath":"/mount/data/analytics/scripts/collection-summary-ingestion-spec.json","druidIngestionUrl":"'$druidIngestionURL'","sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}","sparkCassandraConnectionHost":"{{ core_cassandra_host }}","fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"Collection Summary Report V2"}'
     ;;
     "userinfo-exhaust")
-    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.UserInfoExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"UserInfo Exhaust"}'
+    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.UserInfoExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')","csvColumns":["courseid", "collectionName", "batchid", "batchName", "userid", "username", "state", "district", "orgname", "email", "phone","consentflag", "consentprovideddate", "block", "cluster", "usertype", "usersubtype", "schooludisecode", "schoolname"]},"parallelization":8,"appName":"UserInfo Exhaust"}'
     ;;
     "response-exhaust")
-    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ResponseExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"Response Exhaust"}'
+    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ResponseExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')","csvColumns":["courseid", "collectionName", "batchid", "batchName", "userid", "content_id", "contentname", "attempt_id", "last_attempted_on", "questionid","questiontype", "questiontitle", "questiondescription", "questionduration", "questionscore", "questionmaxscore", "questionoption", "questionresponse"]},"parallelization":8,"appName":"Response Exhaust"}'
     ;;
     "response-exhaust-v2")
-    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ResponseExhaustJobV2","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"Response Exhaust V2"}'
+    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ResponseExhaustJobV2","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')","csvColumns":["courseid", "collectionName", "batchid", "batchName", "userid", "content_id", "contentname", "attempt_id", "last_attempted_on", "questionid", "questiontype", "questiontitle", "questiondescription", "questionduration", "questionscore", "questionmaxscore", "questionoption", "questionresponse"]},"parallelization":8,"appName":"Response Exhaust V2"}'
     ;;
     "progress-exhaust")
-    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ProgressExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"Progress Exhaust"}'
+    echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ProgressExhaustJob","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')","csvColumns":["courseid", "collectionName", "batchid", "batchName", "userid",  "state", "district", "orgname", "schooludisecode", "schoolname", "board", "block", "cluster", "usertype", "usersubtype", "enrolleddate", "completedon", "certificatestatus", "completionPercentage"]},"parallelization":8,"appName":"Progress Exhaust"}'
     ;;
     "progress-exhaust-v2")
     echo '{"search":{"type":"none"},"model":"org.sunbird.lms.exhaust.collection.ProgressExhaustJobV2","modelParams":{"store":"azure","mode":"OnDemand","batchFilters":["TPD"],"searchFilter":{}, "sparkElasticsearchConnectionHost":"{{ sunbird_es_host }}","sparkRedisConnectionHost":"{{ metadata2_redis_host }}","sparkUserDbRedisIndex":"12","sparkUserDbRedisPort":"{{ user_port }}", "sparkCassandraConnectionHost":"{{ core_cassandra_host }}", "fromDate":"$(date --date yesterday '+%Y-%m-%d')","toDate":"$(date --date yesterday '+%Y-%m-%d')"},"parallelization":8,"appName":"Progress Exhaust V2"}'

lern-data-products/pom.xml

@@ -500,4 +500,4 @@
             </plugin>
         </plugins>
 	</build>
-</project>
+</project>

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

@@ -81,4 +81,6 @@ trait BaseReportsJob {
     StorageConfig(store, container, key, Option(storageKey), Option(storageSecret));
   }
 
+  def validateCsvColumns(piiFields: List[String], csvColumns: List[String], level: String): Boolean = true
+
 }

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

@@ -28,23 +28,55 @@ object DataSecurityUtil {
    */
   def getSecurityLevel(jobId: String, orgId: String): String = {
     JobLogger.log(s"getSecurityLevel jobID:: $jobId orgid:: $orgId", None, INFO)(new String())
-    val requestBody = Map("request" -> Map("orgId" -> orgId, "key" -> "dataSecurityPolicy"))
+    val httpResponseBody = getTenantPreferanceDetails(orgId, "dataSecurityPolicy")
+    val responseBody = JSONUtils.deserialize[Map[String, AnyRef]](httpResponseBody)
+    val data = responseBody.getOrElse("result", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+      .getOrElse("response", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+      .getOrElse("data", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+    val globalLevel = data.getOrElse("level", "").asInstanceOf[String]
+    val jobDetail = data.getOrElse("job", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+      .getOrElse(jobId, Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+    val jobLevel = jobDetail.getOrElse("level", "").asInstanceOf[String]
+    if (!StringUtils.isEmpty(jobLevel)) jobLevel else globalLevel
+  }
+
+  /**
+   * fetch the PII fields by calling tenant preference read API using orgId
+   *
+   * @param jobId
+   * @param orgId
+   * @return
+   */
+  def getPIIFieldDetails(jobId: String, orgId: String): List[String] = {
+    JobLogger.log(s"getSecurityLevel jobID:: $jobId orgid:: $orgId", None, INFO)(new String())
+    val httpResponseBody = getTenantPreferanceDetails(orgId, "userPrivateFields")
+    val responseBody = JSONUtils.deserialize[Map[String, AnyRef]](httpResponseBody)
+    val data = responseBody.getOrElse("result", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+      .getOrElse("response", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+      .getOrElse("data", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
+    val piiFields = data.getOrElse("piiFields", List[String]()).asInstanceOf[List[String]]
+    piiFields
+  }
+
+  /**
+   * fetch the job security level by calling tenant preference read API using orgId
+   *
+   * @param jobId
+   * @param orgId
+   * @return
+   */
+  def getTenantPreferanceDetails(orgId: String, key: String): String = {
+    JobLogger.log(s"getTenantPreferanceDetails orgid:: $orgId", None, INFO)(new String())
+    val requestBody = Map("request" -> Map("orgId" -> orgId, "key" -> key))
     val request = JSONUtils.serialize(requestBody)
     val headers: Map[String, String] = Map("Content-Type" -> "application/json")
     val readTenantPrefURL = Constants.TENANT_PREFERENCE_PRIVATE_READ_URL
-    JobLogger.log(s"getSecurityLevel readTenantPrefURL:: $readTenantPrefURL", None, INFO)(new String())
+    JobLogger.log(s"getTenantPreferanceDetails readTenantPrefURL:: $readTenantPrefURL", None, INFO)(new String())
     val httpResponse = httpUtil.post(readTenantPrefURL, request, headers)
     if (httpResponse.status == 200) {
-      JobLogger.log(s"dataSecurityPolicy for org=$orgId, response body=${httpResponse.body}", None, INFO)(new String())
-      val responseBody = JSONUtils.deserialize[Map[String, AnyRef]](httpResponse.body)
-      val data = responseBody.getOrElse("result", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
-        .getOrElse("response", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
-        .getOrElse("data", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
-      val globalLevel = data.getOrElse("level", "").asInstanceOf[String]
-      val jobDetail = data.getOrElse("job", Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
-        .getOrElse(jobId, Map[String, AnyRef]()).asInstanceOf[Map[String, AnyRef]]
-      val jobLevel = jobDetail.getOrElse("level", "").asInstanceOf[String]
-      if (!StringUtils.isEmpty(jobLevel)) jobLevel else globalLevel
+      JobLogger.log(s"getTenantPreferanceDetails for org=$orgId, response body=${httpResponse.body}", None, INFO)(new String())
+      val responseBody = httpResponse.body
+      responseBody
     } else {
       JobLogger.log(s"Error response from Tenant Preferance read API for request :: $requestBody :: response is :: ${httpResponse.status} ::  ${httpResponse.body}", None, ERROR)(new String())
       ""
@@ -169,6 +201,7 @@ object DataSecurityUtil {
     var tempDir = ""
     if (level.nonEmpty) {
       val storageService = fc.getStorageService(storageConfig.store, storageConfig.accountKey.getOrElse(""), storageConfig.secretKey.getOrElse(""));
+      val path = Paths.get(url)
       val filePrefix = storageConfig.store.toLowerCase() match {
         // $COVERAGE-OFF$ Disabling scoverage
         case "s3" =>
@@ -179,19 +212,26 @@ object DataSecurityUtil {
           CommonUtil.getGCloudFile(storageConfig.container, "")
         // $COVERAGE-ON$ for case: local
         case _ =>
-          storageConfig.fileName
+          val filePath = path.toString
+          if (filePath.contains(storageConfig.fileName)){
+            filePath
+          } else {
+            storageConfig.fileName + "/" + filePath
+          }
+
       }
 
       if (!url.isEmpty ) {
         if(request != null) {
           tempDir = AppConf.getConfig("spark_output_temp_dir") + request.request_id + "/"
         } else {
+          val urlSplitArr = url.split("/")
           if (!storageConfig.store.equals("local")) {
-            val urlSplitArr = url.split("/")
             tempDir = AppConf.getConfig("spark_output_temp_dir") + urlSplitArr(3) + "/"
+          } else {
+            tempDir = AppConf.getConfig("spark_output_temp_dir") + urlSplitArr(4) + "/"
           }
         }
-        val path = Paths.get(url)
         objKey = url.replace(filePrefix, "")
         localPath = tempDir + path.getFileName
         fc.getHadoopFileUtil().delete(conf, tempDir)
@@ -237,9 +277,11 @@ object DataSecurityUtil {
         if(request != null) {
           tempDir = AppConf.getConfig("spark_output_temp_dir") + request.request_id + "/"
         } else {
+          val urlSplitArr = url.split("/")
           if (!storageConfig.store.equals("local")) {
-            val urlSplitArr = url.split("/")
             tempDir = AppConf.getConfig("spark_output_temp_dir") + urlSplitArr(3) + "/"
+          } else {
+          tempDir = AppConf.getConfig("spark_output_temp_dir") + urlSplitArr(4) + "/"
           }
         }
         val path = Paths.get(url)

lern-data-products/src/main/scala/org/sunbird/lms/exhaust/collection/BaseCollectionExhaustJob.scala

@@ -19,7 +19,7 @@ import org.joda.time.format.{DateTimeFormat, DateTimeFormatter}
 import org.joda.time.{DateTime, DateTimeZone}
 import org.sunbird.core.util.{DecryptUtil, RedisConnect}
 import org.sunbird.core.exhaust.{BaseReportsJob, JobRequest, OnDemandExhaustJob}
-import org.sunbird.core.util.DataSecurityUtil.{getOrgId, getSecuredExhaustFile, getSecurityLevel}
+import org.sunbird.core.util.DataSecurityUtil.{getOrgId, getPIIFieldDetails, getSecuredExhaustFile, getSecurityLevel}
 import org.sunbird.lms.exhaust.collection.ResponseExhaustJobV2.Question
 
 import java.security.MessageDigest
@@ -151,6 +151,8 @@ trait BaseCollectionExhaustJob extends BaseReportsJob with IJob with OnDemandExh
       JobLogger.log(s"executeOnDemand for channel= "+ request.requested_channel, None, INFO)
       val orgId = request.requested_channel//getOrgId("", request.requested_channel)
       val level = getSecurityLevel(jobId(), orgId)
+      val piiFields = getPIIFieldDetails(jobId(), orgId)
+      val csvColumns = modelParams.getOrElse("csvColumns", List[String]()).asInstanceOf[List[String]]
       JobLogger.log(s"executeOnDemand for url = $orgId and level = $level and channel= $request.requested_channel", None, INFO)
       val reqOrgAndLevel = (request.request_id, orgId, level)
       reqOrgAndLevelDtl :+= reqOrgAndLevel
@@ -161,6 +163,10 @@ trait BaseCollectionExhaustJob extends BaseReportsJob with IJob with OnDemandExh
           JobLogger.log("Channel details at executeOnDemand", Some(Map("channel" -> request.requested_channel, "file size" -> processedSize, "completed batches" -> processedCount)), INFO)
 
           if (checkRequestProcessCriteria(processedCount, processedSize)) {
+            if(!validateCsvColumns(piiFields, csvColumns, level)) {
+              JobLogger.log("Request should have either of batchId, batchFilter, searchFilter or encrption key", Some(Map("requestId" -> request.request_id, "remainingRequest" -> totalRequests.getAndDecrement())), INFO)
+              markRequestAsFailed(request, "Request Security Level is not matching with PII fields or csvColumns CSV columns configured. Please check.")
+            }
             if (validateRequest(request)) {
               val res = processRequest(request, custodianOrgId, userCachedDF, storageConfig, requestsCompleted, orgId, level)
               requestsCompleted.++=(JSONUtils.deserialize[ListBuffer[ProcessedRequest]](res.processed_batches.getOrElse("[]")))

lern-data-products/src/main/scala/org/sunbird/lms/exhaust/collection/ProgressExhaustJob.scala

@@ -5,8 +5,9 @@ import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.ekstep.analytics.framework.Level.INFO
 import org.ekstep.analytics.framework.conf.AppConf
-import org.ekstep.analytics.framework.util.JSONUtils
+import org.ekstep.analytics.framework.util.{JSONUtils, JobLogger}
 import org.ekstep.analytics.framework.{FrameworkContext, JobConfig}
 
 import scala.collection.immutable.Set
@@ -42,8 +43,7 @@ object ProgressExhaustJob extends BaseCollectionExhaustJob {
   private val activityAggDBSettings = Map("table" -> "user_activity_agg", "keyspace" -> AppConf.getConfig("sunbird.courses.keyspace"), "cluster" -> "LMSCluster");
   private val assessmentAggDBSettings = Map("table" -> "assessment_aggregator", "keyspace" -> AppConf.getConfig("sunbird.courses.keyspace"), "cluster" -> "LMSCluster");
   private val contentHierarchyDBSettings = Map("table" -> "content_hierarchy", "keyspace" -> AppConf.getConfig("sunbird.content.hierarchy.keyspace"), "cluster" -> "ContentCluster");
-
-  private val filterColumns = Seq("courseid", "collectionName", "batchid", "batchName", "userid",  "state", "district", "orgname", "schooludisecode", "schoolname", "board", "block", "cluster", "usertype", "usersubtype", "enrolleddate", "completedon", "certificatestatus", "completionPercentage");
+  //private val filterColumns = Seq("courseid", "collectionName", "batchid", "batchName", "userid",  "state", "district", "orgname", "schooludisecode", "schoolname", "board", "block", "cluster", "usertype", "usersubtype", "enrolleddate", "completedon", "certificatestatus", "completionPercentage");
   private val columnsOrder = List("Collection Id", "Collection Name", "Batch Id", "Batch Name", "User UUID", "State", "District", "Org Name", "School Id",
     "School Name", "Block Name", "Cluster Name", "User Type", "User Sub Type", "Declared Board", "Enrolment Date", "Completion Date", "Certificate Status", "Progress", "Total Score")
   private val columnMapping = Map("courseid" -> "Collection Id", "collectionName" -> "Collection Name", "batchid" -> "Batch Id", "batchName" -> "Batch Name", "userid" -> "User UUID",
@@ -52,6 +52,7 @@ object ProgressExhaustJob extends BaseCollectionExhaustJob {
     "completionPercentage" -> "Progress", "total_sum_score" -> "Total Score", "certificatestatus" -> "Certificate Status")
 
   override def processBatch(userEnrolmentDF: DataFrame, collectionBatch: CollectionBatch)(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): DataFrame = {
+    val filterColumns : List[String] = config.modelParams.get.getOrElse("csvColumns", List[String]()).asInstanceOf[List[String]]
     val hierarchyData = loadCollectionHierarchy(collectionBatch.collectionId)
     //val collectionAggDF = getCollectionAggWithModuleData(collectionBatch, hierarchyData).withColumn("batchid", lit(collectionBatch.batchId));
     //val enrolledUsersToBatch = updateCertificateStatus(userEnrolmentDF).select(filterColumns.head, filterColumns.tail: _*)