Issue-1762: User profile updates should be only allowed by admin or bots (#1765)

* Issue-1762: User profile updates should be only allowed by admin or bots

catalog-rest-service/src/main/java/org/openmetadata/catalog/elasticsearch/ElasticSearchIndexDefinition.java

@@ -298,10 +298,12 @@ class TableESIndex extends ElasticSearchIndex {
       }
     }
     ParseTags parseTags = new ParseTags(tags);
+    Long updatedTimestamp = table.getUpdatedAt().getTime();
     TableESIndexBuilder tableESIndexBuilder =  internalBuilder().tableId(tableId)
         .name(tableName)
         .displayName(tableName)
         .description(description)
+        .lastUpdatedTimestamp(updatedTimestamp)
         .fqdn(table.getFullyQualifiedName())
         .suggest(suggest)
         .entityType("table")
@@ -313,7 +315,13 @@ class TableESIndex extends ElasticSearchIndex {
         .tier(parseTags.tierTag);
 
     if (table.getDatabase() != null) {
-      tableESIndexBuilder.database(table.getDatabase().getName());
+      String databaseFQN = table.getDatabase().getName();
+      String[] databaseFQNSplit = databaseFQN.split("\\.");
+      if (databaseFQNSplit.length == 2) {
+        tableESIndexBuilder.database(databaseFQNSplit[1]);
+      } else {
+        tableESIndexBuilder.database(databaseFQNSplit[0]);
+      }
     }
 
     if (table.getService() != null) {
@@ -343,13 +351,13 @@ class TableESIndex extends ElasticSearchIndex {
     ESChangeDescription esChangeDescription = null;
 
     if (table.getChangeDescription() != null) {
-      esChangeDescription = ESChangeDescription.builder().updatedAt(table.getUpdatedAt().getTime())
+      esChangeDescription = ESChangeDescription.builder().updatedAt(updatedTimestamp)
           .updatedBy(table.getUpdatedBy()).build();
       esChangeDescription.setFieldsAdded(table.getChangeDescription().getFieldsAdded());
       esChangeDescription.setFieldsDeleted(table.getChangeDescription().getFieldsDeleted());
       esChangeDescription.setFieldsUpdated(table.getChangeDescription().getFieldsUpdated());
     } else if (responseCode == Response.Status.CREATED.getStatusCode()) {
-      esChangeDescription = ESChangeDescription.builder().updatedAt(table.getUpdatedAt().getTime())
+      esChangeDescription = ESChangeDescription.builder().updatedAt(updatedTimestamp)
           .updatedBy(table.getUpdatedBy())
           .fieldsAdded(new ArrayList<>())
           .fieldsUpdated(new ArrayList<>())
@@ -410,11 +418,13 @@ class TopicESIndex extends ElasticSearchIndex {
       topic.getTags().forEach(tag -> tags.add(tag.getTagFQN()));
     }
     ParseTags parseTags = new ParseTags(tags);
+    Long updatedTimestamp = topic.getUpdatedAt().getTime();
     TopicESIndexBuilder topicESIndexBuilder =  internalBuilder().topicId(topic.getId().toString())
         .name(topic.getName())
         .displayName(topic.getDisplayName())
         .description(topic.getDescription())
         .fqdn(topic.getFullyQualifiedName())
+        .lastUpdatedTimestamp(updatedTimestamp)
         .suggest(suggest)
         .service(topic.getService().getName())
         .serviceType(topic.getServiceType().toString())
@@ -437,13 +447,13 @@ class TopicESIndex extends ElasticSearchIndex {
     ESChangeDescription esChangeDescription = null;
 
     if (topic.getChangeDescription() != null) {
-      esChangeDescription = ESChangeDescription.builder().updatedAt(topic.getUpdatedAt().getTime())
+      esChangeDescription = ESChangeDescription.builder().updatedAt(updatedTimestamp)
           .updatedBy(topic.getUpdatedBy()).build();
       esChangeDescription.setFieldsAdded(topic.getChangeDescription().getFieldsAdded());
       esChangeDescription.setFieldsDeleted(topic.getChangeDescription().getFieldsDeleted());
       esChangeDescription.setFieldsUpdated(topic.getChangeDescription().getFieldsUpdated());
     } else if (responseCode == Response.Status.CREATED.getStatusCode()) {
-      esChangeDescription = ESChangeDescription.builder().updatedAt(topic.getUpdatedAt().getTime())
+      esChangeDescription = ESChangeDescription.builder().updatedAt(updatedTimestamp)
           .updatedBy(topic.getUpdatedBy()).build();
     }
 
@@ -486,7 +496,7 @@ class DashboardESIndex extends ElasticSearchIndex {
     List<ElasticSearchSuggest> suggest = new ArrayList<>();
     suggest.add(ElasticSearchSuggest.builder().input(dashboard.getFullyQualifiedName()).weight(5).build());
     suggest.add(ElasticSearchSuggest.builder().input(dashboard.getDisplayName()).weight(10).build());
-
+    Long updatedTimestamp = dashboard.getUpdatedAt().getTime();
     if (dashboard.getTags() != null) {
       dashboard.getTags().forEach(tag -> tags.add(tag.getTagFQN()));
     }
@@ -501,6 +511,7 @@ class DashboardESIndex extends ElasticSearchIndex {
         .displayName(dashboard.getDisplayName())
         .description(dashboard.getDescription())
         .fqdn(dashboard.getFullyQualifiedName())
+        .lastUpdatedTimestamp(updatedTimestamp)
         .chartNames(chartNames)
         .chartDescriptions(chartDescriptions)
         .entityType("dashboard")
@@ -532,14 +543,14 @@ class DashboardESIndex extends ElasticSearchIndex {
     ESChangeDescription esChangeDescription = null;
     if (dashboard.getChangeDescription() != null) {
       esChangeDescription = ESChangeDescription.builder()
-          .updatedAt(dashboard.getUpdatedAt().getTime())
+          .updatedAt(updatedTimestamp)
           .updatedBy(dashboard.getUpdatedBy()).build();
       esChangeDescription.setFieldsAdded(dashboard.getChangeDescription().getFieldsAdded());
       esChangeDescription.setFieldsDeleted(dashboard.getChangeDescription().getFieldsDeleted());
       esChangeDescription.setFieldsUpdated(dashboard.getChangeDescription().getFieldsUpdated());
     }  else if (responseCode == Response.Status.CREATED.getStatusCode()) {
       esChangeDescription = ESChangeDescription.builder()
-          .updatedAt(dashboard.getUpdatedAt().getTime())
+          .updatedAt(updatedTimestamp)
           .updatedBy(dashboard.getUpdatedBy()).build();
     }
     dashboardESIndexBuilder.changeDescriptions(esChangeDescription != null ? List.of(esChangeDescription) : null);
@@ -577,12 +588,14 @@ class PipelineESIndex extends ElasticSearchIndex {
       taskNames.add(task.getDisplayName());
       taskDescriptions.add(task.getDescription());
     }
+    Long updatedTimestamp = pipeline.getUpdatedAt().getTime();
     ParseTags parseTags = new ParseTags(tags);
     PipelineESIndexBuilder pipelineESIndexBuilder = internalBuilder().pipelineId(pipeline.getId().toString())
         .name(pipeline.getDisplayName())
         .displayName(pipeline.getDisplayName())
         .description(pipeline.getDescription())
         .fqdn(pipeline.getFullyQualifiedName())
+        .lastUpdatedTimestamp(updatedTimestamp)
         .taskNames(taskNames)
         .taskDescriptions(taskDescriptions)
         .entityType("pipeline")
@@ -607,14 +620,14 @@ class PipelineESIndex extends ElasticSearchIndex {
     ESChangeDescription esChangeDescription = null;
     if (pipeline.getChangeDescription() != null) {
       esChangeDescription = ESChangeDescription.builder()
-          .updatedAt(pipeline.getUpdatedAt().getTime())
+          .updatedAt(updatedTimestamp)
           .updatedBy(pipeline.getUpdatedBy()).build();
       esChangeDescription.setFieldsAdded(pipeline.getChangeDescription().getFieldsAdded());
       esChangeDescription.setFieldsDeleted(pipeline.getChangeDescription().getFieldsDeleted());
       esChangeDescription.setFieldsUpdated(pipeline.getChangeDescription().getFieldsUpdated());
     } else if (responseCode == Response.Status.CREATED.getStatusCode()) {
       esChangeDescription = ESChangeDescription.builder()
-          .updatedAt(pipeline.getUpdatedAt().getTime())
+          .updatedAt(updatedTimestamp)
           .updatedBy(pipeline.getUpdatedBy()).build();
     }
     pipelineESIndexBuilder.changeDescriptions(esChangeDescription != null ? List.of(esChangeDescription) : null);

catalog-rest-service/src/main/java/org/openmetadata/catalog/resources/teams/UserResource.java

@@ -41,7 +41,9 @@ import org.openmetadata.catalog.util.ResultList;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.json.JsonObject;
 import javax.json.JsonPatch;
+import javax.json.JsonValue;
 import javax.validation.Valid;
 import javax.validation.constraints.Max;
 import javax.validation.constraints.Min;
@@ -117,7 +119,7 @@ public class UserResource {
                   "parameter to get only necessary fields. Use cursor-based pagination to limit the number " +
                   "entries in the list using `limit` and `before` or `after` query params.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user ",
+              @ApiResponse(responseCode = "200", description = "The user ",
                           content = @Content(mediaType = "application/json",
                           schema = @Schema(implementation = UserList.class)))
   })
@@ -173,11 +175,10 @@ public class UserResource {
   @Operation(summary = "Get a user", tags = "users",
           description = "Get a user by `id`",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user",
+              @ApiResponse(responseCode = "200", description = "The user",
                           content = @Content(mediaType = "application/json",
                           schema = @Schema(implementation = User.class))),
-                  @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")
+              @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")})
-  })
   public User get(@Context UriInfo uriInfo, @Context SecurityContext securityContext, @PathParam("id") String id,
                   @Parameter(description = "Fields requested in the returned resource",
                           schema = @Schema(type = "string", example = FIELDS))
@@ -193,10 +194,10 @@ public class UserResource {
   @Operation(summary = "Get a user by name", tags = "users",
           description = "Get a user by `name`.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user",
+              @ApiResponse(responseCode = "200", description = "The user",
                           content = @Content(mediaType = "application/json",
                           schema = @Schema(implementation = User.class))),
-                  @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")
+              @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")
           })
   public User getByName(@Context UriInfo uriInfo, @Context SecurityContext securityContext,
                       @PathParam("name") String name,
@@ -214,10 +215,10 @@ public class UserResource {
   @Operation(summary = "Get current logged in user", tags = "users",
           description = "Get the user who is authenticated and is currently logged in.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user",
+              @ApiResponse(responseCode = "200", description = "The user",
-                          content = @Content(mediaType = "application/json",
+                      content = @Content(mediaType = "application/json",
-                          schema = @Schema(implementation = User.class))),
+                      schema = @Schema(implementation = User.class))),
-                  @ApiResponse(responseCode = "404", description = "User not found")
+              @ApiResponse(responseCode = "404", description = "User not found")
           })
   public User getCurrentLoggedInUser(@Context UriInfo uriInfo, @Context SecurityContext securityContext,
                                      @Parameter(description = "Fields requested in the returned resource",
@@ -234,10 +235,10 @@ public class UserResource {
   @Operation(summary = "Get a version of the user", tags = "users",
           description = "Get a version of the user by given `id`",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "user",
+              @ApiResponse(responseCode = "200", description = "user",
                           content = @Content(mediaType = "application/json",
-                                  schema = @Schema(implementation = User.class))),
+                          schema = @Schema(implementation = User.class))),
-                  @ApiResponse(responseCode = "404", description = "User for instance {id} and version {version} is " +
+              @ApiResponse(responseCode = "404", description = "User for instance {id} and version {version} is " +
                           "not found")
           })
   public User getVersion(@Context UriInfo uriInfo,
@@ -254,10 +255,10 @@ public class UserResource {
   @Operation(summary = "Create a user", tags = "users",
           description = "Create a new user.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user ",
+              @ApiResponse(responseCode = "200", description = "The user ",
                           content = @Content(mediaType = "application/json",
                           schema = @Schema(implementation = CreateUser.class))),
-                  @ApiResponse(responseCode = "400", description = "Bad request")
+              @ApiResponse(responseCode = "400", description = "Bad request")
           })
   public Response createUser(@Context UriInfo uriInfo, @Context SecurityContext securityContext,
                              @Valid CreateUser create) throws IOException, ParseException {
@@ -273,15 +274,16 @@ public class UserResource {
   @Operation(summary = "Create or Update a user", tags = "users",
           description = "Create or Update a user.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "The user ",
+              @ApiResponse(responseCode = "200", description = "The user ",
-                          content = @Content(mediaType = "application/json",
+                           content = @Content(mediaType = "application/json",
-                                  schema = @Schema(implementation = CreateUser.class))),
+                           schema = @Schema(implementation = CreateUser.class))),
-                  @ApiResponse(responseCode = "400", description = "Bad request")
+              @ApiResponse(responseCode = "400", description = "Bad request")
           })
   public Response createOrUpdateUser(@Context UriInfo uriInfo,
                                      @Context SecurityContext securityContext,
                                      @Valid CreateUser create) throws IOException, ParseException {
-    if (create.getIsAdmin() != null && create.getIsAdmin()) {
+    if ((create.getIsAdmin() != null && create.getIsAdmin()) ||
+        (create.getIsBot() != null && create.getIsBot())) {
       SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
     }
     User user = getUser(securityContext, create);
@@ -307,9 +309,18 @@ public class UserResource {
                                             "{op:add, path: /b, value: val}" +
                                             "]")}))
                             JsonPatch patch) throws IOException, ParseException {
+    for (JsonValue patchOp: patch.toJsonArray()) {
+      JsonObject patchOpObject = patchOp.asJsonObject();
+      if (patchOpObject.containsKey("path") && patchOpObject.containsKey("value")) {
+        String path = patchOpObject.getString("path");
+        if (path.equals("/isAdmin") || path.equals("/isBot")) {
+          SecurityUtil.checkAdminOrBotRole(authorizer, securityContext);
+        }
+      }
+    }
     User user = dao.get(uriInfo, id, new Fields(FIELD_LIST, null));
     SecurityUtil.checkAdminRoleOrPermissions(authorizer, securityContext,
-            new UserEntityInterface(user).getEntityReference());
+        new UserEntityInterface(user).getEntityReference());
     PatchResponse<User> response = dao.patch(uriInfo, UUID.fromString(id),
             securityContext.getUserPrincipal().getName(), patch);
     addHref(uriInfo, response.getEntity());
@@ -322,8 +333,8 @@ public class UserResource {
           description = "Users can't be deleted but are deactivated. The name and display name is prefixed with " +
                   "the string `deactivated`.",
           responses = {
-                  @ApiResponse(responseCode = "200", description = "OK"),
+              @ApiResponse(responseCode = "200", description = "OK"),
-                  @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")
+              @ApiResponse(responseCode = "404", description = "User for instance {id} is not found")
           })
   public Response delete(@Context UriInfo uriInfo, @Context SecurityContext securityContext,
                          @PathParam("id") String id) throws IOException {

catalog-rest-service/src/test/java/org/openmetadata/catalog/resources/teams/UserResourceTest.java

@@ -223,8 +223,8 @@ public class UserResourceTest extends EntityResourceTest<User> {
   public void patch_userNameChange_as_another_user_401(TestInfo test)
           throws HttpResponseException, JsonProcessingException {
     // Ensure user name can't be changed using patch
-    User user = createUser(create(test, 6).withName("test2").withDisplayName("displayName")
+    User user = createUser(create(test, 7).withName("test23").withDisplayName("displayName")
-            .withEmail("test2@email.com"), authHeaders("test2@email.com"));
+            .withEmail("test23@email.com"), authHeaders("test23@email.com"));
     String userJson = JsonUtils.pojoToJson(user);
     user.setDisplayName("newName");
     HttpResponseException exception = assertThrows(HttpResponseException.class, () -> patchUser(userJson, user,
@@ -232,6 +232,19 @@ public class UserResourceTest extends EntityResourceTest<User> {
     assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test100'} does not have permissions");
   }
 
+  @Test
+  public void patch_makeAdmin_as_another_user_401(TestInfo test)
+      throws HttpResponseException, JsonProcessingException {
+    // Ensure user name can't be changed using patch
+    User user = createUser(create(test, 6).withName("test2").withDisplayName("displayName")
+        .withEmail("test2@email.com"), authHeaders("test2@email.com"));
+    String userJson = JsonUtils.pojoToJson(user);
+    user.setIsAdmin(Boolean.TRUE);
+    HttpResponseException exception = assertThrows(HttpResponseException.class, () -> patchUser(userJson, user,
+        authHeaders("test100@email.com")));
+    assertResponse(exception, FORBIDDEN, "Principal: CatalogPrincipal{name='test100'} is not admin");
+  }
+
   @Test
   public void patch_userNameChange_as_same_user_200_ok(TestInfo test) throws HttpResponseException,
           JsonProcessingException {

ingestion/src/metadata/ingestion/sink/elasticsearch.py

@@ -13,6 +13,7 @@ import json
 import logging
 import ssl
 import time
+from datetime import datetime
 from typing import List, Optional
 
 from dateutil import parser
@@ -52,6 +53,10 @@ from metadata.ingestion.sink.elasticsearch_constants import (
 logger = logging.getLogger(__name__)
 
 
+def epoch_ms(dt: datetime):
+    return int(dt.timestamp() * 1000)
+
+
 class ElasticSearchConfig(ConfigModel):
     es_host: str
     es_port: int = 9200
@@ -220,7 +225,7 @@ class ElasticsearchSink(Sink[Entity]):
         column_descriptions = []
         tags = set()
 
-        timestamp = int(table.updatedAt.__root__.timestamp())
+        timestamp = epoch_ms(table.updatedAt.__root__)
         tier = None
         for table_tag in table.tags:
             if "Tier" in table_tag.tagFQN:
@@ -283,7 +288,7 @@ class ElasticsearchSink(Sink[Entity]):
             {"input": [topic_name], "weight": 10},
         ]
         tags = set()
-        timestamp = topic.updatedAt.__root__.timestamp()
+        timestamp = epoch_ms(topic.updatedAt.__root__)
         service_entity = self.metadata.get_by_id(
             entity=MessagingService, entity_id=str(topic.service.id.__root__)
         )
@@ -322,7 +327,7 @@ class ElasticsearchSink(Sink[Entity]):
         dashboard_name = dashboard.name
         suggest = [{"input": [dashboard.displayName], "weight": 10}]
         tags = set()
-        timestamp = dashboard.updatedAt.__root__.timestamp()
+        timestamp = epoch_ms(dashboard.updatedAt.__root__)
         service_entity = self.metadata.get_by_id(
             entity=DashboardService, entity_id=str(dashboard.service.id.__root__)
         )
@@ -383,7 +388,7 @@ class ElasticsearchSink(Sink[Entity]):
         fqdn = pipeline.fullyQualifiedName
         suggest = [{"input": [pipeline.displayName], "weight": 10}]
         tags = set()
-        timestamp = pipeline.updatedAt.__root__.timestamp()
+        timestamp = epoch_ms(pipeline.updatedAt.__root__)
         service_entity = self.metadata.get_by_id(
             entity=PipelineService, entity_id=str(pipeline.service.id.__root__)
         )
@@ -481,7 +486,7 @@ class ElasticsearchSink(Sink[Entity]):
                 version_json = json.loads(version)
                 updatedAt = parser.parse(version_json["updatedAt"])
                 change_description = ChangeDescription(
-                    updatedBy=version_json["updatedBy"], updatedAt=updatedAt.timestamp()
+                    updatedBy=version_json["updatedBy"], updatedAt=epoch_ms(updatedAt)
                 )
                 if "changeDescription" in version_json:
                     change_description.fieldsAdded = version_json["changeDescription"][