http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserAuthoritiesCacheAction.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserAuthoritiesCacheAction.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserAuthoritiesCacheAction.java new file mode 100644 index 0000000..89661b2 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserAuthoritiesCacheAction.java @@ -0,0 +1,73 @@ +/* + * 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.admin.service.action; + +import java.util.Set; +import org.apache.nifi.admin.dao.AuthorityDAO; +import org.apache.nifi.admin.dao.DAOFactory; +import org.apache.nifi.admin.dao.DataAccessException; +import org.apache.nifi.admin.dao.UserDAO; +import org.apache.nifi.admin.service.AccountNotFoundException; +import org.apache.nifi.authorization.Authority; +import org.apache.nifi.authorization.AuthorityProvider; +import org.apache.nifi.user.NiFiUser; +import org.apache.commons.collections4.CollectionUtils; + +/** + * Updates a NiFiUser's authorities. Prior to invoking this action, the user's + * authorities should be set according to the business logic of the service in + * question. This should not be invoked directly when attempting to set user + * authorities as the authorityProvider is not called from this action. + */ +public class UpdateUserAuthoritiesCacheAction extends AbstractUserAction<Void> { + + private final NiFiUser user; + + public UpdateUserAuthoritiesCacheAction(NiFiUser user) { + this.user = user; + } + + @Override + public Void execute(DAOFactory daoFactory, AuthorityProvider authorityProvider) throws DataAccessException { + UserDAO userDao = daoFactory.getUserDAO(); + AuthorityDAO authorityDao = daoFactory.getAuthorityDAO(); + + // get the user + NiFiUser currentUser = userDao.findUserById(user.getId()); + + // ensure the user exists + if (currentUser == null) { + throw new AccountNotFoundException(String.format("Unable to find account with ID %s.", user.getId())); + } + + // determine what authorities need to be added/removed + Set<Authority> authorities = user.getAuthorities(); + Set<Authority> authoritiesToAdd = determineAuthoritiesToAdd(currentUser, authorities); + Set<Authority> authoritiesToRemove = determineAuthoritiesToRemove(currentUser, authorities); + + // update the user authorities locally + if (CollectionUtils.isNotEmpty(authoritiesToAdd)) { + authorityDao.createAuthorities(authoritiesToAdd, user.getId()); + } + if (CollectionUtils.isNotEmpty(authoritiesToRemove)) { + authorityDao.deleteAuthorities(authoritiesToRemove, user.getId()); + } + + return null; + } + +}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserCacheAction.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserCacheAction.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserCacheAction.java new file mode 100644 index 0000000..288e297 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserCacheAction.java @@ -0,0 +1,47 @@ +/* + * 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.admin.service.action; + +import org.apache.nifi.admin.dao.DAOFactory; +import org.apache.nifi.admin.dao.DataAccessException; +import org.apache.nifi.admin.dao.UserDAO; +import org.apache.nifi.authorization.AuthorityProvider; +import org.apache.nifi.user.NiFiUser; + +/** + * Updates a NiFiUser. This will not update the user authorities, they must be + * updated with the UpdateUserAuthoritiesAction. + */ +public class UpdateUserCacheAction extends AbstractUserAction<Void> { + + private final NiFiUser user; + + public UpdateUserCacheAction(NiFiUser user) { + this.user = user; + } + + @Override + public Void execute(DAOFactory daoFactory, AuthorityProvider authorityProvider) throws DataAccessException { + UserDAO userDao = daoFactory.getUserDAO(); + + // update the user + userDao.updateUser(user); + + return null; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserGroupAction.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserGroupAction.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserGroupAction.java new file mode 100644 index 0000000..56b214c --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/action/UpdateUserGroupAction.java @@ -0,0 +1,171 @@ +/* + * 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.admin.service.action; + +import java.util.Date; +import java.util.HashSet; +import java.util.Set; +import org.apache.nifi.admin.dao.DAOFactory; +import org.apache.nifi.admin.dao.DataAccessException; +import org.apache.nifi.admin.dao.UserDAO; +import org.apache.nifi.admin.service.AccountNotFoundException; +import org.apache.nifi.admin.service.AdministrationException; +import org.apache.nifi.authorization.Authority; +import org.apache.nifi.authorization.AuthorityProvider; +import org.apache.nifi.authorization.exception.AuthorityAccessException; +import org.apache.nifi.authorization.exception.UnknownIdentityException; +import org.apache.nifi.user.AccountStatus; +import org.apache.nifi.user.NiFiUser; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Updates all NiFiUser authorities in a specified group. + */ +public class UpdateUserGroupAction extends AbstractUserAction<Void> { + + private static final Logger logger = LoggerFactory.getLogger(UpdateUserGroupAction.class); + + private final String group; + private final Set<String> userIds; + private final Set<Authority> authorities; + + public UpdateUserGroupAction(String group, Set<String> userIds, Set<Authority> authorities) { + this.group = group; + this.userIds = userIds; + this.authorities = authorities; + } + + @Override + public Void execute(DAOFactory daoFactory, AuthorityProvider authorityProvider) throws DataAccessException { + if (userIds == null && authorities == null) { + throw new IllegalArgumentException("Must specify user Ids or authorities."); + } + + UserDAO userDao = daoFactory.getUserDAO(); + + // record the new users being added to this group + final Set<NiFiUser> newUsers = new HashSet<>(); + final Set<String> newUserDns = new HashSet<>(); + + // if the user ids have been specified we need to create/update a group using the specified group name + if (userIds != null) { + if (userIds.isEmpty()) { + throw new IllegalArgumentException("When creating a group, at least one user id must be specified."); + } + + // going to create a group using the specified user ids + for (final String userId : userIds) { + // get the user in question + final NiFiUser user = userDao.findUserById(userId); + + // ensure the user exists + if (user == null) { + throw new AccountNotFoundException(String.format("Unable to find account with ID %s.", userId)); + } + + try { + // if the user is unknown to the authority provider we cannot continue + if (!authorityProvider.doesDnExist(user.getDn()) || AccountStatus.DISABLED.equals(user.getStatus())) { + throw new IllegalStateException(String.format("Unable to group these users because access for '%s' is not %s.", user.getDn(), AccountStatus.ACTIVE.toString())); + } + + // record the user being added to this group + newUsers.add(user); + newUserDns.add(user.getDn()); + } catch (final AuthorityAccessException aae) { + throw new AdministrationException(String.format("Unable to access authority details: %s", aae.getMessage()), aae); + } + } + + try { + // update the authority provider + authorityProvider.setUsersGroup(newUserDns, group); + } catch (UnknownIdentityException uie) { + throw new AccountNotFoundException(String.format("Unable to set user group '%s': %s", StringUtils.join(newUserDns, ", "), uie.getMessage()), uie); + } catch (AuthorityAccessException aae) { + throw new AdministrationException(String.format("Unable to set user group '%s': %s", StringUtils.join(newUserDns, ", "), aae.getMessage()), aae); + } + } + + // get all the users that need to be updated + final Set<NiFiUser> users = new HashSet<>(userDao.findUsersForGroup(group)); + users.addAll(newUsers); + + // ensure the user exists + if (users.isEmpty()) { + throw new AccountNotFoundException(String.format("Unable to find user accounts with group id %s.", group)); + } + + // update each user in this group + for (final NiFiUser user : users) { + // if there are new authorities set them, otherwise refresh them according to the provider + if (authorities != null) { + try { + // update the authority provider as approprivate + authorityProvider.setAuthorities(user.getDn(), authorities); + + // since all the authorities were updated accordingly, set the authorities + user.getAuthorities().clear(); + user.getAuthorities().addAll(authorities); + } catch (UnknownIdentityException uie) { + throw new AccountNotFoundException(String.format("Unable to modify authorities for '%s': %s.", user.getDn(), uie.getMessage()), uie); + } catch (AuthorityAccessException aae) { + throw new AdministrationException(String.format("Unable to access authorities for '%s': %s.", user.getDn(), aae.getMessage()), aae); + } + } else { + try { + // refresh the authorities according to the provider + user.getAuthorities().clear(); + user.getAuthorities().addAll(authorityProvider.getAuthorities(user.getDn())); + } catch (UnknownIdentityException uie) { + throw new AccountNotFoundException(String.format("Unable to determine the authorities for '%s': %s.", user.getDn(), uie.getMessage()), uie); + } catch (AuthorityAccessException aae) { + throw new AdministrationException(String.format("Unable to access authorities for '%s': %s.", user.getDn(), aae.getMessage()), aae); + } + } + + try { + // get the user group + user.setUserGroup(authorityProvider.getGroupForUser(user.getDn())); + } catch (UnknownIdentityException uie) { + throw new AccountNotFoundException(String.format("Unable to determine the group for '%s': %s.", user.getDn(), uie.getMessage()), uie); + } catch (AuthorityAccessException aae) { + throw new AdministrationException(String.format("Unable to access the group for '%s': %s.", user.getDn(), aae.getMessage()), aae); + } + + // update the users status in case they were previously pending or disabled + user.setStatus(AccountStatus.ACTIVE); + + // update the users last verified time - this timestamp shouldn't be recorded + // until the both the user's authorities and group have been synced + Date now = new Date(); + user.setLastVerified(now); + + // persist the user's updates + UpdateUserCacheAction updateUser = new UpdateUserCacheAction(user); + updateUser.execute(daoFactory, authorityProvider); + + // persist the user's authorities + UpdateUserAuthoritiesCacheAction updateUserAuthorities = new UpdateUserAuthoritiesCacheAction(user); + updateUserAuthorities.execute(daoFactory, authorityProvider); + } + + return null; + } +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardAuditService.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardAuditService.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardAuditService.java new file mode 100644 index 0000000..127f1df --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardAuditService.java @@ -0,0 +1,230 @@ +/* + * 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.admin.service.impl; + +import java.io.IOException; +import java.util.Collection; +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import org.apache.nifi.action.Action; +import org.apache.nifi.admin.dao.DataAccessException; +import org.apache.nifi.admin.service.AdministrationException; +import org.apache.nifi.admin.service.AuditService; +import org.apache.nifi.admin.service.action.AddActionsAction; +import org.apache.nifi.admin.service.action.GetActionAction; +import org.apache.nifi.admin.service.action.GetActionsAction; +import org.apache.nifi.admin.service.action.GetPreviousValues; +import org.apache.nifi.admin.service.action.PurgeActionsAction; +import org.apache.nifi.admin.service.transaction.Transaction; +import org.apache.nifi.admin.service.transaction.TransactionBuilder; +import org.apache.nifi.admin.service.transaction.TransactionException; +import org.apache.nifi.history.History; +import org.apache.nifi.history.HistoryQuery; +import org.apache.nifi.history.PreviousValue; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + */ +public class StandardAuditService implements AuditService { + + private static final Logger logger = LoggerFactory.getLogger(StandardAuditService.class); + + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock(); + private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock(); + + private TransactionBuilder transactionBuilder; + + @Override + public void addActions(Collection<Action> actions) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // seed the accounts + AddActionsAction addActions = new AddActionsAction(actions); + transaction.execute(addActions); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public Map<String, List<PreviousValue>> getPreviousValues(String processorId) { + Transaction transaction = null; + Map<String, List<PreviousValue>> previousValues = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // seed the accounts + GetPreviousValues getActions = new GetPreviousValues(processorId); + previousValues = transaction.execute(getActions); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + + return previousValues; + } + + @Override + public History getActions(HistoryQuery query) { + Transaction transaction = null; + History history = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // seed the accounts + GetActionsAction getActions = new GetActionsAction(query); + history = transaction.execute(getActions); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + + return history; + } + + @Override + public Action getAction(Integer actionId) { + Transaction transaction = null; + Action action = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // seed the accounts + GetActionAction getAction = new GetActionAction(actionId); + action = transaction.execute(getAction); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + + return action; + } + + @Override + public void purgeActions(Date end, Action purgeAction) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // purge the action database + PurgeActionsAction purgeActions = new PurgeActionsAction(end, purgeAction); + transaction.execute(purgeActions); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + /** + * Rolls back the specified transaction. + * + * @param transaction + */ + private void rollback(Transaction transaction) { + if (transaction != null) { + transaction.rollback(); + } + } + + /** + * Closes the specified transaction. + * + * @param transaction + */ + private void closeQuietly(final Transaction transaction) { + if (transaction != null) { + try { + transaction.close(); + } catch (final IOException ioe) { + } + } + } + + /* setters */ + public void setTransactionBuilder(TransactionBuilder transactionBuilder) { + this.transactionBuilder = transactionBuilder; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardUserService.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardUserService.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardUserService.java new file mode 100644 index 0000000..63aa93b --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/impl/StandardUserService.java @@ -0,0 +1,629 @@ +/* + * 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.admin.service.impl; + +import java.io.IOException; +import java.util.Collection; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +import org.apache.nifi.admin.dao.DataAccessException; +import org.apache.nifi.admin.service.AccountDisabledException; +import org.apache.nifi.admin.service.AccountPendingException; +import org.apache.nifi.admin.service.AdministrationException; +import org.apache.nifi.admin.service.UserService; +import org.apache.nifi.admin.service.action.AuthorizeUserAction; +import org.apache.nifi.admin.service.action.DeleteUserAction; +import org.apache.nifi.admin.service.action.DisableUserAction; +import org.apache.nifi.admin.service.action.DisableUserGroupAction; +import org.apache.nifi.admin.service.action.FindUserByDnAction; +import org.apache.nifi.admin.service.action.FindUserByIdAction; +import org.apache.nifi.admin.service.action.GetUserGroupAction; +import org.apache.nifi.admin.service.action.GetUsersAction; +import org.apache.nifi.admin.service.action.HasPendingUserAccounts; +import org.apache.nifi.admin.service.action.InvalidateUserAccountAction; +import org.apache.nifi.admin.service.action.InvalidateUserGroupAccountsAction; +import org.apache.nifi.admin.service.action.RequestUserAccountAction; +import org.apache.nifi.admin.service.action.SeedUserAccountsAction; +import org.apache.nifi.admin.service.action.UpdateUserAction; +import org.apache.nifi.admin.service.action.UpdateUserGroupAction; +import org.apache.nifi.admin.service.action.UngroupUserAction; +import org.apache.nifi.admin.service.action.UngroupUserGroupAction; +import org.apache.nifi.admin.service.transaction.Transaction; +import org.apache.nifi.admin.service.transaction.TransactionBuilder; +import org.apache.nifi.admin.service.transaction.TransactionException; +import org.apache.nifi.authorization.Authority; +import org.apache.nifi.user.NiFiUser; +import org.apache.nifi.user.NiFiUserGroup; +import org.apache.nifi.util.FormatUtils; +import org.apache.nifi.util.NiFiProperties; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + */ +public class StandardUserService implements UserService { + + private static final Logger logger = LoggerFactory.getLogger(StandardUserService.class); + + private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); + private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock(); + private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock(); + + private TransactionBuilder transactionBuilder; + private NiFiProperties properties; + + /** + * Seed any users from the authority provider that are not already present. + */ + public void seedUserAccounts() { + // do not seed node's user cache. when/if the node disconnects its + // cache will be populated lazily (as needed) + if (properties.isNode()) { + return; + } + + Transaction transaction = null; + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // seed the accounts + SeedUserAccountsAction seedUserAccounts = new SeedUserAccountsAction(); + transaction.execute(seedUserAccounts); + + // commit the transaction + transaction.commit(); + } catch (AdministrationException ae) { + rollback(transaction); + throw ae; + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUser createPendingUserAccount(String dn, String justification) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // create the account request + RequestUserAccountAction requestUserAccount = new RequestUserAccountAction(dn, justification); + NiFiUser user = transaction.execute(requestUserAccount); + + // commit the transaction + transaction.commit(); + + // return the nifi user + return user; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUserGroup updateGroup(final String group, final Set<String> userIds, final Set<Authority> authorities) { + Transaction transaction = null; + + writeLock.lock(); + try { + // if user ids have been specified, invalidate the user accounts before performing + // the desired updates. if case of an error, this will ensure that these users are + // authorized the next time the access the application + if (userIds != null) { + for (final String userId : userIds) { + invalidateUserAccount(userId); + } + } + + // start the transaction + transaction = transactionBuilder.start(); + + // set the authorities for each user in this group if specified + final UpdateUserGroupAction updateUserGroup = new UpdateUserGroupAction(group, userIds, authorities); + transaction.execute(updateUserGroup); + + // get all the users that are now in this group + final GetUserGroupAction getUserGroup = new GetUserGroupAction(group); + final NiFiUserGroup userGroup = transaction.execute(getUserGroup); + + // commit the transaction + transaction.commit(); + + return userGroup; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public void ungroupUser(String id) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // ungroup the specified user + final UngroupUserAction ungroupUser = new UngroupUserAction(id); + transaction.execute(ungroupUser); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public void ungroup(String group) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // ungroup the specified user + final UngroupUserGroupAction ungroupUserGroup = new UngroupUserGroupAction(group); + transaction.execute(ungroupUserGroup); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUser checkAuthorization(String dn) { + Transaction transaction = null; + + writeLock.lock(); + try { + // create the connection + transaction = transactionBuilder.start(); + + // determine how long the cache is valid for + final int cacheSeconds; + try { + cacheSeconds = (int) FormatUtils.getTimeDuration(properties.getUserCredentialCacheDuration(), TimeUnit.SECONDS); + } catch (IllegalArgumentException iae) { + throw new AdministrationException("User credential cache duration is not configured correctly."); + } + + // attempt to authorize the user + AuthorizeUserAction authorizeUser = new AuthorizeUserAction(dn, cacheSeconds); + NiFiUser user = transaction.execute(authorizeUser); + + // commit the transaction + transaction.commit(); + + // return the nifi user + return user; + } catch (DataAccessException | TransactionException dae) { + rollback(transaction); + throw new AdministrationException(dae); + } catch (AccountDisabledException | AccountPendingException ade) { + rollback(transaction); + throw ade; + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public void deleteUser(String id) { + Transaction transaction = null; + + writeLock.lock(); + try { + // create the connection + transaction = transactionBuilder.start(); + + // delete the user + DeleteUserAction deleteUser = new DeleteUserAction(id); + transaction.execute(deleteUser); + + // commit the transaction + transaction.commit(); + } catch (DataAccessException | TransactionException dae) { + rollback(transaction); + throw new AdministrationException(dae); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUser disable(String id) { + Transaction transaction = null; + + writeLock.lock(); + try { + // create the connection + transaction = transactionBuilder.start(); + + // disable the user + DisableUserAction disableUser = new DisableUserAction(id); + NiFiUser user = transaction.execute(disableUser); + + // commit the transaction + transaction.commit(); + + // return the user + return user; + } catch (DataAccessException | TransactionException dae) { + rollback(transaction); + throw new AdministrationException(dae); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUserGroup disableGroup(String group) { + Transaction transaction = null; + + writeLock.lock(); + try { + // create the connection + transaction = transactionBuilder.start(); + + // disable the user + DisableUserGroupAction disableUser = new DisableUserGroupAction(group); + NiFiUserGroup userGroup = transaction.execute(disableUser); + + // commit the transaction + transaction.commit(); + + // return the user + return userGroup; + } catch (DataAccessException | TransactionException dae) { + rollback(transaction); + throw new AdministrationException(dae); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + @Override + public NiFiUser update(String id, Set<Authority> authorities) { + Transaction transaction = null; + + // may be empty but not null + if (authorities == null) { + throw new IllegalArgumentException("The specified authorities cannot be null."); + } + + writeLock.lock(); + try { + // invalidate the user account in preparation for potential subsequent errors + invalidateUserAccount(id); + + // at this point the current user account has been invalidated so we will + // attempt to update the account. if any part fails we are assured the + // user will be need to be given approval before they access the system at + // a later time + // start the transaction + transaction = transactionBuilder.start(); + + // update the user authorities + UpdateUserAction setUserAuthorities = new UpdateUserAction(id, authorities); + NiFiUser user = transaction.execute(setUserAuthorities); + + // commit the transaction + transaction.commit(); + + // return the user + return user; + } catch (TransactionException | DataAccessException e) { + rollback(transaction); + throw new AdministrationException(e); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + /** + * Invalidates the user with the specified id. This is done to ensure a user + * account will need to be re-validated in case an error occurs while + * modifying a user account. This method should only be invoked from within + * a write lock. + * + * @param id + */ + @Override + public void invalidateUserAccount(String id) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // invalidate the user account + InvalidateUserAccountAction invalidateUserAccount = new InvalidateUserAccountAction(id); + transaction.execute(invalidateUserAccount); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + /** + * Invalidates the user with the specified id. This is done to ensure a user + * account will need to be re-validated in case an error occurs while + * modifying a user account. This method should only be invoked from within + * a write lock. + * + * @param id + */ + @Override + public void invalidateUserGroupAccount(String group) { + Transaction transaction = null; + + writeLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // invalidate the user account + InvalidateUserGroupAccountsAction invalidateUserGroupAccounts = new InvalidateUserGroupAccountsAction(group); + transaction.execute(invalidateUserGroupAccounts); + + // commit the transaction + transaction.commit(); + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + writeLock.unlock(); + } + } + + // ----------------- + // read only methods + // ----------------- + @Override + public Boolean hasPendingUserAccount() { + Transaction transaction = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + final HasPendingUserAccounts hasPendingAccounts = new HasPendingUserAccounts(); + final Boolean hasPendingUserAccounts = transaction.execute(hasPendingAccounts); + + // commit the transaction + transaction.commit(); + + return hasPendingUserAccounts; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + } + + @Override + public Collection<NiFiUser> getUsers() { + Transaction transaction = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // get all users + GetUsersAction getUsers = new GetUsersAction(); + Collection<NiFiUser> users = transaction.execute(getUsers); + + // commit the transaction + transaction.commit(); + + // return the users + return users; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + } + + @Override + public NiFiUser getUserById(String id) { + Transaction transaction = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // return the desired user + FindUserByIdAction findUserById = new FindUserByIdAction(id); + NiFiUser user = transaction.execute(findUserById); + + // commit the transaction + transaction.commit(); + + // return the user + return user; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + } + + @Override + public NiFiUser getUserByDn(String dn) { + Transaction transaction = null; + + readLock.lock(); + try { + // start the transaction + transaction = transactionBuilder.start(); + + // return the desired user + FindUserByDnAction findUserByDn = new FindUserByDnAction(dn); + NiFiUser user = transaction.execute(findUserByDn); + + // commit the transaction + transaction.commit(); + + // return the user + return user; + } catch (TransactionException | DataAccessException te) { + rollback(transaction); + throw new AdministrationException(te); + } catch (Throwable t) { + rollback(transaction); + throw t; + } finally { + closeQuietly(transaction); + readLock.unlock(); + } + } + + /** + * Rolls back the specified transaction. + * + * @param transaction + */ + private void rollback(final Transaction transaction) { + if (transaction != null) { + transaction.rollback(); + } + } + + /** + * Closes the specified transaction. + * + * @param transaction + */ + private void closeQuietly(final Transaction transaction) { + if (transaction != null) { + try { + transaction.close(); + } catch (final IOException ioe) { + } + } + } + + /* + * setters + */ + public void setTransactionBuilder(TransactionBuilder transactionBuilder) { + this.transactionBuilder = transactionBuilder; + } + + public void setProperties(NiFiProperties properties) { + this.properties = properties; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/Transaction.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/Transaction.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/Transaction.java new file mode 100644 index 0000000..edd214b --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/Transaction.java @@ -0,0 +1,49 @@ +/* + * 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.admin.service.transaction; + +import java.io.Closeable; +import org.apache.nifi.admin.service.action.AdministrationAction; + +/** + * Defines a transaction. + */ +public interface Transaction extends Closeable { + + /** + * Executes the specified action within the current transaction. + * + * @param <T> + * @param action + * @return + * @throws IllegalStateException - if there is no current transaction + */ + <T> T execute(AdministrationAction<T> action); + + /** + * Commits the current transaction. + * + * @throws TransactionException - if the transaction is unable to be + * committed + */ + void commit() throws TransactionException; + + /** + * Rolls back the current transaction. + */ + void rollback(); +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionBuilder.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionBuilder.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionBuilder.java new file mode 100644 index 0000000..2d2ef82 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionBuilder.java @@ -0,0 +1,25 @@ +/* + * 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.admin.service.transaction; + +/** + * + */ +public interface TransactionBuilder { + + Transaction start() throws TransactionException; +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionException.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionException.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionException.java new file mode 100644 index 0000000..924e01f --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/TransactionException.java @@ -0,0 +1,40 @@ +/* + * 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.admin.service.transaction; + +/** + * Exception to indicate that the user account is disabled. + */ +public class TransactionException extends RuntimeException { + + public TransactionException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) { + super(message, cause, enableSuppression, writableStackTrace); + } + + public TransactionException(Throwable cause) { + super(cause); + } + + public TransactionException(String message, Throwable cause) { + super(message, cause); + } + + public TransactionException(String message) { + super(message); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransaction.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransaction.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransaction.java new file mode 100644 index 0000000..a3cfb5e --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransaction.java @@ -0,0 +1,93 @@ +/* + * 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.admin.service.transaction.impl; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.SQLException; +import org.apache.nifi.admin.RepositoryUtils; +import org.apache.nifi.admin.dao.DAOFactory; +import org.apache.nifi.admin.dao.impl.DAOFactoryImpl; +import org.apache.nifi.admin.service.action.AdministrationAction; +import org.apache.nifi.admin.service.transaction.TransactionException; +import org.apache.nifi.admin.service.transaction.Transaction; +import org.apache.nifi.authorization.AuthorityProvider; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Transaction implementation that uses the specified SQL Connection and + * AuthorityProvider. + */ +public class StandardTransaction implements Transaction { + + private static final Logger logger = LoggerFactory.getLogger(StandardTransaction.class); + + private final AuthorityProvider authorityProvider; + private Connection connection; + + public StandardTransaction(AuthorityProvider authorityProvider, Connection connection) { + this.authorityProvider = authorityProvider; + this.connection = connection; + } + + @Override + public <T> T execute(AdministrationAction<T> action) { + // ensure the transaction has been started + if (connection == null) { + throw new IllegalStateException("This transaction is not active."); + } + + // create a dao factory + DAOFactory daoFactory = new DAOFactoryImpl(connection); + + // execute the specified action + return action.execute(daoFactory, authorityProvider); + } + + @Override + public void commit() throws TransactionException { + // ensure there is an active transaction + if (connection == null) { + throw new IllegalStateException("No active transaction."); + } + + try { + // commit the transaction + connection.commit(); + } catch (SQLException sqle) { + throw new TransactionException(sqle.getMessage()); + } + } + + @Override + public void rollback() { + // ensure there is an active transaction + if (connection != null) { + // rollback the transaction + RepositoryUtils.rollback(connection, logger); + } + } + + @Override + public void close() throws IOException { + if (connection != null) { + RepositoryUtils.closeQuietly(connection); + connection = null; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransactionBuilder.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransactionBuilder.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransactionBuilder.java new file mode 100644 index 0000000..b6e5a30 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/admin/service/transaction/impl/StandardTransactionBuilder.java @@ -0,0 +1,57 @@ +/* + * 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.admin.service.transaction.impl; + +import java.sql.Connection; +import java.sql.SQLException; +import javax.sql.DataSource; +import org.apache.nifi.admin.service.transaction.Transaction; +import org.apache.nifi.admin.service.transaction.TransactionBuilder; +import org.apache.nifi.admin.service.transaction.TransactionException; +import org.apache.nifi.authorization.AuthorityProvider; + +/** + * + */ +public class StandardTransactionBuilder implements TransactionBuilder { + + private DataSource dataSource; + private AuthorityProvider authorityProvider; + + @Override + public Transaction start() throws TransactionException { + try { + // get a new connection + Connection connection = dataSource.getConnection(); + connection.setAutoCommit(false); + + // create a new transaction + return new StandardTransaction(authorityProvider, connection); + } catch (SQLException sqle) { + throw new TransactionException(sqle.getMessage()); + } + } + + /* setters */ + public void setDataSource(DataSource dataSource) { + this.dataSource = dataSource; + } + + public void setAuthorityProvider(AuthorityProvider authorityProvider) { + this.authorityProvider = authorityProvider; + } +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/AuthorityProviderFactoryBean.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/AuthorityProviderFactoryBean.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/AuthorityProviderFactoryBean.java new file mode 100644 index 0000000..b05d32f --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/AuthorityProviderFactoryBean.java @@ -0,0 +1,516 @@ +/* + * 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.authorization; + +import java.io.File; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import javax.xml.XMLConstants; +import javax.xml.bind.JAXBContext; +import javax.xml.bind.JAXBElement; +import javax.xml.bind.JAXBException; +import javax.xml.bind.Unmarshaller; +import javax.xml.transform.stream.StreamSource; +import javax.xml.validation.Schema; +import javax.xml.validation.SchemaFactory; +import org.apache.nifi.authorization.annotation.AuthorityProviderContext; +import org.apache.nifi.authorization.exception.AuthorityAccessException; +import org.apache.nifi.authorization.exception.IdentityAlreadyExistsException; +import org.apache.nifi.authorization.exception.ProviderCreationException; +import org.apache.nifi.authorization.exception.ProviderDestructionException; +import org.apache.nifi.authorization.exception.UnknownIdentityException; +import org.apache.nifi.authorization.generated.AuthorityProviders; +import org.apache.nifi.authorization.generated.Property; +import org.apache.nifi.authorization.generated.Provider; +import org.apache.nifi.nar.ExtensionManager; +import org.apache.nifi.nar.NarCloseable; +import org.apache.nifi.util.NiFiProperties; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.xml.sax.SAXException; + +/** + * Factory bean for loading the configured authority provider. + */ +public class AuthorityProviderFactoryBean implements FactoryBean, ApplicationContextAware, DisposableBean, AuthorityProviderLookup { + + private static final Logger logger = LoggerFactory.getLogger(AuthorityProviderFactoryBean.class); + private static final String AUTHORITY_PROVIDERS_XSD = "/authority-providers.xsd"; + private static final String JAXB_GENERATED_PATH = "org.apache.nifi.authorization.generated"; + private static final JAXBContext JAXB_CONTEXT = initializeJaxbContext(); + + /** + * Load the JAXBContext. + */ + private static JAXBContext initializeJaxbContext() { + try { + return JAXBContext.newInstance(JAXB_GENERATED_PATH, AuthorityProviderFactoryBean.class.getClassLoader()); + } catch (JAXBException e) { + throw new RuntimeException("Unable to create JAXBContext."); + } + } + + private ApplicationContext applicationContext; + private AuthorityProvider authorityProvider; + private NiFiProperties properties; + private final Map<String, AuthorityProvider> authorityProviders = new HashMap<>(); + + @Override + public AuthorityProvider getAuthorityProvider(String identifier) { + return authorityProviders.get(identifier); + } + + @Override + public Object getObject() throws Exception { + if (authorityProvider == null) { + // look up the authority provider to use + final String authorityProviderIdentifier = properties.getProperty(NiFiProperties.SECURITY_USER_AUTHORITY_PROVIDER); + + // ensure the authority provider class name was specified + if (StringUtils.isBlank(authorityProviderIdentifier)) { + // if configured for ssl, the authority provider must be specified + if (properties.getSslPort() != null) { + throw new Exception("When running securely, the authority provider identifier must be specified in the nifi properties file."); + } + + // use a default provider... only allowable when running not securely + authorityProvider = createDefaultProvider(); + } else { + final AuthorityProviders authorityProviderConfiguration = loadAuthorityProvidersConfiguration(); + + // create each authority provider + for (final Provider provider : authorityProviderConfiguration.getProvider()) { + authorityProviders.put(provider.getIdentifier(), createAuthorityProvider(provider.getIdentifier(), provider.getClazz())); + } + + // configure each authority provider + for (final Provider provider : authorityProviderConfiguration.getProvider()) { + final AuthorityProvider instance = authorityProviders.get(provider.getIdentifier()); + instance.onConfigured(loadAuthorityProviderConfiguration(provider)); + } + + // get the authority provider instance + authorityProvider = getAuthorityProvider(authorityProviderIdentifier); + + // ensure it was found + if (authorityProvider == null) { + throw new Exception(String.format("The specified authority provider '%s' could not be found.", authorityProviderIdentifier)); + } + } + } + + return authorityProvider; + } + + /** + * Loads the authority providers configuration. + * + * @return + * @throws Exception + */ + private AuthorityProviders loadAuthorityProvidersConfiguration() throws Exception { + final File authorityProvidersConfigurationFile = properties.getAuthorityProviderConfiguraitonFile(); + + // load the users from the specified file + if (authorityProvidersConfigurationFile.exists()) { + try { + // find the schema + final SchemaFactory schemaFactory = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI); + final Schema schema = schemaFactory.newSchema(AuthorityProviders.class.getResource(AUTHORITY_PROVIDERS_XSD)); + + // attempt to unmarshal + final Unmarshaller unmarshaller = JAXB_CONTEXT.createUnmarshaller(); + unmarshaller.setSchema(schema); + final JAXBElement<AuthorityProviders> element = unmarshaller.unmarshal(new StreamSource(authorityProvidersConfigurationFile), AuthorityProviders.class); + return element.getValue(); + } catch (SAXException | JAXBException e) { + throw new Exception("Unable to load the authority provider configuration file at: " + authorityProvidersConfigurationFile.getAbsolutePath()); + } + } else { + throw new Exception("Unable to find the authority provider configuration file at " + authorityProvidersConfigurationFile.getAbsolutePath()); + } + } + + /** + * Creates the AuthorityProvider instance for the identifier specified. + * + * @param identifier + * @param authorityProviderClassName + * @return + * @throws Exception + */ + private AuthorityProvider createAuthorityProvider(final String identifier, final String authorityProviderClassName) throws Exception { + // get the classloader for the specified authority provider + final ClassLoader authorityProviderClassLoader = ExtensionManager.getClassLoader(authorityProviderClassName); + if (authorityProviderClassLoader == null) { + throw new Exception(String.format("The specified authority provider class '%s' is not known to this nifi.", authorityProviderClassName)); + } + + // get the current context classloader + final ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader(); + + final AuthorityProvider instance; + try { + // set the appropriate class loader + Thread.currentThread().setContextClassLoader(authorityProviderClassLoader); + + // attempt to load the class + Class<?> rawAuthorityProviderClass = Class.forName(authorityProviderClassName, true, authorityProviderClassLoader); + Class<? extends AuthorityProvider> authorityProviderClass = rawAuthorityProviderClass.asSubclass(AuthorityProvider.class); + + // otherwise create a new instance + Constructor constructor = authorityProviderClass.getConstructor(); + instance = (AuthorityProvider) constructor.newInstance(); + + // method injection + performMethodInjection(instance, authorityProviderClass); + + // field injection + performFieldInjection(instance, authorityProviderClass); + + // call post construction lifecycle event + instance.initialize(new StandardAuthorityProviderInitializationContext(identifier, this)); + } finally { + if (currentClassLoader != null) { + Thread.currentThread().setContextClassLoader(currentClassLoader); + } + } + + return withNarLoader(instance); + } + + /** + * Loads the AuthorityProvider configuration. + * + * @param provider + * @return + */ + private AuthorityProviderConfigurationContext loadAuthorityProviderConfiguration(final Provider provider) { + final Map<String, String> providerProperties = new HashMap<>(); + + for (final Property property : provider.getProperty()) { + providerProperties.put(property.getName(), property.getValue()); + } + + return new StandardAuthorityProviderConfigurationContext(provider.getIdentifier(), providerProperties); + } + + /** + * Performs method injection. + * + * @param instance + * @param authorityProviderClass + * @throws IllegalAccessException + * @throws IllegalArgumentException + * @throws InvocationTargetException + */ + private void performMethodInjection(final AuthorityProvider instance, final Class authorityProviderClass) throws IllegalAccessException, IllegalArgumentException, InvocationTargetException { + for (final Method method : authorityProviderClass.getMethods()) { + if (method.isAnnotationPresent(AuthorityProviderContext.class)) { + // make the method accessible + final boolean isAccessible = method.isAccessible(); + method.setAccessible(true); + + try { + final Class<?>[] argumentTypes = method.getParameterTypes(); + + // look for setters (single argument) + if (argumentTypes.length == 1) { + final Class<?> argumentType = argumentTypes[0]; + + // look for well known types + if (NiFiProperties.class.isAssignableFrom(argumentType)) { + // nifi properties injection + method.invoke(instance, properties); + } else if (ApplicationContext.class.isAssignableFrom(argumentType)) { + // spring application context injection + method.invoke(instance, applicationContext); + } + } + } finally { + method.setAccessible(isAccessible); + } + } + } + + final Class parentClass = authorityProviderClass.getSuperclass(); + if (parentClass != null && AuthorityProvider.class.isAssignableFrom(parentClass)) { + performMethodInjection(instance, parentClass); + } + } + + /** + * Performs field injection. + * + * @param instance + * @param authorityProviderClass + * @throws IllegalArgumentException + * @throws IllegalAccessException + */ + private void performFieldInjection(final AuthorityProvider instance, final Class authorityProviderClass) throws IllegalArgumentException, IllegalAccessException { + for (final Field field : authorityProviderClass.getDeclaredFields()) { + if (field.isAnnotationPresent(AuthorityProviderContext.class)) { + // make the method accessible + final boolean isAccessible = field.isAccessible(); + field.setAccessible(true); + + try { + // get the type + final Class<?> fieldType = field.getType(); + + // only consider this field if it isn't set yet + if (field.get(instance) == null) { + // look for well known types + if (NiFiProperties.class.isAssignableFrom(fieldType)) { + // nifi properties injection + field.set(instance, properties); + } else if (ApplicationContext.class.isAssignableFrom(fieldType)) { + // spring application context injection + field.set(instance, applicationContext); + } + } + + } finally { + field.setAccessible(isAccessible); + } + } + } + + final Class parentClass = authorityProviderClass.getSuperclass(); + if (parentClass != null && AuthorityProvider.class.isAssignableFrom(parentClass)) { + performFieldInjection(instance, parentClass); + } + } + + /** + * Creates a default provider to use when running unsecurely with no + * provider configured. + * + * @return + */ + private AuthorityProvider createDefaultProvider() { + return new AuthorityProvider() { + @Override + public boolean doesDnExist(String dn) throws AuthorityAccessException { + return false; + } + + @Override + public Set<Authority> getAuthorities(String dn) throws UnknownIdentityException, AuthorityAccessException { + return EnumSet.noneOf(Authority.class); + } + + @Override + public void setAuthorities(String dn, Set<Authority> authorities) throws UnknownIdentityException, AuthorityAccessException { + } + + @Override + public Set<String> getUsers(Authority authority) throws AuthorityAccessException { + return new HashSet<>(); + } + + @Override + public void revokeUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + } + + @Override + public void addUser(String dn, String group) throws IdentityAlreadyExistsException, AuthorityAccessException { + } + + @Override + public String getGroupForUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + return null; + } + + @Override + public void revokeGroup(String group) throws UnknownIdentityException, AuthorityAccessException { + } + + @Override + public void setUsersGroup(Set<String> dn, String group) throws UnknownIdentityException, AuthorityAccessException { + } + + @Override + public void ungroupUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + } + + @Override + public void ungroup(String group) throws AuthorityAccessException { + } + + @Override + public void initialize(AuthorityProviderInitializationContext initializationContext) throws ProviderCreationException { + } + + @Override + public void onConfigured(AuthorityProviderConfigurationContext configurationContext) throws ProviderCreationException { + } + + @Override + public void preDestruction() throws ProviderDestructionException { + } + }; + } + + /** + * Decorates the base provider to ensure the nar context classloader is used + * when invoking the underlying methods. + * + * @param baseProvider + * @return + */ + public AuthorityProvider withNarLoader(final AuthorityProvider baseProvider) { + return new AuthorityProvider() { + @Override + public boolean doesDnExist(String dn) throws AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return baseProvider.doesDnExist(dn); + } + } + + @Override + public Set<Authority> getAuthorities(String dn) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return baseProvider.getAuthorities(dn); + } + } + + @Override + public void setAuthorities(String dn, Set<Authority> authorities) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.setAuthorities(dn, authorities); + } + } + + @Override + public Set<String> getUsers(Authority authority) throws AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return baseProvider.getUsers(authority); + } + } + + @Override + public void revokeUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.revokeUser(dn); + } + } + + @Override + public void addUser(String dn, String group) throws IdentityAlreadyExistsException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.addUser(dn, group); + } + } + + @Override + public String getGroupForUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + return baseProvider.getGroupForUser(dn); + } + } + + @Override + public void revokeGroup(String group) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.revokeGroup(group); + } + } + + @Override + public void setUsersGroup(Set<String> dns, String group) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.setUsersGroup(dns, group); + } + } + + @Override + public void ungroupUser(String dn) throws UnknownIdentityException, AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.ungroupUser(dn); + } + } + + @Override + public void ungroup(String group) throws AuthorityAccessException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.ungroup(group); + } + } + + @Override + public void initialize(AuthorityProviderInitializationContext initializationContext) throws ProviderCreationException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.initialize(initializationContext); + } + } + + @Override + public void onConfigured(AuthorityProviderConfigurationContext configurationContext) throws ProviderCreationException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.onConfigured(configurationContext); + } + } + + @Override + public void preDestruction() throws ProviderDestructionException { + try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) { + baseProvider.preDestruction(); + } + } + }; + } + + @Override + public Class getObjectType() { + return AuthorityProvider.class; + } + + @Override + public boolean isSingleton() { + return true; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public void destroy() throws Exception { + if (authorityProvider != null) { + authorityProvider.preDestruction(); + } + } + + public void setProperties(NiFiProperties properties) { + this.properties = properties; + } +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderConfigurationContext.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderConfigurationContext.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderConfigurationContext.java new file mode 100644 index 0000000..0535e27 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderConfigurationContext.java @@ -0,0 +1,50 @@ +/* + * 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.authorization; + +import java.util.Collections; +import java.util.Map; + +/** + * + */ +public class StandardAuthorityProviderConfigurationContext implements AuthorityProviderConfigurationContext { + + private final String identifier; + private final Map<String, String> properties; + + public StandardAuthorityProviderConfigurationContext(String identifier, Map<String, String> properties) { + this.identifier = identifier; + this.properties = properties; + } + + @Override + public String getIdentifier() { + return identifier; + } + + @Override + public Map<String, String> getProperties() { + return Collections.unmodifiableMap(properties); + } + + @Override + public String getProperty(String property) { + return properties.get(property); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4d998c12/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderInitializationContext.java ---------------------------------------------------------------------- diff --git a/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderInitializationContext.java b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderInitializationContext.java new file mode 100644 index 0000000..e4b16c4 --- /dev/null +++ b/nar-bundles/framework-bundle/framework/administration/src/main/java/org/apache/nifi/authorization/StandardAuthorityProviderInitializationContext.java @@ -0,0 +1,42 @@ +/* + * 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.authorization; + +/** + * + */ +public class StandardAuthorityProviderInitializationContext implements AuthorityProviderInitializationContext { + + private final String identifier; + private final AuthorityProviderLookup authorityProviderLookup; + + public StandardAuthorityProviderInitializationContext(String identifier, AuthorityProviderLookup authorityProviderLookup) { + this.identifier = identifier; + this.authorityProviderLookup = authorityProviderLookup; + } + + @Override + public String getIdentifier() { + return identifier; + } + + @Override + public AuthorityProviderLookup getAuthorityProviderLookup() { + return authorityProviderLookup; + } + +}