Merge pull request #1014 from zalando/aruha2132

Added Authorization implementation according latest plugin

build.gradle

@@ -138,7 +138,7 @@ dependencies {
     compile "io.dropwizard.metrics:metrics-servlets:$dropwizardVersion"
     compile "io.dropwizard.metrics:metrics-jvm:$dropwizardVersion"
     compile 'org.apache.commons:commons-lang3:3.8.1'
-    compile 'org.zalando:nakadi-plugin-api:3.1.2'
+    compile 'org.zalando:nakadi-plugin-api:3.2.1'
     compile 'org.echocat.jomon:runtime:1.6.3'
 
     // kafka & zookeeper

src/main/java/org/zalando/nakadi/controller/StoragesController.java

@@ -17,8 +17,8 @@
 import org.zalando.nakadi.exceptions.runtime.NoSuchStorageException;
 import org.zalando.nakadi.exceptions.runtime.StorageIsUsedException;
 import org.zalando.nakadi.exceptions.runtime.UnknownStorageTypeException;
-import org.zalando.nakadi.plugin.api.PluginException;
 import org.zalando.nakadi.plugin.api.authz.AuthorizationService;
+import org.zalando.nakadi.plugin.api.exceptions.PluginException;
 import org.zalando.nakadi.service.AdminService;
 import org.zalando.nakadi.service.StorageService;
 

src/main/java/org/zalando/nakadi/domain/EventType.java

@@ -15,7 +15,6 @@ public EventType(final EventTypeBase eventType, final String version, final Date
         super(eventType);
         this.updatedAt = updatedAt;
         this.createdAt = createdAt;
-
         this.setSchema(new EventTypeSchema(eventType.getSchema(), version, updatedAt));
     }
 

src/main/java/org/zalando/nakadi/domain/EventTypeBase.java

@@ -4,6 +4,8 @@
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.google.common.collect.ImmutableList;
 import org.zalando.nakadi.partitioning.PartitionStrategy;
+import org.zalando.nakadi.plugin.api.authz.EventTypeAuthz;
+import org.zalando.nakadi.plugin.api.authz.Resource;
 
 import javax.annotation.Nullable;
 import javax.validation.Valid;
@@ -15,7 +17,7 @@
 
 import static java.util.Collections.unmodifiableList;
 
-public class EventTypeBase {
+public class EventTypeBase implements EventTypeAuthz {
 
     private static final List<String> EMPTY_PARTITION_KEY_FIELDS = ImmutableList.of();
     private static final List<String> EMPTY_ORDERING_KEY_FIELDS = ImmutableList.of();
@@ -255,4 +257,21 @@ public void setAudience(@Nullable final Audience audience) {
     public void setAuthorization(final ResourceAuthorization authorization) {
         this.authorization = authorization;
     }
+
+    @JsonIgnore
+    @Override
+    public String getAuthCompatibilityMode() {
+        return this.compatibilityMode.toString();
+    }
+
+    @JsonIgnore
+    @Override
+    public String getAuthCleanupPolicy() {
+        return this.cleanupPolicy.toString();
+    }
+
+    @JsonIgnore
+    public Resource<EventTypeBase> asBaseResource() {
+        return new ResourceImpl<>(getName(), ResourceImpl.EVENT_TYPE_RESOURCE, getAuthorization(), this);
+    }
 }

src/main/java/org/zalando/nakadi/domain/ResourceAuthorization.java

@@ -104,13 +104,12 @@ public Optional<List<AuthorizationAttribute>> getAttributesForOperation(
         }
     }
 
-
     @Override
     public Map<String, List<AuthorizationAttribute>> asMapValue() {
         return ImmutableMap.of(
-                "admins", getAdmins(),
-                "readers", getReaders(),
-                "writers", getWriters());
+                AuthorizationService.Operation.ADMIN.toString(), getAdmins(),
+                AuthorizationService.Operation.READ.toString(), getReaders(),
+                AuthorizationService.Operation.WRITE.toString(), getWriters());
     }
 
     @Override

src/main/java/org/zalando/nakadi/domain/ResourceImpl.java

@@ -5,6 +5,7 @@
 import org.zalando.nakadi.plugin.api.authz.Resource;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 public class ResourceImpl<T> implements Resource<T> {
@@ -13,6 +14,7 @@ public class ResourceImpl<T> implements Resource<T> {
     public static final String ADMIN_RESOURCE = "nakadi";
     public static final String EVENT_TYPE_RESOURCE = "event-type";
     public static final String SUBSCRIPTION_RESOURCE = "subscription";
+    public static final String PERMISSION_RESOURCE = "permission";
 
     private final T resource;
     private final String name;
@@ -40,9 +42,20 @@ public String getType() {
     @Override
     public Optional<List<AuthorizationAttribute>> getAttributesForOperation(
             final AuthorizationService.Operation operation) {
+        if (null == authorization) {
+            return Optional.empty();
+        }
         return authorization.getAttributesForOperation(operation);
     }
 
+    @Override
+    public Map<String, List<AuthorizationAttribute>> getAuthorization() {
+        if (authorization != null) {
+            return authorization.asMapValue();
+        }
+        return null;
+    }
+
     @Override
     public T get() {
         return resource;

src/main/java/org/zalando/nakadi/domain/SubscriptionAuthorization.java

@@ -64,8 +64,8 @@ public int hashCode() {
     @Override
     public Map<String, List<AuthorizationAttribute>> asMapValue() {
         return ImmutableMap.of(
-                "admins", getAdmins(),
-                "readers", getReaders()
+                AuthorizationService.Operation.ADMIN.toString(), getAdmins(),
+                AuthorizationService.Operation.READ.toString(), getReaders()
         );
     }
 

src/main/java/org/zalando/nakadi/domain/SubscriptionBase.java

@@ -1,6 +1,7 @@
 package org.zalando.nakadi.domain;
 
 import com.google.common.collect.ImmutableList;
+import org.zalando.nakadi.plugin.api.authz.Resource;
 import org.zalando.nakadi.view.SubscriptionCursorWithoutToken;
 
 import javax.annotation.Nullable;
@@ -126,4 +127,8 @@ public boolean equals(final Object o) {
     public int hashCode() {
         return Objects.hash(owningApplication, eventTypes, consumerGroup, readFrom, initialCursors);
     }
+
+    public Resource<SubscriptionBase> asBaseResource(final String id) {
+        return new ResourceImpl<>(id, ResourceImpl.SUBSCRIPTION_RESOURCE, getAuthorization(), this);
+    }
 }

src/main/java/org/zalando/nakadi/exceptions/runtime/AccessDeniedException.java

@@ -12,6 +12,11 @@ public AccessDeniedException(final AuthorizationService.Operation operation, fin
         this.operation = operation;
     }
 
+    public AccessDeniedException(final Resource resource) {
+        this.resource = resource;
+        this.operation = null;
+    }
+
     public Resource getResource() {
         return resource;
     }

src/main/java/org/zalando/nakadi/plugin/auth/DefaultAuthorizationService.java

@@ -1,10 +1,12 @@
 package org.zalando.nakadi.plugin.auth;
 
-import org.zalando.nakadi.plugin.api.PluginException;
-import org.zalando.nakadi.plugin.api.authz.AuthorizationAttribute;
+
 import org.zalando.nakadi.plugin.api.authz.AuthorizationService;
 import org.zalando.nakadi.plugin.api.authz.Resource;
 import org.zalando.nakadi.plugin.api.authz.Subject;
+import org.zalando.nakadi.plugin.api.exceptions.AuthorizationInvalidException;
+import org.zalando.nakadi.plugin.api.exceptions.OperationOnResourceNotPermittedException;
+import org.zalando.nakadi.plugin.api.exceptions.PluginException;
 
 import javax.annotation.Nullable;
 import java.util.List;
@@ -13,13 +15,17 @@
 public class DefaultAuthorizationService implements AuthorizationService {
 
     @Override
-    public boolean isAuthorized(final Operation operation, final Resource resource) {
+    public boolean isAuthorized(final Operation operation, final Resource resource) throws PluginException {
         return true;
     }
 
     @Override
-    public boolean isAuthorizationAttributeValid(final AuthorizationAttribute authorizationAttribute) {
-        return true;
+    public void isAuthorizationForResourceValid(final Resource resource) throws PluginException,
+            AuthorizationInvalidException, OperationOnResourceNotPermittedException {
+    }
+
+    public DefaultAuthorizationService() {
+        super();
     }
 
     @Override

src/main/java/org/zalando/nakadi/service/AdminService.java

@@ -13,9 +13,11 @@
 import org.zalando.nakadi.domain.ResourceImpl;
 import org.zalando.nakadi.exceptions.runtime.DbWriteOperationsBlockedException;
 import org.zalando.nakadi.exceptions.runtime.UnableProcessException;
-import org.zalando.nakadi.plugin.api.PluginException;
+import org.zalando.nakadi.exceptions.runtime.UnprocessableEntityException;
 import org.zalando.nakadi.plugin.api.authz.AuthorizationService;
 import org.zalando.nakadi.plugin.api.authz.Resource;
+import org.zalando.nakadi.plugin.api.exceptions.AuthorizationInvalidException;
+import org.zalando.nakadi.plugin.api.exceptions.PluginException;
 import org.zalando.nakadi.repository.db.AuthorizationDbRepository;
 
 import java.util.List;
@@ -27,6 +29,7 @@
 
 import static org.zalando.nakadi.domain.ResourceImpl.ADMIN_RESOURCE;
 import static org.zalando.nakadi.domain.ResourceImpl.ALL_DATA_ACCESS_RESOURCE;
+import static org.zalando.nakadi.domain.ResourceImpl.PERMISSION_RESOURCE;
 
 @Service
 public class AdminService {
@@ -115,16 +118,11 @@ private List<Permission> removeDefaultAdmin(final List<Permission> permissions)
     }
 
     private void validateAllAdmins(final List<Permission> admins) throws UnableProcessException, PluginException {
-        final List<Permission> invalid = admins.stream().filter(permission ->
-                !authorizationService.isAuthorizationAttributeValid(permission.getAuthorizationAttribute()))
-                .collect(Collectors.toList());
-        if (!invalid.isEmpty()) {
-            final String message = invalid.stream()
-                    .map(permission -> String.format("authorization attribute %s:%s is invalid",
-                            permission.getAuthorizationAttribute().getDataType(),
-                            permission.getAuthorizationAttribute().getValue()))
-                    .collect(Collectors.joining(", "));
-            throw new UnableProcessException(message);
+        try {
+            authorizationService.isAuthorizationForResourceValid(new ResourceImpl<>(PERMISSION_RESOURCE,
+                    PERMISSION_RESOURCE, ResourceAuthorization.fromPermissionsList(admins), null));
+        } catch (AuthorizationInvalidException e) {
+            throw new UnprocessableEntityException(e.getMessage());
         }
     }
 }

src/main/java/org/zalando/nakadi/service/AuthorizationValidator.java

@@ -5,19 +5,20 @@
 import org.springframework.stereotype.Service;
 import org.zalando.nakadi.domain.EventType;
 import org.zalando.nakadi.domain.Subscription;
-import org.zalando.nakadi.domain.ValidatableAuthorization;
 import org.zalando.nakadi.exceptions.runtime.AccessDeniedException;
+import org.zalando.nakadi.exceptions.runtime.ForbiddenOperationException;
 import org.zalando.nakadi.exceptions.runtime.InternalNakadiException;
 import org.zalando.nakadi.exceptions.runtime.ServiceTemporarilyUnavailableException;
 import org.zalando.nakadi.exceptions.runtime.UnableProcessException;
-import org.zalando.nakadi.plugin.api.PluginException;
+import org.zalando.nakadi.exceptions.runtime.UnprocessableEntityException;
 import org.zalando.nakadi.plugin.api.authz.AuthorizationAttribute;
 import org.zalando.nakadi.plugin.api.authz.AuthorizationService;
 import org.zalando.nakadi.plugin.api.authz.Resource;
+import org.zalando.nakadi.plugin.api.exceptions.AuthorizationInvalidException;
+import org.zalando.nakadi.plugin.api.exceptions.OperationOnResourceNotPermittedException;
+import org.zalando.nakadi.plugin.api.exceptions.PluginException;
 import org.zalando.nakadi.repository.EventTypeRepository;
 
-import javax.annotation.Nullable;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -44,12 +45,11 @@ public AuthorizationValidator(
         this.adminService = adminService;
     }
 
-    public void validateAuthorization(@Nullable final ValidatableAuthorization auth) throws UnableProcessException,
+    public void validateAuthorization(final Resource resource) throws UnableProcessException,
             ServiceTemporarilyUnavailableException {
-        if (auth != null) {
-            final Map<String, List<AuthorizationAttribute>> authorization = auth.asMapValue();
-            checkAuthAttributesAreValid(authorization);
-            checkAuthAttributesNoDuplicates(authorization);
+        checkAuthorisationForResourceAreValid(resource);
+        if (resource.getAuthorization() != null) {
+            checkAuthAttributesNoDuplicates(resource.getAuthorization());
         }
     }
 
@@ -86,24 +86,21 @@ private void checkAuthAttributesNoDuplicates(final Map<String, List<Authorizatio
         }
     }
 
-    private void checkAuthAttributesAreValid(final Map<String, List<AuthorizationAttribute>> allAttributes)
-            throws UnableProcessException, ServiceTemporarilyUnavailableException {
+    private void checkAuthorisationForResourceAreValid(final Resource resource)
+            throws UnableProcessException, ServiceTemporarilyUnavailableException, ForbiddenOperationException {
+
         try {
-            final String errorMessage = allAttributes.values().stream()
-                    .flatMap(Collection::stream)
-                    .filter(attr -> !authorizationService.isAuthorizationAttributeValid(attr))
-                    .map(attr -> String.format("authorization attribute %s:%s is invalid",
-                            attr.getDataType(), attr.getValue()))
-                    .collect(Collectors.joining(", "));
-
-            if (!Strings.isNullOrEmpty(errorMessage)) {
-                throw new UnableProcessException(errorMessage);
-            }
-        } catch (final PluginException e) {
+            authorizationService.isAuthorizationForResourceValid(resource);
+        } catch (OperationOnResourceNotPermittedException e) {
+            throw new ForbiddenOperationException(e.getMessage());
+        } catch (AuthorizationInvalidException e) {
+            throw new UnprocessableEntityException(e.getMessage());
+        } catch (PluginException e) {
             throw new ServiceTemporarilyUnavailableException("Error calling authorization plugin", e);
         }
     }
 
+
     public void authorizeEventTypeWrite(final EventType eventType)
             throws AccessDeniedException, ServiceTemporarilyUnavailableException {
         if (eventType.getAuthorization() == null) {
@@ -148,28 +145,13 @@ public void authorizeStreamRead(final EventType eventType) throws AccessDeniedEx
         }
 
         final Resource resource = eventType.asResource();
-        try {
-            if (!authorizationService.isAuthorized(AuthorizationService.Operation.READ, resource)
-                    && !adminService.hasAllDataAccess(AuthorizationService.Operation.READ)) {
-                throw new AccessDeniedException(AuthorizationService.Operation.READ, resource);
-            }
-        } catch (final PluginException e) {
-            throw new ServiceTemporarilyUnavailableException("Error calling authorization plugin", e);
-        }
+        checkResourceAuthorization(resource);
     }
 
     public void authorizeSubscriptionRead(final Subscription subscription) throws AccessDeniedException {
         if (null != subscription.getAuthorization()) {
             final Resource resource = subscription.asResource();
-            try {
-                // In this case operation for read will invoke
-                if (!authorizationService.isAuthorized(AuthorizationService.Operation.READ, resource)
-                        && !adminService.hasAllDataAccess(AuthorizationService.Operation.READ)) {
-                    throw new AccessDeniedException(AuthorizationService.Operation.READ, resource);
-                }
-            } catch (final PluginException e) {
-                throw new ServiceTemporarilyUnavailableException("Error calling authorization plugin", e);
-            }
+            checkResourceAuthorization(resource);
         }
         subscription.getEventTypes().forEach(
                 (eventTypeName) -> {
@@ -187,6 +169,12 @@ public void authorizeSubscriptionCommit(final Subscription subscription) throws
             return;
         }
         final Resource resource = subscription.asResource();
+        checkResourceAuthorization(resource);
+
+    }
+
+    private void checkResourceAuthorization(final Resource resource)
+            throws ServiceTemporarilyUnavailableException, AccessDeniedException {
         try {
             if (!authorizationService.isAuthorized(AuthorizationService.Operation.READ, resource)
                     && !adminService.hasAllDataAccess(AuthorizationService.Operation.READ)) {
@@ -204,14 +192,16 @@ public void authorizeSubscriptionAdmin(final Subscription subscription) throws A
         authorizeResourceAdmin(subscription.asResource());
     }
 
-    public void validateAuthorization(final ValidatableAuthorization oldValue, final ValidatableAuthorization newValue)
+    public void validateAuthorization(final Resource oldValue, final Resource newValue)
             throws UnableProcessException, ServiceTemporarilyUnavailableException {
-        if (oldValue != null && newValue == null) {
+        final Map<String, List<AuthorizationAttribute>> oldAuth = oldValue.getAuthorization();
+        final Map<String, List<AuthorizationAttribute>> newAuth = newValue.getAuthorization();
+        if (oldAuth != null && newAuth == null) {
             throw new UnableProcessException(
                     "Changing authorization object to `null` is not possible due to existing one");
         }
 
-        if (oldValue != null && oldValue.equals(newValue)) {
+        if (oldAuth != null && oldAuth.equals(newAuth)) {
             return;
         }
 

src/main/java/org/zalando/nakadi/service/EventTypeService.java

@@ -158,7 +158,7 @@ public void create(final EventTypeBase eventType)
         validateCompaction(eventType);
         enrichment.validate(eventType);
         partitionResolver.validate(eventType);
-        authorizationValidator.validateAuthorization(eventType.getAuthorization());
+        authorizationValidator.validateAuthorization(eventType.asBaseResource());
 
         eventTypeRepository.saveEventType(eventType);
 
@@ -348,7 +348,7 @@ public void update(final String eventTypeName,
                 eventTypeOptionsValidator.checkRetentionTime(eventTypeBase.getOptions());
                 authorizationValidator.authorizeEventTypeAdmin(original);
             }
-            authorizationValidator.validateAuthorization(original.getAuthorization(), eventTypeBase.getAuthorization());
+            authorizationValidator.validateAuthorization(original.asResource(), eventTypeBase.asBaseResource());
             validateName(eventTypeName, eventTypeBase);
             validateCompactionUpdate(original, eventTypeBase);
             validateSchema(eventTypeBase);