SNOW-1543690 Make credentialRotationRequired optional when creating a…

… principal (apache#5)

Description
Rename the "OTP" settings to CREDENTIAL_ROTATION_REQUIRED to be more
clear and also to deprecate any old flags already in persistent storage.

Allow choosing whether to force initial credential rotation at
createPrincipal time to facilitate simpler admin flows in cases where
such rotation is not desired.

Testing

<!-- Please describe your change here and remove this comment -->

## Pre-review checklist
- [ ] I attest that this change meets the bar for low risk without
security requirements as defined in the [Accelerated Risk Assessment
Criteria](https://developer-handbook.m1.us-west-2.aws.app.snowflake.com/docs/reference/security-review/accelerated-risk-assessment/#eligibility)
and I have taken the [Risk Assessment Training in
Workday](https://wd5.myworkday.com/snowflake/learning/course/6c613806284a1001f111fedf3e4e0000).
- Checking this checkbox is mandatory if using the [Accelerated Risk
Assessment](https://developer-handbook.m1.us-west-2.aws.app.snowflake.com/docs/reference/security-review/accelerated-risk-assessment/)
to risk assess the changes in this Pull Request.
- If this change does not meet the bar for low risk without security
requirements (as confirmed by the peer reviewers of this pull request)
then a [formal Risk
Assessment](https://developer-handbook.m1.us-west-2.aws.app.snowflake.com/docs/reference/security-review/risk-assessment/)
must be completed. Please note that a formal Risk Assessment will
require you to spend extra time performing a security review for this
change. Please account for this extra time earlier rather than later to
avoid unnecessary delays in the release process.
- [ ] This change has code coverage for the new code added

polaris/build.gradle

@@ -85,7 +85,7 @@ subprojects {
             }
         }
         java {
-            target 'src/main/java/**/*.java', 'src/test/java/**/*.java'
+            target 'src/main/java/**/*.java', 'src/testFixtures/java/**/*.java', 'src/test/java/**/*.java'
             googleJavaFormat()
             indentWithSpaces(2)
             removeUnusedImports()

polaris/polaris-core/src/main/java/io/polaris/core/PolarisConfiguration.java

@@ -2,7 +2,8 @@
 
 public class PolarisConfiguration {
 
-  public static final String ENFORCE_PRINCIPAL_OTP_CHECKING = "ENFORCE_PRINCIPAL_OTP_CHECKING";
+  public static final String ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING =
+      "ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING";
 
   private PolarisConfiguration() {}
 }

polaris/polaris-core/src/main/java/io/polaris/core/auth/PolarisAuthorizer.java

@@ -482,19 +482,22 @@ public void authorizeOrThrow(
       @NotNull PolarisAuthorizableOperation authzOp,
       @Nullable List<PolarisResolvedPathWrapper> targets,
       @Nullable List<PolarisResolvedPathWrapper> secondaries) {
-    boolean enforceOtpFlagState =
+    boolean enforceCredentialRotationRequiredState =
         featureConfig.getConfiguration(
             CallContext.getCurrentContext().getPolarisCallContext(),
-            PolarisConfiguration.ENFORCE_PRINCIPAL_OTP_CHECKING,
+            PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING,
             false);
-    if ((enforceOtpFlagState
-            && authenticatedPrincipal
-                .getPrincipalEntity()
-                .getInternalPropertiesAsMap()
-                .containsKey(PolarisEntityConstants.PRINCIPAL_OTP_STATE)
-            && authzOp != PolarisAuthorizableOperation.ROTATE_CREDENTIALS)
-        || !isAuthorized(
-            authenticatedPrincipal, activatedGranteeIds, authzOp, targets, secondaries)) {
+    if (enforceCredentialRotationRequiredState
+        && authenticatedPrincipal
+            .getPrincipalEntity()
+            .getInternalPropertiesAsMap()
+            .containsKey(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)
+        && authzOp != PolarisAuthorizableOperation.ROTATE_CREDENTIALS) {
+      throw new ForbiddenException(
+          "Principal '%s' is not authorized for op %s due to PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE",
+          authenticatedPrincipal.getName(), authzOp);
+    } else if (!isAuthorized(
+        authenticatedPrincipal, activatedGranteeIds, authzOp, targets, secondaries)) {
       throw new ForbiddenException(
           "Principal '%s' with activated PrincipalRoles '%s' and activated ids '%s' is not authorized for op %s",
           authenticatedPrincipal.getName(),

polaris/polaris-core/src/main/java/io/polaris/core/entity/PolarisEntityConstants.java

@@ -39,7 +39,8 @@ public class PolarisEntityConstants {
 
   private static final String PRINCIPAL_TYPE_NAME = "principal_type_name";
 
-  public static final String PRINCIPAL_OTP_STATE = "OTP_ONLY";
+  public static final String PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE =
+      "CREDENTIAL_ROTATION_REQUIRED";
 
   /**
    * Name format of storage integration for polaris entity: POLARIS_<catalog_id>_<entity_id> . This

polaris/polaris-core/src/main/java/io/polaris/core/entity/PrincipalEntity.java

@@ -62,6 +62,12 @@ public Builder setClientId(String clientId) {
       return this;
     }
 
+    public Builder setCredentialRotationRequiredState() {
+      internalProperties.put(
+          PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE, "true");
+      return this;
+    }
+
     public PrincipalEntity build() {
       return new PrincipalEntity(buildBase());
     }

polaris/polaris-core/src/main/java/io/polaris/core/persistence/PolarisMetaStoreManagerImpl.java

@@ -906,9 +906,6 @@ private Map<String, String> deserializeProperties(PolarisCallContext callCtx, St
     internalProperties.put(
         PolarisEntityConstants.getClientIdPropertyName(), principalSecrets.getPrincipalClientId());
 
-    // set the OTP flag
-    internalProperties.put(PolarisEntityConstants.PRINCIPAL_OTP_STATE, "true");
-
     // remember client id
     principal.setInternalProperties(this.serializeProperties(callCtx, internalProperties));
 
@@ -973,18 +970,25 @@ private Map<String, String> deserializeProperties(PolarisCallContext callCtx, St
             principal.getInternalProperties() == null ? "{}" : principal.getInternalProperties());
 
     boolean doReset =
-        reset || internalProps.get(PolarisEntityConstants.PRINCIPAL_OTP_STATE) != null;
+        reset
+            || internalProps.get(
+                    PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)
+                != null;
     PolarisPrincipalSecrets secrets =
         ms.rotatePrincipalSecrets(callCtx, clientId, principalId, masterSecret, doReset);
 
-    if (reset && !internalProps.containsKey(PolarisEntityConstants.PRINCIPAL_OTP_STATE)) {
-      internalProps.put(PolarisEntityConstants.PRINCIPAL_OTP_STATE, "true");
+    if (reset
+        && !internalProps.containsKey(
+            PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)) {
+      internalProps.put(
+          PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE, "true");
       principal.setInternalProperties(
           PolarisObjectMapperUtil.serializeProperties(callCtx, internalProps));
       principal.setEntityVersion(principal.getEntityVersion() + 1);
       writeEntity(callCtx, ms, principal, true);
-    } else if (internalProps.containsKey(PolarisEntityConstants.PRINCIPAL_OTP_STATE)) {
-      internalProps.remove(PolarisEntityConstants.PRINCIPAL_OTP_STATE);
+    } else if (internalProps.containsKey(
+        PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)) {
+      internalProps.remove(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE);
       principal.setInternalProperties(
           PolarisObjectMapperUtil.serializeProperties(callCtx, internalProps));
       principal.setEntityVersion(principal.getEntityVersion() + 1);

polaris/polaris-core/src/testFixtures/java/io/polaris/core/persistence/PolarisTestMetaStoreManager.java

@@ -429,6 +429,10 @@ PolarisBaseEntity createPrincipal(String name) {
             PolarisEntitySubType.NULL_SUBTYPE,
             PolarisEntityConstants.getRootEntityId(),
             name);
+    principalEntity.setInternalProperties(
+        PolarisObjectMapperUtil.serializeProperties(
+            this.polarisCallContext,
+            Map.of(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE, "true")));
     PolarisMetaStoreManager.CreatePrincipalResult createPrincipalResult =
         polarisMetaStoreManager.createPrincipal(this.polarisCallContext, principalEntity);
     Assertions.assertNotNull(createPrincipalResult);
@@ -469,7 +473,9 @@ PolarisBaseEntity createPrincipal(String name) {
     Map<String, String> internalProperties =
         PolarisObjectMapperUtil.deserializeProperties(
             this.polarisCallContext, createPrincipalResult.getPrincipal().getInternalProperties());
-    Assertions.assertNotNull(internalProperties.get(PolarisEntityConstants.PRINCIPAL_OTP_STATE));
+    Assertions.assertNotNull(
+        internalProperties.get(
+            PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE));
 
     // simulate retry if we are asked to
     if (this.doRetry) {
@@ -508,7 +514,9 @@ PolarisBaseEntity createPrincipal(String name) {
     internalProperties =
         PolarisObjectMapperUtil.deserializeProperties(
             this.polarisCallContext, reloadPrincipal.getInternalProperties());
-    Assertions.assertNull(internalProperties.get(PolarisEntityConstants.PRINCIPAL_OTP_STATE));
+    Assertions.assertNull(
+        internalProperties.get(
+            PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE));
 
     // rotate the secrets, twice!
     polarisMetaStoreManager.rotatePrincipalSecrets(
@@ -551,9 +559,12 @@ PolarisBaseEntity createPrincipal(String name) {
     internalProperties =
         PolarisObjectMapperUtil.deserializeProperties(
             this.polarisCallContext, newPrincipal.getInternalProperties());
-    Assertions.assertNotNull(internalProperties.get(PolarisEntityConstants.PRINCIPAL_OTP_STATE));
+    Assertions.assertNotNull(
+        internalProperties.get(
+            PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE));
 
-    // reset again. we should get new secrets and the OTP flag should be gone
+    // reset again. we should get new secrets and the CREDENTIAL_ROTATION_REQUIRED flag should be
+    // gone
     polarisMetaStoreManager.rotatePrincipalSecrets(
         this.polarisCallContext,
         clientId,
@@ -579,7 +590,9 @@ PolarisBaseEntity createPrincipal(String name) {
     internalProperties =
         PolarisObjectMapperUtil.deserializeProperties(
             this.polarisCallContext, finalPrincipal.getInternalProperties());
-    Assertions.assertNull(internalProperties.get(PolarisEntityConstants.PRINCIPAL_OTP_STATE));
+    Assertions.assertNull(
+        internalProperties.get(
+            PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE));
 
     // return it
     return finalPrincipal;

polaris/polaris-server.yml

@@ -51,7 +51,7 @@ server:
 baseCatalogType: "polaris"
 
 featureConfiguration:
-  ENFORCE_PRINCIPAL_OTP_CHECKING: false
+  ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING: false
   DISABLE_TOKEN_GENERATION_FOR_USER_PRINCIPALS: true
   SUPPORTED_CATALOG_STORAGE_TYPES:
     - S3

polaris/polaris-service/src/main/java/io/polaris/service/admin/PolarisServiceImpl.java

@@ -160,8 +160,12 @@ public Response listCatalogs(SecurityContext securityContext) {
   @Override
   public Response createPrincipal(CreatePrincipalRequest request, SecurityContext securityContext) {
     PolarisAdminService adminService = newAdminService(securityContext);
-    PrincipalWithCredentials createdPrincipal =
-        adminService.createPrincipal(PrincipalEntity.fromPrincipal(request.getPrincipal()));
+    PrincipalEntity principal = PrincipalEntity.fromPrincipal(request.getPrincipal());
+    if (Boolean.TRUE.equals(request.getCredentialRotationRequired())) {
+      principal =
+          new PrincipalEntity.Builder(principal).setCredentialRotationRequiredState().build();
+    }
+    PrincipalWithCredentials createdPrincipal = adminService.createPrincipal(principal);
     LOG.info("Created new principal {}", createdPrincipal);
     return Response.status(Response.Status.CREATED).entity(createdPrincipal).build();
   }

polaris/polaris-service/src/main/java/io/polaris/service/config/Serializers.java

@@ -80,6 +80,9 @@ public CreatePrincipalRequest deserialize(JsonParser p, DeserializationContext c
         return CreatePrincipalRequest.builder()
             .setPrincipal(
                 ctxt.readTreeAsValue((JsonNode) treeNode.get("principal"), Principal.class))
+            .setCredentialRotationRequired(
+                ctxt.readTreeAsValue(
+                    (JsonNode) treeNode.get("credentialRotationRequired"), Boolean.class))
             .build();
       } else {
         return CreatePrincipalRequest.builder()

polaris/polaris-service/src/test/java/io/polaris/service/admin/PolarisAuthzTestBase.java

@@ -105,7 +105,9 @@ public abstract class PolarisAuthzTestBase {
   protected final PolarisAuthorizer polarisAuthorizer =
       new PolarisAuthorizer(
           new DefaultConfigurationStore(
-              Map.of(PolarisConfiguration.ENFORCE_PRINCIPAL_OTP_CHECKING, true)));
+              Map.of(
+                  PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING,
+                  true)));
 
   protected BasePolarisCatalog baseCatalog;
   protected PolarisAdminService adminService;
@@ -268,7 +270,7 @@ public void after() {
         credentials.getClientId(),
         lookupEntity.getEntity().getId(),
         credentials.getClientSecret(),
-        true);
+        false);
 
     return new PrincipalEntity(
         PolarisEntity.of(

polaris/polaris-service/src/test/java/io/polaris/service/admin/PolarisServiceImplIntegrationTest.java

@@ -235,12 +235,12 @@ public void testCreateCatalogWithNullBaseLocation() {
     ObjectMapper mapper = new ObjectMapper();
     JsonNode storageConfig = mapper.valueToTree(awsConfigModel);
     ObjectNode catalogNode = mapper.createObjectNode();
-    catalogNode.put("storageConfigInfo", storageConfig);
+    catalogNode.set("storageConfigInfo", storageConfig);
     catalogNode.put("name", "my-catalog");
     catalogNode.put("type", "INTERNAL");
-    catalogNode.put("properties", mapper.createObjectNode());
+    catalogNode.set("properties", mapper.createObjectNode());
     ObjectNode requestNode = mapper.createObjectNode();
-    requestNode.put("catalog", catalogNode);
+    requestNode.set("catalog", catalogNode);
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/catalogs")
             .post(Entity.json(requestNode))) {
@@ -262,11 +262,11 @@ public void testCreateCatalogWithoutProperties() {
     ObjectMapper mapper = new ObjectMapper();
     JsonNode storageConfig = mapper.valueToTree(awsConfigModel);
     ObjectNode catalogNode = mapper.createObjectNode();
-    catalogNode.put("storageConfigInfo", storageConfig);
+    catalogNode.set("storageConfigInfo", storageConfig);
     catalogNode.put("name", "my-catalog");
     catalogNode.put("type", "INTERNAL");
     ObjectNode requestNode = mapper.createObjectNode();
-    requestNode.put("catalog", catalogNode);
+    requestNode.set("catalog", catalogNode);
 
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/catalogs", "Bearer " + userToken)
@@ -469,14 +469,14 @@ public void testCreateCatalogWithoutDefaultLocation() {
     ObjectMapper mapper = new ObjectMapper();
     JsonNode storageConfig = mapper.valueToTree(awsConfigModel);
     ObjectNode catalogNode = mapper.createObjectNode();
-    catalogNode.put("storageConfigInfo", storageConfig);
+    catalogNode.set("storageConfigInfo", storageConfig);
     catalogNode.put("name", "my-catalog");
     catalogNode.put("type", "INTERNAL");
     ObjectNode properties = mapper.createObjectNode();
-    properties.put("default-base-location", mapper.nullNode());
-    catalogNode.put("properties", properties);
+    properties.set("default-base-location", mapper.nullNode());
+    catalogNode.set("properties", properties);
     ObjectNode requestNode = mapper.createObjectNode();
-    requestNode.put("catalog", catalogNode);
+    requestNode.set("catalog", catalogNode);
 
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/catalogs")
@@ -756,7 +756,7 @@ public void testCreatePrincipalAndRotateCredentials() {
     Principal returnedPrincipal = null;
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/principals")
-            .post(Entity.json(new CreatePrincipalRequest(principal)))) {
+            .post(Entity.json(new CreatePrincipalRequest(principal, true)))) {
       assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
       PrincipalWithCredentials parsed = response.readEntity(PrincipalWithCredentials.class);
       creds = parsed.getCredentials();
@@ -776,10 +776,27 @@ public void testCreatePrincipalAndRotateCredentials() {
       assertThat(response).returns(Response.Status.FORBIDDEN.getStatusCode(), Response::getStatus);
     }
 
-    // Now try to rotate using a fresh token associated with the principal itself.
+    // Get a fresh token associate with the principal itself.
     String newPrincipalToken =
         TokenUtils.getTokenFromSecrets(
             EXT.client(), EXT.getLocalPort(), oldClientId, oldSecret, realm);
+
+    // Any call should initially fail with error indicating that rotation is needed.
+    try (Response response =
+        newRequest(
+                "http://localhost:%d/api/management/v1/principals/myprincipal",
+                "Bearer " + newPrincipalToken)
+            .get()) {
+      assertThat(response).returns(Response.Status.FORBIDDEN.getStatusCode(), Response::getStatus);
+      ErrorResponse error = response.readEntity(ErrorResponse.class);
+      assertThat(error)
+          .isNotNull()
+          .extracting(ErrorResponse::message)
+          .asString()
+          .contains("PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE");
+    }
+
+    // Now try to rotate using the principal's token.
     try (Response response =
         newRequest(
                 "http://localhost:%d/api/management/v1/principals/myprincipal/rotate",
@@ -810,14 +827,14 @@ public void testCreateListUpdateAndDeletePrincipal() {
             .build();
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/principals")
-            .post(Entity.json(new CreatePrincipalRequest(principal)))) {
+            .post(Entity.json(new CreatePrincipalRequest(principal, null)))) {
       assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
     }
 
     // Second attempt to create the same entity should fail with CONFLICT.
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/principals")
-            .post(Entity.json(new CreatePrincipalRequest(principal)))) {
+            .post(Entity.json(new CreatePrincipalRequest(principal, false)))) {
       assertThat(response).returns(Response.Status.CONFLICT.getStatusCode(), Response::getStatus);
     }
 
@@ -1156,15 +1173,15 @@ public void testAssignListAndRevokePrincipalRoles() {
     Principal principal1 = new Principal(Principal.TypeEnum.SERVICE, "myprincipal1");
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/principals")
-            .post(Entity.json(new CreatePrincipalRequest(principal1)))) {
+            .post(Entity.json(new CreatePrincipalRequest(principal1, false)))) {
 
       assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
     }
 
     Principal principal2 = new Principal(Principal.TypeEnum.SERVICE, "myprincipal2");
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/principals")
-            .post(Entity.json(new CreatePrincipalRequest(principal2)))) {
+            .post(Entity.json(new CreatePrincipalRequest(principal2, false)))) {
 
       assertThat(response).returns(Response.Status.CREATED.getStatusCode(), Response::getStatus);
     }

polaris/polaris-service/src/test/java/io/polaris/service/catalog/PolarisCatalogHandlerWrapperAuthzTest.java

@@ -187,6 +187,7 @@ public void testInsufficientPermissionsPriorToSecretRotation() {
                 new PrincipalEntity.Builder()
                     .setName(principalName)
                     .setCreateTimestamp(Instant.now().toEpochMilli())
+                    .setCredentialRotationRequiredState()
                     .build());
     adminService.assignPrincipalRole(principalName, PRINCIPAL_ROLE1);
     adminService.assignPrincipalRole(principalName, PRINCIPAL_ROLE2);

polaris/polaris-service/src/test/resources/polaris-server-integrationtest.yml

@@ -51,7 +51,7 @@ server:
 baseCatalogType: "polaris"
 
 featureConfiguration:
-  ENFORCE_PRINCIPAL_OTP_CHECKING: true
+  ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING: true
   DISABLE_TOKEN_GENERATION_FOR_USER_PRINCIPALS: true
   ALLOW_WILDCARD_LOCATION: true
   SUPPORTED_CATALOG_STORAGE_TYPES:

polaris/spec/polaris-management-service.yml

@@ -903,6 +903,9 @@ components:
       properties:
         principal:
           $ref: '#/components/schemas/Principal'
+        credentialRotationRequired:
+          type: boolean
+          description: If true, the initial credentials can only be used to call rotateCredentials
 
     Principal:
       description: A Polaris principal.

snowflake-polaris/src/test/resources/polaris-server-integrationtest.yml

@@ -51,7 +51,7 @@ server:
 baseCatalogType: "polaris"
 
 featureConfiguration:
-  ENFORCE_PRINCIPAL_OTP_CHECKING: true
+  ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING: true
   DISABLE_TOKEN_GENERATION_FOR_USER_PRINCIPALS: true
   ALLOW_WILDCARD_LOCATION: true
   SUPPORTED_CATALOG_STORAGE_TYPES: