This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6c8fed169ad NIFI-15939: Adding support to specify the Process Group
used to locat… (#11247)
6c8fed169ad is described below
commit 6c8fed169adc598a3a02ce552435b9d4a03eef67
Author: Matt Gilman <[email protected]>
AuthorDate: Tue May 19 12:45:31 2026 -0400
NIFI-15939: Adding support to specify the Process Group used to locat…
(#11247)
* NIFI-15939: Adding support to specify the Process Group used to locate
descedent components when authorizing bulletins.
---
.../apache/nifi/web/StandardNiFiServiceFacade.java | 109 +++++++++++--
.../nifi/web/StandardNiFiServiceFacadeTest.java | 171 +++++++++++++++++++++
2 files changed, 263 insertions(+), 17 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
index dc82f31094a..9de055e4aa7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
@@ -4674,36 +4674,105 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
}
private boolean authorizeBulletin(final Bulletin bulletin) {
+ return authorizeBulletin(bulletin, null);
+ }
+
+ /**
+ * Resolves the bulletin's source authorizable and checks READ for the
current user.
+ *
+ * <p>When {@code group} is non-null the source is resolved within that
group's
+ * component hierarchy via the {@code findX} APIs first. This is required
for
+ * bulletins generated inside a connector's managed flow: the standard
+ * {@link AuthorizableLookup} is backed by {@link
org.apache.nifi.web.dao.impl.StandardProcessorDAO}
+ * (and its peers) which only walks the {@link
org.apache.nifi.controller.flow.FlowManager}'s
+ * root group, so any source that lives inside a connector's managed flow
context
+ * would otherwise resolve to {@link ResourceNotFoundException} and report
+ * {@code canRead = false} for every bulletin -- preventing the canvas from
+ * rendering bulletin icons for the connector's components and child
groups.
+ * Resolving via the live group preserves the source's authorizable parent
chain
+ * (which terminates at the connector node via
+ * {@link
org.apache.nifi.groups.ProcessGroup#setExplicitParentAuthorizable(Authorizable)}),
+ * so READ on the connector correctly grants READ on its inner
bulletins.</p>
+ *
+ * <p>The fallback to {@code authorizableLookup} handles source types that
are
+ * not addressable via a {@link ProcessGroup} (controller-scoped
components such
+ * as reporting tasks and flow analysis rules) and the standard root-flow
case
+ * when no owning group is supplied.</p>
+ */
+ private boolean authorizeBulletin(final Bulletin bulletin, final
ProcessGroup group) {
final String sourceId = bulletin.getSourceId();
final ComponentType type = bulletin.getSourceType();
final Authorizable authorizable;
try {
- authorizable = switch (type) {
- case PROCESSOR ->
authorizableLookup.getProcessor(sourceId).getAuthorizable();
- case REPORTING_TASK ->
authorizableLookup.getReportingTask(sourceId).getAuthorizable();
- case FLOW_ANALYSIS_RULE ->
authorizableLookup.getFlowAnalysisRule(sourceId).getAuthorizable();
- case FLOW_REGISTRY_CLIENT ->
authorizableLookup.getFlowRegistryClient(sourceId).getAuthorizable();
- case PARAMETER_PROVIDER ->
authorizableLookup.getParameterProvider(sourceId).getAuthorizable();
- case CONTROLLER_SERVICE ->
authorizableLookup.getControllerService(sourceId).getAuthorizable();
- case FLOW_CONTROLLER -> controllerFacade;
- case INPUT_PORT -> authorizableLookup.getInputPort(sourceId);
- case OUTPUT_PORT -> authorizableLookup.getOutputPort(sourceId);
- case REMOTE_PROCESS_GROUP ->
authorizableLookup.getRemoteProcessGroup(sourceId);
- case PROCESS_GROUP ->
authorizableLookup.getProcessGroup(sourceId).getAuthorizable();
- case CONNECTOR -> authorizableLookup.getConnector(sourceId);
- default -> throw new IllegalArgumentException("Unexpected
ComponentType: " + type);
- };
+ authorizable = resolveBulletinAuthorizable(sourceId, type, group);
} catch (final ResourceNotFoundException e) {
// if the underlying component is gone, disallow
return false;
}
+ if (authorizable == null) {
+ return false;
+ }
+
// perform the authorization
final AuthorizationResult result =
authorizable.checkAuthorization(authorizer, RequestAction.READ,
NiFiUserUtils.getNiFiUser());
return Result.Approved.equals(result.getResult());
}
+ /**
+ * Locates the {@link ProcessGroup} that owns the bulletin, including
connector-managed groups.
+ * When the group cannot be found, returns {@code null} so authorization
falls back to the
+ * global {@link AuthorizableLookup}.
+ */
+ private ProcessGroup resolveOwningProcessGroupForBulletin(final Bulletin
bulletin) {
+ final String groupId = bulletin.getGroupId();
+ if (groupId == null) {
+ return null;
+ }
+
+ try {
+ return processGroupDAO.getProcessGroup(groupId, true);
+ } catch (final ResourceNotFoundException e) {
+ // Owning group was removed; fall back to global authorizable
lookup.
+ return null;
+ }
+ }
+
+ private Authorizable resolveBulletinAuthorizable(final String sourceId,
final ComponentType type, final ProcessGroup group) {
+ if (group != null) {
+ final Authorizable found = switch (type) {
+ case PROCESSOR -> group.findProcessor(sourceId);
+ case INPUT_PORT -> group.findInputPort(sourceId);
+ case OUTPUT_PORT -> group.findOutputPort(sourceId);
+ case REMOTE_PROCESS_GROUP ->
group.findRemoteProcessGroup(sourceId);
+ case CONTROLLER_SERVICE ->
group.findControllerService(sourceId, true, true);
+ case PROCESS_GROUP -> sourceId.equals(group.getIdentifier()) ?
group : group.findProcessGroup(sourceId);
+ default -> null;
+ };
+
+ if (found != null) {
+ return found;
+ }
+ }
+
+ return switch (type) {
+ case PROCESSOR ->
authorizableLookup.getProcessor(sourceId).getAuthorizable();
+ case REPORTING_TASK ->
authorizableLookup.getReportingTask(sourceId).getAuthorizable();
+ case FLOW_ANALYSIS_RULE ->
authorizableLookup.getFlowAnalysisRule(sourceId).getAuthorizable();
+ case FLOW_REGISTRY_CLIENT ->
authorizableLookup.getFlowRegistryClient(sourceId).getAuthorizable();
+ case PARAMETER_PROVIDER ->
authorizableLookup.getParameterProvider(sourceId).getAuthorizable();
+ case CONTROLLER_SERVICE ->
authorizableLookup.getControllerService(sourceId).getAuthorizable();
+ case FLOW_CONTROLLER -> controllerFacade;
+ case INPUT_PORT -> authorizableLookup.getInputPort(sourceId);
+ case OUTPUT_PORT -> authorizableLookup.getOutputPort(sourceId);
+ case REMOTE_PROCESS_GROUP ->
authorizableLookup.getRemoteProcessGroup(sourceId);
+ case PROCESS_GROUP ->
authorizableLookup.getProcessGroup(sourceId).getAuthorizable();
+ case CONNECTOR -> authorizableLookup.getConnector(sourceId);
+ default -> throw new IllegalArgumentException("Unexpected
ComponentType: " + type);
+ };
+ }
+
@Override
public BulletinBoardDTO getBulletinBoard(final BulletinQueryDTO query) {
// build the query
@@ -4724,7 +4793,8 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
final List<BulletinEntity> bulletinEntities = new ArrayList<>();
for (final ListIterator<Bulletin> bulletinIter =
results.listIterator(results.size()); bulletinIter.hasPrevious();) {
final Bulletin bulletin = bulletinIter.previous();
-
bulletinEntities.add(entityFactory.createBulletinEntity(dtoFactory.createBulletinDto(bulletin,
true), authorizeBulletin(bulletin)));
+ final ProcessGroup owningGroup =
resolveOwningProcessGroupForBulletin(bulletin);
+
bulletinEntities.add(entityFactory.createBulletinEntity(dtoFactory.createBulletinDto(bulletin,
true), authorizeBulletin(bulletin, owningGroup)));
}
// create the bulletin board
@@ -5417,9 +5487,14 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
bulletins.addAll(bulletinRepository.findBulletinsForGroupBySource(descendantGroup.getIdentifier()));
}
+ // Pass the owning group so authorizeBulletin can resolve sources via
the live
+ // group hierarchy. The standard AuthorizableLookup only walks the
FlowManager's
+ // root group and cannot resolve sources that live inside a
connector's managed
+ // flow context, which would otherwise cause every bulletin in the
connector
+ // canvas to report canRead=false.
List<BulletinEntity> bulletinEntities = new ArrayList<>();
for (final Bulletin bulletin : bulletins) {
-
bulletinEntities.add(entityFactory.createBulletinEntity(dtoFactory.createBulletinDto(bulletin,
false), authorizeBulletin(bulletin)));
+
bulletinEntities.add(entityFactory.createBulletinEntity(dtoFactory.createBulletinDto(bulletin,
false), authorizeBulletin(bulletin, group)));
}
return pruneAndSortBulletins(bulletinEntities,
BulletinRepository.MAX_BULLETINS_PER_COMPONENT);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java
index aba36685007..640894bcf24 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java
@@ -87,6 +87,7 @@ import
org.apache.nifi.registry.flow.mapping.InstantiatedVersionedProcessGroup;
import org.apache.nifi.registry.flow.mapping.VersionedComponentFlowMapper;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinFactory;
+import org.apache.nifi.reporting.BulletinQuery;
import org.apache.nifi.reporting.ComponentType;
import org.apache.nifi.reporting.UserAwareEventAccess;
import org.apache.nifi.services.FlowService;
@@ -94,6 +95,8 @@ import org.apache.nifi.util.MockBulletinRepository;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.validation.RuleViolation;
import org.apache.nifi.validation.RuleViolationsManager;
+import org.apache.nifi.web.api.dto.BulletinBoardDTO;
+import org.apache.nifi.web.api.dto.BulletinQueryDTO;
import org.apache.nifi.web.api.dto.ComponentStateDTO;
import org.apache.nifi.web.api.dto.ConnectorDTO;
import org.apache.nifi.web.api.dto.CounterDTO;
@@ -1009,6 +1012,169 @@ public class StandardNiFiServiceFacadeTest {
assertEquals(groupId, result.getBulletins().get(0).getGroupId());
}
+ /**
+ * Regression test for the connector canvas case: bulletins generated by a
processor that
+ * lives inside a connector's managed flow cannot be resolved by the
standard
+ * {@link AuthorizableLookup} (which only walks the FlowManager's root
group). Without
+ * resolving the source through the owning {@link ProcessGroup}, every
bulletin would
+ * report {@code canRead=false} and the canvas would never render the
bulletin icon
+ * for the connector or its child groups.
+ */
+ @Test
+ public void
testUpdateProcessGroup_BulletinAuthorizedViaGroupLookupWhenAuthorizableLookupMisses()
{
+ final Authentication authentication = new NiFiAuthenticationToken(new
NiFiUserDetails(new Builder().identity(USER_1).build()));
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final String groupId = UUID.randomUUID().toString();
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(processGroup.getIdentifier()).thenReturn(groupId);
+
+ // Simulate the source residing inside a connector-managed flow
context: not findable
+ // through the global authorizable lookup (PROCESSOR_ID_2 throws
ResourceNotFoundException
+ // in the test fixture) but resolvable through the owning ProcessGroup
with an
+ // Authorizable that approves READ for the current user. The processor
mock is built
+ // before the outer stubbing call so its own stubbing does not
interleave with this
+ // when()/thenReturn() pair.
+ final ProcessorNode approvingProcessor = approvingProcessorNode();
+
when(processGroup.findProcessor(PROCESSOR_ID_2)).thenReturn(approvingProcessor);
+
+ final ProcessGroupEntity result =
invokeUpdateProcessGroupWithBulletin(groupId, processGroup, PROCESSOR_ID_2,
PROCESSOR_NAME_2, BULLETIN_MESSAGE_2);
+
+ assertNotNull(result);
+ assertEquals(1, result.getBulletins().size());
+ assertTrue(result.getBulletins().get(0).getCanRead(),
+ "Bulletin canRead should be true when the source is resolved
via the owning ProcessGroup");
+ }
+
+ /**
+ * When the bulletin source cannot be located inside the owning {@link
ProcessGroup}
+ * (the standard root-flow case), authorization must fall back to the
global
+ * {@link AuthorizableLookup} so existing non-connector behavior is
preserved.
+ */
+ @Test
+ public void
testUpdateProcessGroup_BulletinAuthorizationFallsBackToAuthorizableLookupWhenGroupLookupMisses()
{
+ final Authentication authentication = new NiFiAuthenticationToken(new
NiFiUserDetails(new Builder().identity(USER_1).build()));
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final String groupId = UUID.randomUUID().toString();
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(processGroup.getIdentifier()).thenReturn(groupId);
+ // findProcessor returns null by default; the fallback to
authorizableLookup must succeed.
+
+ // PROCESSOR_ID_1 resolves through the global authorizable lookup and
is approved for USER_1.
+ final ProcessGroupEntity result =
invokeUpdateProcessGroupWithBulletin(groupId, processGroup, PROCESSOR_ID_1,
PROCESSOR_NAME_1, BULLETIN_MESSAGE_1);
+
+ assertNotNull(result);
+ assertEquals(1, result.getBulletins().size());
+ assertTrue(result.getBulletins().get(0).getCanRead(),
+ "Bulletin canRead should be true when the global authorizable
lookup approves the source");
+ }
+
+ /**
+ * When neither the owning {@link ProcessGroup} nor the global {@link
AuthorizableLookup}
+ * can resolve the bulletin source (the source component has been
deleted), authorization
+ * must deny so the bulletin is reported as unreadable rather than
surfacing the
+ * underlying {@link ResourceNotFoundException} as a 500.
+ */
+ @Test
+ public void
testUpdateProcessGroup_BulletinDeniedWhenNeitherLookupResolvesSource() {
+ final Authentication authentication = new NiFiAuthenticationToken(new
NiFiUserDetails(new Builder().identity(USER_1).build()));
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final String groupId = UUID.randomUUID().toString();
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(processGroup.getIdentifier()).thenReturn(groupId);
+ // findProcessor returns null, and PROCESSOR_ID_2 throws
ResourceNotFoundException in the global lookup.
+
+ final ProcessGroupEntity result =
invokeUpdateProcessGroupWithBulletin(groupId, processGroup, PROCESSOR_ID_2,
PROCESSOR_NAME_2, BULLETIN_MESSAGE_2);
+
+ assertNotNull(result);
+ assertEquals(1, result.getBulletins().size());
+ assertFalse(result.getBulletins().get(0).getCanRead(),
+ "Bulletin canRead should be false when the source cannot be
resolved by either lookup");
+ }
+
+ /**
+ * The bulletin board authorizes each row independently of process group
flow responses. Connector-managed
+ * sources must resolve through the bulletin's owning group (including
connector-managed process groups)
+ * or the board reports {@code canRead=false} even when the user can read
the connector.
+ */
+ @Test
+ public void
testGetBulletinBoard_BulletinAuthorizedViaOwningProcessGroupWhenAuthorizableLookupMisses()
{
+ final Authentication authentication = new NiFiAuthenticationToken(new
NiFiUserDetails(new Builder().identity(USER_1).build()));
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+
+ final String groupId = UUID.randomUUID().toString();
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(processGroup.getIdentifier()).thenReturn(groupId);
+
+ final ProcessorNode approvingProcessor = approvingProcessorNode();
+
when(processGroup.findProcessor(PROCESSOR_ID_2)).thenReturn(approvingProcessor);
+ when(processGroupDAO.getProcessGroup(groupId,
true)).thenReturn(processGroup);
+
+ final StandardNiFiServiceFacade serviceFacadeSpy = spy(serviceFacade);
+ final MockTestBulletinRepository bulletinRepository = new
MockTestBulletinRepository();
+ serviceFacadeSpy.setBulletinRepository(bulletinRepository);
+
+ bulletinRepository.addBulletin(
+ BulletinFactory.createBulletin(groupId, GROUP_NAME_1,
PROCESSOR_ID_2,
+ ComponentType.PROCESSOR, PROCESSOR_NAME_2,
+ BULLETIN_CATEGORY, BULLETIN_SEVERITY,
BULLETIN_MESSAGE_2, PATH_TO_GROUP_1));
+
+ final BulletinBoardDTO board = serviceFacadeSpy.getBulletinBoard(new
BulletinQueryDTO());
+
+ assertNotNull(board);
+ assertEquals(1, board.getBulletins().size());
+ assertTrue(board.getBulletins().get(0).getCanRead(),
+ "Bulletin board canRead should be true when the source is
resolved via the owning ProcessGroup");
+ verify(processGroupDAO).getProcessGroup(groupId, true);
+ }
+
+ private ProcessGroupEntity invokeUpdateProcessGroupWithBulletin(final
String groupId, final ProcessGroup processGroup,
+ final
String sourceId, final String sourceName, final String message) {
+ final ProcessGroupStatus processGroupStatus = new ProcessGroupStatus();
+ processGroupStatus.setId(groupId);
+ processGroupStatus.setName(GROUP_NAME_1);
+ processGroupStatus.setStatelessActiveThreadCount(0);
+
+ final ControllerFacade controllerFacade = mock(ControllerFacade.class);
+
when(controllerFacade.getProcessGroupStatus(any())).thenReturn(processGroupStatus);
+
+ final StandardNiFiServiceFacade serviceFacadeSpy = spy(serviceFacade);
+ serviceFacadeSpy.setControllerFacade(controllerFacade);
+
+ final ProcessGroupDTO processGroupDTO = new ProcessGroupDTO();
+ processGroupDTO.setId(groupId);
+
when(processGroupDAO.getProcessGroup(groupId)).thenReturn(processGroup);
+
when(processGroupDAO.updateProcessGroup(processGroupDTO)).thenReturn(processGroup);
+
+ final RevisionManager revisionManager = mock(RevisionManager.class);
+ final Revision revision = new Revision(1L, "a", "b");
+ final FlowModification lastModification = new
FlowModification(revision, "a");
+ final RevisionUpdate<Object> snapshot = new
StandardRevisionUpdate<>(processGroupDTO, lastModification);
+ when(revisionManager.updateRevision(any(), any(),
any())).thenReturn(snapshot);
+ serviceFacadeSpy.setRevisionManager(revisionManager);
+
+ final MockTestBulletinRepository bulletinRepository = new
MockTestBulletinRepository();
+ serviceFacadeSpy.setBulletinRepository(bulletinRepository);
+
+ bulletinRepository.addBulletin(
+ BulletinFactory.createBulletin(groupId, GROUP_NAME_1, sourceId,
+ ComponentType.PROCESSOR, sourceName,
+ BULLETIN_CATEGORY, BULLETIN_SEVERITY, message,
PATH_TO_GROUP_1));
+
+ return serviceFacadeSpy.updateProcessGroup(revision, processGroupDTO);
+ }
+
+ private ProcessorNode approvingProcessorNode() {
+ // The production code calls Authorizable#checkAuthorization on
whatever findProcessor
+ // returns. Stubbing it directly keeps the test focused on the
resolution path and
+ // avoids reproducing the entire AbstractComponentNode authorization
machinery.
+ final ProcessorNode processorNode = mock(ProcessorNode.class);
+ when(processorNode.checkAuthorization(any(Authorizer.class), any(),
any(NiFiUser.class))).thenReturn(AuthorizationResult.approved());
+ return processorNode;
+ }
+
@Test
public void testSearchTenantsNullQuery() {
setupSearchTenants();
@@ -1116,6 +1282,11 @@ public class StandardNiFiServiceFacadeTest {
bulletinList.add(bulletin);
}
+ @Override
+ public List<Bulletin> findBulletins(BulletinQuery bulletinQuery) {
+ return new ArrayList<>(bulletinList);
+ }
+
@Override
public List<Bulletin> findBulletinsForSource(String sourceId) {
List<Bulletin> ans = new ArrayList<>();
