NIFI-2307: - Enforcing connection permissions based on the source and destination comonent. - Removing connection specific access policies. NIFI-2265: - Filtering out sensitive details in component status and status history when appropriate. NIFI-1800: - Adding parent process group id to the Controller Services table. NIFI-2077: - Removing some old un-used icons following the UI refresh. NIFI-2242: - Requiring write permissions for all components in a selection. NIFI-2080: - Updating style of the name in the selection context to handle scroll bars and use available width. NIFI-2331: - Addressing issue when removing a user/group which was causing the tenant policy to be removed. NIFI-2335: - Ensuring the flow is saved after starting/stopping a process group. NIFI-2235: - Ensuring we use consistent conditions between the context menu and the operate palette.

- Allowing users with read only access to the tenants page.
- Fixing current user integration test.
- Ensuring schedule methods are locked appropriately.
- Addressing comments from PR.

This closes #698

Signed-off-by: jpercivall <joepercivall@yahoo.com>
This commit is contained in:
Matt Gilman 2016-07-21 22:53:00 -04:00 committed by jpercivall
parent 0dbba811f3
commit 4a4d60e6af
111 changed files with 1149 additions and 616 deletions

View File

@ -97,7 +97,7 @@ public interface Authorizable {
if (parent == null) { if (parent == null) {
return AuthorizationResult.denied(); return AuthorizationResult.denied();
} else { } else {
return parent.checkAuthorization(authorizer, action, user); return parent.checkAuthorization(authorizer, action, user, resourceContext);
} }
} else { } else {
return result; return result;
@ -152,7 +152,7 @@ public interface Authorizable {
if (parent == null) { if (parent == null) {
throw new AccessDeniedException("Access is denied"); throw new AccessDeniedException("Access is denied");
} else { } else {
parent.authorize(authorizer, action, user); parent.authorize(authorizer, action, user, resourceContext);
} }
} else if (Result.Denied.equals(result.getResult())) { } else if (Result.Denied.equals(result.getResult())) {
throw new AccessDeniedException(result.getExplanation()); throw new AccessDeniedException(result.getExplanation());

View File

@ -16,16 +16,24 @@
*/ */
package org.apache.nifi.controller.status.history; package org.apache.nifi.controller.status.history;
import java.util.Date;
import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessGroupStatus;
import java.util.Date;
/** /**
* A repository for storing and retrieving components' historical status * A repository for storing and retrieving components' historical status
* information * information
*/ */
public interface ComponentStatusRepository { public interface ComponentStatusRepository {
String COMPONENT_DETAIL_ID = "Id";
String COMPONENT_DETAIL_GROUP_ID = "Group Id";
String COMPONENT_DETAIL_NAME = "Name";
String COMPONENT_DETAIL_TYPE = "Type";
String COMPONENT_DETAIL_SOURCE_NAME = "Source Name";
String COMPONENT_DETAIL_DESTINATION_NAME = "Destination Name";
String COMPONENT_DETAIL_URI = "Uri";
/** /**
* Captures the status information provided in the given report * Captures the status information provided in the given report
* *

View File

@ -17,14 +17,13 @@
package org.apache.nifi.web.api.dto.status; package org.apache.nifi.web.api.dto.status;
import java.util.Date; import com.wordnik.swagger.annotations.ApiModelProperty;
import java.util.List; import org.apache.nifi.web.api.dto.util.TimeAdapter;
import javax.xml.bind.annotation.XmlType; import javax.xml.bind.annotation.XmlType;
import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter; import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
import java.util.Date;
import com.wordnik.swagger.annotations.ApiModelProperty; import java.util.List;
import org.apache.nifi.web.api.dto.util.TimeAdapter;
@XmlType(name = "remoteProcessGroupStatus") @XmlType(name = "remoteProcessGroupStatus")
public class RemoteProcessGroupStatusDTO { public class RemoteProcessGroupStatusDTO {
@ -35,8 +34,6 @@ public class RemoteProcessGroupStatusDTO {
private String transmissionStatus; private String transmissionStatus;
private Date statsLastRefreshed; private Date statsLastRefreshed;
private List<String> authorizationIssues;
private RemoteProcessGroupStatusSnapshotDTO aggregateSnapshot; private RemoteProcessGroupStatusSnapshotDTO aggregateSnapshot;
private List<NodeRemoteProcessGroupStatusSnapshotDTO> nodeSnapshots; private List<NodeRemoteProcessGroupStatusSnapshotDTO> nodeSnapshots;
@ -76,16 +73,6 @@ public class RemoteProcessGroupStatusDTO {
this.transmissionStatus = transmissionStatus; this.transmissionStatus = transmissionStatus;
} }
@ApiModelProperty("Any remote authorization issues for the remote process group.")
public List<String> getAuthorizationIssues() {
return authorizationIssues;
}
public void setAuthorizationIssues(List<String> authorizationIssues) {
this.authorizationIssues = authorizationIssues;
}
@ApiModelProperty("The URI of the target system.") @ApiModelProperty("The URI of the target system.")
public String getTargetUri() { public String getTargetUri() {
return targetUri; return targetUri;

View File

@ -16,13 +16,10 @@
*/ */
package org.apache.nifi.web.api.dto.status; package org.apache.nifi.web.api.dto.status;
import java.util.ArrayList; import com.wordnik.swagger.annotations.ApiModelProperty;
import java.util.List;
import javax.xml.bind.annotation.XmlType; import javax.xml.bind.annotation.XmlType;
import com.wordnik.swagger.annotations.ApiModelProperty;
/** /**
* The status of a remote process group in this NiFi. * The status of a remote process group in this NiFi.
*/ */
@ -36,8 +33,6 @@ public class RemoteProcessGroupStatusSnapshotDTO implements Cloneable {
private String transmissionStatus; private String transmissionStatus;
private Integer activeThreadCount; private Integer activeThreadCount;
private List<String> authorizationIssues;
private Integer flowFilesSent = 0; private Integer flowFilesSent = 0;
private Long bytesSent = 0L; private Long bytesSent = 0L;
private String sent; private String sent;
@ -118,18 +113,6 @@ public class RemoteProcessGroupStatusSnapshotDTO implements Cloneable {
this.activeThreadCount = activeThreadCount; this.activeThreadCount = activeThreadCount;
} }
/**
* @return any remote authorization issues for this remote process group
*/
@ApiModelProperty("Any remote authorization issues for the remote process group.")
public List<String> getAuthorizationIssues() {
return authorizationIssues;
}
public void setAuthorizationIssues(List<String> authorizationIssues) {
this.authorizationIssues = authorizationIssues;
}
/** /**
* @return Formatted description of the amount of data sent to this remote process group * @return Formatted description of the amount of data sent to this remote process group
*/ */
@ -201,7 +184,6 @@ public class RemoteProcessGroupStatusSnapshotDTO implements Cloneable {
other.setTargetUri(getTargetUri()); other.setTargetUri(getTargetUri());
other.setTransmissionStatus(getTransmissionStatus()); other.setTransmissionStatus(getTransmissionStatus());
other.setActiveThreadCount(getActiveThreadCount()); other.setActiveThreadCount(getActiveThreadCount());
other.setAuthorizationIssues(getAuthorizationIssues() == null ? null : new ArrayList<String>(getAuthorizationIssues()));
other.setFlowFilesSent(getFlowFilesSent()); other.setFlowFilesSent(getFlowFilesSent());
other.setBytesSent(getBytesSent()); other.setBytesSent(getBytesSent());
other.setFlowFilesReceived(getFlowFilesReceived()); other.setFlowFilesReceived(getFlowFilesReceived());

View File

@ -34,10 +34,13 @@ public class ConnectionEntity extends ComponentEntity {
private ConnectionStatusDTO status; private ConnectionStatusDTO status;
private List<PositionDTO> bends; private List<PositionDTO> bends;
private Integer labelIndex; private Integer labelIndex;
private Long zIndex;
private String sourceId; private String sourceId;
private String sourceGroupId; private String sourceGroupId;
private String sourceType;
private String destinationId; private String destinationId;
private String destinationGroupId; private String destinationGroupId;
private String destinationType;
/** /**
* @return RelationshipDTO that is being serialized * @return RelationshipDTO that is being serialized
@ -92,6 +95,20 @@ public class ConnectionEntity extends ComponentEntity {
this.labelIndex = labelIndex; this.labelIndex = labelIndex;
} }
/**
* @return z index for this connection
*/
@ApiModelProperty(
value = "The z index of the connection."
)
public Long getzIndex() {
return zIndex;
}
public void setzIndex(Long zIndex) {
this.zIndex = zIndex;
}
/** /**
* @return The identifier of the source of this connection * @return The identifier of the source of this connection
*/ */
@ -134,6 +151,22 @@ public class ConnectionEntity extends ComponentEntity {
this.sourceGroupId = sourceGroupId; this.sourceGroupId = sourceGroupId;
} }
/**
* @return type of this source connectable component
*/
@ApiModelProperty(
value = "The type of component the source connectable is.",
required = true,
allowableValues = "PROCESSOR, REMOTE_INPUT_PORT, REMOTE_OUTPUT_PORT, INPUT_PORT, OUTPUT_PORT, FUNNEL"
)
public String getSourceType() {
return sourceType;
}
public void setSourceType(String sourceType) {
this.sourceType = sourceType;
}
/** /**
* @return The identifier of the group of the destination of this connection * @return The identifier of the group of the destination of this connection
*/ */
@ -147,4 +180,20 @@ public class ConnectionEntity extends ComponentEntity {
public void setDestinationGroupId(String destinationGroupId) { public void setDestinationGroupId(String destinationGroupId) {
this.destinationGroupId = destinationGroupId; this.destinationGroupId = destinationGroupId;
} }
/**
* @return type of this destination connectable component
*/
@ApiModelProperty(
value = "The type of component the destination connectable is.",
required = true,
allowableValues = "PROCESSOR, REMOTE_INPUT_PORT, REMOTE_OUTPUT_PORT, INPUT_PORT, OUTPUT_PORT, FUNNEL"
)
public String getDestinationType() {
return destinationType;
}
public void setDestinationType(String destinationType) {
this.destinationType = destinationType;
}
} }

View File

@ -16,6 +16,7 @@
*/ */
package org.apache.nifi.web.api.entity; package org.apache.nifi.web.api.entity;
import com.wordnik.swagger.annotations.ApiModelProperty;
import org.apache.nifi.web.api.dto.ProcessorDTO; import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO; import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO;
@ -28,6 +29,7 @@ import javax.xml.bind.annotation.XmlRootElement;
public class ProcessorEntity extends ComponentEntity { public class ProcessorEntity extends ComponentEntity {
private ProcessorDTO component; private ProcessorDTO component;
private String inputRequirement;
private ProcessorStatusDTO status; private ProcessorStatusDTO status;
/** /**
@ -55,4 +57,18 @@ public class ProcessorEntity extends ComponentEntity {
public void setStatus(ProcessorStatusDTO status) { public void setStatus(ProcessorStatusDTO status) {
this.status = status; this.status = status;
} }
/**
* @return the input requirement of this processor
*/
@ApiModelProperty(
value = "The input requirement for this processor."
)
public String getInputRequirement() {
return inputRequirement;
}
public void setInputRequirement(String inputRequirement) {
this.inputRequirement = inputRequirement;
}
} }

View File

@ -22,18 +22,6 @@ import java.util.Objects;
public final class ResourceFactory { public final class ResourceFactory {
private final static Resource CONNECTION_RESOURCE = new Resource() {
@Override
public String getIdentifier() {
return ResourceType.Connection.getValue();
}
@Override
public String getName() {
return "Connection";
}
};
private final static Resource CONTROLLER_RESOURCE = new Resource() { private final static Resource CONTROLLER_RESOURCE = new Resource() {
@Override @Override
public String getIdentifier() { public String getIdentifier() {
@ -299,15 +287,6 @@ public final class ResourceFactory {
} }
}; };
/**
* Gets the Resource for accessing Connections.
*
* @return The resource for accessing connections
*/
public static Resource getConnectionResource() {
return CONNECTION_RESOURCE;
}
/** /**
* Gets the Resource for accessing the Controller. This includes Controller level configuration, bulletins, reporting tasks, and the cluster. * Gets the Resource for accessing the Controller. This includes Controller level configuration, bulletins, reporting tasks, and the cluster.
* *
@ -602,30 +581,6 @@ public final class ResourceFactory {
}; };
} }
/**
* Gets a Resource fo accessing a flowfile queue for the specified connection.
*
* @param connectionIdentifier The identifier of the connection
* @param connectionName The name of the connection
* @return The resource
*/
public static Resource getFlowFileQueueResource(final String connectionIdentifier, final String connectionName) {
Objects.requireNonNull(connectionIdentifier, "The connection identifier must be specified.");
Objects.requireNonNull(connectionName, "The connection name must be specified.");
return new Resource() {
@Override
public String getIdentifier() {
return String.format("/flowfile-queue/%s", connectionIdentifier);
}
@Override
public String getName() {
return connectionName + " queue";
}
};
}
/** /**
* Prevent outside instantiation. * Prevent outside instantiation.
*/ */

View File

@ -17,11 +17,10 @@
package org.apache.nifi.authorization.resource; package org.apache.nifi.authorization.resource;
public enum ResourceType { public enum ResourceType {
Connection("/connections"),
Controller("/controller"), Controller("/controller"),
ControllerService("/controller-services"), ControllerService("/controller-services"),
Counters("/counters"), Counters("/counters"),
Funnel("/funnel"), Funnel("/funnels"),
Flow("/flow"), Flow("/flow"),
InputPort("/input-ports"), InputPort("/input-ports"),
Label("/labels"), Label("/labels"),

View File

@ -21,14 +21,10 @@ import org.apache.nifi.cluster.manager.StatusMerger;
import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.web.api.dto.status.NodeProcessGroupStatusSnapshotDTO; import org.apache.nifi.web.api.dto.status.NodeProcessGroupStatusSnapshotDTO;
import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO; import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ProcessGroupStatusSnapshotDTO;
import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusSnapshotDTO;
import org.apache.nifi.web.api.entity.ProcessGroupStatusEntity; import org.apache.nifi.web.api.entity.ProcessGroupStatusEntity;
import java.net.URI; import java.net.URI;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map; import java.util.Map;
import java.util.regex.Pattern; import java.util.regex.Pattern;
@ -70,18 +66,6 @@ public class GroupStatusEndpointMerger extends AbstractNodeStatusEndpoint<Proces
continue; continue;
} }
final ProcessGroupStatusSnapshotDTO nodeSnapshot = nodeProcessGroupStatus.getAggregateSnapshot();
for (final RemoteProcessGroupStatusSnapshotDTO remoteProcessGroupStatus : nodeSnapshot.getRemoteProcessGroupStatusSnapshots()) {
final List<String> nodeAuthorizationIssues = remoteProcessGroupStatus.getAuthorizationIssues();
if (nodeAuthorizationIssues != null && !nodeAuthorizationIssues.isEmpty()) {
for (final ListIterator<String> iter = nodeAuthorizationIssues.listIterator(); iter.hasNext();) {
final String Issue = iter.next();
iter.set("[" + nodeId.getApiAddress() + ":" + nodeId.getApiPort() + "] -- " + Issue);
}
remoteProcessGroupStatus.setAuthorizationIssues(nodeAuthorizationIssues);
}
}
StatusMerger.merge(mergedProcessGroupStatus, nodeProcessGroupStatus, nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort()); StatusMerger.merge(mergedProcessGroupStatus, nodeProcessGroupStatus, nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort());
} }
} }

View File

@ -378,15 +378,6 @@ public class StatusMerger {
target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount()); target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
final List<String> authIssues = new ArrayList<>();
if (target.getAuthorizationIssues() != null) {
authIssues.addAll(target.getAuthorizationIssues());
}
if (toMerge.getAuthorizationIssues() != null) {
authIssues.addAll(toMerge.getAuthorizationIssues());
}
target.setAuthorizationIssues(authIssues);
target.setFlowFilesSent(target.getFlowFilesSent() + toMerge.getFlowFilesSent()); target.setFlowFilesSent(target.getFlowFilesSent() + toMerge.getFlowFilesSent());
target.setBytesSent(target.getBytesSent() + toMerge.getBytesSent()); target.setBytesSent(target.getBytesSent() + toMerge.getBytesSent());
target.setFlowFilesReceived(target.getFlowFilesReceived() + toMerge.getFlowFilesReceived()); target.setFlowFilesReceived(target.getFlowFilesReceived() + toMerge.getFlowFilesReceived());

View File

@ -20,10 +20,14 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.EqualsBuilder; import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder; import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.nifi.authorization.AccessDeniedException;
import org.apache.nifi.authorization.AuthorizationResult;
import org.apache.nifi.authorization.AuthorizationResult.Result;
import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.Resource; import org.apache.nifi.authorization.Resource;
import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.resource.ResourceFactory; import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.authorization.resource.ResourceType;
import org.apache.nifi.controller.ProcessScheduler; import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.StandardFlowFileQueue; import org.apache.nifi.controller.StandardFlowFileQueue;
import org.apache.nifi.controller.queue.FlowFileQueue; import org.apache.nifi.controller.queue.FlowFileQueue;
@ -42,6 +46,7 @@ import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
@ -103,12 +108,20 @@ public final class StandardConnection implements Connection {
@Override @Override
public Authorizable getParentAuthorizable() { public Authorizable getParentAuthorizable() {
return getSource(); return null;
} }
@Override @Override
public Resource getResource() { public Resource getResource() {
String name = getName(); return new Resource() {
@Override
public String getIdentifier() {
return "/connections/" + StandardConnection.this.getIdentifier();
}
@Override
public String getName() {
String name = StandardConnection.this.getName();
final Collection<Relationship> relationships = getRelationships(); final Collection<Relationship> relationships = getRelationships();
if (name == null && CollectionUtils.isNotEmpty(relationships)) { if (name == null && CollectionUtils.isNotEmpty(relationships)) {
@ -119,7 +132,27 @@ public final class StandardConnection implements Connection {
name = "Connection"; name = "Connection";
} }
return ResourceFactory.getComponentResource(ResourceType.Connection, getIdentifier(), name); return name;
}
};
}
@Override
public AuthorizationResult checkAuthorization(Authorizer authorizer, RequestAction action, NiFiUser user, Map<String, String> resourceContext) {
// check the source
final AuthorizationResult sourceResult = getSource().checkAuthorization(authorizer, action, user, resourceContext);
if (Result.Denied.equals(sourceResult.getResult())) {
return sourceResult;
}
// check the destination
return getDestination().checkAuthorization(authorizer, action, user, resourceContext);
}
@Override
public void authorize(Authorizer authorizer, RequestAction action, NiFiUser user, Map<String, String> resourceContext) throws AccessDeniedException {
getSource().authorize(authorizer, action, user, resourceContext);
getDestination().authorize(authorizer, action, user, resourceContext);
} }
@Override @Override

View File

@ -16,39 +16,8 @@
*/ */
package org.apache.nifi.controller; package org.apache.nifi.controller;
import static java.util.Objects.requireNonNull; import com.sun.jersey.api.client.ClientHandlerException;
import org.apache.commons.collections4.Predicate;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.net.ssl.SSLContext;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.action.Action; import org.apache.nifi.action.Action;
import org.apache.nifi.admin.service.AuditService; import org.apache.nifi.admin.service.AuditService;
@ -59,6 +28,7 @@ import org.apache.nifi.annotation.lifecycle.OnShutdown;
import org.apache.nifi.annotation.notification.OnPrimaryNodeStateChange; import org.apache.nifi.annotation.notification.OnPrimaryNodeStateChange;
import org.apache.nifi.annotation.notification.PrimaryNodeState; import org.apache.nifi.annotation.notification.PrimaryNodeState;
import org.apache.nifi.authorization.Authorizer; import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.Resource; import org.apache.nifi.authorization.Resource;
import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.resource.ProvenanceEventAuthorizable; import org.apache.nifi.authorization.resource.ProvenanceEventAuthorizable;
@ -183,10 +153,10 @@ import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.SimpleProcessLogger; import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessorInitializationContext; import org.apache.nifi.processor.StandardProcessorInitializationContext;
import org.apache.nifi.processor.StandardValidationContextFactory; import org.apache.nifi.processor.StandardValidationContextFactory;
import org.apache.nifi.provenance.ProvenanceRepository; import org.apache.nifi.provenance.ProvenanceAuthorizableFactory;
import org.apache.nifi.provenance.ProvenanceEventRecord; import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType; import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.ProvenanceAuthorizableFactory; import org.apache.nifi.provenance.ProvenanceRepository;
import org.apache.nifi.provenance.StandardProvenanceEventRecord; import org.apache.nifi.provenance.StandardProvenanceEventRecord;
import org.apache.nifi.remote.HttpRemoteSiteListener; import org.apache.nifi.remote.HttpRemoteSiteListener;
import org.apache.nifi.remote.RemoteGroupPort; import org.apache.nifi.remote.RemoteGroupPort;
@ -236,7 +206,37 @@ import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.sun.jersey.api.client.ClientHandlerException; import javax.net.ssl.SSLContext;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static java.util.Objects.requireNonNull;
public class FlowController implements EventAccess, ControllerServiceProvider, ReportingTaskProvider, QueueProvider, Authorizable, ProvenanceAuthorizableFactory, NodeTypeProvider { public class FlowController implements EventAccess, ControllerServiceProvider, ReportingTaskProvider, QueueProvider, Authorizable, ProvenanceAuthorizableFactory, NodeTypeProvider {
@ -2128,28 +2128,85 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
return root == null ? null : root.findProcessGroup(searchId); return root == null ? null : root.findProcessGroup(searchId);
} }
/**
* Returns the status of all components in the controller. This request is not in the context of a user so the results will be unfiltered.
*
* @return the component status
*/
@Override @Override
public ProcessGroupStatus getControllerStatus() { public ProcessGroupStatus getControllerStatus() {
return getGroupStatus(getRootGroupId()); return getGroupStatus(getRootGroupId());
} }
/**
* Returns the status of all components in the specified group. This request is not in the context of a user so the results will be unfiltered.
*
* @param groupId group id
* @return the component status
*/
public ProcessGroupStatus getGroupStatus(final String groupId) { public ProcessGroupStatus getGroupStatus(final String groupId) {
return getGroupStatus(groupId, getProcessorStats()); return getGroupStatus(groupId, getProcessorStats());
} }
public ProcessGroupStatus getGroupStatus(final String groupId, final RepositoryStatusReport statusReport) { /**
final ProcessGroup group = getGroup(groupId); * Returns the status for components in the specified group. This request is made by the specified user so the results will be filtered accordingly.
return getGroupStatus(group, statusReport); *
* @param groupId group id
* @param user user making request
* @return the component status
*/
public ProcessGroupStatus getGroupStatus(final String groupId, final NiFiUser user) {
return getGroupStatus(groupId, getProcessorStats(), user);
} }
public ProcessGroupStatus getGroupStatus(final ProcessGroup group, final RepositoryStatusReport statusReport) { /**
* Returns the status for the components in the specified group with the specified report. This request is not in the context of a user so the results
* will be unfiltered.
*
* @param groupId group id
* @param statusReport report
* @return the component status
*/
public ProcessGroupStatus getGroupStatus(final String groupId, final RepositoryStatusReport statusReport) {
final ProcessGroup group = getGroup(groupId);
// this was invoked with no user context so the results will be unfiltered... necessary for aggregating status history
return getGroupStatus(group, statusReport, authorizable -> true);
}
/**
* Returns the status for the components in the specified group with the specified report. This request is made by the specified user
* so the results will be filtered accordingly.
*
* @param groupId group id
* @param statusReport report
* @param user user making request
* @return the component status
*/
public ProcessGroupStatus getGroupStatus(final String groupId, final RepositoryStatusReport statusReport, final NiFiUser user) {
final ProcessGroup group = getGroup(groupId);
// on demand status request for a specific user... require authorization per component and filter results as appropriate
return getGroupStatus(group, statusReport, authorizable -> authorizable.isAuthorized(authorizer, RequestAction.READ, user));
}
/**
* Returns the status for the components in the specified group with the specified report. The results will be filtered by executing
* the specified predicate.
*
* @param group group id
* @param statusReport report
* @param isAuthorized is authorized check
* @return the component status
*/
public ProcessGroupStatus getGroupStatus(final ProcessGroup group, final RepositoryStatusReport statusReport, final Predicate<Authorizable> isAuthorized) {
if (group == null) { if (group == null) {
return null; return null;
} }
final ProcessGroupStatus status = new ProcessGroupStatus(); final ProcessGroupStatus status = new ProcessGroupStatus();
status.setId(group.getIdentifier()); status.setId(group.getIdentifier());
status.setName(group.getName()); status.setName(isAuthorized.evaluate(group) ? group.getName() : group.getIdentifier());
int activeGroupThreads = 0; int activeGroupThreads = 0;
long bytesRead = 0L; long bytesRead = 0L;
long bytesWritten = 0L; long bytesWritten = 0L;
@ -2170,7 +2227,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final Collection<ProcessorStatus> processorStatusCollection = new ArrayList<>(); final Collection<ProcessorStatus> processorStatusCollection = new ArrayList<>();
status.setProcessorStatus(processorStatusCollection); status.setProcessorStatus(processorStatusCollection);
for (final ProcessorNode procNode : group.getProcessors()) { for (final ProcessorNode procNode : group.getProcessors()) {
final ProcessorStatus procStat = getProcessorStatus(statusReport, procNode); final ProcessorStatus procStat = getProcessorStatus(statusReport, procNode, isAuthorized);
processorStatusCollection.add(procStat); processorStatusCollection.add(procStat);
activeGroupThreads += procStat.getActiveThreadCount(); activeGroupThreads += procStat.getActiveThreadCount();
bytesRead += procStat.getBytesRead(); bytesRead += procStat.getBytesRead();
@ -2186,7 +2243,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final Collection<ProcessGroupStatus> localChildGroupStatusCollection = new ArrayList<>(); final Collection<ProcessGroupStatus> localChildGroupStatusCollection = new ArrayList<>();
status.setProcessGroupStatus(localChildGroupStatusCollection); status.setProcessGroupStatus(localChildGroupStatusCollection);
for (final ProcessGroup childGroup : group.getProcessGroups()) { for (final ProcessGroup childGroup : group.getProcessGroups()) {
final ProcessGroupStatus childGroupStatus = getGroupStatus(childGroup, statusReport); final ProcessGroupStatus childGroupStatus = getGroupStatus(childGroup, statusReport, isAuthorized);
localChildGroupStatusCollection.add(childGroupStatus); localChildGroupStatusCollection.add(childGroupStatus);
activeGroupThreads += childGroupStatus.getActiveThreadCount(); activeGroupThreads += childGroupStatus.getActiveThreadCount();
bytesRead += childGroupStatus.getBytesRead(); bytesRead += childGroupStatus.getBytesRead();
@ -2207,7 +2264,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final Collection<RemoteProcessGroupStatus> remoteProcessGroupStatusCollection = new ArrayList<>(); final Collection<RemoteProcessGroupStatus> remoteProcessGroupStatusCollection = new ArrayList<>();
status.setRemoteProcessGroupStatus(remoteProcessGroupStatusCollection); status.setRemoteProcessGroupStatus(remoteProcessGroupStatusCollection);
for (final RemoteProcessGroup remoteGroup : group.getRemoteProcessGroups()) { for (final RemoteProcessGroup remoteGroup : group.getRemoteProcessGroups()) {
final RemoteProcessGroupStatus remoteStatus = createRemoteGroupStatus(remoteGroup, statusReport); final RemoteProcessGroupStatus remoteStatus = createRemoteGroupStatus(remoteGroup, statusReport, isAuthorized);
if (remoteStatus != null) { if (remoteStatus != null) {
remoteProcessGroupStatusCollection.add(remoteStatus); remoteProcessGroupStatusCollection.add(remoteStatus);
@ -2224,13 +2281,17 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
// get the connection and remote port status // get the connection and remote port status
for (final Connection conn : group.getConnections()) { for (final Connection conn : group.getConnections()) {
final boolean isConnectionAuthorized = isAuthorized.evaluate(conn);
final boolean isSourceAuthorized = isAuthorized.evaluate(conn.getSource());
final boolean isDestinationAuthorized = isAuthorized.evaluate(conn.getDestination());
final ConnectionStatus connStatus = new ConnectionStatus(); final ConnectionStatus connStatus = new ConnectionStatus();
connStatus.setId(conn.getIdentifier()); connStatus.setId(conn.getIdentifier());
connStatus.setGroupId(conn.getProcessGroup().getIdentifier()); connStatus.setGroupId(conn.getProcessGroup().getIdentifier());
connStatus.setSourceId(conn.getSource().getIdentifier()); connStatus.setSourceId(conn.getSource().getIdentifier());
connStatus.setSourceName(conn.getSource().getName()); connStatus.setSourceName(isSourceAuthorized ? conn.getSource().getName() : conn.getSource().getIdentifier());
connStatus.setDestinationId(conn.getDestination().getIdentifier()); connStatus.setDestinationId(conn.getDestination().getIdentifier());
connStatus.setDestinationName(conn.getDestination().getName()); connStatus.setDestinationName(isDestinationAuthorized ? conn.getDestination().getName() : conn.getDestination().getIdentifier());
connStatus.setBackPressureDataSizeThreshold(conn.getFlowFileQueue().getBackPressureDataSizeThreshold()); connStatus.setBackPressureDataSizeThreshold(conn.getFlowFileQueue().getBackPressureDataSizeThreshold());
connStatus.setBackPressureObjectThreshold(conn.getFlowFileQueue().getBackPressureObjectThreshold()); connStatus.setBackPressureObjectThreshold(conn.getFlowFileQueue().getBackPressureObjectThreshold());
@ -2245,6 +2306,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
bytesTransferred += connectionStatusReport.getContentSizeIn() + connectionStatusReport.getContentSizeOut(); bytesTransferred += connectionStatusReport.getContentSizeIn() + connectionStatusReport.getContentSizeOut();
} }
if (isConnectionAuthorized) {
if (StringUtils.isNotBlank(conn.getName())) { if (StringUtils.isNotBlank(conn.getName())) {
connStatus.setName(conn.getName()); connStatus.setName(conn.getName());
} else if (conn.getRelationships() != null && !conn.getRelationships().isEmpty()) { } else if (conn.getRelationships() != null && !conn.getRelationships().isEmpty()) {
@ -2254,6 +2316,9 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
} }
connStatus.setName(StringUtils.join(relationships, ", ")); connStatus.setName(StringUtils.join(relationships, ", "));
} }
} else {
connStatus.setName(conn.getIdentifier());
}
final QueueSize queueSize = conn.getFlowFileQueue().size(); final QueueSize queueSize = conn.getFlowFileQueue().size();
final int connectionQueuedCount = queueSize.getObjectCount(); final int connectionQueuedCount = queueSize.getObjectCount();
@ -2285,10 +2350,12 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final Set<Port> inputPorts = group.getInputPorts(); final Set<Port> inputPorts = group.getInputPorts();
for (final Port port : inputPorts) { for (final Port port : inputPorts) {
final boolean isInputPortAuthorized = isAuthorized.evaluate(port);
final PortStatus portStatus = new PortStatus(); final PortStatus portStatus = new PortStatus();
portStatus.setId(port.getIdentifier()); portStatus.setId(port.getIdentifier());
portStatus.setGroupId(port.getProcessGroup().getIdentifier()); portStatus.setGroupId(port.getProcessGroup().getIdentifier());
portStatus.setName(port.getName()); portStatus.setName(isInputPortAuthorized ? port.getName() : port.getIdentifier());
portStatus.setActiveThreadCount(processScheduler.getActiveThreadCount(port)); portStatus.setActiveThreadCount(processScheduler.getActiveThreadCount(port));
// determine the run status // determine the run status
@ -2343,10 +2410,12 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final Set<Port> outputPorts = group.getOutputPorts(); final Set<Port> outputPorts = group.getOutputPorts();
for (final Port port : outputPorts) { for (final Port port : outputPorts) {
final boolean isOutputPortAuthorized = isAuthorized.evaluate(port);
final PortStatus portStatus = new PortStatus(); final PortStatus portStatus = new PortStatus();
portStatus.setId(port.getIdentifier()); portStatus.setId(port.getIdentifier());
portStatus.setGroupId(port.getProcessGroup().getIdentifier()); portStatus.setGroupId(port.getProcessGroup().getIdentifier());
portStatus.setName(port.getName()); portStatus.setName(isOutputPortAuthorized ? port.getName() : port.getIdentifier());
portStatus.setActiveThreadCount(processScheduler.getActiveThreadCount(port)); portStatus.setActiveThreadCount(processScheduler.getActiveThreadCount(port));
// determine the run status // determine the run status
@ -2419,7 +2488,9 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
return status; return status;
} }
private RemoteProcessGroupStatus createRemoteGroupStatus(final RemoteProcessGroup remoteGroup, final RepositoryStatusReport statusReport) { private RemoteProcessGroupStatus createRemoteGroupStatus(final RemoteProcessGroup remoteGroup, final RepositoryStatusReport statusReport, final Predicate<Authorizable> isAuthorized) {
final boolean isRemoteProcessGroupAuthorized = isAuthorized.evaluate(remoteGroup);
int receivedCount = 0; int receivedCount = 0;
long receivedContentSize = 0L; long receivedContentSize = 0L;
int sentCount = 0; int sentCount = 0;
@ -2430,8 +2501,8 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
final RemoteProcessGroupStatus status = new RemoteProcessGroupStatus(); final RemoteProcessGroupStatus status = new RemoteProcessGroupStatus();
status.setGroupId(remoteGroup.getProcessGroup().getIdentifier()); status.setGroupId(remoteGroup.getProcessGroup().getIdentifier());
status.setName(remoteGroup.getName()); status.setName(isRemoteProcessGroupAuthorized ? remoteGroup.getName() : remoteGroup.getIdentifier());
status.setTargetUri(remoteGroup.getTargetUri().toString()); status.setTargetUri(isRemoteProcessGroupAuthorized ? remoteGroup.getTargetUri().toString() : null);
long lineageMillis = 0L; long lineageMillis = 0L;
int flowFilesRemoved = 0; int flowFilesRemoved = 0;
@ -2499,12 +2570,14 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
return status; return status;
} }
private ProcessorStatus getProcessorStatus(final RepositoryStatusReport report, final ProcessorNode procNode) { private ProcessorStatus getProcessorStatus(final RepositoryStatusReport report, final ProcessorNode procNode, final Predicate<Authorizable> isAuthorized) {
final boolean isProcessorAuthorized = isAuthorized.evaluate(procNode);
final ProcessorStatus status = new ProcessorStatus(); final ProcessorStatus status = new ProcessorStatus();
status.setId(procNode.getIdentifier()); status.setId(procNode.getIdentifier());
status.setGroupId(procNode.getProcessGroup().getIdentifier()); status.setGroupId(procNode.getProcessGroup().getIdentifier());
status.setName(procNode.getName()); status.setName(isProcessorAuthorized ? procNode.getName() : procNode.getIdentifier());
status.setType(procNode.getComponentType()); status.setType(isProcessorAuthorized ? procNode.getComponentType() : "Processor");
final FlowFileEvent entry = report.getReportEntries().get(procNode.getIdentifier()); final FlowFileEvent entry = report.getReportEntries().get(procNode.getIdentifier());
if (entry == null) { if (entry == null) {

View File

@ -16,8 +16,6 @@
*/ */
package org.apache.nifi.controller.status.history; package org.apache.nifi.controller.status.history;
import java.util.Date;
import org.apache.nifi.controller.status.ConnectionStatus; import org.apache.nifi.controller.status.ConnectionStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.controller.status.ProcessorStatus;
@ -30,6 +28,8 @@ import org.apache.nifi.util.RingBuffer.ForEachEvaluator;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.util.Date;
public class VolatileComponentStatusRepository implements ComponentStatusRepository { public class VolatileComponentStatusRepository implements ComponentStatusRepository {
public static final String NUM_DATA_POINTS_PROPERTY = "nifi.components.status.repository.buffer.size"; public static final String NUM_DATA_POINTS_PROPERTY = "nifi.components.status.repository.buffer.size";
@ -69,7 +69,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
@Override @Override
public StatusHistory getProcessorStatusHistory(final String processorId, final Date start, final Date end, final int preferredDataPoints) { public StatusHistory getProcessorStatusHistory(final String processorId, final Date start, final Date end, final int preferredDataPoints) {
final StandardStatusHistory history = new StandardStatusHistory(); final StandardStatusHistory history = new StandardStatusHistory();
history.setComponentDetail("Id", processorId); history.setComponentDetail(COMPONENT_DETAIL_ID, processorId);
captures.forEach(new ForEachEvaluator<Capture>() { captures.forEach(new ForEachEvaluator<Capture>() {
@Override @Override
@ -80,9 +80,9 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
return true; return true;
} }
history.setComponentDetail("Group Id", status.getGroupId()); history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId());
history.setComponentDetail("Name", status.getName()); history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName());
history.setComponentDetail("Type", status.getType()); history.setComponentDetail(COMPONENT_DETAIL_TYPE, status.getType());
final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); final StandardStatusSnapshot snapshot = new StandardStatusSnapshot();
snapshot.setTimestamp(capture.getCaptureDate()); snapshot.setTimestamp(capture.getCaptureDate());
@ -102,7 +102,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
@Override @Override
public StatusHistory getConnectionStatusHistory(final String connectionId, final Date start, final Date end, final int preferredDataPoints) { public StatusHistory getConnectionStatusHistory(final String connectionId, final Date start, final Date end, final int preferredDataPoints) {
final StandardStatusHistory history = new StandardStatusHistory(); final StandardStatusHistory history = new StandardStatusHistory();
history.setComponentDetail("Id", connectionId); history.setComponentDetail(COMPONENT_DETAIL_ID, connectionId);
captures.forEach(new ForEachEvaluator<Capture>() { captures.forEach(new ForEachEvaluator<Capture>() {
@Override @Override
@ -113,10 +113,10 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
return true; return true;
} }
history.setComponentDetail("Group Id", status.getGroupId()); history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId());
history.setComponentDetail("Name", status.getName()); history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName());
history.setComponentDetail("Source Name", status.getSourceName()); history.setComponentDetail(COMPONENT_DETAIL_SOURCE_NAME, status.getSourceName());
history.setComponentDetail("Destination Name", status.getDestinationName()); history.setComponentDetail(COMPONENT_DETAIL_DESTINATION_NAME, status.getDestinationName());
final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); final StandardStatusSnapshot snapshot = new StandardStatusSnapshot();
snapshot.setTimestamp(capture.getCaptureDate()); snapshot.setTimestamp(capture.getCaptureDate());
@ -136,7 +136,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
@Override @Override
public StatusHistory getProcessGroupStatusHistory(final String processGroupId, final Date start, final Date end, final int preferredDataPoints) { public StatusHistory getProcessGroupStatusHistory(final String processGroupId, final Date start, final Date end, final int preferredDataPoints) {
final StandardStatusHistory history = new StandardStatusHistory(); final StandardStatusHistory history = new StandardStatusHistory();
history.setComponentDetail("Id", processGroupId); history.setComponentDetail(COMPONENT_DETAIL_ID, processGroupId);
captures.forEach(new ForEachEvaluator<Capture>() { captures.forEach(new ForEachEvaluator<Capture>() {
@Override @Override
@ -147,7 +147,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
return true; return true;
} }
history.setComponentDetail("Name", status.getName()); history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName());
final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); final StandardStatusSnapshot snapshot = new StandardStatusSnapshot();
snapshot.setTimestamp(capture.getCaptureDate()); snapshot.setTimestamp(capture.getCaptureDate());
@ -167,7 +167,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
@Override @Override
public StatusHistory getRemoteProcessGroupStatusHistory(final String remoteGroupId, final Date start, final Date end, final int preferredDataPoints) { public StatusHistory getRemoteProcessGroupStatusHistory(final String remoteGroupId, final Date start, final Date end, final int preferredDataPoints) {
final StandardStatusHistory history = new StandardStatusHistory(); final StandardStatusHistory history = new StandardStatusHistory();
history.setComponentDetail("Id", remoteGroupId); history.setComponentDetail(COMPONENT_DETAIL_ID, remoteGroupId);
captures.forEach(new ForEachEvaluator<Capture>() { captures.forEach(new ForEachEvaluator<Capture>() {
@Override @Override
@ -178,9 +178,9 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
return true; return true;
} }
history.setComponentDetail("Group Id", status.getGroupId()); history.setComponentDetail(COMPONENT_DETAIL_GROUP_ID, status.getGroupId());
history.setComponentDetail("Name", status.getName()); history.setComponentDetail(COMPONENT_DETAIL_NAME, status.getName());
history.setComponentDetail("Uri", status.getTargetUri()); history.setComponentDetail(COMPONENT_DETAIL_URI, status.getTargetUri());
final StandardStatusSnapshot snapshot = new StandardStatusSnapshot(); final StandardStatusSnapshot snapshot = new StandardStatusSnapshot();
snapshot.setTimestamp(capture.getCaptureDate()); snapshot.setTimestamp(capture.getCaptureDate());

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.nifi.web; package org.apache.nifi.authorization;
import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.controller.Snippet; import org.apache.nifi.controller.Snippet;
@ -72,7 +72,7 @@ public interface AuthorizableLookup {
* @param id connection id * @param id connection id
* @return authorizable * @return authorizable
*/ */
Authorizable getConnection(String id); ConnectionAuthorizable getConnection(String id);
/** /**
* Get the authorizable ProcessGroup. * Get the authorizable ProcessGroup.

View File

@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.nifi.web; package org.apache.nifi.authorization;
public interface AuthorizeAccess { public interface AuthorizeAccess {
void authorize(AuthorizableLookup lookup); void authorize(AuthorizableLookup lookup);

View File

@ -0,0 +1,54 @@
/*
* 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 org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.groups.ProcessGroup;
/**
* Authorizable for a Connection and its Group, Source, and Destination.
*/
public interface ConnectionAuthorizable {
/**
* Returns the authorizable for this connection. Non null
*
* @return authorizable
*/
Authorizable getAuthorizable();
/**
* Returns the source.
*
* @return source
*/
Connectable getSource();
/**
* Returns the destination.
*
* @return destination
*/
Connectable getDestination();
/**
* Returns the parent process group.
*
* @return parent
*/
ProcessGroup getParentGroup();
}

View File

@ -14,11 +14,9 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.nifi.web; package org.apache.nifi.authorization;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.authorization.AccessPolicy;
import org.apache.nifi.authorization.Resource;
import org.apache.nifi.authorization.resource.AccessPolicyAuthorizable; import org.apache.nifi.authorization.resource.AccessPolicyAuthorizable;
import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.resource.DataTransferAuthorizable; import org.apache.nifi.authorization.resource.DataTransferAuthorizable;
@ -26,12 +24,15 @@ import org.apache.nifi.authorization.resource.ProvenanceEventAuthorizable;
import org.apache.nifi.authorization.resource.ResourceFactory; import org.apache.nifi.authorization.resource.ResourceFactory;
import org.apache.nifi.authorization.resource.ResourceType; import org.apache.nifi.authorization.resource.ResourceType;
import org.apache.nifi.authorization.resource.TenantAuthorizable; import org.apache.nifi.authorization.resource.TenantAuthorizable;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.controller.ConfiguredComponent; import org.apache.nifi.controller.ConfiguredComponent;
import org.apache.nifi.controller.Snippet; import org.apache.nifi.controller.Snippet;
import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceReference; import org.apache.nifi.controller.service.ControllerServiceReference;
import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup; import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.web.ResourceNotFoundException;
import org.apache.nifi.web.controller.ControllerFacade; import org.apache.nifi.web.controller.ControllerFacade;
import org.apache.nifi.web.dao.AccessPolicyDAO; import org.apache.nifi.web.dao.AccessPolicyDAO;
import org.apache.nifi.web.dao.ConnectionDAO; import org.apache.nifi.web.dao.ConnectionDAO;
@ -125,8 +126,29 @@ class StandardAuthorizableLookup implements AuthorizableLookup {
} }
@Override @Override
public Authorizable getConnection(final String id) { public ConnectionAuthorizable getConnection(final String id) {
return connectionDAO.getConnection(id); final Connection connection = connectionDAO.getConnection(id);
return new ConnectionAuthorizable() {
@Override
public Authorizable getAuthorizable() {
return connection;
}
@Override
public Connectable getSource() {
return connection.getSource();
}
@Override
public Connectable getDestination() {
return connection.getDestination();
}
@Override
public ProcessGroup getParentGroup() {
return connection.getProcessGroup();
}
};
} }
@Override @Override
@ -288,9 +310,6 @@ class StandardAuthorizableLookup implements AuthorizableLookup {
private Authorizable getAccessPolicyByResource(final ResourceType resourceType, final String componentId) { private Authorizable getAccessPolicyByResource(final ResourceType resourceType, final String componentId) {
Authorizable authorizable = null; Authorizable authorizable = null;
switch (resourceType) { switch (resourceType) {
case Connection:
authorizable = getConnection(componentId);
break;
case ControllerService: case ControllerService:
authorizable = getControllerService(componentId); authorizable = getControllerService(componentId);
break; break;

View File

@ -16,6 +16,7 @@
*/ */
package org.apache.nifi.web; package org.apache.nifi.web;
import org.apache.nifi.authorization.AuthorizeAccess;
import org.apache.nifi.authorization.RequestAction; import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.user.NiFiUser; import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.controller.ScheduledState; import org.apache.nifi.controller.ScheduledState;

View File

@ -16,11 +16,12 @@
*/ */
package org.apache.nifi.web; package org.apache.nifi.web;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect; import org.aspectj.lang.annotation.Aspect;
import java.util.concurrent.locks.ReentrantReadWriteLock;
/** /**
* Aspect to limit access into the core. * Aspect to limit access into the core.
*/ */
@ -142,6 +143,17 @@ public class NiFiServiceFacadeLock {
} }
} }
@Around("within(org.apache.nifi.web.NiFiServiceFacade+) && "
+ "execution(* schedule*(..))")
public Object scheduleLock(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
writeLock.lock();
try {
return proceedingJoinPoint.proceed();
} finally {
writeLock.unlock();
}
}
@Around("within(org.apache.nifi.web.NiFiServiceFacade+) && " @Around("within(org.apache.nifi.web.NiFiServiceFacade+) && "
+ "execution(* get*(..))") + "execution(* get*(..))")
public Object getLock(ProceedingJoinPoint proceedingJoinPoint) throws Throwable { public Object getLock(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {

View File

@ -23,10 +23,13 @@ import org.apache.nifi.action.FlowChangeAction;
import org.apache.nifi.action.Operation; import org.apache.nifi.action.Operation;
import org.apache.nifi.action.details.FlowChangePurgeDetails; import org.apache.nifi.action.details.FlowChangePurgeDetails;
import org.apache.nifi.admin.service.AuditService; import org.apache.nifi.admin.service.AuditService;
import org.apache.nifi.authorization.AbstractPolicyBasedAuthorizer;
import org.apache.nifi.authorization.AccessDeniedException; import org.apache.nifi.authorization.AccessDeniedException;
import org.apache.nifi.authorization.AccessPolicy; import org.apache.nifi.authorization.AccessPolicy;
import org.apache.nifi.authorization.AuthorizableLookup;
import org.apache.nifi.authorization.AuthorizationResult; import org.apache.nifi.authorization.AuthorizationResult;
import org.apache.nifi.authorization.AuthorizationResult.Result; import org.apache.nifi.authorization.AuthorizationResult.Result;
import org.apache.nifi.authorization.AuthorizeAccess;
import org.apache.nifi.authorization.Authorizer; import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.Group; import org.apache.nifi.authorization.Group;
import org.apache.nifi.authorization.RequestAction; import org.apache.nifi.authorization.RequestAction;
@ -747,6 +750,9 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
updatedRevisions.put(revision.getComponentId(), currentRevision.incrementRevision(revision.getClientId())); updatedRevisions.put(revision.getComponentId(), currentRevision.incrementRevision(revision.getClientId()));
} }
// save
controllerFacade.save();
// gather details for response // gather details for response
final ScheduleComponentsEntity entity = new ScheduleComponentsEntity(); final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
entity.setId(processGroupId); entity.setId(processGroupId);
@ -903,11 +909,22 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
.map(g -> g.getIdentifier()).map(mapUserGroupIdToTenantEntity()).collect(Collectors.toSet()) : null; .map(g -> g.getIdentifier()).map(mapUserGroupIdToTenantEntity()).collect(Collectors.toSet()) : null;
final Set<AccessPolicySummaryEntity> policyEntities = user != null ? userGroupDAO.getAccessPoliciesForUser(userId).stream() final Set<AccessPolicySummaryEntity> policyEntities = user != null ? userGroupDAO.getAccessPoliciesForUser(userId).stream()
.map(ap -> createAccessPolicySummaryEntity(ap)).collect(Collectors.toSet()) : null; .map(ap -> createAccessPolicySummaryEntity(ap)).collect(Collectors.toSet()) : null;
final UserDTO snapshot = deleteComponent(
revision, final RevisionClaim claim = new StandardRevisionClaim(revision);
authorizableLookup.getTenant(), final NiFiUser nifiUser = NiFiUserUtils.getNiFiUser();
() -> userDAO.deleteUser(userId),
dtoFactory.createUserDto(user, userGroups, policyEntities)); // perform the deletion
final UserDTO snapshot = revisionManager.deleteRevision(claim, nifiUser, () -> {
logger.debug("Attempting to delete component {} with claim {}", user, claim);
userDAO.deleteUser(userId);
// save the flow
controllerFacade.save();
logger.debug("Deletion of component {} was successful", user);
return dtoFactory.createUserDto(user, userGroups, policyEntities);
});
return entityFactory.createUserEntity(snapshot, null, null); return entityFactory.createUserEntity(snapshot, null, null);
} }
@ -918,11 +935,22 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
final Set<TenantEntity> users = userGroup != null ? userGroup.getUsers().stream() final Set<TenantEntity> users = userGroup != null ? userGroup.getUsers().stream()
.map(mapUserIdToTenantEntity()).collect(Collectors.toSet()) : .map(mapUserIdToTenantEntity()).collect(Collectors.toSet()) :
null; null;
final UserGroupDTO snapshot = deleteComponent(
revision, final RevisionClaim claim = new StandardRevisionClaim(revision);
authorizableLookup.getTenant(), final NiFiUser nifiUser = NiFiUserUtils.getNiFiUser();
() -> userGroupDAO.deleteUserGroup(userGroupId),
dtoFactory.createUserGroupDto(userGroup, users)); // perform the deletion
final UserGroupDTO snapshot = revisionManager.deleteRevision(claim, nifiUser, () -> {
logger.debug("Attempting to delete component {} with claim {}", userGroup, claim);
userGroupDAO.deleteUserGroup(userGroupId);
// save the flow
controllerFacade.save();
logger.debug("Deletion of component {} was successful", userGroup);
return dtoFactory.createUserGroupDto(userGroup, users);
});
return entityFactory.createUserGroupEntity(snapshot, null, null); return entityFactory.createUserGroupEntity(snapshot, null, null);
} }
@ -976,18 +1004,30 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
controllerFacade.save(); controllerFacade.save();
logger.debug("Deletion of component {} was successful", authorizable); logger.debug("Deletion of component {} was successful", authorizable);
// if configured with a policy based authorizer, attempt to remove the corresponding policies
if (authorizer instanceof AbstractPolicyBasedAuthorizer) {
try { try {
// since the component is being deleted, also delete any relevant access policies // since the component is being deleted, also delete any relevant read access policies
final AccessPolicy readPolicy = accessPolicyDAO.getAccessPolicy(RequestAction.READ, authorizable); final AccessPolicy readPolicy = accessPolicyDAO.getAccessPolicy(RequestAction.READ, authorizable);
if (authorizable.getResource().getIdentifier().equals(readPolicy.getResource())) { if (authorizable.getResource().getIdentifier().equals(readPolicy.getResource())) {
accessPolicyDAO.deleteAccessPolicy(readPolicy.getIdentifier()); accessPolicyDAO.deleteAccessPolicy(readPolicy.getIdentifier());
} }
} catch (final ResourceNotFoundException e) {
// no policy exists for this component... no worries
} catch (final Exception e) {
logger.warn(String.format("Unable to remove access policy for %s %s after component removal.", RequestAction.READ, authorizable.getResource().getIdentifier()), e);
}
try {
// since the component is being deleted, also delete any relevant write access policies
final AccessPolicy writePolicy = accessPolicyDAO.getAccessPolicy(RequestAction.WRITE, authorizable); final AccessPolicy writePolicy = accessPolicyDAO.getAccessPolicy(RequestAction.WRITE, authorizable);
if (authorizable.getResource().getIdentifier().equals(writePolicy.getResource())) { if (authorizable.getResource().getIdentifier().equals(writePolicy.getResource())) {
accessPolicyDAO.deleteAccessPolicy(writePolicy.getIdentifier()); accessPolicyDAO.deleteAccessPolicy(writePolicy.getIdentifier());
} }
} catch (final ResourceNotFoundException e) {
// no policy exists for this component... no worries
} catch (final Exception e) { } catch (final Exception e) {
logger.warn(String.format("Unable to remove access policy for %s after component removal.", authorizable.getResource().getIdentifier()), e); logger.warn(String.format("Unable to remove access policy for %s %s after component removal.", RequestAction.WRITE, authorizable.getResource().getIdentifier()), e);
}
} }
return dto; return dto;
@ -2660,7 +2700,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
authorizable = authorizableLookup.getFunnel(sourceId); authorizable = authorizableLookup.getFunnel(sourceId);
break; break;
case Connection: case Connection:
authorizable = authorizableLookup.getConnection(sourceId); authorizable = authorizableLookup.getConnection(sourceId).getAuthorizable();
break; break;
case AccessPolicy: case AccessPolicy:
authorizable = authorizableLookup.getAccessPolicyById(sourceId); authorizable = authorizableLookup.getAccessPolicyById(sourceId);

View File

@ -42,8 +42,8 @@ import org.apache.nifi.remote.protocol.ResponseCode;
import org.apache.nifi.remote.protocol.http.HttpHeaders; import org.apache.nifi.remote.protocol.http.HttpHeaders;
import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.TypeOneUUIDGenerator; import org.apache.nifi.util.TypeOneUUIDGenerator;
import org.apache.nifi.web.AuthorizableLookup; import org.apache.nifi.authorization.AuthorizableLookup;
import org.apache.nifi.web.AuthorizeAccess; import org.apache.nifi.authorization.AuthorizeAccess;
import org.apache.nifi.web.NiFiServiceFacade; import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision; import org.apache.nifi.web.Revision;
import org.apache.nifi.web.api.dto.RevisionDTO; import org.apache.nifi.web.api.dto.RevisionDTO;
@ -420,8 +420,7 @@ public abstract class ApplicationResource {
snippet.getProcessors().keySet().stream().map(id -> lookup.getProcessor(id)).forEach(authorize); snippet.getProcessors().keySet().stream().map(id -> lookup.getProcessor(id)).forEach(authorize);
snippet.getInputPorts().keySet().stream().map(id -> lookup.getInputPort(id)).forEach(authorize); snippet.getInputPorts().keySet().stream().map(id -> lookup.getInputPort(id)).forEach(authorize);
snippet.getOutputPorts().keySet().stream().map(id -> lookup.getOutputPort(id)).forEach(authorize); snippet.getOutputPorts().keySet().stream().map(id -> lookup.getOutputPort(id)).forEach(authorize);
snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(authorize); snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(connAuth -> authorize.accept(connAuth.getAuthorizable()));
snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(authorize);
snippet.getFunnels().keySet().stream().map(id -> lookup.getFunnel(id)).forEach(authorize); snippet.getFunnels().keySet().stream().map(id -> lookup.getFunnel(id)).forEach(authorize);
} }
@ -440,8 +439,7 @@ public abstract class ApplicationResource {
snippet.getProcessors().keySet().stream().map(id -> lookup.getProcessor(id)).forEach(authorize); snippet.getProcessors().keySet().stream().map(id -> lookup.getProcessor(id)).forEach(authorize);
snippet.getInputPorts().keySet().stream().map(id -> lookup.getInputPort(id)).forEach(authorize); snippet.getInputPorts().keySet().stream().map(id -> lookup.getInputPort(id)).forEach(authorize);
snippet.getOutputPorts().keySet().stream().map(id -> lookup.getOutputPort(id)).forEach(authorize); snippet.getOutputPorts().keySet().stream().map(id -> lookup.getOutputPort(id)).forEach(authorize);
snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(authorize); snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(connAuth -> authorize.accept(connAuth.getAuthorizable()));
snippet.getConnections().keySet().stream().map(id -> lookup.getConnection(id)).forEach(authorize);
snippet.getFunnels().keySet().stream().map(id -> lookup.getFunnel(id)).forEach(authorize); snippet.getFunnels().keySet().stream().map(id -> lookup.getFunnel(id)).forEach(authorize);
} }

View File

@ -24,12 +24,16 @@ import com.wordnik.swagger.annotations.ApiResponses;
import com.wordnik.swagger.annotations.Authorization; import com.wordnik.swagger.annotations.Authorization;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.authorization.Authorizer; import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.ConnectionAuthorizable;
import org.apache.nifi.authorization.RequestAction; import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.resource.Authorizable; import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.user.NiFiUserUtils; import org.apache.nifi.authorization.user.NiFiUserUtils;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.web.NiFiServiceFacade; import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision; import org.apache.nifi.web.Revision;
import org.apache.nifi.web.api.dto.ConnectionDTO; import org.apache.nifi.web.api.dto.ConnectionDTO;
import org.apache.nifi.web.api.dto.FlowFileSummaryDTO;
import org.apache.nifi.web.api.dto.ListingRequestDTO;
import org.apache.nifi.web.api.entity.ConnectionEntity; import org.apache.nifi.web.api.entity.ConnectionEntity;
import org.apache.nifi.web.api.request.ClientIdParameter; import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.LongParameter; import org.apache.nifi.web.api.request.LongParameter;
@ -87,6 +91,38 @@ public class ConnectionResource extends ApplicationResource {
return connectionEntity; return connectionEntity;
} }
/**
* Populate the URIs for the specified flowfile listing.
*
* @param connectionId connection
* @param flowFileListing flowfile listing
* @return dto
*/
public ListingRequestDTO populateRemainingFlowFileListingContent(final String connectionId, final ListingRequestDTO flowFileListing) {
// uri of the listing
flowFileListing.setUri(generateResourceUri("connections", connectionId, "listing-requests", flowFileListing.getId()));
// uri of each flowfile
if (flowFileListing.getFlowFileSummaries() != null) {
for (final FlowFileSummaryDTO flowFile : flowFileListing.getFlowFileSummaries()) {
populateRemainingFlowFileContent(connectionId, flowFile);
}
}
return flowFileListing;
}
/**
* Populate the URIs for the specified flowfile.
*
* @param connectionId the connection id
* @param flowFile the flowfile
* @return the dto
*/
public FlowFileSummaryDTO populateRemainingFlowFileContent(final String connectionId, final FlowFileSummaryDTO flowFile) {
flowFile.setUri(generateResourceUri("connections", connectionId, "flowfiles", flowFile.getUuid()));
return flowFile;
}
/** /**
* Retrieves the specified connection. * Retrieves the specified connection.
* *
@ -130,8 +166,9 @@ public class ConnectionResource extends ApplicationResource {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable conn = lookup.getConnection(id); // ensure read access to this connection (checks source and destination)
conn.authorize(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser()); final Authorizable authorizable = lookup.getConnection(id).getAuthorizable();
authorizable.authorize(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser());
}); });
// get the specified relationship // get the specified relationship
@ -200,6 +237,10 @@ public class ConnectionResource extends ApplicationResource {
+ "requested resource (%s).", connection.getId(), id)); + "requested resource (%s).", connection.getId(), id));
} }
if (connection.getDestination() != null && connection.getDestination().getId() == null) {
throw new IllegalArgumentException("When specifying a destination component, the destination id is required.");
}
if (isReplicateRequest()) { if (isReplicateRequest()) {
return replicate(HttpMethod.PUT, connectionEntity); return replicate(HttpMethod.PUT, connectionEntity);
} }
@ -209,8 +250,20 @@ public class ConnectionResource extends ApplicationResource {
serviceFacade, serviceFacade,
revision, revision,
lookup -> { lookup -> {
Authorizable authorizable = lookup.getConnection(id); // verifies write access to this connection (this checks the current source and destination)
authorizable.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); ConnectionAuthorizable connAuth = lookup.getConnection(id);
connAuth.getAuthorizable().authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
// if a destination has been specified and is different
final Connectable currentDestination = connAuth.getDestination();
if (connection.getDestination() != null && currentDestination.getIdentifier().equals(connection.getDestination().getId())) {
// verify access of the new destination (current destination was already authorized as part of the connection check)
final Authorizable newDestinationAuthorizable = lookup.getConnectable(connection.getDestination().getId());
newDestinationAuthorizable.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
// verify access of the parent group (this is the same check that is performed when creating the connection)
connAuth.getParentGroup().authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}
}, },
() -> serviceFacade.verifyUpdateConnection(connection), () -> serviceFacade.verifyUpdateConnection(connection),
() -> { () -> {
@ -284,8 +337,9 @@ public class ConnectionResource extends ApplicationResource {
serviceFacade, serviceFacade,
revision, revision,
lookup -> { lookup -> {
final Authorizable conn = lookup.getConnection(id); // verifies write access to the source and destination
conn.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); final Authorizable authorizable = lookup.getConnection(id).getAuthorizable();
authorizable.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}, },
() -> serviceFacade.verifyDeleteConnection(id), () -> serviceFacade.verifyDeleteConnection(id),
() -> { () -> {

View File

@ -16,29 +16,12 @@
*/ */
package org.apache.nifi.web.api; package org.apache.nifi.web.api;
import java.io.IOException; import com.wordnik.swagger.annotations.Api;
import java.io.InputStream; import com.wordnik.swagger.annotations.ApiOperation;
import java.io.OutputStream; import com.wordnik.swagger.annotations.ApiParam;
import java.net.URI; import com.wordnik.swagger.annotations.ApiResponse;
import com.wordnik.swagger.annotations.ApiResponses;
import javax.servlet.http.HttpServletRequest; import com.wordnik.swagger.annotations.Authorization;
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.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.StreamingOutput;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.authorization.Authorizer; import org.apache.nifi.authorization.Authorizer;
import org.apache.nifi.authorization.RequestAction; import org.apache.nifi.authorization.RequestAction;
@ -58,12 +41,27 @@ import org.apache.nifi.web.api.entity.FlowFileEntity;
import org.apache.nifi.web.api.entity.ListingRequestEntity; import org.apache.nifi.web.api.entity.ListingRequestEntity;
import org.apache.nifi.web.api.request.ClientIdParameter; import org.apache.nifi.web.api.request.ClientIdParameter;
import com.wordnik.swagger.annotations.Api; import javax.servlet.http.HttpServletRequest;
import com.wordnik.swagger.annotations.ApiOperation; import javax.ws.rs.Consumes;
import com.wordnik.swagger.annotations.ApiParam; import javax.ws.rs.DELETE;
import com.wordnik.swagger.annotations.ApiResponse; import javax.ws.rs.DefaultValue;
import com.wordnik.swagger.annotations.ApiResponses; import javax.ws.rs.GET;
import com.wordnik.swagger.annotations.Authorization; import javax.ws.rs.HttpMethod;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.StreamingOutput;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
/** /**
* RESTful endpoint for managing a flowfile queue. * RESTful endpoint for managing a flowfile queue.
@ -174,7 +172,7 @@ public class FlowFileQueueResource extends ApplicationResource {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
@ -259,7 +257,7 @@ public class FlowFileQueueResource extends ApplicationResource {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
@ -338,7 +336,7 @@ public class FlowFileQueueResource extends ApplicationResource {
if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(id); final Authorizable connection = lookup.getConnection(id).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
} }
@ -409,7 +407,7 @@ public class FlowFileQueueResource extends ApplicationResource {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
@ -475,7 +473,7 @@ public class FlowFileQueueResource extends ApplicationResource {
if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
} }
@ -545,7 +543,7 @@ public class FlowFileQueueResource extends ApplicationResource {
if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(id); final Authorizable connection = lookup.getConnection(id).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
} }
@ -615,7 +613,7 @@ public class FlowFileQueueResource extends ApplicationResource {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
@ -681,7 +679,7 @@ public class FlowFileQueueResource extends ApplicationResource {
if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
final Authorizable connection = lookup.getConnection(connectionId); final Authorizable connection = lookup.getConnection(connectionId).getAuthorizable();
connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); connection.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
} }

View File

@ -31,7 +31,7 @@ import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.user.NiFiUserUtils; import org.apache.nifi.authorization.user.NiFiUserUtils;
import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.controller.Snippet; import org.apache.nifi.controller.Snippet;
import org.apache.nifi.web.AuthorizableLookup; import org.apache.nifi.authorization.AuthorizableLookup;
import org.apache.nifi.web.NiFiServiceFacade; import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision; import org.apache.nifi.web.Revision;
import org.apache.nifi.web.api.dto.ConnectionDTO; import org.apache.nifi.web.api.dto.ConnectionDTO;
@ -1533,20 +1533,37 @@ public class ProcessGroupResource extends ApplicationResource {
} }
connectionEntity.getComponent().setParentGroupId(groupId); connectionEntity.getComponent().setParentGroupId(groupId);
// get the connection
final ConnectionDTO connection = connectionEntity.getComponent();
if (connection.getSource() == null || connection.getSource().getId() == null) {
throw new IllegalArgumentException("The source of the connection must be specified.");
}
if (connection.getDestination() == null || connection.getDestination().getId() == null) {
throw new IllegalArgumentException("The destination of the connection must be specified.");
}
if (isReplicateRequest()) { if (isReplicateRequest()) {
return replicate(HttpMethod.POST, connectionEntity); return replicate(HttpMethod.POST, connectionEntity);
} }
// get the connection
final ConnectionDTO connection = connectionEntity.getComponent();
// handle expects request (usually from the cluster manager) // handle expects request (usually from the cluster manager)
final boolean validationPhase = isValidationPhase(httpServletRequest); final boolean validationPhase = isValidationPhase(httpServletRequest);
if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) { if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
// authorize access // authorize access
serviceFacade.authorizeAccess(lookup -> { serviceFacade.authorizeAccess(lookup -> {
// ensure write access to the group
final Authorizable processGroup = lookup.getProcessGroup(groupId); final Authorizable processGroup = lookup.getProcessGroup(groupId);
processGroup.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); processGroup.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
// ensure write access to the source
final Authorizable source = lookup.getConnectable(connection.getSource().getId());
source.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
// ensure write access to the destination
final Authorizable destination = lookup.getConnectable(connection.getDestination().getId());
destination.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser());
}); });
} }
if (validationPhase) { if (validationPhase) {

View File

@ -631,9 +631,11 @@ public final class DtoFactory {
return null; return null;
} }
boolean isAuthorized = connectable.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser());
final ConnectableDTO dto = new ConnectableDTO(); final ConnectableDTO dto = new ConnectableDTO();
dto.setId(connectable.getIdentifier()); dto.setId(connectable.getIdentifier());
dto.setName(connectable.getName()); dto.setName(isAuthorized ? connectable.getName() : connectable.getIdentifier());
dto.setType(connectable.getConnectableType().name()); dto.setType(connectable.getConnectableType().name());
if (connectable instanceof RemoteGroupPort) { if (connectable instanceof RemoteGroupPort) {
@ -643,12 +645,16 @@ public final class DtoFactory {
dto.setRunning(remoteGroupPort.isTargetRunning()); dto.setRunning(remoteGroupPort.isTargetRunning());
dto.setTransmitting(remoteGroupPort.isRunning()); dto.setTransmitting(remoteGroupPort.isRunning());
dto.setExists(remoteGroupPort.getTargetExists()); dto.setExists(remoteGroupPort.getTargetExists());
if (isAuthorized) {
dto.setComments(remoteGroup.getComments()); dto.setComments(remoteGroup.getComments());
}
} else { } else {
dto.setGroupId(connectable.getProcessGroup().getIdentifier()); dto.setGroupId(connectable.getProcessGroup().getIdentifier());
dto.setRunning(connectable.isRunning()); dto.setRunning(connectable.isRunning());
if (isAuthorized) {
dto.setComments(connectable.getComments()); dto.setComments(connectable.getComments());
} }
}
return dto; return dto;
} }

View File

@ -78,6 +78,7 @@ public final class EntityFactory {
entity.setPermissions(permissions); entity.setPermissions(permissions);
entity.setStatus(status); entity.setStatus(status);
entity.setId(dto.getId()); entity.setId(dto.getId());
entity.setInputRequirement(dto.getInputRequirement());
entity.setPosition(dto.getPosition()); entity.setPosition(dto.getPosition());
if (permissions != null && permissions.getCanRead()) { if (permissions != null && permissions.getCanRead()) {
entity.setComponent(dto); entity.setComponent(dto);
@ -245,10 +246,13 @@ public final class EntityFactory {
entity.setPosition(dto.getPosition()); entity.setPosition(dto.getPosition());
entity.setBends(dto.getBends()); entity.setBends(dto.getBends());
entity.setLabelIndex(dto.getLabelIndex()); entity.setLabelIndex(dto.getLabelIndex());
entity.setzIndex(dto.getzIndex());
entity.setSourceId(dto.getSource().getId()); entity.setSourceId(dto.getSource().getId());
entity.setSourceGroupId(dto.getSource().getGroupId()); entity.setSourceGroupId(dto.getSource().getGroupId());
entity.setSourceType(dto.getSource().getType());
entity.setDestinationId(dto.getDestination().getId()); entity.setDestinationId(dto.getDestination().getId());
entity.setDestinationGroupId(dto.getDestination().getGroupId()); entity.setDestinationGroupId(dto.getDestination().getGroupId());
entity.setDestinationType(dto.getDestination().getType());
if (permissions != null && permissions.getCanRead()) { if (permissions != null && permissions.getCanRead()) {
entity.setComponent(dto); entity.setComponent(dto);
} }

View File

@ -57,6 +57,7 @@ import org.apache.nifi.controller.status.PortStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus; import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus; import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.controller.status.RemoteProcessGroupStatus; import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
import org.apache.nifi.controller.status.history.ComponentStatusRepository;
import org.apache.nifi.diagnostics.SystemDiagnostics; import org.apache.nifi.diagnostics.SystemDiagnostics;
import org.apache.nifi.flowfile.FlowFilePrioritizer; import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.flowfile.attributes.CoreAttributes;
@ -268,7 +269,15 @@ public class ControllerFacade implements Authorizable {
throw new ResourceNotFoundException(String.format("Unable to locate processor with id '%s'.", processorId)); throw new ResourceNotFoundException(String.format("Unable to locate processor with id '%s'.", processorId));
} }
return flowController.getProcessorStatusHistory(processorId); final StatusHistoryDTO statusHistory = flowController.getProcessorStatusHistory(processorId);
// if not authorized
if (!processor.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser())) {
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_NAME, processorId);
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_TYPE, "Processor");
}
return statusHistory;
} }
/** /**
@ -286,7 +295,16 @@ public class ControllerFacade implements Authorizable {
throw new ResourceNotFoundException(String.format("Unable to locate connection with id '%s'.", connectionId)); throw new ResourceNotFoundException(String.format("Unable to locate connection with id '%s'.", connectionId));
} }
return flowController.getConnectionStatusHistory(connectionId); final StatusHistoryDTO statusHistory = flowController.getConnectionStatusHistory(connectionId);
// if not authorized
if (!connection.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser())) {
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_NAME, connectionId);
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_SOURCE_NAME, connection.getSource().getIdentifier());
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_DESTINATION_NAME, connection.getDestination().getIdentifier());
}
return statusHistory;
} }
/** /**
@ -305,7 +323,14 @@ public class ControllerFacade implements Authorizable {
throw new ResourceNotFoundException(String.format("Unable to locate process group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate process group with id '%s'.", groupId));
} }
return flowController.getProcessGroupStatusHistory(groupId); final StatusHistoryDTO statusHistory = flowController.getProcessGroupStatusHistory(groupId);
// if not authorized
if (!group.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser())) {
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_NAME, groupId);
}
return statusHistory;
} }
/** /**
@ -323,7 +348,15 @@ public class ControllerFacade implements Authorizable {
throw new ResourceNotFoundException(String.format("Unable to locate remote process group with id '%s'.", remoteProcessGroupId)); throw new ResourceNotFoundException(String.format("Unable to locate remote process group with id '%s'.", remoteProcessGroupId));
} }
return flowController.getRemoteProcessGroupStatusHistory(remoteProcessGroupId); final StatusHistoryDTO statusHistory = flowController.getRemoteProcessGroupStatusHistory(remoteProcessGroupId);
// if not authorized
if (!remoteProcessGroup.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser())) {
statusHistory.getComponentDetails().put(ComponentStatusRepository.COMPONENT_DETAIL_NAME, remoteProcessGroupId);
statusHistory.getComponentDetails().remove(ComponentStatusRepository.COMPONENT_DETAIL_URI);
}
return statusHistory;
} }
/** /**
@ -512,10 +545,11 @@ public class ControllerFacade implements Authorizable {
* @return the status for the specified process group * @return the status for the specified process group
*/ */
public ProcessGroupStatus getProcessGroupStatus(final String groupId) { public ProcessGroupStatus getProcessGroupStatus(final String groupId) {
final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (processGroupStatus == null) { if (processGroupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
return processGroupStatus; return processGroupStatus;
} }
@ -536,7 +570,7 @@ public class ControllerFacade implements Authorizable {
// calculate the process group status // calculate the process group status
final String groupId = processor.getProcessGroup().getIdentifier(); final String groupId = processor.getProcessGroup().getIdentifier();
final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (processGroupStatus == null) { if (processGroupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
@ -566,7 +600,7 @@ public class ControllerFacade implements Authorizable {
// calculate the process group status // calculate the process group status
final String groupId = connection.getProcessGroup().getIdentifier(); final String groupId = connection.getProcessGroup().getIdentifier();
final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (processGroupStatus == null) { if (processGroupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
@ -595,7 +629,7 @@ public class ControllerFacade implements Authorizable {
} }
final String groupId = port.getProcessGroup().getIdentifier(); final String groupId = port.getProcessGroup().getIdentifier();
final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (processGroupStatus == null) { if (processGroupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
@ -624,7 +658,7 @@ public class ControllerFacade implements Authorizable {
} }
final String groupId = port.getProcessGroup().getIdentifier(); final String groupId = port.getProcessGroup().getIdentifier();
final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus processGroupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (processGroupStatus == null) { if (processGroupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
@ -653,7 +687,7 @@ public class ControllerFacade implements Authorizable {
} }
final String groupId = remoteProcessGroup.getProcessGroup().getIdentifier(); final String groupId = remoteProcessGroup.getProcessGroup().getIdentifier();
final ProcessGroupStatus groupStatus = flowController.getGroupStatus(groupId); final ProcessGroupStatus groupStatus = flowController.getGroupStatus(groupId, NiFiUserUtils.getNiFiUser());
if (groupStatus == null) { if (groupStatus == null) {
throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId)); throw new ResourceNotFoundException(String.format("Unable to locate group with id '%s'.", groupId));
} }
@ -736,12 +770,6 @@ public class ControllerFacade implements Authorizable {
resources.add(ResourceFactory.getProvenanceEventResource(processor.getResource())); resources.add(ResourceFactory.getProvenanceEventResource(processor.getResource()));
} }
// add each connection
for (final Connection connection : root.findAllConnections()) {
resources.add(ResourceFactory.getComponentResource(ResourceType.Connection, connection.getIdentifier(), connection.getName()));
resources.add(ResourceFactory.getFlowFileQueueResource(connection.getIdentifier(), connection.getName()));
}
// add each label // add each label
for (final Label label : root.findAllLabels()) { for (final Label label : root.findAllLabels()) {
resources.add(ResourceFactory.getComponentResource(ResourceType.Label, label.getIdentifier(), label.getValue())); resources.add(ResourceFactory.getComponentResource(ResourceType.Label, label.getIdentifier(), label.getValue()));

View File

@ -18,7 +18,6 @@ package org.apache.nifi.web.dao.impl;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.authorization.AccessDeniedException; import org.apache.nifi.authorization.AccessDeniedException;
import org.apache.nifi.authorization.AuthorizationRequest;
import org.apache.nifi.authorization.AuthorizationResult; import org.apache.nifi.authorization.AuthorizationResult;
import org.apache.nifi.authorization.AuthorizationResult.Result; import org.apache.nifi.authorization.AuthorizationResult.Result;
import org.apache.nifi.authorization.Authorizer; import org.apache.nifi.authorization.Authorizer;
@ -26,6 +25,7 @@ import org.apache.nifi.authorization.RequestAction;
import org.apache.nifi.authorization.UserContextKeys; import org.apache.nifi.authorization.UserContextKeys;
import org.apache.nifi.authorization.user.NiFiUser; import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.authorization.user.NiFiUserUtils; import org.apache.nifi.authorization.user.NiFiUserUtils;
import org.apache.nifi.authorization.user.StandardNiFiUser;
import org.apache.nifi.connectable.Connectable; import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType; import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection; import org.apache.nifi.connectable.Connection;
@ -611,18 +611,8 @@ public class StandardConnectionDAO extends ComponentDAO implements ConnectionDAO
userContext = null; userContext = null;
} }
final AuthorizationRequest request = new AuthorizationRequest.Builder() final NiFiUser chainUser = new StandardNiFiUser(identity, user.getClientAddress());
.identity(identity) final AuthorizationResult result = connection.checkAuthorization(authorizer, RequestAction.WRITE, chainUser, attributes);
.anonymous(user.isAnonymous())
.accessAttempt(false)
.action(RequestAction.WRITE)
.resource(connection.getResource())
.resourceContext(attributes)
.userContext(userContext)
.build();
// perform the authorization
final AuthorizationResult result = authorizer.authorize(request);
if (!Result.Approved.equals(result.getResult())) { if (!Result.Approved.equals(result.getResult())) {
throw new AccessDeniedException(result.getExplanation()); throw new AccessDeniedException(result.getExplanation());
} }

View File

@ -119,7 +119,7 @@
<property name="dtoFactory" ref="dtoFactory"/> <property name="dtoFactory" ref="dtoFactory"/>
<property name="bulletinRepository" ref="bulletinRepository"/> <property name="bulletinRepository" ref="bulletinRepository"/>
</bean> </bean>
<bean id="authorizableLookup" class="org.apache.nifi.web.StandardAuthorizableLookup"> <bean id="authorizableLookup" class="org.apache.nifi.authorization.StandardAuthorizableLookup">
<property name="controllerFacade" ref="controllerFacade"/> <property name="controllerFacade" ref="controllerFacade"/>
<property name="processorDAO" ref="processorDAO"/> <property name="processorDAO" ref="processorDAO"/>
<property name="inputPortDAO" ref="inputPortDAO"/> <property name="inputPortDAO" ref="inputPortDAO"/>

View File

@ -66,7 +66,7 @@ public class ITFlowAccessControl {
*/ */
@Test @Test
public void testGetIdentity() throws Exception { public void testGetIdentity() throws Exception {
helper.testGenericGetUri(helper.getBaseUrl() + "/flow/identity"); helper.testGenericGetUri(helper.getBaseUrl() + "/flow/current-user");
} }
/** /**

View File

@ -151,7 +151,7 @@
<md-menu-item layout-align="space-around center"> <md-menu-item layout-align="space-around center">
<a id="users-link" layout="row" <a id="users-link" layout="row"
ng-click="appCtrl.serviceProvider.headerCtrl.globalMenuCtrl.users.shell.launch();" ng-click="appCtrl.serviceProvider.headerCtrl.globalMenuCtrl.users.shell.launch();"
ng-class="{disabled: !appCtrl.nf.Common.canModifyTenants()}"> ng-class="{disabled: !appCtrl.nf.Common.canAccessTenants()}">
<i class="fa fa-users"></i>Users <i class="fa fa-users"></i>Users
</a> </a>
</md-menu-item> </md-menu-item>

View File

@ -15,7 +15,7 @@
limitations under the License. limitations under the License.
--%> --%>
<%@ page contentType="text/html" pageEncoding="UTF-8" session="false" %> <%@ page contentType="text/html" pageEncoding="UTF-8" session="false" %>
<div id="drop-request-status-dialog" layout="column" class="hidden large-dialog"> <div id="drop-request-status-dialog" layout="column" class="hidden small-dialog">
<div class="dialog-content"> <div class="dialog-content">
<div class="setting"> <div class="setting">
<div class="setting-field"> <div class="setting-field">

View File

@ -95,14 +95,14 @@
<div id="operation-buttons"> <div id="operation-buttons">
<div> <div>
<div id="operate-configure" class="action-button" title="Configuration"> <div id="operate-configure" class="action-button" title="Configuration">
<button ng-click="appCtrl.nf.Actions['showConfiguration'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.serviceProvider.graphControlsCtrl.openConfigureOrDetailsView();"
ng-disabled="false"> ng-disabled="!(appCtrl.serviceProvider.graphControlsCtrl.canConfigureOrOpenDetails())">
<div class="graph-control-action-icon fa fa-gear"></div></button> <div class="graph-control-action-icon fa fa-gear"></div></button>
</div> </div>
<div class="button-spacer-small">&nbsp;</div> <div class="button-spacer-small">&nbsp;</div>
<div id="operate-policy" class="action-button" title="Access Policies"> <div id="operate-policy" class="action-button" title="Access Policies">
<button ng-click="appCtrl.nf.Actions['managePolicies'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.nf.Actions['managePolicies'](appCtrl.nf.CanvasUtils.getSelection());"
ng-disabled="!(appCtrl.nf.CanvasUtils.getSelection().size() <= 1 && appCtrl.nf.Common.canAccessTenants())"> ng-disabled="!(appCtrl.serviceProvider.graphControlsCtrl.canManagePolicies())">
<div class="graph-control-action-icon fa fa-key"></div></button> <div class="graph-control-action-icon fa fa-key"></div></button>
</div> </div>
<div class="button-spacer-large">&nbsp;</div> <div class="button-spacer-large">&nbsp;</div>
@ -120,13 +120,13 @@
<div class="button-spacer-large">&nbsp;</div> <div class="button-spacer-large">&nbsp;</div>
<div id="operate-start" class="action-button" title="Start"> <div id="operate-start" class="action-button" title="Start">
<button ng-click="appCtrl.nf.Actions['start'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.nf.Actions['start'](appCtrl.nf.CanvasUtils.getSelection());"
ng-disabled="!appCtrl.nf.CanvasUtils.getSelection().empty() && !appCtrl.nf.CanvasUtils.canModify(appCtrl.nf.CanvasUtils.getSelection());"> ng-disabled="!appCtrl.nf.CanvasUtils.areRunnable(appCtrl.nf.CanvasUtils.getSelection());">
<div class="graph-control-action-icon fa fa-play"></div></button> <div class="graph-control-action-icon fa fa-play"></div></button>
</div> </div>
<div class="button-spacer-small">&nbsp;</div> <div class="button-spacer-small">&nbsp;</div>
<div id="operate-stop" class="action-button" title="Stop"> <div id="operate-stop" class="action-button" title="Stop">
<button ng-click="appCtrl.nf.Actions['stop'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.nf.Actions['stop'](appCtrl.nf.CanvasUtils.getSelection());"
ng-disabled="!appCtrl.nf.CanvasUtils.getSelection().empty() && !appCtrl.nf.CanvasUtils.canModify(appCtrl.nf.CanvasUtils.getSelection());"> ng-disabled="!appCtrl.nf.CanvasUtils.areStoppable(appCtrl.nf.CanvasUtils.getSelection());">
<div class="graph-control-action-icon fa fa-stop"></div></button> <div class="graph-control-action-icon fa fa-stop"></div></button>
</div> </div>
<div class="button-spacer-large">&nbsp;</div> <div class="button-spacer-large">&nbsp;</div>
@ -146,13 +146,13 @@
<div style="margin-top: 5px;"> <div style="margin-top: 5px;">
<div id="operate-copy" class="action-button" title="Copy"> <div id="operate-copy" class="action-button" title="Copy">
<button ng-click="appCtrl.nf.Actions['copy'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.nf.Actions['copy'](appCtrl.nf.CanvasUtils.getSelection());"
ng-disabled="!appCtrl.nf.CanvasUtils.isCopyable(appCtrl.nf.CanvasUtils.getSelection()) || !appCtrl.nf.CanvasUtils.canRead(appCtrl.nf.CanvasUtils.getSelection());"> ng-disabled="!appCtrl.nf.CanvasUtils.isCopyable(appCtrl.nf.CanvasUtils.getSelection());">
<div class="graph-control-action-icon fa fa-copy"></div></button> <div class="graph-control-action-icon fa fa-copy"></div></button>
</div> </div>
<div class="button-spacer-small">&nbsp;</div> <div class="button-spacer-small">&nbsp;</div>
<div id="operate-paste" class="action-button" title="Paste"> <div id="operate-paste" class="action-button" title="Paste">
<button ng-click="appCtrl.nf.Actions['paste'](appCtrl.nf.CanvasUtils.getSelection());" <button ng-click="appCtrl.nf.Actions['paste'](appCtrl.nf.CanvasUtils.getSelection());"
ng-disabled="!appCtrl.nf.Clipboard.isCopied()"> ng-disabled="!appCtrl.nf.CanvasUtils.isPastable()">
<div class="graph-control-action-icon fa fa-paste"></div></button> <div class="graph-control-action-icon fa fa-paste"></div></button>
</div> </div>
<div class="button-spacer-large">&nbsp;</div> <div class="button-spacer-large">&nbsp;</div>

View File

@ -23,7 +23,7 @@
<div id="users-filter-status" class="filter-status"> <div id="users-filter-status" class="filter-status">
Displaying&nbsp;<span id="displayed-users"></span>&nbsp;of&nbsp;<span id="total-users"></span> Displaying&nbsp;<span id="displayed-users"></span>&nbsp;of&nbsp;<span id="total-users"></span>
</div> </div>
<div id="users-filter-container" class="filter-container"> <div id="users-filter-container">
<input type="text" placeholder="Filter" id="users-filter" class="filter"/> <input type="text" placeholder="Filter" id="users-filter" class="filter"/>
<div id="users-filter-type" class="filter-type"></div> <div id="users-filter-type" class="filter-type"></div>
</div> </div>

View File

@ -234,7 +234,7 @@ g.connection path.connection-path.unauthorized {
stroke-dasharray: 3,3; stroke-dasharray: 3,3;
} }
text.connection-from-run-status, text.connection-to-run-status { text.connection-from-run-status, text.connection-to-run-status, text.expiration-icon {
fill: #728e9b; fill: #728e9b;
font-family: FontAwesome; font-family: FontAwesome;
font-size: 10px; font-size: 10px;

View File

@ -145,9 +145,9 @@ div.graph-control-header-action {
font-size: 15px; font-size: 15px;
font-family: Roboto; font-family: Roboto;
color: #262626; color: #262626;
width: 210px; width: 230px;
text-overflow: ellipsis; text-overflow: ellipsis;
overflow-x: hidden; overflow: hidden;
white-space: nowrap; white-space: nowrap;
} }

View File

@ -130,7 +130,7 @@ div.policy-selected-component-name {
font-size: 15px; font-size: 15px;
font-family: Roboto; font-family: Roboto;
color: #262626; color: #262626;
max-width: 300px; width: 300px;
text-overflow: ellipsis; text-overflow: ellipsis;
overflow-x: hidden; overflow-x: hidden;
white-space: nowrap; white-space: nowrap;

View File

@ -193,7 +193,7 @@ nf.ng.Canvas.GlobalMenuCtrl = function (serviceProvider) {
* Launch the users shell. * Launch the users shell.
*/ */
launch: function () { launch: function () {
if (nf.Common.canModifyTenants()) { if (nf.Common.canAccessTenants()) {
nf.Shell.showPage('users'); nf.Shell.showPage('users');
} }
} }

View File

@ -286,6 +286,56 @@ nf.ng.Canvas.GraphControlsCtrl = function (serviceProvider, navigateCtrl, operat
} }
} }
}, },
/**
* Determines whether the user can configure or open the details dialog.
*/
canConfigureOrOpenDetails: function () {
var selection = nf.CanvasUtils.getSelection();
if (selection.empty()) {
return nf.Canvas.canRead() || nf.Canvas.canWrite();
}
return nf.CanvasUtils.isConfigurable(selection) || nf.CanvasUtils.hasDetails(selection);
},
/**
* Opens either the configuration or details view based on the current state.
*/
openConfigureOrDetailsView: function () {
var selection = nf.CanvasUtils.getSelection();
if (selection.empty()) {
nf.ProcessGroupConfiguration.showConfiguration(nf.Canvas.getGroupId());
}
if (nf.CanvasUtils.isConfigurable(selection)) {
nf.Actions.showConfiguration(selection);
} else if (nf.CanvasUtils.hasDetails(selection)) {
nf.Actions.showDetails(selection);
}
},
/**
* Determines whether the user can configure or open the policy management page.
*/
canManagePolicies: function () {
var selection = nf.CanvasUtils.getSelection();
// ensure 0 or 1 components selected
if (selection.size() <= 1) {
// if something is selected, ensure it's not a connection
if (!selection.empty() && nf.CanvasUtils.isConnection(selection)) {
return false;
}
// ensure access to read tenants
return nf.Common.canAccessTenants();
}
return false;
}
} }
var graphControlsCtrl = new GraphControlsCtrl(navigateCtrl, operateCtrl); var graphControlsCtrl = new GraphControlsCtrl(navigateCtrl, operateCtrl);

View File

@ -202,16 +202,16 @@ nf.Actions = (function () {
var selectionData = selection.datum(); var selectionData = selection.datum();
// the source is in the current group // the source is in the current group
if (selectionData.component.source.groupId === nf.Canvas.getGroupId()) { if (selectionData.sourceGroupId === nf.Canvas.getGroupId()) {
var source = d3.select('#id-' + selectionData.component.source.id); var source = d3.select('#id-' + selectionData.sourceId);
nf.Actions.show(source); nf.Actions.show(source);
} else if (selectionData.component.source.type === 'REMOTE_OUTPUT_PORT') { } else if (selectionData.sourceType === 'REMOTE_OUTPUT_PORT') {
// if the source is remote // if the source is remote
var remoteSource = d3.select('#id-' + selectionData.component.source.groupId); var remoteSource = d3.select('#id-' + selectionData.sourceGroupId);
nf.Actions.show(remoteSource); nf.Actions.show(remoteSource);
} else { } else {
// if the source is local but in a sub group // if the source is local but in a sub group
nf.CanvasUtils.showComponent(selectionData.component.source.groupId, selectionData.component.source.id); nf.CanvasUtils.showComponent(selectionData.sourceGroupId, selectionData.sourceId);
} }
} }
}, },
@ -226,16 +226,16 @@ nf.Actions = (function () {
var selectionData = selection.datum(); var selectionData = selection.datum();
// the destination is in the current group or its remote // the destination is in the current group or its remote
if (selectionData.component.destination.groupId === nf.Canvas.getGroupId()) { if (selectionData.destinationGroupId === nf.Canvas.getGroupId()) {
var destination = d3.select('#id-' + selectionData.component.destination.id); var destination = d3.select('#id-' + selectionData.destinationId);
nf.Actions.show(destination); nf.Actions.show(destination);
} else if (selectionData.component.destination.type === 'REMOTE_INPUT_PORT') { } else if (selectionData.destinationType === 'REMOTE_INPUT_PORT') {
// if the destination is remote // if the destination is remote
var remoteDestination = d3.select('#id-' + selectionData.component.destination.groupId); var remoteDestination = d3.select('#id-' + selectionData.destinationGroupId);
nf.Actions.show(remoteDestination); nf.Actions.show(remoteDestination);
} else { } else {
// if the destination is local but in a sub group // if the destination is local but in a sub group
nf.CanvasUtils.showComponent(selectionData.component.destination.groupId, selectionData.component.destination.id); nf.CanvasUtils.showComponent(selectionData.destinationGroupId, selectionData.destinationId);
} }
} }
}, },
@ -1083,7 +1083,7 @@ nf.Actions = (function () {
var processor = selection.datum(); var processor = selection.datum();
// view the state for the selected processor // view the state for the selected processor
nf.ComponentState.showState(processor, nf.CanvasUtils.supportsModification(selection)); nf.ComponentState.showState(processor, nf.CanvasUtils.isConfigurable(selection));
}, },
/** /**
@ -1403,8 +1403,8 @@ nf.Actions = (function () {
// determine the current max zIndex // determine the current max zIndex
var maxZIndex = -1; var maxZIndex = -1;
$.each(nf.Connection.get(), function (_, otherConnection) { $.each(nf.Connection.get(), function (_, otherConnection) {
if (connection.id !== otherConnection.id && otherConnection.component.zIndex > maxZIndex) { if (connection.id !== otherConnection.id && otherConnection.zIndex > maxZIndex) {
maxZIndex = otherConnection.component.zIndex; maxZIndex = otherConnection.zIndex;
} }
}); });

Some files were not shown because too many files have changed in this diff Show More