Support private extractor registration

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## Unreleased
 
+### Added
+- Add "when" parameter in a few GET API endpoints to enable pagination [#266](https://github.com/clowder-framework/clowder/issues/266)
+- Extractors can now specify an extractor_key and an owner (email address) when sending a
+registration or heartbeat to Clowder that will restrict use of that extractor to them.
+
 ## Fixed
 - Updated lastModifiesDate when updating file or metadata to a dataset, added lastModified to UI  [386](https://github.com/clowder-framework/clowder/issues/386)
 - Disabled button while create dataset ajax call is still going on [#311](https://github.com/clowder-framework/clowder/issues/311)

app/api/Extractions.scala

@@ -127,36 +127,6 @@ class Extractions @Inject()(
       }
   }
 
-  /**
-   *
-   * Given a file id (UUID), submit this file for extraction
-   */
-  def submitExtraction(id: UUID) = PermissionAction(Permission.ViewFile, Some(ResourceRef(ResourceRef.file, id)))(parse.json) { implicit request =>
-    if (UUID.isValid(id.stringify)) {
-      files.get(id) match {
-        case Some(file) => {
-          // FIXME dataset not available?
-          routing.fileCreated(file, None, Utils.baseUrl(request).toString, request.apiKey) match {
-            case Some(jobId) => {
-              Ok(Json.obj("status" -> "OK", "job_id" -> jobId))
-            }
-            case None => {
-              val message = "No jobId found for Extraction"
-              Logger.error(message)
-              InternalServerError(toJson(Map("status" -> "KO", "message" -> message)))
-            }
-          }
-        }
-        case None => {
-          Logger.error("Could not retrieve file that was just saved.")
-          InternalServerError("Error uploading file")
-        }
-      } //file match
-    } else {
-      BadRequest("Not valid id")
-    }
-  }
-
   /**
    * For a given file id, checks for the status of all extractors processing that file.
    * REST endpoint  GET /api/extractions/:id/status
@@ -404,24 +374,24 @@ class Extractions @Inject()(
     Ok(jarr)
   }
 
-  def listExtractors(categories: List[String]) = AuthenticatedAction  { implicit request =>
-    Ok(Json.toJson(extractors.listExtractorsInfo(categories)))
+  def listExtractors(categories: List[String], space: Option[UUID]) = AuthenticatedAction  { implicit request =>
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
+    Ok(Json.toJson(extractors.listExtractorsInfo(categories, userid)))
   }
 
-  def getExtractorInfo(extractorName: String) = AuthenticatedAction { implicit request =>
-    extractors.getExtractorInfo(extractorName) match {
+  def getExtractorInfo(extractorName: String, extractor_key: Option[String]) = AuthenticatedAction { implicit request =>
+    extractors.getExtractorInfo(extractorName, extractor_key, request.user) match {
       case Some(info) => Ok(Json.toJson(info))
       case None => NotFound(Json.obj("status" -> "KO", "message" -> "Extractor info not found"))
     }
   }
 
-  def deleteExtractor(extractorName: String) = ServerAdminAction { implicit request =>
-    extractors.deleteExtractor(extractorName)
+  def deleteExtractor(extractorName: String, extractor_key: Option[String]) = ServerAdminAction { implicit request =>
+    extractors.deleteExtractor(extractorName, extractor_key)
     Ok(toJson(Map("status" -> "success")))
   }
 
-  def addExtractorInfo() = AuthenticatedAction(parse.json) { implicit request =>
-
+  def addExtractorInfo(extractor_key: Option[String], user: Option[String]) = AuthenticatedAction(parse.json) { implicit request =>
     // If repository is of type object, change it into an array.
     // This is for backward compatibility with requests from existing extractors.
     var requestJson = request.body \ "repository" match {
@@ -438,34 +408,66 @@ class Extractions @Inject()(
         BadRequest(Json.obj("status" -> "KO", "message" -> JsError.toFlatJson(errors)))
       },
       info => {
-        extractors.updateExtractorInfo(info) match {
-          case Some(u) => {
-            // Create/assign any default labels for this extractor
-            u.defaultLabels.foreach(labelStr => {
-              val segments = labelStr.split("/")
-              val (labelName, labelCategory) = if (segments.length > 1) {
-                (segments(1), segments(0))
-              } else {
-                (segments(0), "Other")
+        // Check private extractor flags
+        val submissionInfo: Option[ExtractorInfo] = extractor_key match {
+          case Some(ek) => {
+            user match {
+              case None => {
+                Logger.error("Extractors with a private key must also specify a user email.")
+                None
               }
-              extractors.getExtractorsLabel(labelName) match {
-                case None => {
-                  // Label does not exist - create and assign it
-                  val createdLabel = extractors.createExtractorsLabel(labelName, Some(labelCategory), List[String](u.name))
-                }
-                case Some(lbl) => {
-                  // Label already exists, assign it
-                  if (!lbl.extractors.contains(u.name)) {
-                    val label = ExtractorsLabel(lbl.id, lbl.name, lbl.category, lbl.extractors ++ List[String](u.name))
-                    val updatedLabel = extractors.updateExtractorsLabel(label)
+              case Some(userEmail) => {
+                userservice.findByEmail(userEmail) match {
+                  case Some(u) => {
+                    val perms = List(new ResourceRef('user, u.id))
+                    Some(info.copy(unique_key=Some(ek), permissions=perms))
+                  }
+                  case None => {
+                    Logger.error("No user found with email "+userEmail)
+                    None
                   }
                 }
               }
-            })
+            }
+          }
+          case None => Some(info)
+        }
+
+        // TODO: Check user permissions if the extractor_key has already been registered
+
+        submissionInfo match {
+          case None => BadRequest("Extractors with a private key must also specify a non-anonymous user.")
+          case Some(subInfo) => {
+            extractors.updateExtractorInfo(subInfo) match {
+              case Some(u) => {
+                // Create/assign any default labels for this extractor
+                u.defaultLabels.foreach(labelStr => {
+                  val segments = labelStr.split("/")
+                  val (labelName, labelCategory) = if (segments.length > 1) {
+                    (segments(1), segments(0))
+                  } else {
+                    (segments(0), "Other")
+                  }
+                  extractors.getExtractorsLabel(labelName) match {
+                    case None => {
+                      // Label does not exist - create and assign it
+                      val createdLabel = extractors.createExtractorsLabel(labelName, Some(labelCategory), List[String](u.name))
+                    }
+                    case Some(lbl) => {
+                      // Label already exists, assign it
+                      if (!lbl.extractors.contains(u.name)) {
+                        val label = ExtractorsLabel(lbl.id, lbl.name, lbl.category, lbl.extractors ++ List[String](u.name))
+                        val updatedLabel = extractors.updateExtractorsLabel(label)
+                      }
+                    }
+                  }
+                })
 
-            Ok(Json.obj("status" -> "OK", "message" -> ("Extractor info updated. ID = " + u.id)))
+                Ok(Json.obj("status" -> "OK", "message" -> ("Extractor info updated. ID = " + u.id)))
+              }
+              case None => BadRequest(Json.obj("status" -> "KO", "message" -> "Error updating extractor info"))
+            }
           }
-          case None => BadRequest(Json.obj("status" -> "KO", "message" -> "Error updating extractor info"))
         }
       }
     )
@@ -518,11 +520,14 @@ class Extractions @Inject()(
           }
           // if extractor_id is not specified default to execution of all extractors matching mime type
           (request.body \ "extractor").asOpt[String] match {
-            case Some(extractorId) =>
+            case Some(extractorId) => {
+              val extractorKey = (request.body \ "extractor").asOpt[String]
+              extractors.getExtractorInfo(extractorId, extractorKey, request.user)
               val job_id = routing.submitFileManually(new UUID(originalId), file, Utils.baseUrl(request), extractorId, extra,
                 datasetId, newFlags, request.apiKey, request.user)
               sink.logSubmitFileToExtractorEvent(file, extractorId, request.user)
               Ok(Json.obj("status" -> "OK", "job_id" -> job_id))
+            }
             case None => {
               routing.fileCreated(file, None, Utils.baseUrl(request).toString, request.apiKey) match {
                 case Some(job_id) => {

app/controllers/Extractors.scala

@@ -39,9 +39,9 @@ class Extractors  @Inject() (extractions: ExtractionService,
   /**
    * Gets a map of all updates from all jobs given to this extractor.
    */
-  def showJobHistory(extractorName: String) = AuthenticatedAction { implicit request =>
+  def showJobHistory(extractorName: String, extractor_key: Option[String]) = AuthenticatedAction { implicit request =>
     implicit val user = request.user
-    extractorService.getExtractorInfo(extractorName) match {
+    extractorService.getExtractorInfo(extractorName, extractor_key, user) match {
       case None => NotFound(s"No extractor found with name=${extractorName}")
       case Some(info) => {
         val allExtractions = extractions.findAll()
@@ -56,9 +56,10 @@ class Extractors  @Inject() (extractions: ExtractionService,
     */
   def selectExtractors() = AuthenticatedAction { implicit request =>
     implicit val user = request.user
-
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
     // Filter extractors by user filters necessary
-    var runningExtractors: List[ExtractorInfo] = extractorService.listExtractorsInfo(List.empty)
+    // TODO: Filter by multiple spaces
+    var runningExtractors: List[ExtractorInfo] = extractorService.listExtractorsInfo(List.empty, userid)
     val selectedExtractors: List[String] = extractorService.getEnabledExtractors()
     val groups = extractions.groupByType(extractions.findAll())
     val allLabels = extractorService.listExtractorsLabels()
@@ -166,7 +167,7 @@ class Extractors  @Inject() (extractions: ExtractionService,
   def manageLabels = ServerAdminAction { implicit request =>
     implicit val user = request.user
     val categories = List[String]("EXTRACT")
-    val extractors = extractorService.listExtractorsInfo(categories)
+    val extractors = extractorService.listExtractorsInfo(categories, None)
     val labels = extractorService.listExtractorsLabels()
 
     Ok(views.html.extractorLabels(labels, extractors))
@@ -211,7 +212,8 @@ class Extractors  @Inject() (extractions: ExtractionService,
 
   def showExtractorInfo(extractorName: String) = AuthenticatedAction { implicit request =>
     implicit val user = request.user
-    val targetExtractor = extractorService.listExtractorsInfo(List.empty).find(p => p.name == extractorName)
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
+    val targetExtractor = extractorService.listExtractorsInfo(List.empty, userid).find(p => p.name == extractorName)
     targetExtractor match {
       case Some(extractor) => {
         val labels = extractorService.getLabelsForExtractor(extractor.name)
@@ -223,6 +225,7 @@ class Extractors  @Inject() (extractions: ExtractionService,
 
   def showExtractorMetrics(extractorName: String) = AuthenticatedAction { implicit request =>
     implicit val user = request.user
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
 
     val dateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS")
     val todaydate = dateFormatter.format(new java.util.Date())
@@ -299,7 +302,7 @@ class Extractors  @Inject() (extractions: ExtractionService,
     }
     Logger.warn("last 10 average: " + lastTenAverage)
 
-    val targetExtractor = extractorService.listExtractorsInfo(List.empty).find(p => p.name == extractorName)
+    val targetExtractor = extractorService.listExtractorsInfo(List.empty, userid).find(p => p.name == extractorName)
     targetExtractor match {
       case Some(extractor) => Ok(views.html.extractorMetrics(extractorName, average.toString, lastTenAverage.toString, lastweeksubmitted, lastmonthsubmitted))
       case None => InternalServerError("Extractor Info not found: " + extractorName)
@@ -308,11 +311,19 @@ class Extractors  @Inject() (extractions: ExtractionService,
 
   def submitFileExtraction(file_id: UUID) = PermissionAction(Permission.EditFile, Some(ResourceRef(ResourceRef.file, file_id))) { implicit request =>
     implicit val user = request.user
-    val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"))
-    val extractors = all_extractors.filter(!_.process.file.isEmpty)
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
     fileService.get(file_id) match {
-
       case Some(file) => {
+        val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"), userid)
+        var extractors = all_extractors.filter(!_.process.file.isEmpty)
+
+        val user_extra = userid match {
+          case Some(uid) => all_extractors.filter(_.permissions.contains(ResourceRef('user, uid)))
+          case None => List.empty
+        }
+
+        extractors = (extractors ++ user_extra).distinct
+
         val foldersContainingFile = folders.findByFileId(file.id).sortBy(_.name)
         var folderHierarchy = new ListBuffer[Folder]()
         if(foldersContainingFile.length > 0) {
@@ -352,7 +363,8 @@ class Extractors  @Inject() (extractions: ExtractionService,
 
   def submitSelectedExtractions(ds_id: UUID) = PermissionAction(Permission.EditDataset, Some(ResourceRef(ResourceRef.dataset, ds_id))) { implicit request =>
     implicit val user = request.user
-    val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"))
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
+    val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"), userid)
     val extractors = all_extractors.filter(!_.process.file.isEmpty)
     datasets.get(ds_id) match {
       case Some(dataset) => {
@@ -372,10 +384,13 @@ class Extractors  @Inject() (extractions: ExtractionService,
 
   def submitDatasetExtraction(ds_id: UUID) = PermissionAction(Permission.EditDataset, Some(ResourceRef(ResourceRef.dataset, ds_id))) { implicit request =>
     implicit val user = request.user
-    val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"))
-    val extractors = all_extractors.filter(!_.process.dataset.isEmpty)
+    val userid = request.user.map(u => Some(u.id)).getOrElse(None)
     datasetService.get(ds_id) match {
-      case Some(ds) => Ok(views.html.extractions.submitDatasetExtraction(extractors, ds))
+      case Some(ds) => {
+        val all_extractors = extractorService.listExtractorsInfo(List("EXTRACT", "CONVERT"), userid)
+        val extractors = all_extractors.filter(!_.process.dataset.isEmpty)
+        Ok(views.html.extractions.submitDatasetExtraction(extractors, ds))
+      }
       case None => InternalServerError("Dataset not found")
     }
   }

app/controllers/Spaces.scala

@@ -83,10 +83,11 @@ class Spaces @Inject() (spaces: SpaceService, users: UserService, events: EventS
   def selectExtractors(id: UUID) = AuthenticatedAction {
     implicit request =>
       implicit val user = request.user
+      val userid = request.user.map(u => Some(u.id)).getOrElse(None)
       spaces.get(id) match {
         case Some(s) => {
           // get list of registered extractors
-          val runningExtractors: List[ExtractorInfo] = extractors.listExtractorsInfo(List.empty)
+          val runningExtractors: List[ExtractorInfo] = extractors.listExtractorsInfo(List.empty, userid)
           // list of extractors enabled globally
           val globalSelections: List[String] = extractors.getEnabledExtractors()
           // get list of extractors registered with a specific space

app/models/Extraction.scala

@@ -78,6 +78,7 @@ case class ExtractorDetail(
  *
  * @param id id internal to the system
  * @param name lower case, no spaces, can use dashes
+ * @param uniqueName name+suffix to uniquely identify extractor for private use e.g. clowder.extractor.v2.johndoe123
  * @param version the version, for example 1.3.5
  * @param updated date when this information was last updated
  * @param description short description of what the extractor does
@@ -117,7 +118,9 @@ case class ExtractorInfo(
   defaultLabels: List[String] = List[String](),
   process: ExtractorProcessTriggers = new ExtractorProcessTriggers(),
   categories: List[String] = List[String](ExtractorCategory.EXTRACT.toString),
-  parameters: JsValue = JsObject(Seq())
+  parameters: JsValue = JsObject(Seq()),
+  unique_key: Option[String] = None,
+  permissions: List[ResourceRef] =List[ResourceRef]()
 )
 
 /** what are the categories of the extractor?
@@ -170,7 +173,9 @@ object ExtractorInfo {
       (JsPath \ "labels").read[List[String]].orElse(Reads.pure(List.empty)) and
       (JsPath \ "process").read[ExtractorProcessTriggers].orElse(Reads.pure(new ExtractorProcessTriggers())) and
       (JsPath \ "categories").read[List[String]].orElse(Reads.pure(List[String](ExtractorCategory.EXTRACT.toString))) and
-      (JsPath \ "parameters").read[JsValue].orElse(Reads.pure(JsObject(Seq())))
+      (JsPath \ "parameters").read[JsValue].orElse(Reads.pure(JsObject(Seq()))) and
+      (JsPath \ "unique_key").read[Option[String]].orElse(Reads.pure(None)) and
+      (JsPath \ "permissions").read[List[ResourceRef]].orElse(Reads.pure(List.empty))
     )(ExtractorInfo.apply _)
 }