http://git-wip-us.apache.org/repos/asf/nifi/blob/f47be77b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java index af3f8a4..50fa505 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java @@ -25,9 +25,12 @@ import org.apache.nifi.action.Operation; import org.apache.nifi.action.details.FlowChangePurgeDetails; import org.apache.nifi.admin.service.AuditService; import org.apache.nifi.admin.service.KeyService; +import org.apache.nifi.authorization.AccessPolicy; import org.apache.nifi.authorization.Authorizer; +import org.apache.nifi.authorization.Group; import org.apache.nifi.authorization.RequestAction; import org.apache.nifi.authorization.Resource; +import org.apache.nifi.authorization.User; import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.user.NiFiUser; import org.apache.nifi.authorization.user.NiFiUserUtils; @@ -114,6 +117,8 @@ import org.apache.nifi.web.api.dto.RevisionDTO; import org.apache.nifi.web.api.dto.SnippetDTO; import org.apache.nifi.web.api.dto.SystemDiagnosticsDTO; import org.apache.nifi.web.api.dto.TemplateDTO; +import org.apache.nifi.web.api.dto.UserDTO; +import org.apache.nifi.web.api.dto.UserGroupDTO; import org.apache.nifi.web.api.dto.action.ActionDTO; import org.apache.nifi.web.api.dto.action.HistoryDTO; import org.apache.nifi.web.api.dto.action.HistoryQueryDTO; @@ -130,6 +135,7 @@ import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO; import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO; import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO; import org.apache.nifi.web.api.dto.status.StatusHistoryDTO; +import org.apache.nifi.web.api.entity.AccessPolicyEntity; import org.apache.nifi.web.api.entity.ConnectionEntity; import org.apache.nifi.web.api.entity.ControllerConfigurationEntity; import org.apache.nifi.web.api.entity.ControllerServiceEntity; @@ -147,7 +153,10 @@ import org.apache.nifi.web.api.entity.RemoteProcessGroupPortEntity; import org.apache.nifi.web.api.entity.ReportingTaskEntity; import org.apache.nifi.web.api.entity.ScheduleComponentsEntity; import org.apache.nifi.web.api.entity.SnippetEntity; +import org.apache.nifi.web.api.entity.UserEntity; +import org.apache.nifi.web.api.entity.UserGroupEntity; import org.apache.nifi.web.controller.ControllerFacade; +import org.apache.nifi.web.dao.AccessPolicyDAO; import org.apache.nifi.web.dao.ConnectionDAO; import org.apache.nifi.web.dao.ControllerServiceDAO; import org.apache.nifi.web.dao.FunnelDAO; @@ -159,6 +168,8 @@ import org.apache.nifi.web.dao.RemoteProcessGroupDAO; import org.apache.nifi.web.dao.ReportingTaskDAO; import org.apache.nifi.web.dao.SnippetDAO; import org.apache.nifi.web.dao.TemplateDAO; +import org.apache.nifi.web.dao.UserDAO; +import org.apache.nifi.web.dao.UserGroupDAO; import org.apache.nifi.web.revision.DeleteRevisionTask; import org.apache.nifi.web.revision.ExpiredRevisionClaimException; import org.apache.nifi.web.revision.ReadOnlyRevisionCallback; @@ -178,6 +189,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.HashSet; @@ -220,6 +232,9 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { private ControllerServiceDAO controllerServiceDAO; private ReportingTaskDAO reportingTaskDAO; private TemplateDAO templateDAO; + private UserDAO userDAO; + private UserGroupDAO userGroupDAO; + private AccessPolicyDAO accessPolicyDAO; private ClusterCoordinator clusterCoordinator; private HeartbeatMonitor heartbeatMonitor; @@ -234,105 +249,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { private Authorizer authorizer; - private final AuthorizableLookup authorizableLookup = new AuthorizableLookup() { - @Override - public Authorizable getProcessor(final String id) { - return processorDAO.getProcessor(id); - } - - @Override - public Authorizable getInputPort(final String id) { - return inputPortDAO.getPort(id); - } - - @Override - public Authorizable getOutputPort(final String id) { - return outputPortDAO.getPort(id); - } - - @Override - public Authorizable getConnection(final String id) { - return connectionDAO.getConnection(id); - } - - @Override - public Authorizable getProcessGroup(final String id) { - return processGroupDAO.getProcessGroup(id); - } - - @Override - public Authorizable getRemoteProcessGroup(final String id) { - return remoteProcessGroupDAO.getRemoteProcessGroup(id); - } - - @Override - public Authorizable getRemoteProcessGroupInputPort(final String remoteProcessGroupId, final String id) { - final RemoteProcessGroup remoteProcessGroup = remoteProcessGroupDAO.getRemoteProcessGroup(remoteProcessGroupId); - return remoteProcessGroup.getInputPort(id); - } - - @Override - public Authorizable getRemoteProcessGroupOutputPort(final String remoteProcessGroupId, final String id) { - final RemoteProcessGroup remoteProcessGroup = remoteProcessGroupDAO.getRemoteProcessGroup(remoteProcessGroupId); - return remoteProcessGroup.getOutputPort(id); - } - - @Override - public Authorizable getLabel(final String id) { - return labelDAO.getLabel(id); - } - - @Override - public Authorizable getFunnel(final String id) { - return funnelDAO.getFunnel(id); - } - - @Override - public Authorizable getControllerService(final String id) { - return controllerServiceDAO.getControllerService(id); - } - - @Override - public Authorizable getControllerServiceReferencingComponent(final String controllerSeriveId, final String id) { - final ControllerServiceNode controllerService = controllerServiceDAO.getControllerService(controllerSeriveId); - final ControllerServiceReference referencingComponents = controllerService.getReferences(); - - ConfiguredComponent reference = null; - for (final ConfiguredComponent component : referencingComponents.getReferencingComponents()) { - if (component.getIdentifier().equals(id)) { - reference = component; - break; - } - } - - if (reference == null) { - throw new ResourceNotFoundException("Unable to find referencing component with id " + id); - } - - return reference; - } - - @Override - public Authorizable getReportingTask(final String id) { - return reportingTaskDAO.getReportingTask(id); - } - - @Override - public Snippet getSnippet(final String id) { - return snippetDAO.getSnippet(id); - } - - @Override - public Authorizable getTemplate(final String id) { - return templateDAO.getTemplate(id); - } - - @Override - public Authorizable getConnectable(final String id) { - final ProcessGroup group = processGroupDAO.getProcessGroup(controllerFacade.getRootGroupId()); - return group.findConnectable(id); - } - }; + private AuthorizableLookup authorizableLookup; // ----------------------------------------- // Synchronization methods @@ -598,6 +515,61 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { // Write Operations // ----------------------------------------- @Override + public UpdateResult<AccessPolicyEntity> updateAccessPolicy(final Revision revision, final AccessPolicyDTO accessPolicyDTO) { + // if access policy does not exist, then create new access policy + if (!accessPolicyDAO.hasAccessPolicy(accessPolicyDTO.getId())) { + return new UpdateResult<>(createAccessPolicy(revision, accessPolicyDTO), false); + } + + final Authorizable accessPolicyAuthorizable = authorizableLookup.getAccessPolicyAuthorizable(accessPolicyDTO.getId()); + final RevisionUpdate<AccessPolicyDTO> snapshot = updateComponent(revision, + accessPolicyAuthorizable, + () -> accessPolicyDAO.updateAccessPolicy(accessPolicyDTO), + accessPolicy -> { + final Set<UserEntity> users = accessPolicy.getUsers().stream().map(userId -> getUser(userId, true) ).collect(Collectors.toSet()); + final Set<UserGroupEntity> userGroups = accessPolicy.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true) ).collect(Collectors.toSet()); + return dtoFactory.createAccessPolicyDto(accessPolicy, userGroups, users); + }); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(accessPolicyAuthorizable); + return new UpdateResult<>(entityFactory.createAccessPolicyEntity(snapshot.getComponent(), dtoFactory.createRevisionDTO(snapshot.getLastModification()), accessPolicy), false); + } + + @Override + public UpdateResult<UserEntity> updateUser(final Revision revision, final UserDTO userDTO) { + // if user does not exist, then create new user + if (!userDAO.hasUser(userDTO.getId())) { + return new UpdateResult<>(createUser(revision, userDTO), false); + } + + final Authorizable usersAuthorizable = authorizableLookup.getUsersAuthorizable(); + final RevisionUpdate<UserDTO> snapshot = updateComponent(revision, + usersAuthorizable, + () -> userDAO.updateUser(userDTO), + user -> dtoFactory.createUserDto(user, user.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet()))); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(usersAuthorizable); + return new UpdateResult<>(entityFactory.createUserEntity(snapshot.getComponent(), dtoFactory.createRevisionDTO(snapshot.getLastModification()), accessPolicy), false); + } + + @Override + public UpdateResult<UserGroupEntity> updateUserGroup(final Revision revision, final UserGroupDTO userGroupDTO) { + // if user group does not exist, then create new user group + if (!userGroupDAO.hasUserGroup(userGroupDTO.getId())) { + return new UpdateResult<>(createUserGroup(revision, userGroupDTO), false); + } + + final Authorizable userGroupsAuthorizable = authorizableLookup.getUserGroupsAuthorizable(); + final RevisionUpdate<UserGroupDTO> snapshot = updateComponent(revision, + userGroupsAuthorizable, + () -> userGroupDAO.updateUserGroup(userGroupDTO), + userGroup -> dtoFactory.createUserGroupDto(userGroup, userGroup.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet()))); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(userGroupsAuthorizable); + return new UpdateResult<>(entityFactory.createUserGroupEntity(snapshot.getComponent(), dtoFactory.createRevisionDTO(snapshot.getLastModification()), accessPolicy), false); + } + + @Override public UpdateResult<ConnectionEntity> updateConnection(final Revision revision, final ConnectionDTO connectionDTO) { // if connection does not exist, then create new connection if (connectionDAO.hasConnection(connectionDTO.getId()) == false) { @@ -685,7 +657,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { */ private <D, C> RevisionUpdate<D> updateComponent(final Revision revision, final Authorizable authorizable, final Supplier<C> daoUpdate, final Function<C, D> dtoCreation) { final NiFiUser user = NiFiUserUtils.getNiFiUser(); - final String modifier = user.getUserName(); + final String userName = NiFiUserUtils.getNiFiUserName(); try { final RevisionUpdate<D> updatedComponent = revisionManager.updateRevision(new StandardRevisionClaim(revision), user, new UpdateRevisionTask<D>() { @Override @@ -699,7 +671,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { final Revision updatedRevision = incrementRevision(revision); final D dto = dtoCreation.apply(component); - final FlowModification lastModification = new FlowModification(updatedRevision, modifier); + final FlowModification lastModification = new FlowModification(updatedRevision, userName); return new StandardRevisionUpdate<>(dto, lastModification); } }); @@ -730,7 +702,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { final NiFiUser user = NiFiUserUtils.getNiFiUser(); final RevisionClaim revisionClaim = new StandardRevisionClaim(revisions); - RevisionUpdate<SnippetDTO> snapshot; + final RevisionUpdate<SnippetDTO> snapshot; try { snapshot = revisionManager.updateRevision(revisionClaim, user, new UpdateRevisionTask<SnippetDTO>() { @Override @@ -905,19 +877,19 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { @Override public ControllerConfigurationEntity updateControllerConfiguration(final Revision revision, final ControllerConfigurationDTO controllerConfigurationDTO) { final RevisionUpdate<ControllerConfigurationDTO> updatedComponent = updateComponent( - revision, - controllerFacade, - () -> { - if (controllerConfigurationDTO.getMaxTimerDrivenThreadCount() != null) { - controllerFacade.setMaxTimerDrivenThreadCount(controllerConfigurationDTO.getMaxTimerDrivenThreadCount()); - } - if (controllerConfigurationDTO.getMaxEventDrivenThreadCount() != null) { - controllerFacade.setMaxEventDrivenThreadCount(controllerConfigurationDTO.getMaxEventDrivenThreadCount()); - } + revision, + controllerFacade, + () -> { + if (controllerConfigurationDTO.getMaxTimerDrivenThreadCount() != null) { + controllerFacade.setMaxTimerDrivenThreadCount(controllerConfigurationDTO.getMaxTimerDrivenThreadCount()); + } + if (controllerConfigurationDTO.getMaxEventDrivenThreadCount() != null) { + controllerFacade.setMaxEventDrivenThreadCount(controllerConfigurationDTO.getMaxEventDrivenThreadCount()); + } - return controllerConfigurationDTO; - }, - controller -> dtoFactory.createControllerConfigurationDto(controllerFacade, properties.getAutoRefreshInterval())); + return controllerConfigurationDTO; + }, + controller -> dtoFactory.createControllerConfigurationDto(controllerFacade, properties.getAutoRefreshInterval())); final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(controllerFacade); final RevisionDTO updateRevision = dtoFactory.createRevisionDTO(updatedComponent.getLastModification()); @@ -1068,6 +1040,48 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { } @Override + public UserEntity deleteUser(final Revision revision, final String userId) { + final User user = userDAO.getUser(userId); + final Set<UserGroupEntity> userGroups = user != null ? user.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet()) : null; + final UserDTO snapshot = deleteComponent( + revision, + authorizableLookup.getUsersAuthorizable(), + () -> userDAO.deleteUser(userId), + dtoFactory.createUserDto(user, userGroups)); + + return entityFactory.createUserEntity(snapshot, null, null); + } + + @Override + public UserGroupEntity deleteUserGroup(final Revision revision, final String userGroupId) { + final Group userGroup = userGroupDAO.getUserGroup(userGroupId); + final Set<UserEntity> users = userGroup != null ? userGroup.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet()) : + null; + final UserGroupDTO snapshot = deleteComponent( + revision, + authorizableLookup.getUserGroupsAuthorizable(), + () -> userGroupDAO.deleteUserGroup(userGroupId), + dtoFactory.createUserGroupDto(userGroup, users)); + + return entityFactory.createUserGroupEntity(snapshot, null, null); + } + + @Override + public AccessPolicyEntity deleteAccessPolicy(final Revision revision, final String accessPolicyId) { + final AccessPolicy accessPolicy = accessPolicyDAO.getAccessPolicy(accessPolicyId); + final Set<UserGroupEntity> userGroups = accessPolicy != null ? accessPolicy.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet()) : null; + final Set<UserEntity> users = accessPolicy != null ? accessPolicy.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet()) : null; + final AccessPolicyDTO snapshot = deleteComponent( + revision, + authorizableLookup.getAccessPolicyAuthorizable(accessPolicyId), + () -> accessPolicyDAO.deleteAccessPolicy(accessPolicyId), + dtoFactory.createAccessPolicyDto(accessPolicy, userGroups, + users)); + + return entityFactory.createAccessPolicyEntity(snapshot, null, null); + } + + @Override public FunnelEntity deleteFunnel(final Revision revision, final String funnelId) { final Funnel funnel = funnelDAO.getFunnel(funnelId); final FunnelDTO snapshot = deleteComponent( @@ -1306,6 +1320,47 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { return entityFactory.createFunnelEntity(snapshot.getComponent(), dtoFactory.createRevisionDTO(snapshot.getLastModification()), accessPolicy); } + @Override + public AccessPolicyEntity createAccessPolicy(final Revision revision, final AccessPolicyDTO accessPolicyDTO) { + final String creator = NiFiUserUtils.getNiFiUserName(); + if (revision.getVersion() != 0) { + throw new IllegalArgumentException("The revision must start at 0."); + } + final AccessPolicy newAccessPolicy = accessPolicyDAO.createAccessPolicy(accessPolicyDTO); + final AccessPolicyDTO newAccessPolicyDto = dtoFactory.createAccessPolicyDto(newAccessPolicy, + newAccessPolicy.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet()), + newAccessPolicy.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet())); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(authorizableLookup.getAccessPolicyAuthorizable(newAccessPolicy.getIdentifier())); + return entityFactory.createAccessPolicyEntity(newAccessPolicyDto, dtoFactory.createRevisionDTO(new FlowModification(revision, creator)), accessPolicy); + } + + @Override + public UserEntity createUser(final Revision revision, final UserDTO userDTO) { + final String creator = NiFiUserUtils.getNiFiUserName(); + if (revision.getVersion() != 0) { + throw new IllegalArgumentException("The revision must start at 0."); + } + final User newUser = userDAO.createUser(userDTO); + final UserDTO newUserDto = dtoFactory.createUserDto(newUser, newUser.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet())); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(authorizableLookup.getUsersAuthorizable()); + return entityFactory.createUserEntity(newUserDto, dtoFactory.createRevisionDTO(new FlowModification(revision, creator)), accessPolicy); + } + + @Override + public UserGroupEntity createUserGroup(final Revision revision, final UserGroupDTO userGroupDTO) { + final String creator = NiFiUserUtils.getNiFiUserName(); + if (revision.getVersion() != 0) { + throw new IllegalArgumentException("The revision must start at 0."); + } + final Group newUserGroup = userGroupDAO.createUserGroup(userGroupDTO); + final UserGroupDTO newUserGroupDto = dtoFactory.createUserGroupDto(newUserGroup, newUserGroup.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet())); + + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(authorizableLookup.getUserGroupsAuthorizable()); + return entityFactory.createUserGroupEntity(newUserGroupDto, dtoFactory.createRevisionDTO(new FlowModification(revision, creator)), accessPolicy); + } + private void validateSnippetContents(final FlowSnippetDTO flow) { // validate any processors if (flow.getProcessors() != null) { @@ -2349,6 +2404,78 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { } @Override + public AccessPolicyEntity getAccessPolicy(final String accessPolicyId) { + return revisionManager.get(accessPolicyId, rev -> { + final Authorizable accessPolicyAuthorizable = authorizableLookup.getAccessPolicyAuthorizable(accessPolicyId); + accessPolicyAuthorizable.authorize(authorizer, RequestAction.READ); + + final RevisionDTO revision = dtoFactory.createRevisionDTO(rev); + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(accessPolicyAuthorizable); + final AccessPolicy requestedAccessPolicy = accessPolicyDAO.getAccessPolicy(accessPolicyId); + return entityFactory.createAccessPolicyEntity( + dtoFactory.createAccessPolicyDto(requestedAccessPolicy, + requestedAccessPolicy.getGroups().stream().map(userGroupId -> getUserGroup(userGroupId, true)).collect(Collectors.toSet()), + requestedAccessPolicy.getUsers().stream().map(userId -> getUser(userId, true)).collect(Collectors.toSet())), + revision, accessPolicy); + }); + } + + @Override + public UserEntity getUser(final String userId, final boolean prune) { + return revisionManager.get(userId, rev -> { + final Authorizable usersAuthorizable = authorizableLookup.getUsersAuthorizable(); + usersAuthorizable.authorize(authorizer, RequestAction.READ); + + final RevisionDTO revision = dtoFactory.createRevisionDTO(rev); + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(usersAuthorizable); + final User user = userDAO.getUser(userId); + final Set<UserGroupEntity> userGroups = user.getGroups().stream() + .map(userGroupId -> prune ? getUserGroupPruned(userGroupId) : getUserGroup(userGroupId, false)) + .collect(Collectors.toSet()); + return entityFactory.createUserEntity(dtoFactory.createUserDto(user, userGroups), revision, accessPolicy); + }); + } + + private UserEntity getUserPruned(final String userId) { + return revisionManager.get(userId, rev -> { + final Authorizable usersAuthorizable = authorizableLookup.getUsersAuthorizable(); + usersAuthorizable.authorize(authorizer, RequestAction.READ); + + final RevisionDTO revision = dtoFactory.createRevisionDTO(rev); + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(usersAuthorizable); + final User user = userDAO.getUser(userId); + return entityFactory.createUserEntity(dtoFactory.createUserDto(user, Collections.emptySet()), revision, accessPolicy); + }); + } + + @Override + public UserGroupEntity getUserGroup(final String userGroupId, final boolean prune) { + return revisionManager.get(userGroupId, rev -> { + final Authorizable userGroupsAuthorizable = authorizableLookup.getUserGroupsAuthorizable(); + userGroupsAuthorizable.authorize(authorizer, RequestAction.READ); + + final RevisionDTO revision = dtoFactory.createRevisionDTO(rev); + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(userGroupsAuthorizable); + final Group userGroup = userGroupDAO.getUserGroup(userGroupId); + final Set<UserEntity> users = userGroup.getUsers().stream().map(userId -> prune ? getUserPruned(userId) : getUser(userId, false)).collect(Collectors.toSet()); + return entityFactory.createUserGroupEntity(dtoFactory.createUserGroupDto(userGroup, users), + revision, accessPolicy); + }); + } + + private UserGroupEntity getUserGroupPruned(final String userGroupId) { + return revisionManager.get(userGroupId, rev -> { + final Authorizable userGroupsAuthorizable = authorizableLookup.getUserGroupsAuthorizable(); + userGroupsAuthorizable.authorize(authorizer, RequestAction.READ); + + final RevisionDTO revision = dtoFactory.createRevisionDTO(rev); + final AccessPolicyDTO accessPolicy = dtoFactory.createAccessPolicyDto(userGroupsAuthorizable); + final Group userGroup = userGroupDAO.getUserGroup(userGroupId); + return entityFactory.createUserGroupEntity(dtoFactory.createUserGroupDto(userGroup, Collections.emptySet()), revision, accessPolicy); + }); + } + + @Override public Set<LabelEntity> getLabels(final String groupId) { final ProcessGroup group = processGroupDAO.getProcessGroup(groupId); group.authorize(authorizer, RequestAction.READ); @@ -2944,10 +3071,25 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { this.snippetUtils = snippetUtils; } + public void setAuthorizableLookup(final AuthorizableLookup authorizableLookup) { + this.authorizableLookup = authorizableLookup; + } + public void setAuthorizer(final Authorizer authorizer) { this.authorizer = authorizer; } + public void setUserDAO(final UserDAO userDAO) { + this.userDAO = userDAO; + } + + public void setUserGroupDAO(final UserGroupDAO userGroupDAO) { + this.userGroupDAO = userGroupDAO; + } + + public void setAccessPolicyDAO(final AccessPolicyDAO accessPolicyDAO) { + this.accessPolicyDAO = accessPolicyDAO; + } public void setClusterCoordinator(final ClusterCoordinator coordinator) { this.clusterCoordinator = coordinator; } @@ -2956,7 +3098,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade { this.heartbeatMonitor = heartbeatMonitor; } - public void setBulletinRepository(BulletinRepository bulletinRepository) { + public void setBulletinRepository(final BulletinRepository bulletinRepository) { this.bulletinRepository = bulletinRepository; } }
http://git-wip-us.apache.org/repos/asf/nifi/blob/f47be77b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/AccessPolicyResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/AccessPolicyResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/AccessPolicyResource.java new file mode 100644 index 0000000..a00cf18 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/AccessPolicyResource.java @@ -0,0 +1,440 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.web.api; + +import com.wordnik.swagger.annotations.Api; +import com.wordnik.swagger.annotations.ApiOperation; +import com.wordnik.swagger.annotations.ApiParam; +import com.wordnik.swagger.annotations.ApiResponse; +import com.wordnik.swagger.annotations.ApiResponses; +import com.wordnik.swagger.annotations.Authorization; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.authorization.Authorizer; +import org.apache.nifi.authorization.RequestAction; +import org.apache.nifi.authorization.resource.Authorizable; +import org.apache.nifi.cluster.coordination.ClusterCoordinator; +import org.apache.nifi.cluster.coordination.http.replication.RequestReplicator; +import org.apache.nifi.util.NiFiProperties; +import org.apache.nifi.web.NiFiServiceFacade; +import org.apache.nifi.web.Revision; +import org.apache.nifi.web.UpdateResult; +import org.apache.nifi.web.api.dto.AccessPolicyDTO; +import org.apache.nifi.web.api.dto.RevisionDTO; +import org.apache.nifi.web.api.entity.AccessPolicyEntity; +import org.apache.nifi.web.api.request.ClientIdParameter; +import org.apache.nifi.web.api.request.LongParameter; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.GET; +import javax.ws.rs.HttpMethod; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.net.URI; + +/** + * RESTful endpoint for managing access policies. + */ +@Path("/policies") +@Api( + value = "/policies", + description = "Endpoint for managing access policies." +) +public class AccessPolicyResource extends ApplicationResource { + + private final NiFiServiceFacade serviceFacade; + private final Authorizer authorizer; + + public AccessPolicyResource(NiFiServiceFacade serviceFacade, Authorizer authorizer, NiFiProperties properties, RequestReplicator requestReplicator, ClusterCoordinator clusterCoordinator) { + this.serviceFacade = serviceFacade; + this.authorizer = authorizer; + setProperties(properties); + setRequestReplicator(requestReplicator); + setClusterCoordinator(clusterCoordinator); + } + + /** + * Populates the uri for the specified access policy. + * + * @param accessPolicyEntity accessPolicyEntity + * @return accessPolicyEntity + */ + public AccessPolicyEntity populateRemainingAccessPolicyEntityContent(AccessPolicyEntity accessPolicyEntity) { + if (accessPolicyEntity.getComponent() != null) { + populateRemainingAccessPolicyContent(accessPolicyEntity.getComponent()); + } + return accessPolicyEntity; + } + + /** + * Populates the uri for the specified accessPolicy. + */ + public AccessPolicyDTO populateRemainingAccessPolicyContent(AccessPolicyDTO accessPolicy) { + // populate the access policy href + accessPolicy.setUri(generateResourceUri("policies", accessPolicy.getId())); + return accessPolicy; + } + + /** + * Creates a new access policy. + * + * @param httpServletRequest request + * @param accessPolicyEntity An accessPolicyEntity. + * @return An accessPolicyEntity. + */ + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Creates an access policy", + notes = " Available resources:\n" + + " /flow - READ - allows user/entity to load the UI and see the flow structure\n" + + " - WRITE - NA\n" + + " /resource - READ - allows user/entity to retrieve the available resources\n" + + " - WRITE - NA\n" + + " /system - READ - allows user/entity to retrieve system level diagnostics (CPU load, disk utilization, etc)\n" + + " - WRITE - NA\n" + + " /controller - READ - allows user/entity to retrieve configuration details for the controller (controller bulletins, thread pool, reporting tasks, etc)\n" + + " - WRITE - allows user/entity to modify configuration details for the controller\n" + + " /provenance - READ - allows user/entity to perform provenance requests. results will be filtered based on access to provenance data per component\n" + + " - WRITE - NA\n" + + " /token - READ - NA\n" + + " - WRITE - allows user/entity to create a token for access the REST API\n" + + " /site-to-site - READ - allows user/entity to retrieve configuration details for performing site to site data transfers with this NiFi\n" + + " - WRITE - NA\n" + + " /proxy - READ - NA\n" + + " - WRITE - allows user/entity to create a proxy request on behalf of another user\n" + + " /process-groups/{id} - READ - allows user/entity to retrieve configuration details for the process group and all descendant components without explicit " + + "access policies\n" + + " - WRITE - allows user/entity to create/update/delete configuration details for the process group and all descendant components without " + + "explicit access policies\n" + + " /processors/{id} - READ - allows user/entity to retrieve configuration details for the processor overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to update/delete the processor overriding any inherited authorizations from an ancestor process group\n" + + " /input-ports/{id} - READ - allows user/entity to retrieve configuration details for the input port overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to update/delete the input port overriding any inherited authorizations from an ancestor process group\n" + + " /output-ports/{id} - READ - allows user/entity to retrieve configuration details for the output port overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to update/delete the output port overriding any inherited authorizations from an ancestor process group\n" + + " /labels/{id} - READ - allows user/entity to retrieve configuration details for the label overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to update/delete the label overriding any inherited authorizations from an ancestor process group\n" + + " /connections/{id} - READ - allows user/entity to retrieve configuration details for the connection overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to update/delete the label overriding any inherited authorizations from an ancestor process group\n" + + " /remote-process-groups/{id} - READ - allows user/entity to retrieve configuration details for the remote process group overriding any inherited authorizations from an " + + "ancestor process group\n" + + " - WRITE - allows user/entity to update/delete the remote process group overriding any inherited authorizations from an ancestor process " + + "group\n" + + " /templates/{id} - READ - allows user/entity to retrieve configuration details for the template overriding any inherited authorizations from an ancestor " + + "process group\n" + + " - WRITE - allows user/entity to create/update/delete the template overriding any inherited authorizations from an ancestor process group\n" + + " /controller-services/{id} - READ - allows user/entity to retrieve configuration details for the controller service overriding any inherited authorizations from an " + + "ancestor process group\n" + + " - WRITE - allows user/entity to update/delete the controller service overriding any inherited authorizations from an ancestor process " + + "group\n" + + " /reporting-tasks/{id} - READ - allows user/entity to retrieve configuration details for the reporting tasks overriding any inherited authorizations from the " + + "controller\n" + + " - WRITE - allows user/entity to create/update/delete the reporting tasks overriding any inherited authorizations from the controller\n" + + " /{type}/{id}/provenance - READ - allows user/entity to view provenance data from the underlying component\n" + + " - WRITE - NA\n", + response = AccessPolicyEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response createAccessPolicy( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The access policy configuration details.", + required = true + ) final AccessPolicyEntity accessPolicyEntity) { + + if (accessPolicyEntity == null || accessPolicyEntity.getComponent() == null) { + throw new IllegalArgumentException("Access policy details must be specified."); + } + + if (accessPolicyEntity.getComponent().getId() != null) { + throw new IllegalArgumentException("Access policy ID cannot be specified."); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.POST, accessPolicyEntity); + } + + // handle expects request (usually from the cluster manager) + final boolean validationPhase = isValidationPhase(httpServletRequest); + if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable accessPolicies = lookup.getAccessPoliciesAuthorizable(); + accessPolicies.authorize(authorizer, RequestAction.WRITE); + }); + } + if (validationPhase) { + return generateContinueResponse().build(); + } + + // set the access policy id as appropriate + accessPolicyEntity.getComponent().setId(generateUuid()); + + // get revision from the config + final RevisionDTO revisionDTO = accessPolicyEntity.getRevision(); + Revision revision = new Revision(revisionDTO.getVersion(), revisionDTO.getClientId(), accessPolicyEntity.getComponent().getId()); + + // create the access policy and generate the json + final AccessPolicyEntity entity = serviceFacade.createAccessPolicy(revision, accessPolicyEntity.getComponent()); + populateRemainingAccessPolicyEntityContent(entity); + + // build the response + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } + + /** + * Retrieves the specified access policy. + * + * @param id The id of the access policy to retrieve + * @return An accessPolicyEntity. + */ + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')") + @ApiOperation( + value = "Gets an access policy", + response = AccessPolicyEntity.class, + authorizations = { + @Authorization(value = "Read Only", type = "ROLE_MONITOR"), + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM"), + @Authorization(value = "Administrator", type = "ROLE_ADMIN") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response getAccessPolicy( + @ApiParam( + value = "The access policy id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable accessPolicy = lookup.getAccessPolicyAuthorizable(id); + accessPolicy.authorize(authorizer, RequestAction.READ); + }); + + // get the access policy + final AccessPolicyEntity entity = serviceFacade.getAccessPolicy(id); + populateRemainingAccessPolicyEntityContent(entity); + + return clusterContext(generateOkResponse(entity)).build(); + } + + /** + * Updates an access policy. + * + * @param httpServletRequest request + * @param id The id of the access policy to update. + * @param accessPolicyEntity An accessPolicyEntity. + * @return An accessPolicyEntity. + */ + @PUT + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Updates a access policy", + response = AccessPolicyEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response updateAccessPolicy( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The access policy id.", + required = true + ) + @PathParam("id") final String id, + @ApiParam( + value = "The access policy configuration details.", + required = true + ) final AccessPolicyEntity accessPolicyEntity) { + + if (accessPolicyEntity == null || accessPolicyEntity.getComponent() == null) { + throw new IllegalArgumentException("Access policy details must be specified."); + } + + if (accessPolicyEntity.getRevision() == null) { + throw new IllegalArgumentException("Revision must be specified."); + } + + // ensure the ids are the same + final AccessPolicyDTO accessPolicyDTO = accessPolicyEntity.getComponent(); + if (!id.equals(accessPolicyDTO.getId())) { + throw new IllegalArgumentException(String.format("The access policy id (%s) in the request body does not equal the " + + "access policy id of the requested resource (%s).", accessPolicyDTO.getId(), id)); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.PUT, accessPolicyEntity); + } + + // Extract the revision + final Revision revision = getRevision(accessPolicyEntity, id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable accessPolicy = lookup.getAccessPolicyAuthorizable(id); + accessPolicy.authorize(authorizer, RequestAction.WRITE); + }, + null, + () -> { + // update the access policy + final UpdateResult<AccessPolicyEntity> updateResult = serviceFacade.updateAccessPolicy(revision, accessPolicyDTO); + + // get the results + final AccessPolicyEntity entity = updateResult.getResult(); + populateRemainingAccessPolicyEntityContent(entity); + + if (updateResult.isNew()) { + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } else { + return clusterContext(generateOkResponse(entity)).build(); + } + } + ); + } + + /** + * Removes the specified access policy. + * + * @param httpServletRequest request + * @param version The revision is used to verify the client is working with + * the latest version of the flow. + * @param clientId Optional client id. If the client id is not specified, a + * new one will be generated. This value (whether specified or generated) is + * included in the response. + * @param id The id of the access policy to remove. + * @return A entity containing the client id and an updated revision. + */ + @DELETE + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Deletes an access policy", + response = AccessPolicyEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response removeAccessPolicy( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The revision is used to verify the client is working with the latest version of the flow.", + required = false + ) + @QueryParam(VERSION) final LongParameter version, + @ApiParam( + value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.", + required = false + ) + @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) final ClientIdParameter clientId, + @ApiParam( + value = "The access policy id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.DELETE); + } + + // handle expects request (usually from the cluster manager) + final Revision revision = new Revision(version == null ? null : version.getLong(), clientId.getClientId(), id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable accessPolicy = lookup.getAccessPolicyAuthorizable(id); + accessPolicy.authorize(authorizer, RequestAction.READ); + }, + () -> { + }, + () -> { + // delete the specified access policy + final AccessPolicyEntity entity = serviceFacade.deleteAccessPolicy(revision, id); + return clusterContext(generateOkResponse(entity)).build(); + } + ); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/f47be77b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UserGroupsResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UserGroupsResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UserGroupsResource.java new file mode 100644 index 0000000..553f8c6 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UserGroupsResource.java @@ -0,0 +1,385 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.web.api; + +import com.wordnik.swagger.annotations.Api; +import com.wordnik.swagger.annotations.ApiOperation; +import com.wordnik.swagger.annotations.ApiParam; +import com.wordnik.swagger.annotations.ApiResponse; +import com.wordnik.swagger.annotations.ApiResponses; +import com.wordnik.swagger.annotations.Authorization; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.authorization.Authorizer; +import org.apache.nifi.authorization.RequestAction; +import org.apache.nifi.authorization.resource.Authorizable; +import org.apache.nifi.cluster.coordination.ClusterCoordinator; +import org.apache.nifi.cluster.coordination.http.replication.RequestReplicator; +import org.apache.nifi.util.NiFiProperties; +import org.apache.nifi.web.NiFiServiceFacade; +import org.apache.nifi.web.Revision; +import org.apache.nifi.web.UpdateResult; +import org.apache.nifi.web.api.dto.RevisionDTO; +import org.apache.nifi.web.api.dto.UserGroupDTO; +import org.apache.nifi.web.api.entity.UserGroupEntity; +import org.apache.nifi.web.api.request.ClientIdParameter; +import org.apache.nifi.web.api.request.LongParameter; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.GET; +import javax.ws.rs.HttpMethod; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.net.URI; + +@Path("/user-groups") +@Api( + value = "/user-groups", + description = "Endpoint for managing user groups." +) +public class UserGroupsResource extends ApplicationResource { + + private final NiFiServiceFacade serviceFacade; + private final Authorizer authorizer; + + public UserGroupsResource(NiFiServiceFacade serviceFacade, Authorizer authorizer, NiFiProperties properties, RequestReplicator requestReplicator, ClusterCoordinator clusterCoordinator) { + this.serviceFacade = serviceFacade; + this.authorizer = authorizer; + setProperties(properties); + setRequestReplicator(requestReplicator); + setClusterCoordinator(clusterCoordinator); + } + + /** + * Populates the uri for the specified user group. + * + * @param userGroupEntity userGroupEntity + * @return userGroupEntity + */ + public UserGroupEntity populateRemainingUserGroupEntityContent(UserGroupEntity userGroupEntity) { + if (userGroupEntity.getComponent() != null) { + populateRemainingUserGroupContent(userGroupEntity.getComponent()); + } + return userGroupEntity; + } + + /** + * Populates the uri for the specified userGroup. + */ + public UserGroupDTO populateRemainingUserGroupContent(UserGroupDTO userGroup) { + // populate the user group href + userGroup.setUri(generateResourceUri("user-groups", userGroup.getId())); + return userGroup; + } + + /** + * Creates a new user group. + * + * @param httpServletRequest request + * @param userGroupEntity An userGroupEntity. + * @return An userGroupEntity. + */ + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Creates a user group", + response = UserGroupEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response createUserGroup( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The user group configuration details.", + required = true + ) final UserGroupEntity userGroupEntity) { + + if (userGroupEntity == null || userGroupEntity.getComponent() == null) { + throw new IllegalArgumentException("User group details must be specified."); + } + + if (userGroupEntity.getComponent().getId() != null) { + throw new IllegalArgumentException("User group ID cannot be specified."); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.POST, userGroupEntity); + } + + // handle expects request (usually from the cluster manager) + final boolean validationPhase = isValidationPhase(httpServletRequest); + if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable userGroups = lookup.getUserGroupsAuthorizable(); + userGroups.authorize(authorizer, RequestAction.WRITE); + }); + } + if (validationPhase) { + return generateContinueResponse().build(); + } + + // set the user group id as appropriate + userGroupEntity.getComponent().setId(generateUuid()); + + // get revision from the config + final RevisionDTO revisionDTO = userGroupEntity.getRevision(); + Revision revision = new Revision(revisionDTO.getVersion(), revisionDTO.getClientId(), userGroupEntity.getComponent().getId()); + + // create the user group and generate the json + final UserGroupEntity entity = serviceFacade.createUserGroup(revision, userGroupEntity.getComponent()); + populateRemainingUserGroupEntityContent(entity); + + // build the response + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } + + /** + * Retrieves the specified user group. + * + * @param id The id of the user group to retrieve + * @return An userGroupEntity. + */ + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')") + @ApiOperation( + value = "Gets a user group", + response = UserGroupEntity.class, + authorizations = { + @Authorization(value = "Read Only", type = "ROLE_MONITOR"), + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM"), + @Authorization(value = "Administrator", type = "ROLE_ADMIN") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response getUserGroup( + @ApiParam( + value = "The user group id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable userGroups = lookup.getUserGroupsAuthorizable(); + userGroups.authorize(authorizer, RequestAction.READ); + }); + + // get the user group + final UserGroupEntity entity = serviceFacade.getUserGroup(id, true); + populateRemainingUserGroupEntityContent(entity); + + return clusterContext(generateOkResponse(entity)).build(); + } + + /** + * Updates a user group. + * + * @param httpServletRequest request + * @param id The id of the user group to update. + * @param userGroupEntity An userGroupEntity. + * @return An userGroupEntity. + */ + @PUT + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Updates a user group", + response = UserGroupEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response updateUserGroup( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The user group id.", + required = true + ) + @PathParam("id") final String id, + @ApiParam( + value = "The user group configuration details.", + required = true + ) final UserGroupEntity userGroupEntity) { + + if (userGroupEntity == null || userGroupEntity.getComponent() == null) { + throw new IllegalArgumentException("User group details must be specified."); + } + + if (userGroupEntity.getRevision() == null) { + throw new IllegalArgumentException("Revision must be specified."); + } + + // ensure the ids are the same + final UserGroupDTO userGroupDTO = userGroupEntity.getComponent(); + if (!id.equals(userGroupDTO.getId())) { + throw new IllegalArgumentException(String.format("The user group id (%s) in the request body does not equal the " + + "user group id of the requested resource (%s).", userGroupDTO.getId(), id)); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.PUT, userGroupEntity); + } + + // Extract the revision + final Revision revision = getRevision(userGroupEntity, id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable userGroups = lookup.getUserGroupsAuthorizable(); + userGroups.authorize(authorizer, RequestAction.WRITE); + }, + null, + () -> { + // update the user group + final UpdateResult<UserGroupEntity> updateResult = serviceFacade.updateUserGroup(revision, userGroupDTO); + + // get the results + final UserGroupEntity entity = updateResult.getResult(); + populateRemainingUserGroupEntityContent(entity); + + if (updateResult.isNew()) { + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } else { + return clusterContext(generateOkResponse(entity)).build(); + } + } + ); + } + + /** + * Removes the specified user group. + * + * @param httpServletRequest request + * @param version The revision is used to verify the client is working with + * the latest version of the flow. + * @param clientId Optional client id. If the client id is not specified, a + * new one will be generated. This value (whether specified or generated) is + * included in the response. + * @param id The id of the user group to remove. + * @return A entity containing the client id and an updated revision. + */ + @DELETE + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Deletes a user group", + response = UserGroupEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response removeUserGroup( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The revision is used to verify the client is working with the latest version of the flow.", + required = false + ) + @QueryParam(VERSION) final LongParameter version, + @ApiParam( + value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.", + required = false + ) + @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) final ClientIdParameter clientId, + @ApiParam( + value = "The user group id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.DELETE); + } + + // handle expects request (usually from the cluster manager) + final Revision revision = new Revision(version == null ? null : version.getLong(), clientId.getClientId(), id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable userGroups = lookup.getUserGroupsAuthorizable(); + userGroups.authorize(authorizer, RequestAction.READ); + }, + () -> { + }, + () -> { + // delete the specified user group + final UserGroupEntity entity = serviceFacade.deleteUserGroup(revision, id); + return clusterContext(generateOkResponse(entity)).build(); + } + ); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/f47be77b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UsersResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UsersResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UsersResource.java new file mode 100644 index 0000000..01deb93 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/UsersResource.java @@ -0,0 +1,385 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.web.api; + +import com.wordnik.swagger.annotations.Api; +import com.wordnik.swagger.annotations.ApiOperation; +import com.wordnik.swagger.annotations.ApiParam; +import com.wordnik.swagger.annotations.ApiResponse; +import com.wordnik.swagger.annotations.ApiResponses; +import com.wordnik.swagger.annotations.Authorization; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.authorization.Authorizer; +import org.apache.nifi.authorization.RequestAction; +import org.apache.nifi.authorization.resource.Authorizable; +import org.apache.nifi.cluster.coordination.ClusterCoordinator; +import org.apache.nifi.cluster.coordination.http.replication.RequestReplicator; +import org.apache.nifi.util.NiFiProperties; +import org.apache.nifi.web.NiFiServiceFacade; +import org.apache.nifi.web.Revision; +import org.apache.nifi.web.UpdateResult; +import org.apache.nifi.web.api.dto.RevisionDTO; +import org.apache.nifi.web.api.dto.UserDTO; +import org.apache.nifi.web.api.entity.UserEntity; +import org.apache.nifi.web.api.request.ClientIdParameter; +import org.apache.nifi.web.api.request.LongParameter; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.GET; +import javax.ws.rs.HttpMethod; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.net.URI; + +@Path("/users") +@Api( + value = "/users", + description = "Endpoint for managing users." +) +public class UsersResource extends ApplicationResource { + + private final NiFiServiceFacade serviceFacade; + private final Authorizer authorizer; + + public UsersResource(NiFiServiceFacade serviceFacade, Authorizer authorizer, NiFiProperties properties, RequestReplicator requestReplicator, ClusterCoordinator clusterCoordinator) { + this.serviceFacade = serviceFacade; + this.authorizer = authorizer; + setProperties(properties); + setRequestReplicator(requestReplicator); + setClusterCoordinator(clusterCoordinator); + } + + /** + * Populates the uri for the specified user. + * + * @param userEntity userEntity + * @return userEntity + */ + public UserEntity populateRemainingUserEntityContent(UserEntity userEntity) { + if (userEntity.getComponent() != null) { + populateRemainingUserContent(userEntity.getComponent()); + } + return userEntity; + } + + /** + * Populates the uri for the specified user. + */ + public UserDTO populateRemainingUserContent(UserDTO user) { + // populate the user href + user.setUri(generateResourceUri("users", user.getId())); + return user; + } + + /** + * Creates a new user. + * + * @param httpServletRequest request + * @param userEntity An userEntity. + * @return An userEntity. + */ + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Creates a user", + response = UserEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response createUser( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The user configuration details.", + required = true + ) final UserEntity userEntity) { + + if (userEntity == null || userEntity.getComponent() == null) { + throw new IllegalArgumentException("User details must be specified."); + } + + if (userEntity.getComponent().getId() != null) { + throw new IllegalArgumentException("User ID cannot be specified."); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.POST, userEntity); + } + + // handle expects request (usually from the cluster manager) + final boolean validationPhase = isValidationPhase(httpServletRequest); + if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable users = lookup.getUsersAuthorizable(); + users.authorize(authorizer, RequestAction.WRITE); + }); + } + if (validationPhase) { + return generateContinueResponse().build(); + } + + // set the user id as appropriate + userEntity.getComponent().setId(generateUuid()); + + // get revision from the config + final RevisionDTO revisionDTO = userEntity.getRevision(); + Revision revision = new Revision(revisionDTO.getVersion(), revisionDTO.getClientId(), userEntity.getComponent().getId()); + + // create the user and generate the json + final UserEntity entity = serviceFacade.createUser(revision, userEntity.getComponent()); + populateRemainingUserEntityContent(entity); + + // build the response + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } + + /** + * Retrieves the specified user. + * + * @param id The id of the user to retrieve + * @return An userEntity. + */ + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')") + @ApiOperation( + value = "Gets a user", + response = UserEntity.class, + authorizations = { + @Authorization(value = "Read Only", type = "ROLE_MONITOR"), + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM"), + @Authorization(value = "Administrator", type = "ROLE_ADMIN") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response getUser( + @ApiParam( + value = "The user id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable users = lookup.getUsersAuthorizable(); + users.authorize(authorizer, RequestAction.READ); + }); + + // get the user + final UserEntity entity = serviceFacade.getUser(id, true); + populateRemainingUserEntityContent(entity); + + return clusterContext(generateOkResponse(entity)).build(); + } + + /** + * Updates a user. + * + * @param httpServletRequest request + * @param id The id of the user to update. + * @param userEntity An userEntity. + * @return An userEntity. + */ + @PUT + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Updates a user", + response = UserEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response updateUser( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The user id.", + required = true + ) + @PathParam("id") final String id, + @ApiParam( + value = "The user configuration details.", + required = true + ) final UserEntity userEntity) { + + if (userEntity == null || userEntity.getComponent() == null) { + throw new IllegalArgumentException("User details must be specified."); + } + + if (userEntity.getRevision() == null) { + throw new IllegalArgumentException("Revision must be specified."); + } + + // ensure the ids are the same + final UserDTO userDTO = userEntity.getComponent(); + if (!id.equals(userDTO.getId())) { + throw new IllegalArgumentException(String.format("The user id (%s) in the request body does not equal the " + + "user id of the requested resource (%s).", userDTO.getId(), id)); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.PUT, userEntity); + } + + // Extract the revision + final Revision revision = getRevision(userEntity, id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable users = lookup.getUsersAuthorizable(); + users.authorize(authorizer, RequestAction.WRITE); + }, + null, + () -> { + // update the user + final UpdateResult<UserEntity> updateResult = serviceFacade.updateUser(revision, userDTO); + + // get the results + final UserEntity entity = updateResult.getResult(); + populateRemainingUserEntityContent(entity); + + if (updateResult.isNew()) { + return clusterContext(generateCreatedResponse(URI.create(entity.getComponent().getUri()), entity)).build(); + } else { + return clusterContext(generateOkResponse(entity)).build(); + } + } + ); + } + + /** + * Removes the specified user. + * + * @param httpServletRequest request + * @param version The revision is used to verify the client is working with + * the latest version of the flow. + * @param clientId Optional client id. If the client id is not specified, a + * new one will be generated. This value (whether specified or generated) is + * included in the response. + * @param id The id of the user to remove. + * @return A entity containing the client id and an updated revision. + */ + @DELETE + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + // TODO - @PreAuthorize("hasRole('ROLE_DFM')") + @ApiOperation( + value = "Deletes a user", + response = UserEntity.class, + authorizations = { + @Authorization(value = "Data Flow Manager", type = "ROLE_DFM") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response removeUser( + @Context final HttpServletRequest httpServletRequest, + @ApiParam( + value = "The revision is used to verify the client is working with the latest version of the flow.", + required = false + ) + @QueryParam(VERSION) final LongParameter version, + @ApiParam( + value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.", + required = false + ) + @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) final ClientIdParameter clientId, + @ApiParam( + value = "The user id.", + required = true + ) + @PathParam("id") final String id) { + + if (isReplicateRequest()) { + return replicate(HttpMethod.DELETE); + } + + // handle expects request (usually from the cluster manager) + final Revision revision = new Revision(version == null ? null : version.getLong(), clientId.getClientId(), id); + return withWriteLock( + serviceFacade, + revision, + lookup -> { + final Authorizable users = lookup.getUsersAuthorizable(); + users.authorize(authorizer, RequestAction.READ); + }, + () -> { + }, + () -> { + // delete the specified user + final UserEntity entity = serviceFacade.deleteUser(revision, id); + return clusterContext(generateOkResponse(entity)).build(); + } + ); + } +}
