You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2022/08/30 13:59:08 UTC
[GitHub] [ignite] ololo3000 opened a new pull request, #10226: IGNITE-15322 wip.
ololo3000 opened a new pull request, #10226:
URL: https://github.com/apache/ignite/pull/10226
Thank you for submitting the pull request to the Apache Ignite.
In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:
### The Contribution Checklist
- [ ] There is a single JIRA ticket related to the pull request.
- [ ] The web-link to the pull request is attached to the JIRA ticket.
- [ ] The JIRA ticket has the _Patch Available_ state.
- [ ] The pull request body describes changes that have been made.
The description explains _WHAT_ and _WHY_ was made instead of _HOW_.
- [ ] The pull request title is treated as the final commit message.
The following pattern must be used: `IGNITE-XXXX Change summary` where `XXXX` - number of JIRA issue.
- [ ] A reviewer has been mentioned through the JIRA comments
(see [the Maintainers list](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute#HowtoContribute-ReviewProcessandMaintainers))
- [ ] The pull request has been checked by the Teamcity Bot and
the `green visa` attached to the JIRA ticket (see [TC.Bot: Check PR](https://mtcga.gridgain.com/prs.html))
### Notes
- [How to Contribute](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute)
- [Coding abbreviation rules](https://cwiki.apache.org/confluence/display/IGNITE/Abbreviation+Rules)
- [Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines)
- [Apache Ignite Teamcity Bot](https://cwiki.apache.org/confluence/display/IGNITE/Apache+Ignite+Teamcity+Bot)
If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com _#ignite_ channel.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1092310510
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -203,7 +203,7 @@ public void setKernalContext(GridKernalContext ctx) {
/**
* @return {@link IgniteCompute} for this cluster group.
*/
- public final IgniteCompute compute() {
+ public final IgniteComputeImpl compute() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg merged pull request #10226: IGNITE-15322 Stop authorization of internal tasks by their class name.
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg merged PR #10226:
URL: https://github.com/apache/ignite/pull/10226
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "anton-vinogradov (via GitHub)" <gi...@apache.org>.
anton-vinogradov commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1093398466
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -201,16 +204,29 @@ public void setKernalContext(GridKernalContext ctx) {
}
/**
- * @return {@link IgniteCompute} for this cluster group.
+ * @return Internal {@link IgniteCompute} facade for this cluster group.
+ */
+ public final IgniteCompute internalComputeFacade() {
Review Comment:
"facade" postfix looks suspicious
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -75,8 +75,11 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable {
/** Kernal context. */
protected transient GridKernalContext ctx;
- /** Compute. */
- private transient IgniteComputeImpl compute;
+ /** Internal compute facade. */
+ private transient IgniteCompute internalComputeFacade;
+
+ /** Public compute facade. */
+ private transient IgniteCompute publicComputeFacade;
Review Comment:
"facade" postfix looks suspicious
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1092307037
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
Review Comment:
Fixed. The name was changed to SecurityAwareJob.
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -203,7 +203,7 @@ public void setKernalContext(GridKernalContext ctx) {
/**
* @return {@link IgniteCompute} for this cluster group.
*/
- public final IgniteCompute compute() {
+ public final IgniteComputeImpl compute() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1105497519
##########
modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcResultSet.java:
##########
@@ -268,9 +268,10 @@ void closeInternal() throws SQLException {
JdbcConnection conn = (JdbcConnection)stmt.getConnection();
if (conn.isCloseCursorTaskSupported()) {
- Ignite ignite = conn.ignite();
+ IgniteEx ignite = conn.ignite();
- ignite.compute(ignite.cluster().forNodeId(conn.nodeId())).call(new JdbcCloseCursorTask(uuid));
+ ignite.internalCompute(ignite.cluster().forNodeId(conn.nodeId()))
+ .call(new JdbcCloseCursorTask(uuid));
Review Comment:
Fixed.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java:
##########
@@ -385,4 +393,27 @@ public static SecurityContext authenticateLocalNode(
return secCtx;
}
+
+ /** */
+ public static void authorizeAll(IgniteSecurity security, SecurityPermissionSet permissions) {
+ if (!security.enabled())
+ return;
+
+ if (!F.isEmpty(permissions.systemPermissions())) {
Review Comment:
Fixed.
##########
modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java:
##########
@@ -0,0 +1,624 @@
+/*
+ * 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.ignite.common;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientAuthorizationException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.GridClient;
+import org.apache.ignite.internal.client.GridClientConfiguration;
+import org.apache.ignite.internal.client.GridClientFactory;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataCollectorTask;
+import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
+import org.apache.ignite.internal.processors.security.compute.ComputePermissionCheckTest;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityData;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityPluginProvider;
+import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.lang.gridfunc.AtomicIntegerFactoryCallable;
+import org.apache.ignite.internal.util.lang.gridfunc.RunnableWrapperClosure;
+import org.apache.ignite.internal.util.lang.gridfunc.ToStringClosure;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.plugin.security.SecurityCredentials;
+import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
+import org.apache.ignite.plugin.security.SecurityException;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static java.util.Collections.singletonList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.common.AbstractEventSecurityContextTest.sendRestRequest;
+import static org.apache.ignite.internal.processors.rest.GridRestCommand.EXE;
+import static org.apache.ignite.plugin.security.SecurityPermission.CACHE_CREATE;
+import static org.apache.ignite.plugin.security.SecurityPermission.JOIN_AS_SERVER;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_CANCEL;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_EXECUTE;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class ComputeTaskPermissionsTest extends AbstractSecurityTest {
+ /** */
+ private static final IgniteCallable SYSTEM_CALLABLE = new AtomicIntegerFactoryCallable();
+
+ /** */
+ private static final IgniteRunnable SYSTEM_RUNNABLE = new RunnableWrapperClosure(null);
+
+ /** */
+ private static final IgniteClosure SYSTEM_CLOSURE = new ToStringClosure<>();
+
+ /** */
+ private static final ComputeTask SYSTEM_TASK = new SnapshotMetadataCollectorTask();
+
+ /** */
+ private static final AtomicInteger EXECUTED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final AtomicInteger CANCELLED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final String CACHE = DEFAULT_CACHE_NAME;
+
+ /** */
+ private static final int SRV_NODES_CNT = 2;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ for (int idx = 0; idx < SRV_NODES_CNT; idx++)
+ startGrid(idx, false);
+
+ startGrid(SRV_NODES_CNT, true);
+
+ grid(0).createCache(CACHE);
+ }
+
+ /** */
+ private IgniteEx startGrid(int idx, boolean isClient) throws Exception {
+ String login = getTestIgniteInstanceName(idx);
+
+ SecurityPermissionSetBuilder permsBuilder = new SecurityPermissionSetBuilder()
+ .defaultAllowAll(false)
+ .appendTaskPermissions(AllowedCallable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedCallableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedComputeTask.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedClosure.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(CancelAllowedTask.class.getName(), TASK_EXECUTE, TASK_CANCEL)
+ .appendTaskPermissions(CancelForbiddenTask.class.getName(), TASK_EXECUTE);
+
+ if (!isClient)
+ permsBuilder.appendSystemPermissions(CACHE_CREATE, JOIN_AS_SERVER);
+
+ SecurityPermissionSet permissions = permsBuilder.build();
+
+ IgniteConfiguration cfg = getConfiguration(
+ login,
+ new TestSecurityPluginProvider(
+ login,
+ "",
+ permissions,
+ null,
+ false,
+ new TestSecurityData("ignite-client", permissions),
+ new TestSecurityData("grid-client", permissions),
+ new TestSecurityData("rest-client", permissions)
+ )).setClientMode(isClient);
+
+ cfg.setConnectorConfiguration(new ConnectorConfiguration()
+ .setJettyPath("modules/clients/src/test/resources/jetty/rest-jetty.xml"))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(1)));
+
+ return startGrid(cfg);
+ }
+
+ /** */
+ @Test
+ public void testNode() throws Exception {
+ doNodeTest(0, 0);
+ doNodeTest(0, 1);
+ doNodeTest(2, 1);
+ }
+
+ /** */
+ @Test
+ public void testIgniteClient() throws Exception {
+ try (
+ IgniteClient cli = Ignition.startClient(new ClientConfiguration()
+ .setUserName("ignite-client")
+ .setUserPassword("")
+ .setAddresses("127.0.0.1:10800"))
+ ) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync2(t, a).get());
+
+ testTaskCancelFailed(t -> cli.compute().executeAsync2(t, null));
+ testTaskCancelSucceeded(t -> cli.compute().executeAsync2(t, null));
+ }
+ }
+
+ /** */
+ @Test
+ public void testGridClient() throws Exception {
+ GridClientConfiguration cfg = new GridClientConfiguration()
+ .setServers(singletonList("127.0.0.1:11211"))
+ .setSecurityCredentialsProvider(new SecurityCredentialsBasicProvider(new SecurityCredentials("grid-client", "")));
+
+ try (GridClient cli = GridClientFactory.start(cfg)) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecute(t, CACHE, "key", a));
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecuteAsync(t, CACHE, "key", a).get());
+ }
+ }
+
+ /** */
+ @Test
+ public void testRestClient() {
+ for (Ignite ignite : Ignition.allGrids()) {
+ checkTask(String.class, (t, a) -> {
+ JsonNode resp = sendRestRequest(
+ EXE,
+ Arrays.asList(
+ "name=" + t,
+ "destId=" + ignite.cluster().localNode().id()
+ ),
+ "rest-client"
+ );
+
+ if (0 != resp.get("successStatus").intValue())
+ throw new IgniteException(resp.get("error").textValue());
+ });
+ }
+ }
+
+ /** */
+ private void doNodeTest(int initiator, int executor) throws Exception {
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(String.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(String.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkCallable(t -> compute(initiator, executor).call(t));
+ checkCallable(t -> compute(initiator, executor).callAsync(t).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c));
+ checkCallables(c -> compute(initiator, executor).callAsync(c).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c, new TestReducer()));
+ checkCallables(c -> compute(initiator, executor).callAsync(c, new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).run(r));
+ checkRunnable(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnables(r -> compute(initiator, executor).run(r));
+ checkRunnables(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnable(r -> compute(initiator, executor).broadcast(r));
+ checkRunnable(r -> compute(initiator, executor).broadcastAsync(r).get());
+
+ checkCallable(c -> compute(initiator, executor).broadcast(c));
+ checkCallable(c -> compute(initiator, executor).broadcastAsync(c).get());
+
+ checkClosure(c -> compute(initiator, executor).broadcast(c, null));
+ checkClosure(c -> compute(initiator, executor).broadcastAsync(c, null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, (Object)null));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, (Object)null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, singletonList(null), new TestReducer()));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, singletonList(null), new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(CACHE, keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(CACHE, keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), partitionForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(CACHE, keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(CACHE, keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), partitionForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkRunnable(r -> executorService(initiator, executor).submit(r).get());
+ checkRunnable(r -> executorService(initiator, executor).submit(r, null).get());
+ checkCallable(c -> executorService(initiator, executor).submit(c).get());
+
+ checkCallable(c -> executorService(initiator, executor).invokeAll(singletonList(c)).forEach(ComputePermissionCheckTest::getQuiet));
+ checkCallable(c ->
+ executorService(initiator, executor).invokeAll(
+ singletonList(c),
+ getTestTimeout(),
+ MILLISECONDS
+ ).forEach(ComputePermissionCheckTest::getQuiet)
+ );
+
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c)));
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c), getTestTimeout(), MILLISECONDS));
+ }
+
+ /** */
+ private IgniteCompute compute(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.compute(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private ExecutorService executorService(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.executorService(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private void checkRunnable(ConsumerX<IgniteRunnable> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedRunnable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenRunnable()));
+ assertFailed(() -> consumer.accept(SYSTEM_RUNNABLE));
+ }
+
+ /** */
+ private void checkRunnables(ConsumerX<Collection<IgniteRunnable>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new AllowedRunnableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new ForbiddenRunnable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), SYSTEM_RUNNABLE)));
+ }
+
+ /** */
+ private void checkCallable(ConsumerX<IgniteCallable<AtomicInteger>> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedCallable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenCallable()));
+ assertFailed(() -> consumer.accept(SYSTEM_CALLABLE));
+ }
+
+ /** */
+ private void checkCallables(ConsumerX<Collection<IgniteCallable<AtomicInteger>>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new AllowedCallableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new ForbiddenCallable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), SYSTEM_CALLABLE)));
+ }
+
+ /** */
+ private void checkClosure(Consumer<IgniteClosure> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedClosure()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenClosure()));
+ assertFailed(() -> consumer.accept(SYSTEM_CLOSURE));
+ }
+
+ /** */
+ private <T> void checkTask(Class<T> cls, BiConsumerX<T, Object> consumer) {
+ assertCompleted(() -> consumer.accept(cast(new AllowedComputeTask(), cls), null), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(cast(new ForbiddenComputeTask(), cls), null));
+ assertFailed(() -> consumer.accept(cast(SYSTEM_TASK, cls), null));
+ }
+
+ /** */
+ private void testTaskCancelFailed(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ EXECUTED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelForbiddenTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ GridTestUtils.assertThrowsWithCause(() -> fut.cancel(true), ClientAuthorizationException.class);
+
+ assertFalse(fut.isCancelled());
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == EXECUTED_TASK_CNTR.get(), getTestTimeout()));
+ }
+
+ /** */
+ private void testTaskCancelSucceeded(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ CANCELLED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelAllowedTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ assertTrue(fut.cancel(true));
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == CANCELLED_TASK_CNTR.get(), getTestTimeout()));
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+ }
+
+ /** */
+ private void assertCompleted(RunnableX r, int expCnt) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ r.run();
+
+ assertEquals(expCnt, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ private void assertFailed(RunnableX r) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ try {
+ r.run();
+ }
+ catch (Exception e) {
+ if (X.hasCause(e, ClientAuthorizationException.class) || X.hasCause(e, SecurityException.class))
+ return;
+
+ if (e instanceof IgniteException) {
+ assertTrue(
+ e.getMessage().contains("Authorization failed")
+ || e.getMessage().contains("Access to Ignite Internal tasks is restricted")
+ || e.getMessage().contains("Failed to get any task completion")
+ );
+
+ return;
+ }
+ }
+
+ fail();
+
+ assertEquals(0, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ public <T> T cast(Object obj, Class<T> cls) {
+ if (String.class.equals(cls))
+ return (T)obj.getClass().getName();
+ else if (Class.class.equals(cls))
+ return (T)obj.getClass();
+ else if (ComputeTask.class.equals(cls))
+ return (T)obj;
+ else
+ throw new IllegalStateException();
+ }
+
+ /** */
+ private int keyForNode(int nodeIdx) {
+ return keyForNode(grid(0).affinity(CACHE), new AtomicInteger(0), grid(nodeIdx).cluster().localNode());
+ }
+
+ /** */
+ private int partitionForNode(int nodeIdx) {
+ return grid(0).affinity(CACHE).partition(keyForNode(nodeIdx));
+ }
+
+ /** */
+ private static class AllowedComputeTask extends AbstractTask { }
+
+ /** */
+ private static class ForbiddenComputeTask extends AbstractTask { }
+
+ /** */
+ private static class AllowedCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedCallableSecond extends AbstractCallable { }
+
+ /** */
+ private static class ForbiddenCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedRunnableSecond extends AbstractRunnable { }
+
+ /** */
+ private static class ForbiddenRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedClosure extends AbstractClosure { }
+
+ /** */
+ private static class ForbiddenClosure extends AbstractClosure { }
+
+ /** */
+ private static class CancelAllowedTask extends AbstractHangingTask { }
+
+ /** */
+ private static class CancelForbiddenTask extends AbstractHangingTask { }
+
+ /** */
+ private static class AbstractRunnable implements IgniteRunnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+ }
+ }
+
+ /** */
+ private static class AbstractCallable implements IgniteCallable<AtomicInteger> {
+ /** {@inheritDoc} */
+ @Override public AtomicInteger call() throws Exception {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return new AtomicInteger(0);
+ }
+ }
+
+ /** */
+ private static class AbstractClosure implements IgniteClosure<Boolean, Boolean> {
+ /** {@inheritDoc} */
+ @Override public Boolean apply(Boolean o) {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+ }
+
+ /** */
+ private static class AbstractHangingTask extends AbstractTask {
+ /** */
+ public static CountDownLatch taskStartedLatch;
+
Review Comment:
Fixed.
##########
modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java:
##########
@@ -0,0 +1,624 @@
+/*
+ * 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.ignite.common;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientAuthorizationException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.GridClient;
+import org.apache.ignite.internal.client.GridClientConfiguration;
+import org.apache.ignite.internal.client.GridClientFactory;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataCollectorTask;
+import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
+import org.apache.ignite.internal.processors.security.compute.ComputePermissionCheckTest;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityData;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityPluginProvider;
+import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.lang.gridfunc.AtomicIntegerFactoryCallable;
+import org.apache.ignite.internal.util.lang.gridfunc.RunnableWrapperClosure;
+import org.apache.ignite.internal.util.lang.gridfunc.ToStringClosure;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.plugin.security.SecurityCredentials;
+import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
+import org.apache.ignite.plugin.security.SecurityException;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static java.util.Collections.singletonList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.common.AbstractEventSecurityContextTest.sendRestRequest;
+import static org.apache.ignite.internal.processors.rest.GridRestCommand.EXE;
+import static org.apache.ignite.plugin.security.SecurityPermission.CACHE_CREATE;
+import static org.apache.ignite.plugin.security.SecurityPermission.JOIN_AS_SERVER;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_CANCEL;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_EXECUTE;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class ComputeTaskPermissionsTest extends AbstractSecurityTest {
+ /** */
+ private static final IgniteCallable SYSTEM_CALLABLE = new AtomicIntegerFactoryCallable();
+
+ /** */
+ private static final IgniteRunnable SYSTEM_RUNNABLE = new RunnableWrapperClosure(null);
+
+ /** */
+ private static final IgniteClosure SYSTEM_CLOSURE = new ToStringClosure<>();
+
+ /** */
+ private static final ComputeTask SYSTEM_TASK = new SnapshotMetadataCollectorTask();
+
+ /** */
+ private static final AtomicInteger EXECUTED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final AtomicInteger CANCELLED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final String CACHE = DEFAULT_CACHE_NAME;
+
+ /** */
+ private static final int SRV_NODES_CNT = 2;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ for (int idx = 0; idx < SRV_NODES_CNT; idx++)
+ startGrid(idx, false);
+
+ startGrid(SRV_NODES_CNT, true);
+
+ grid(0).createCache(CACHE);
+ }
+
+ /** */
+ private IgniteEx startGrid(int idx, boolean isClient) throws Exception {
+ String login = getTestIgniteInstanceName(idx);
+
+ SecurityPermissionSetBuilder permsBuilder = new SecurityPermissionSetBuilder()
+ .defaultAllowAll(false)
+ .appendTaskPermissions(AllowedCallable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedCallableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedComputeTask.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedClosure.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(CancelAllowedTask.class.getName(), TASK_EXECUTE, TASK_CANCEL)
+ .appendTaskPermissions(CancelForbiddenTask.class.getName(), TASK_EXECUTE);
+
+ if (!isClient)
+ permsBuilder.appendSystemPermissions(CACHE_CREATE, JOIN_AS_SERVER);
+
+ SecurityPermissionSet permissions = permsBuilder.build();
+
+ IgniteConfiguration cfg = getConfiguration(
+ login,
+ new TestSecurityPluginProvider(
+ login,
+ "",
+ permissions,
+ null,
+ false,
+ new TestSecurityData("ignite-client", permissions),
+ new TestSecurityData("grid-client", permissions),
+ new TestSecurityData("rest-client", permissions)
+ )).setClientMode(isClient);
+
+ cfg.setConnectorConfiguration(new ConnectorConfiguration()
+ .setJettyPath("modules/clients/src/test/resources/jetty/rest-jetty.xml"))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(1)));
+
+ return startGrid(cfg);
+ }
+
+ /** */
+ @Test
+ public void testNode() throws Exception {
+ doNodeTest(0, 0);
+ doNodeTest(0, 1);
+ doNodeTest(2, 1);
+ }
+
+ /** */
+ @Test
+ public void testIgniteClient() throws Exception {
+ try (
+ IgniteClient cli = Ignition.startClient(new ClientConfiguration()
+ .setUserName("ignite-client")
+ .setUserPassword("")
+ .setAddresses("127.0.0.1:10800"))
+ ) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync2(t, a).get());
+
+ testTaskCancelFailed(t -> cli.compute().executeAsync2(t, null));
+ testTaskCancelSucceeded(t -> cli.compute().executeAsync2(t, null));
+ }
+ }
+
+ /** */
+ @Test
+ public void testGridClient() throws Exception {
+ GridClientConfiguration cfg = new GridClientConfiguration()
+ .setServers(singletonList("127.0.0.1:11211"))
+ .setSecurityCredentialsProvider(new SecurityCredentialsBasicProvider(new SecurityCredentials("grid-client", "")));
+
+ try (GridClient cli = GridClientFactory.start(cfg)) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecute(t, CACHE, "key", a));
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecuteAsync(t, CACHE, "key", a).get());
+ }
+ }
+
+ /** */
+ @Test
+ public void testRestClient() {
+ for (Ignite ignite : Ignition.allGrids()) {
+ checkTask(String.class, (t, a) -> {
+ JsonNode resp = sendRestRequest(
+ EXE,
+ Arrays.asList(
+ "name=" + t,
+ "destId=" + ignite.cluster().localNode().id()
+ ),
+ "rest-client"
+ );
+
+ if (0 != resp.get("successStatus").intValue())
+ throw new IgniteException(resp.get("error").textValue());
+ });
+ }
+ }
+
+ /** */
+ private void doNodeTest(int initiator, int executor) throws Exception {
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(String.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(String.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkCallable(t -> compute(initiator, executor).call(t));
+ checkCallable(t -> compute(initiator, executor).callAsync(t).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c));
+ checkCallables(c -> compute(initiator, executor).callAsync(c).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c, new TestReducer()));
+ checkCallables(c -> compute(initiator, executor).callAsync(c, new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).run(r));
+ checkRunnable(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnables(r -> compute(initiator, executor).run(r));
+ checkRunnables(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnable(r -> compute(initiator, executor).broadcast(r));
+ checkRunnable(r -> compute(initiator, executor).broadcastAsync(r).get());
+
+ checkCallable(c -> compute(initiator, executor).broadcast(c));
+ checkCallable(c -> compute(initiator, executor).broadcastAsync(c).get());
+
+ checkClosure(c -> compute(initiator, executor).broadcast(c, null));
+ checkClosure(c -> compute(initiator, executor).broadcastAsync(c, null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, (Object)null));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, (Object)null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, singletonList(null), new TestReducer()));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, singletonList(null), new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(CACHE, keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(CACHE, keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), partitionForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(CACHE, keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(CACHE, keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), partitionForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkRunnable(r -> executorService(initiator, executor).submit(r).get());
+ checkRunnable(r -> executorService(initiator, executor).submit(r, null).get());
+ checkCallable(c -> executorService(initiator, executor).submit(c).get());
+
+ checkCallable(c -> executorService(initiator, executor).invokeAll(singletonList(c)).forEach(ComputePermissionCheckTest::getQuiet));
+ checkCallable(c ->
+ executorService(initiator, executor).invokeAll(
+ singletonList(c),
+ getTestTimeout(),
+ MILLISECONDS
+ ).forEach(ComputePermissionCheckTest::getQuiet)
+ );
+
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c)));
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c), getTestTimeout(), MILLISECONDS));
+ }
+
+ /** */
+ private IgniteCompute compute(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.compute(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private ExecutorService executorService(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.executorService(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private void checkRunnable(ConsumerX<IgniteRunnable> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedRunnable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenRunnable()));
+ assertFailed(() -> consumer.accept(SYSTEM_RUNNABLE));
+ }
+
+ /** */
+ private void checkRunnables(ConsumerX<Collection<IgniteRunnable>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new AllowedRunnableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new ForbiddenRunnable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), SYSTEM_RUNNABLE)));
+ }
+
+ /** */
+ private void checkCallable(ConsumerX<IgniteCallable<AtomicInteger>> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedCallable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenCallable()));
+ assertFailed(() -> consumer.accept(SYSTEM_CALLABLE));
+ }
+
+ /** */
+ private void checkCallables(ConsumerX<Collection<IgniteCallable<AtomicInteger>>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new AllowedCallableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new ForbiddenCallable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), SYSTEM_CALLABLE)));
+ }
+
+ /** */
+ private void checkClosure(Consumer<IgniteClosure> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedClosure()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenClosure()));
+ assertFailed(() -> consumer.accept(SYSTEM_CLOSURE));
+ }
+
+ /** */
+ private <T> void checkTask(Class<T> cls, BiConsumerX<T, Object> consumer) {
+ assertCompleted(() -> consumer.accept(cast(new AllowedComputeTask(), cls), null), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(cast(new ForbiddenComputeTask(), cls), null));
+ assertFailed(() -> consumer.accept(cast(SYSTEM_TASK, cls), null));
+ }
+
+ /** */
+ private void testTaskCancelFailed(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ EXECUTED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelForbiddenTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ GridTestUtils.assertThrowsWithCause(() -> fut.cancel(true), ClientAuthorizationException.class);
+
+ assertFalse(fut.isCancelled());
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == EXECUTED_TASK_CNTR.get(), getTestTimeout()));
+ }
+
+ /** */
+ private void testTaskCancelSucceeded(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ CANCELLED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelAllowedTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ assertTrue(fut.cancel(true));
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == CANCELLED_TASK_CNTR.get(), getTestTimeout()));
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+ }
+
+ /** */
+ private void assertCompleted(RunnableX r, int expCnt) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ r.run();
+
+ assertEquals(expCnt, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ private void assertFailed(RunnableX r) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ try {
+ r.run();
+ }
+ catch (Exception e) {
+ if (X.hasCause(e, ClientAuthorizationException.class) || X.hasCause(e, SecurityException.class))
+ return;
+
+ if (e instanceof IgniteException) {
+ assertTrue(
+ e.getMessage().contains("Authorization failed")
+ || e.getMessage().contains("Access to Ignite Internal tasks is restricted")
+ || e.getMessage().contains("Failed to get any task completion")
+ );
+
+ return;
+ }
+ }
+
+ fail();
+
+ assertEquals(0, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ public <T> T cast(Object obj, Class<T> cls) {
+ if (String.class.equals(cls))
+ return (T)obj.getClass().getName();
+ else if (Class.class.equals(cls))
+ return (T)obj.getClass();
+ else if (ComputeTask.class.equals(cls))
+ return (T)obj;
+ else
+ throw new IllegalStateException();
+ }
+
+ /** */
+ private int keyForNode(int nodeIdx) {
+ return keyForNode(grid(0).affinity(CACHE), new AtomicInteger(0), grid(nodeIdx).cluster().localNode());
+ }
+
+ /** */
+ private int partitionForNode(int nodeIdx) {
+ return grid(0).affinity(CACHE).partition(keyForNode(nodeIdx));
+ }
+
+ /** */
+ private static class AllowedComputeTask extends AbstractTask { }
+
+ /** */
+ private static class ForbiddenComputeTask extends AbstractTask { }
+
+ /** */
+ private static class AllowedCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedCallableSecond extends AbstractCallable { }
+
+ /** */
+ private static class ForbiddenCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedRunnableSecond extends AbstractRunnable { }
+
+ /** */
+ private static class ForbiddenRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedClosure extends AbstractClosure { }
+
+ /** */
+ private static class ForbiddenClosure extends AbstractClosure { }
+
+ /** */
+ private static class CancelAllowedTask extends AbstractHangingTask { }
+
+ /** */
+ private static class CancelForbiddenTask extends AbstractHangingTask { }
+
+ /** */
+ private static class AbstractRunnable implements IgniteRunnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+ }
+ }
+
+ /** */
+ private static class AbstractCallable implements IgniteCallable<AtomicInteger> {
+ /** {@inheritDoc} */
+ @Override public AtomicInteger call() throws Exception {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return new AtomicInteger(0);
+ }
+ }
+
+ /** */
+ private static class AbstractClosure implements IgniteClosure<Boolean, Boolean> {
+ /** {@inheritDoc} */
+ @Override public Boolean apply(Boolean o) {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+ }
+
+ /** */
+ private static class AbstractHangingTask extends AbstractTask {
+ /** */
+ public static CountDownLatch taskStartedLatch;
+
+ /** */
+ public static CountDownLatch taskUnblockedLatch;
+
+ /** {@inheritDoc} */
+ @Override protected ComputeJob job() {
+ return new Job();
+ }
+
+ /** */
+ private static class Job implements ComputeJob {
+ /** {@inheritDoc} */
+ @Override public Object execute() {
+ LockSupport.parkNanos(10_000_000);
+
+ taskStartedLatch.countDown();
+
+ try {
+ taskUnblockedLatch.await(5_000, MILLISECONDS);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ throw new IgniteException(e);
+ }
+
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void cancel() {
+ CANCELLED_TASK_CNTR.incrementAndGet();
+ }
+ }
+ }
+
+ /** */
+ private static class AbstractTask extends ComputeTaskAdapter<Object, Object> {
+ /** {@inheritDoc} */
+ @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ @Nullable Object arg
+ ) throws IgniteException {
+ return subgrid.stream().filter(g -> !g.isClient()).collect(Collectors.toMap(g -> job(), g -> g, (l, r) -> l));
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1105497102
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/TaskExecutionOptions.java:
##########
@@ -176,13 +176,13 @@ public TaskExecutionOptions asSystemTask() {
}
/** */
- public boolean isAuthenticationDisabled() {
- return isAuthDisabled;
+ public boolean isPublicRequest() {
+ return isPublicReq;
}
/** */
- public TaskExecutionOptions withAuthenticationDisabled() {
- isAuthDisabled = true;
+ public TaskExecutionOptions asPublicRequest() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1092310182
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
+ /** */
+ public SecurityPermissionSet NO_PERMISSIONS = SecurityPermissionSetBuilder.create().build();
+
+ /** */
+ public SecurityPermissionSet requiredPermissions();
+
+ /** */
+ public static SecurityPermissionSet cachePermissions(String cacheName, SecurityPermission... perms) {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "anton-vinogradov (via GitHub)" <gi...@apache.org>.
anton-vinogradov commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1090814381
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -203,7 +203,7 @@ public void setKernalContext(GridKernalContext ctx) {
/**
* @return {@link IgniteCompute} for this cluster group.
*/
- public final IgniteCompute compute() {
+ public final IgniteComputeImpl compute() {
Review Comment:
Keep interfaces at API when possible.
No need to provide implementation here.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
+ /** */
+ public SecurityPermissionSet NO_PERMISSIONS = SecurityPermissionSetBuilder.create().build();
Review Comment:
Incorrect definition place.
Should be relocated to SecurityPermissionSetBuilder.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1736,4 +1763,51 @@ else if (!getLocJobStatistics && locNodeId.equals(jobResult.getNode().id()))
return S.toString(GridTaskWorker.class, this);
}
}
+
+ /** */
+ private void authorizeSystemJobStart(ComputeJob job) {
+ Object executable = unwrap(job);
+
+ if (!isSystemType(ctx, executable.getClass())) {
+ ctx.security().authorize(executable.getClass().getName(), TASK_EXECUTE);
+
+ return;
+ }
+
+ if (executable instanceof SecurityPermissionAware)
+ authorizeAll(ctx.security(), ((SecurityPermissionAware)executable).requiredPermissions());
+ else if (opts.isSystemTaskGuard()) {
+ throw new SecurityException("Access to Ignite Internal tasks is restricted" +
+ " [task=" + task.getClass().getName() + ", job=" + executable.getClass() + "]");
+ }
+ }
+
+ /** */
+ public void authorizeTaskCancel() {
+ if (!ctx.security().enabled())
+ return;
+
+ if (!isSystemType(ctx, task.getClass())) {
+ ctx.security().authorize(task.getClass().getName(), TASK_CANCEL);
+
+ return;
+ }
Review Comment:
Lines above are duplicated several times (5) in different forms.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
+ /** */
+ public SecurityPermissionSet NO_PERMISSIONS = SecurityPermissionSetBuilder.create().build();
+
+ /** */
+ public SecurityPermissionSet requiredPermissions();
+
+ /** */
+ public static SecurityPermissionSet cachePermissions(String cacheName, SecurityPermission... perms) {
Review Comment:
Incorrect definition place.
Should be relocated to SecurityPermissionSetBuilder.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1736,4 +1763,51 @@ else if (!getLocJobStatistics && locNodeId.equals(jobResult.getNode().id()))
return S.toString(GridTaskWorker.class, this);
}
}
+
+ /** */
+ private void authorizeSystemJobStart(ComputeJob job) {
+ Object executable = unwrap(job);
+
+ if (!isSystemType(ctx, executable.getClass())) {
+ ctx.security().authorize(executable.getClass().getName(), TASK_EXECUTE);
+
+ return;
+ }
+
+ if (executable instanceof SecurityPermissionAware)
+ authorizeAll(ctx.security(), ((SecurityPermissionAware)executable).requiredPermissions());
+ else if (opts.isSystemTaskGuard()) {
+ throw new SecurityException("Access to Ignite Internal tasks is restricted" +
+ " [task=" + task.getClass().getName() + ", job=" + executable.getClass() + "]");
+ }
+ }
+
+ /** */
+ public void authorizeTaskCancel() {
+ if (!ctx.security().enabled())
+ return;
+
+ if (!isSystemType(ctx, task.getClass())) {
+ ctx.security().authorize(task.getClass().getName(), TASK_CANCEL);
+
+ return;
Review Comment:
Could return be replaced with else if?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/TaskExecutionOptions.java:
##########
@@ -176,13 +176,13 @@ public TaskExecutionOptions asSystemTask() {
}
/** */
- public boolean isAuthenticationDisabled() {
- return isAuthDisabled;
+ public boolean isSystemTaskGuard() {
+ return isSystemTaskGuard;
}
/** */
- public TaskExecutionOptions withAuthenticationDisabled() {
- isAuthDisabled = true;
+ public TaskExecutionOptions withSystemTaskGuard() {
+ isSystemTaskGuard = true;
return this;
}
Review Comment:
Looks like, this spetifies option for current thread only, is it safe?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1736,4 +1763,51 @@ else if (!getLocJobStatistics && locNodeId.equals(jobResult.getNode().id()))
return S.toString(GridTaskWorker.class, this);
}
}
+
+ /** */
+ private void authorizeSystemJobStart(ComputeJob job) {
+ Object executable = unwrap(job);
+
+ if (!isSystemType(ctx, executable.getClass())) {
+ ctx.security().authorize(executable.getClass().getName(), TASK_EXECUTE);
+
+ return;
Review Comment:
Could return be replaced with else if?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java:
##########
@@ -1589,4 +1579,36 @@ public Map<ComputeJobStatusEnum, Long> jobStatuses(IgniteUuid sesId) {
else
return taskWorker.jobStatuses();
}
+
+ /** */
+ private void authorizeTaskExecution(String taskName, Class<?> taskCls, ComputeTask<?, ?> task) {
+ taskCls = resolveTaskClass(taskName, taskCls, task);
+
+ if (taskCls != null && SecurityUtils.isSystemType(ctx, taskCls))
+ return;
Review Comment:
Could return be replaced with else if?
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
Review Comment:
Looks like correct mnemonic is about what it may do/be.
For example, PublicJob.
##########
modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java:
##########
@@ -68,10 +68,11 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
/** Custom executor name. */
private String execName;
+ /** */
+ private boolean isSysTaskGuard;
Review Comment:
Could this be replaced with something explains it's a public/private api?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1092306679
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/TaskExecutionOptions.java:
##########
@@ -176,13 +176,13 @@ public TaskExecutionOptions asSystemTask() {
}
/** */
- public boolean isAuthenticationDisabled() {
- return isAuthDisabled;
+ public boolean isSystemTaskGuard() {
+ return isSystemTaskGuard;
}
/** */
- public TaskExecutionOptions withAuthenticationDisabled() {
- isAuthDisabled = true;
+ public TaskExecutionOptions withSystemTaskGuard() {
+ isSystemTaskGuard = true;
return this;
}
Review Comment:
The option itself is not thread local.
Take a look at IgniteComputeImpl, please. We do use TaskExecutionOptions as thread local variable here, but isPublicFacade flag is stored separately and applied to each compute execution method. So it is safe.
##########
modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java:
##########
@@ -68,10 +68,11 @@ public class IgniteComputeImpl extends AsyncSupportAdapter<IgniteCompute>
/** Custom executor name. */
private String execName;
+ /** */
+ private boolean isSysTaskGuard;
Review Comment:
Fixed. Changed to isPublicFacade.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1736,4 +1763,51 @@ else if (!getLocJobStatistics && locNodeId.equals(jobResult.getNode().id()))
return S.toString(GridTaskWorker.class, this);
}
}
+
+ /** */
+ private void authorizeSystemJobStart(ComputeJob job) {
+ Object executable = unwrap(job);
+
+ if (!isSystemType(ctx, executable.getClass())) {
+ ctx.security().authorize(executable.getClass().getName(), TASK_EXECUTE);
+
+ return;
+ }
+
+ if (executable instanceof SecurityPermissionAware)
+ authorizeAll(ctx.security(), ((SecurityPermissionAware)executable).requiredPermissions());
+ else if (opts.isSystemTaskGuard()) {
+ throw new SecurityException("Access to Ignite Internal tasks is restricted" +
+ " [task=" + task.getClass().getName() + ", job=" + executable.getClass() + "]");
+ }
+ }
+
+ /** */
+ public void authorizeTaskCancel() {
+ if (!ctx.security().enabled())
+ return;
+
+ if (!isSystemType(ctx, task.getClass())) {
+ ctx.security().authorize(task.getClass().getName(), TASK_CANCEL);
+
+ return;
Review Comment:
Fixed.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java:
##########
@@ -1589,4 +1579,36 @@ public Map<ComputeJobStatusEnum, Long> jobStatuses(IgniteUuid sesId) {
else
return taskWorker.jobStatuses();
}
+
+ /** */
+ private void authorizeTaskExecution(String taskName, Class<?> taskCls, ComputeTask<?, ?> task) {
+ taskCls = resolveTaskClass(taskName, taskCls, task);
+
+ if (taskCls != null && SecurityUtils.isSystemType(ctx, taskCls))
+ return;
Review Comment:
Fixed.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java:
##########
@@ -1736,4 +1763,51 @@ else if (!getLocJobStatistics && locNodeId.equals(jobResult.getNode().id()))
return S.toString(GridTaskWorker.class, this);
}
}
+
+ /** */
+ private void authorizeSystemJobStart(ComputeJob job) {
+ Object executable = unwrap(job);
+
+ if (!isSystemType(ctx, executable.getClass())) {
+ ctx.security().authorize(executable.getClass().getName(), TASK_EXECUTE);
+
+ return;
Review Comment:
Fixed.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
+ /** */
+ public SecurityPermissionSet NO_PERMISSIONS = SecurityPermissionSetBuilder.create().build();
+
+ /** */
+ public SecurityPermissionSet requiredPermissions();
+
+ /** */
+ public static SecurityPermissionSet cachePermissions(String cacheName, SecurityPermission... perms) {
Review Comment:
Fixed.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityPermissionAware.java:
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.ignite.internal.processors.security;
+
+import org.apache.ignite.plugin.security.SecurityPermission;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+
+/** */
+public interface SecurityPermissionAware {
+ /** */
+ public SecurityPermissionSet NO_PERMISSIONS = SecurityPermissionSetBuilder.create().build();
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "anton-vinogradov (via GitHub)" <gi...@apache.org>.
anton-vinogradov commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1095772220
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/security/SecurityUtils.java:
##########
@@ -385,4 +393,27 @@ public static SecurityContext authenticateLocalNode(
return secCtx;
}
+
+ /** */
+ public static void authorizeAll(IgniteSecurity security, SecurityPermissionSet permissions) {
+ if (!security.enabled())
+ return;
+
+ if (!F.isEmpty(permissions.systemPermissions())) {
Review Comment:
null check seems to be enough
##########
modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java:
##########
@@ -0,0 +1,624 @@
+/*
+ * 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.ignite.common;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientAuthorizationException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.GridClient;
+import org.apache.ignite.internal.client.GridClientConfiguration;
+import org.apache.ignite.internal.client.GridClientFactory;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataCollectorTask;
+import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
+import org.apache.ignite.internal.processors.security.compute.ComputePermissionCheckTest;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityData;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityPluginProvider;
+import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.lang.gridfunc.AtomicIntegerFactoryCallable;
+import org.apache.ignite.internal.util.lang.gridfunc.RunnableWrapperClosure;
+import org.apache.ignite.internal.util.lang.gridfunc.ToStringClosure;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.plugin.security.SecurityCredentials;
+import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
+import org.apache.ignite.plugin.security.SecurityException;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static java.util.Collections.singletonList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.common.AbstractEventSecurityContextTest.sendRestRequest;
+import static org.apache.ignite.internal.processors.rest.GridRestCommand.EXE;
+import static org.apache.ignite.plugin.security.SecurityPermission.CACHE_CREATE;
+import static org.apache.ignite.plugin.security.SecurityPermission.JOIN_AS_SERVER;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_CANCEL;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_EXECUTE;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class ComputeTaskPermissionsTest extends AbstractSecurityTest {
+ /** */
+ private static final IgniteCallable SYSTEM_CALLABLE = new AtomicIntegerFactoryCallable();
+
+ /** */
+ private static final IgniteRunnable SYSTEM_RUNNABLE = new RunnableWrapperClosure(null);
+
+ /** */
+ private static final IgniteClosure SYSTEM_CLOSURE = new ToStringClosure<>();
+
+ /** */
+ private static final ComputeTask SYSTEM_TASK = new SnapshotMetadataCollectorTask();
+
+ /** */
+ private static final AtomicInteger EXECUTED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final AtomicInteger CANCELLED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final String CACHE = DEFAULT_CACHE_NAME;
+
+ /** */
+ private static final int SRV_NODES_CNT = 2;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ for (int idx = 0; idx < SRV_NODES_CNT; idx++)
+ startGrid(idx, false);
+
+ startGrid(SRV_NODES_CNT, true);
+
+ grid(0).createCache(CACHE);
+ }
+
+ /** */
+ private IgniteEx startGrid(int idx, boolean isClient) throws Exception {
+ String login = getTestIgniteInstanceName(idx);
+
+ SecurityPermissionSetBuilder permsBuilder = new SecurityPermissionSetBuilder()
+ .defaultAllowAll(false)
+ .appendTaskPermissions(AllowedCallable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedCallableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedComputeTask.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedClosure.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(CancelAllowedTask.class.getName(), TASK_EXECUTE, TASK_CANCEL)
+ .appendTaskPermissions(CancelForbiddenTask.class.getName(), TASK_EXECUTE);
+
+ if (!isClient)
+ permsBuilder.appendSystemPermissions(CACHE_CREATE, JOIN_AS_SERVER);
+
+ SecurityPermissionSet permissions = permsBuilder.build();
+
+ IgniteConfiguration cfg = getConfiguration(
+ login,
+ new TestSecurityPluginProvider(
+ login,
+ "",
+ permissions,
+ null,
+ false,
+ new TestSecurityData("ignite-client", permissions),
+ new TestSecurityData("grid-client", permissions),
+ new TestSecurityData("rest-client", permissions)
+ )).setClientMode(isClient);
+
+ cfg.setConnectorConfiguration(new ConnectorConfiguration()
+ .setJettyPath("modules/clients/src/test/resources/jetty/rest-jetty.xml"))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(1)));
+
+ return startGrid(cfg);
+ }
+
+ /** */
+ @Test
+ public void testNode() throws Exception {
+ doNodeTest(0, 0);
+ doNodeTest(0, 1);
+ doNodeTest(2, 1);
+ }
+
+ /** */
+ @Test
+ public void testIgniteClient() throws Exception {
+ try (
+ IgniteClient cli = Ignition.startClient(new ClientConfiguration()
+ .setUserName("ignite-client")
+ .setUserPassword("")
+ .setAddresses("127.0.0.1:10800"))
+ ) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync2(t, a).get());
+
+ testTaskCancelFailed(t -> cli.compute().executeAsync2(t, null));
+ testTaskCancelSucceeded(t -> cli.compute().executeAsync2(t, null));
+ }
+ }
+
+ /** */
+ @Test
+ public void testGridClient() throws Exception {
+ GridClientConfiguration cfg = new GridClientConfiguration()
+ .setServers(singletonList("127.0.0.1:11211"))
+ .setSecurityCredentialsProvider(new SecurityCredentialsBasicProvider(new SecurityCredentials("grid-client", "")));
+
+ try (GridClient cli = GridClientFactory.start(cfg)) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecute(t, CACHE, "key", a));
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecuteAsync(t, CACHE, "key", a).get());
+ }
+ }
+
+ /** */
+ @Test
+ public void testRestClient() {
+ for (Ignite ignite : Ignition.allGrids()) {
+ checkTask(String.class, (t, a) -> {
+ JsonNode resp = sendRestRequest(
+ EXE,
+ Arrays.asList(
+ "name=" + t,
+ "destId=" + ignite.cluster().localNode().id()
+ ),
+ "rest-client"
+ );
+
+ if (0 != resp.get("successStatus").intValue())
+ throw new IgniteException(resp.get("error").textValue());
+ });
+ }
+ }
+
+ /** */
+ private void doNodeTest(int initiator, int executor) throws Exception {
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(String.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(String.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkCallable(t -> compute(initiator, executor).call(t));
+ checkCallable(t -> compute(initiator, executor).callAsync(t).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c));
+ checkCallables(c -> compute(initiator, executor).callAsync(c).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c, new TestReducer()));
+ checkCallables(c -> compute(initiator, executor).callAsync(c, new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).run(r));
+ checkRunnable(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnables(r -> compute(initiator, executor).run(r));
+ checkRunnables(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnable(r -> compute(initiator, executor).broadcast(r));
+ checkRunnable(r -> compute(initiator, executor).broadcastAsync(r).get());
+
+ checkCallable(c -> compute(initiator, executor).broadcast(c));
+ checkCallable(c -> compute(initiator, executor).broadcastAsync(c).get());
+
+ checkClosure(c -> compute(initiator, executor).broadcast(c, null));
+ checkClosure(c -> compute(initiator, executor).broadcastAsync(c, null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, (Object)null));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, (Object)null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, singletonList(null), new TestReducer()));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, singletonList(null), new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(CACHE, keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(CACHE, keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), partitionForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(CACHE, keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(CACHE, keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), partitionForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkRunnable(r -> executorService(initiator, executor).submit(r).get());
+ checkRunnable(r -> executorService(initiator, executor).submit(r, null).get());
+ checkCallable(c -> executorService(initiator, executor).submit(c).get());
+
+ checkCallable(c -> executorService(initiator, executor).invokeAll(singletonList(c)).forEach(ComputePermissionCheckTest::getQuiet));
+ checkCallable(c ->
+ executorService(initiator, executor).invokeAll(
+ singletonList(c),
+ getTestTimeout(),
+ MILLISECONDS
+ ).forEach(ComputePermissionCheckTest::getQuiet)
+ );
+
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c)));
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c), getTestTimeout(), MILLISECONDS));
+ }
+
+ /** */
+ private IgniteCompute compute(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.compute(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private ExecutorService executorService(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.executorService(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private void checkRunnable(ConsumerX<IgniteRunnable> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedRunnable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenRunnable()));
+ assertFailed(() -> consumer.accept(SYSTEM_RUNNABLE));
+ }
+
+ /** */
+ private void checkRunnables(ConsumerX<Collection<IgniteRunnable>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new AllowedRunnableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new ForbiddenRunnable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), SYSTEM_RUNNABLE)));
+ }
+
+ /** */
+ private void checkCallable(ConsumerX<IgniteCallable<AtomicInteger>> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedCallable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenCallable()));
+ assertFailed(() -> consumer.accept(SYSTEM_CALLABLE));
+ }
+
+ /** */
+ private void checkCallables(ConsumerX<Collection<IgniteCallable<AtomicInteger>>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new AllowedCallableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new ForbiddenCallable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), SYSTEM_CALLABLE)));
+ }
+
+ /** */
+ private void checkClosure(Consumer<IgniteClosure> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedClosure()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenClosure()));
+ assertFailed(() -> consumer.accept(SYSTEM_CLOSURE));
+ }
+
+ /** */
+ private <T> void checkTask(Class<T> cls, BiConsumerX<T, Object> consumer) {
+ assertCompleted(() -> consumer.accept(cast(new AllowedComputeTask(), cls), null), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(cast(new ForbiddenComputeTask(), cls), null));
+ assertFailed(() -> consumer.accept(cast(SYSTEM_TASK, cls), null));
+ }
+
+ /** */
+ private void testTaskCancelFailed(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ EXECUTED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelForbiddenTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ GridTestUtils.assertThrowsWithCause(() -> fut.cancel(true), ClientAuthorizationException.class);
+
+ assertFalse(fut.isCancelled());
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == EXECUTED_TASK_CNTR.get(), getTestTimeout()));
+ }
+
+ /** */
+ private void testTaskCancelSucceeded(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ CANCELLED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelAllowedTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ assertTrue(fut.cancel(true));
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == CANCELLED_TASK_CNTR.get(), getTestTimeout()));
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+ }
+
+ /** */
+ private void assertCompleted(RunnableX r, int expCnt) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ r.run();
+
+ assertEquals(expCnt, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ private void assertFailed(RunnableX r) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ try {
+ r.run();
+ }
+ catch (Exception e) {
+ if (X.hasCause(e, ClientAuthorizationException.class) || X.hasCause(e, SecurityException.class))
+ return;
+
+ if (e instanceof IgniteException) {
+ assertTrue(
+ e.getMessage().contains("Authorization failed")
+ || e.getMessage().contains("Access to Ignite Internal tasks is restricted")
+ || e.getMessage().contains("Failed to get any task completion")
+ );
+
+ return;
+ }
+ }
+
+ fail();
+
+ assertEquals(0, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ public <T> T cast(Object obj, Class<T> cls) {
+ if (String.class.equals(cls))
+ return (T)obj.getClass().getName();
+ else if (Class.class.equals(cls))
+ return (T)obj.getClass();
+ else if (ComputeTask.class.equals(cls))
+ return (T)obj;
+ else
+ throw new IllegalStateException();
+ }
+
+ /** */
+ private int keyForNode(int nodeIdx) {
+ return keyForNode(grid(0).affinity(CACHE), new AtomicInteger(0), grid(nodeIdx).cluster().localNode());
+ }
+
+ /** */
+ private int partitionForNode(int nodeIdx) {
+ return grid(0).affinity(CACHE).partition(keyForNode(nodeIdx));
+ }
+
+ /** */
+ private static class AllowedComputeTask extends AbstractTask { }
+
+ /** */
+ private static class ForbiddenComputeTask extends AbstractTask { }
+
+ /** */
+ private static class AllowedCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedCallableSecond extends AbstractCallable { }
+
+ /** */
+ private static class ForbiddenCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedRunnableSecond extends AbstractRunnable { }
+
+ /** */
+ private static class ForbiddenRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedClosure extends AbstractClosure { }
+
+ /** */
+ private static class ForbiddenClosure extends AbstractClosure { }
+
+ /** */
+ private static class CancelAllowedTask extends AbstractHangingTask { }
+
+ /** */
+ private static class CancelForbiddenTask extends AbstractHangingTask { }
+
+ /** */
+ private static class AbstractRunnable implements IgniteRunnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+ }
+ }
+
+ /** */
+ private static class AbstractCallable implements IgniteCallable<AtomicInteger> {
+ /** {@inheritDoc} */
+ @Override public AtomicInteger call() throws Exception {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return new AtomicInteger(0);
+ }
+ }
+
+ /** */
+ private static class AbstractClosure implements IgniteClosure<Boolean, Boolean> {
+ /** {@inheritDoc} */
+ @Override public Boolean apply(Boolean o) {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+ }
+
+ /** */
+ private static class AbstractHangingTask extends AbstractTask {
+ /** */
+ public static CountDownLatch taskStartedLatch;
+
Review Comment:
abstrast class should be marked as abstract
##########
modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java:
##########
@@ -0,0 +1,624 @@
+/*
+ * 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.ignite.common;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCompute;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.client.ClientAuthorizationException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeTask;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ConnectorConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.client.GridClient;
+import org.apache.ignite.internal.client.GridClientConfiguration;
+import org.apache.ignite.internal.client.GridClientFactory;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataCollectorTask;
+import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
+import org.apache.ignite.internal.processors.security.compute.ComputePermissionCheckTest;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityData;
+import org.apache.ignite.internal.processors.security.impl.TestSecurityPluginProvider;
+import org.apache.ignite.internal.util.lang.RunnableX;
+import org.apache.ignite.internal.util.lang.gridfunc.AtomicIntegerFactoryCallable;
+import org.apache.ignite.internal.util.lang.gridfunc.RunnableWrapperClosure;
+import org.apache.ignite.internal.util.lang.gridfunc.ToStringClosure;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.lang.IgniteCallable;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgniteReducer;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.plugin.security.SecurityCredentials;
+import org.apache.ignite.plugin.security.SecurityCredentialsBasicProvider;
+import org.apache.ignite.plugin.security.SecurityException;
+import org.apache.ignite.plugin.security.SecurityPermissionSet;
+import org.apache.ignite.plugin.security.SecurityPermissionSetBuilder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static java.util.Collections.singletonList;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.ignite.common.AbstractEventSecurityContextTest.sendRestRequest;
+import static org.apache.ignite.internal.processors.rest.GridRestCommand.EXE;
+import static org.apache.ignite.plugin.security.SecurityPermission.CACHE_CREATE;
+import static org.apache.ignite.plugin.security.SecurityPermission.JOIN_AS_SERVER;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_CANCEL;
+import static org.apache.ignite.plugin.security.SecurityPermission.TASK_EXECUTE;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/** */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class ComputeTaskPermissionsTest extends AbstractSecurityTest {
+ /** */
+ private static final IgniteCallable SYSTEM_CALLABLE = new AtomicIntegerFactoryCallable();
+
+ /** */
+ private static final IgniteRunnable SYSTEM_RUNNABLE = new RunnableWrapperClosure(null);
+
+ /** */
+ private static final IgniteClosure SYSTEM_CLOSURE = new ToStringClosure<>();
+
+ /** */
+ private static final ComputeTask SYSTEM_TASK = new SnapshotMetadataCollectorTask();
+
+ /** */
+ private static final AtomicInteger EXECUTED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final AtomicInteger CANCELLED_TASK_CNTR = new AtomicInteger();
+
+ /** */
+ private static final String CACHE = DEFAULT_CACHE_NAME;
+
+ /** */
+ private static final int SRV_NODES_CNT = 2;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ for (int idx = 0; idx < SRV_NODES_CNT; idx++)
+ startGrid(idx, false);
+
+ startGrid(SRV_NODES_CNT, true);
+
+ grid(0).createCache(CACHE);
+ }
+
+ /** */
+ private IgniteEx startGrid(int idx, boolean isClient) throws Exception {
+ String login = getTestIgniteInstanceName(idx);
+
+ SecurityPermissionSetBuilder permsBuilder = new SecurityPermissionSetBuilder()
+ .defaultAllowAll(false)
+ .appendTaskPermissions(AllowedCallable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedCallableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnable.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedRunnableSecond.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedComputeTask.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(AllowedClosure.class.getName(), TASK_EXECUTE)
+ .appendTaskPermissions(CancelAllowedTask.class.getName(), TASK_EXECUTE, TASK_CANCEL)
+ .appendTaskPermissions(CancelForbiddenTask.class.getName(), TASK_EXECUTE);
+
+ if (!isClient)
+ permsBuilder.appendSystemPermissions(CACHE_CREATE, JOIN_AS_SERVER);
+
+ SecurityPermissionSet permissions = permsBuilder.build();
+
+ IgniteConfiguration cfg = getConfiguration(
+ login,
+ new TestSecurityPluginProvider(
+ login,
+ "",
+ permissions,
+ null,
+ false,
+ new TestSecurityData("ignite-client", permissions),
+ new TestSecurityData("grid-client", permissions),
+ new TestSecurityData("rest-client", permissions)
+ )).setClientMode(isClient);
+
+ cfg.setConnectorConfiguration(new ConnectorConfiguration()
+ .setJettyPath("modules/clients/src/test/resources/jetty/rest-jetty.xml"))
+ .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+ .setThinClientConfiguration(new ThinClientConfiguration()
+ .setMaxActiveComputeTasksPerConnection(1)));
+
+ return startGrid(cfg);
+ }
+
+ /** */
+ @Test
+ public void testNode() throws Exception {
+ doNodeTest(0, 0);
+ doNodeTest(0, 1);
+ doNodeTest(2, 1);
+ }
+
+ /** */
+ @Test
+ public void testIgniteClient() throws Exception {
+ try (
+ IgniteClient cli = Ignition.startClient(new ClientConfiguration()
+ .setUserName("ignite-client")
+ .setUserPassword("")
+ .setAddresses("127.0.0.1:10800"))
+ ) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync2(t, a).get());
+
+ testTaskCancelFailed(t -> cli.compute().executeAsync2(t, null));
+ testTaskCancelSucceeded(t -> cli.compute().executeAsync2(t, null));
+ }
+ }
+
+ /** */
+ @Test
+ public void testGridClient() throws Exception {
+ GridClientConfiguration cfg = new GridClientConfiguration()
+ .setServers(singletonList("127.0.0.1:11211"))
+ .setSecurityCredentialsProvider(new SecurityCredentialsBasicProvider(new SecurityCredentials("grid-client", "")));
+
+ try (GridClient cli = GridClientFactory.start(cfg)) {
+ checkTask(String.class, (t, a) -> cli.compute().execute(t, a));
+ checkTask(String.class, (t, a) -> cli.compute().executeAsync(t, a).get());
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecute(t, CACHE, "key", a));
+ checkTask(String.class, (t, a) -> cli.compute().affinityExecuteAsync(t, CACHE, "key", a).get());
+ }
+ }
+
+ /** */
+ @Test
+ public void testRestClient() {
+ for (Ignite ignite : Ignition.allGrids()) {
+ checkTask(String.class, (t, a) -> {
+ JsonNode resp = sendRestRequest(
+ EXE,
+ Arrays.asList(
+ "name=" + t,
+ "destId=" + ignite.cluster().localNode().id()
+ ),
+ "rest-client"
+ );
+
+ if (0 != resp.get("successStatus").intValue())
+ throw new IgniteException(resp.get("error").textValue());
+ });
+ }
+ }
+
+ /** */
+ private void doNodeTest(int initiator, int executor) throws Exception {
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(ComputeTask.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(Class.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkTask(String.class, (t, a) -> grid(initiator).compute().execute(t, a));
+ checkTask(String.class, (t, a) -> grid(initiator).compute().executeAsync(t, a).get());
+
+ checkCallable(t -> compute(initiator, executor).call(t));
+ checkCallable(t -> compute(initiator, executor).callAsync(t).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c));
+ checkCallables(c -> compute(initiator, executor).callAsync(c).get());
+
+ checkCallables(c -> compute(initiator, executor).call(c, new TestReducer()));
+ checkCallables(c -> compute(initiator, executor).callAsync(c, new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).run(r));
+ checkRunnable(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnables(r -> compute(initiator, executor).run(r));
+ checkRunnables(r -> compute(initiator, executor).runAsync(r).get());
+
+ checkRunnable(r -> compute(initiator, executor).broadcast(r));
+ checkRunnable(r -> compute(initiator, executor).broadcastAsync(r).get());
+
+ checkCallable(c -> compute(initiator, executor).broadcast(c));
+ checkCallable(c -> compute(initiator, executor).broadcastAsync(c).get());
+
+ checkClosure(c -> compute(initiator, executor).broadcast(c, null));
+ checkClosure(c -> compute(initiator, executor).broadcastAsync(c, null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, (Object)null));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, (Object)null).get());
+
+ checkClosure(c -> compute(initiator, executor).apply(c, singletonList(null), new TestReducer()));
+ checkClosure(c -> compute(initiator, executor).applyAsync(c, singletonList(null), new TestReducer()).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(CACHE, keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(CACHE, keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), keyForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkRunnable(r -> compute(initiator, executor).affinityRun(singletonList(CACHE), partitionForNode(executor), r));
+ checkRunnable(r -> compute(initiator, executor).affinityRunAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(CACHE, keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(CACHE, keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), keyForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), keyForNode(executor), r).get());
+
+ checkCallable(r -> compute(initiator, executor).affinityCall(singletonList(CACHE), partitionForNode(executor), r));
+ checkCallable(r -> compute(initiator, executor).affinityCallAsync(singletonList(CACHE), partitionForNode(executor), r).get());
+
+ checkRunnable(r -> executorService(initiator, executor).submit(r).get());
+ checkRunnable(r -> executorService(initiator, executor).submit(r, null).get());
+ checkCallable(c -> executorService(initiator, executor).submit(c).get());
+
+ checkCallable(c -> executorService(initiator, executor).invokeAll(singletonList(c)).forEach(ComputePermissionCheckTest::getQuiet));
+ checkCallable(c ->
+ executorService(initiator, executor).invokeAll(
+ singletonList(c),
+ getTestTimeout(),
+ MILLISECONDS
+ ).forEach(ComputePermissionCheckTest::getQuiet)
+ );
+
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c)));
+ checkCallable(c -> executorService(initiator, executor).invokeAny(singletonList(c), getTestTimeout(), MILLISECONDS));
+ }
+
+ /** */
+ private IgniteCompute compute(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.compute(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private ExecutorService executorService(int initiator, int executor) {
+ IgniteEx ignite = grid(initiator);
+
+ return ignite.executorService(ignite.cluster().forNodeId(grid(executor).localNode().id()));
+ }
+
+ /** */
+ private void checkRunnable(ConsumerX<IgniteRunnable> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedRunnable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenRunnable()));
+ assertFailed(() -> consumer.accept(SYSTEM_RUNNABLE));
+ }
+
+ /** */
+ private void checkRunnables(ConsumerX<Collection<IgniteRunnable>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new AllowedRunnableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), new ForbiddenRunnable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedRunnable(), SYSTEM_RUNNABLE)));
+ }
+
+ /** */
+ private void checkCallable(ConsumerX<IgniteCallable<AtomicInteger>> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedCallable()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenCallable()));
+ assertFailed(() -> consumer.accept(SYSTEM_CALLABLE));
+ }
+
+ /** */
+ private void checkCallables(ConsumerX<Collection<IgniteCallable<AtomicInteger>>> consumer) {
+ assertCompleted(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new AllowedCallableSecond())), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), new ForbiddenCallable())));
+ assertFailed(() -> consumer.accept(Arrays.asList(new AllowedCallable(), SYSTEM_CALLABLE)));
+ }
+
+ /** */
+ private void checkClosure(Consumer<IgniteClosure> consumer) {
+ assertCompleted(() -> consumer.accept(new AllowedClosure()), 1);
+ assertFailed(() -> consumer.accept(new ForbiddenClosure()));
+ assertFailed(() -> consumer.accept(SYSTEM_CLOSURE));
+ }
+
+ /** */
+ private <T> void checkTask(Class<T> cls, BiConsumerX<T, Object> consumer) {
+ assertCompleted(() -> consumer.accept(cast(new AllowedComputeTask(), cls), null), SRV_NODES_CNT);
+ assertFailed(() -> consumer.accept(cast(new ForbiddenComputeTask(), cls), null));
+ assertFailed(() -> consumer.accept(cast(SYSTEM_TASK, cls), null));
+ }
+
+ /** */
+ private void testTaskCancelFailed(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ EXECUTED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelForbiddenTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ GridTestUtils.assertThrowsWithCause(() -> fut.cancel(true), ClientAuthorizationException.class);
+
+ assertFalse(fut.isCancelled());
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == EXECUTED_TASK_CNTR.get(), getTestTimeout()));
+ }
+
+ /** */
+ private void testTaskCancelSucceeded(Function<String, Future<?>> taskStarter) throws Exception {
+ AbstractHangingTask.taskStartedLatch = new CountDownLatch(SRV_NODES_CNT);
+ AbstractHangingTask.taskUnblockedLatch = new CountDownLatch(1);
+
+ CANCELLED_TASK_CNTR.set(0);
+
+ Future<?> fut = taskStarter.apply(CancelAllowedTask.class.getName());
+
+ assertTrue(AbstractHangingTask.taskStartedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ assertTrue(fut.cancel(true));
+
+ assertTrue(waitForCondition(() -> SRV_NODES_CNT == CANCELLED_TASK_CNTR.get(), getTestTimeout()));
+
+ AbstractHangingTask.taskUnblockedLatch.countDown();
+ }
+
+ /** */
+ private void assertCompleted(RunnableX r, int expCnt) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ r.run();
+
+ assertEquals(expCnt, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ private void assertFailed(RunnableX r) {
+ EXECUTED_TASK_CNTR.set(0);
+
+ try {
+ r.run();
+ }
+ catch (Exception e) {
+ if (X.hasCause(e, ClientAuthorizationException.class) || X.hasCause(e, SecurityException.class))
+ return;
+
+ if (e instanceof IgniteException) {
+ assertTrue(
+ e.getMessage().contains("Authorization failed")
+ || e.getMessage().contains("Access to Ignite Internal tasks is restricted")
+ || e.getMessage().contains("Failed to get any task completion")
+ );
+
+ return;
+ }
+ }
+
+ fail();
+
+ assertEquals(0, EXECUTED_TASK_CNTR.get());
+ }
+
+ /** */
+ public <T> T cast(Object obj, Class<T> cls) {
+ if (String.class.equals(cls))
+ return (T)obj.getClass().getName();
+ else if (Class.class.equals(cls))
+ return (T)obj.getClass();
+ else if (ComputeTask.class.equals(cls))
+ return (T)obj;
+ else
+ throw new IllegalStateException();
+ }
+
+ /** */
+ private int keyForNode(int nodeIdx) {
+ return keyForNode(grid(0).affinity(CACHE), new AtomicInteger(0), grid(nodeIdx).cluster().localNode());
+ }
+
+ /** */
+ private int partitionForNode(int nodeIdx) {
+ return grid(0).affinity(CACHE).partition(keyForNode(nodeIdx));
+ }
+
+ /** */
+ private static class AllowedComputeTask extends AbstractTask { }
+
+ /** */
+ private static class ForbiddenComputeTask extends AbstractTask { }
+
+ /** */
+ private static class AllowedCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedCallableSecond extends AbstractCallable { }
+
+ /** */
+ private static class ForbiddenCallable extends AbstractCallable { }
+
+ /** */
+ private static class AllowedRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedRunnableSecond extends AbstractRunnable { }
+
+ /** */
+ private static class ForbiddenRunnable extends AbstractRunnable { }
+
+ /** */
+ private static class AllowedClosure extends AbstractClosure { }
+
+ /** */
+ private static class ForbiddenClosure extends AbstractClosure { }
+
+ /** */
+ private static class CancelAllowedTask extends AbstractHangingTask { }
+
+ /** */
+ private static class CancelForbiddenTask extends AbstractHangingTask { }
+
+ /** */
+ private static class AbstractRunnable implements IgniteRunnable {
+ /** {@inheritDoc} */
+ @Override public void run() {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+ }
+ }
+
+ /** */
+ private static class AbstractCallable implements IgniteCallable<AtomicInteger> {
+ /** {@inheritDoc} */
+ @Override public AtomicInteger call() throws Exception {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return new AtomicInteger(0);
+ }
+ }
+
+ /** */
+ private static class AbstractClosure implements IgniteClosure<Boolean, Boolean> {
+ /** {@inheritDoc} */
+ @Override public Boolean apply(Boolean o) {
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+ }
+
+ /** */
+ private static class AbstractHangingTask extends AbstractTask {
+ /** */
+ public static CountDownLatch taskStartedLatch;
+
+ /** */
+ public static CountDownLatch taskUnblockedLatch;
+
+ /** {@inheritDoc} */
+ @Override protected ComputeJob job() {
+ return new Job();
+ }
+
+ /** */
+ private static class Job implements ComputeJob {
+ /** {@inheritDoc} */
+ @Override public Object execute() {
+ LockSupport.parkNanos(10_000_000);
+
+ taskStartedLatch.countDown();
+
+ try {
+ taskUnblockedLatch.await(5_000, MILLISECONDS);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+
+ throw new IgniteException(e);
+ }
+
+ EXECUTED_TASK_CNTR.incrementAndGet();
+
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void cancel() {
+ CANCELLED_TASK_CNTR.incrementAndGet();
+ }
+ }
+ }
+
+ /** */
+ private static class AbstractTask extends ComputeTaskAdapter<Object, Object> {
+ /** {@inheritDoc} */
+ @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ @Nullable Object arg
+ ) throws IgniteException {
+ return subgrid.stream().filter(g -> !g.isClient()).collect(Collectors.toMap(g -> job(), g -> g, (l, r) -> l));
Review Comment:
`g` is for everything? %)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] petrov-mg commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "petrov-mg (via GitHub)" <gi...@apache.org>.
petrov-mg commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1094354071
##########
modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java:
##########
@@ -201,16 +204,29 @@ public void setKernalContext(GridKernalContext ctx) {
}
/**
- * @return {@link IgniteCompute} for this cluster group.
+ * @return Internal {@link IgniteCompute} facade for this cluster group.
+ */
+ public final IgniteCompute internalComputeFacade() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "anton-vinogradov (via GitHub)" <gi...@apache.org>.
anton-vinogradov commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1095578728
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/task/TaskExecutionOptions.java:
##########
@@ -176,13 +176,13 @@ public TaskExecutionOptions asSystemTask() {
}
/** */
- public boolean isAuthenticationDisabled() {
- return isAuthDisabled;
+ public boolean isPublicRequest() {
+ return isPublicReq;
}
/** */
- public TaskExecutionOptions withAuthenticationDisabled() {
- isAuthDisabled = true;
+ public TaskExecutionOptions asPublicRequest() {
Review Comment:
Please fill javadoc explains this option
##########
modules/core/src/main/java/org/apache/ignite/internal/IgniteComputeImpl.java:
##########
@@ -84,20 +85,8 @@ public IgniteComputeImpl() {
* @param ctx Kernal context.
* @param prj Projection.
*/
- public IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj) {
- this(ctx, prj, false);
- }
-
- /**
- * @param ctx Kernal context.
- * @param prj Projection.
- * @param async Async support flag.
- */
- private IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, boolean async) {
- super(async);
-
- this.ctx = ctx;
- this.prj = prj;
+ public IgniteComputeImpl(GridKernalContext ctx, ClusterGroupAdapter prj, boolean isPublicFacade) {
Review Comment:
Let's replace with some SecurityAwareIgniteComputeImpl extends IgniteComputeImpl wrapper instead of dedicated constructor.
##########
modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcResultSet.java:
##########
@@ -268,9 +268,10 @@ void closeInternal() throws SQLException {
JdbcConnection conn = (JdbcConnection)stmt.getConnection();
if (conn.isCloseCursorTaskSupported()) {
- Ignite ignite = conn.ignite();
+ IgniteEx ignite = conn.ignite();
- ignite.compute(ignite.cluster().forNodeId(conn.nodeId())).call(new JdbcCloseCursorTask(uuid));
+ ignite.internalCompute(ignite.cluster().forNodeId(conn.nodeId()))
+ .call(new JdbcCloseCursorTask(uuid));
Review Comment:
no reason fro newline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a diff in pull request #10226: IGNITE-15322 WIP
Posted by "anton-vinogradov (via GitHub)" <gi...@apache.org>.
anton-vinogradov commented on code in PR #10226:
URL: https://github.com/apache/ignite/pull/10226#discussion_r1095580442
##########
modules/core/src/main/java/org/apache/ignite/internal/jdbc2/JdbcResultSet.java:
##########
@@ -268,9 +268,10 @@ void closeInternal() throws SQLException {
JdbcConnection conn = (JdbcConnection)stmt.getConnection();
if (conn.isCloseCursorTaskSupported()) {
- Ignite ignite = conn.ignite();
+ IgniteEx ignite = conn.ignite();
- ignite.compute(ignite.cluster().forNodeId(conn.nodeId())).call(new JdbcCloseCursorTask(uuid));
+ ignite.internalCompute(ignite.cluster().forNodeId(conn.nodeId()))
+ .call(new JdbcCloseCursorTask(uuid));
Review Comment:
no reason for newline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org