IGNITE-28753 : Replace SecurityAwareMessage with a common OperationContextMessage - test fix

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -28,7 +28,6 @@
 import org.apache.ignite.internal.managers.communication.CompressedMessage;
 import org.apache.ignite.internal.managers.communication.ErrorMessage;
 import org.apache.ignite.internal.managers.communication.GridIoMessage;
-import org.apache.ignite.internal.managers.communication.GridIoSecurityAwareMessage;
 import org.apache.ignite.internal.managers.communication.GridIoUserMessage;
 import org.apache.ignite.internal.managers.communication.IgniteIoTestMessage;
 import org.apache.ignite.internal.managers.communication.SessionChannelMessage;
@@ -240,6 +239,7 @@
 import org.apache.ignite.internal.processors.rollingupgrade.RollingUpgradeNodeData;
 import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureSet;
 import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteProductFeatures;
+import org.apache.ignite.internal.processors.security.SecurityContextImpl;
 import org.apache.ignite.internal.processors.service.ServiceChangeBatchRequest;
 import org.apache.ignite.internal.processors.service.ServiceClusterDeploymentResult;
 import org.apache.ignite.internal.processors.service.ServiceClusterDeploymentResultBatch;
@@ -608,12 +608,14 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         // [11500 - 11600]:  IO, networking messages.
         msgIdx = NODE_ID_MSG_TYPE;
         withNoSchema(NodeIdMessage.class);
+        msgIdx = HANDSHAKE_MSG_TYPE;
         withNoSchema(HandshakeMessage.class);
+        msgIdx = HANDSHAKE_WAIT_MSG_TYPE;
         withNoSchema(HandshakeWaitMessage.class);
         withNoSchema(GridIoMessage.class);
         withNoSchema(IgniteIoTestMessage.class);
         withSchema(GridIoUserMessage.class);
-        withSchema(GridIoSecurityAwareMessage.class);
+        ++msgIdx; // Former GridIoSecurityAwareMessage
         withNoSchema(RecoveryLastReceivedMessage.class);
         withNoSchema(TcpInverseConnectionResponseMessage.class);
         withNoSchema(SessionChannelMessage.class);
@@ -690,9 +692,10 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         // [13400 - 13500]: Operation context messages.
         msgIdx = 13400;
         withNoSchema(OperationContextMessage.class);
+        withNoSchema(SecurityContextImpl.class);
 
-        // [13500 - 13600]: Rolling Upgrade messages.
-        msgIdx = 13500;
+        // [13600 - 13700]: Rolling Upgrade messages.
+        msgIdx = 13600;
         withNoSchema(IgniteFeatureSet.class);
         withNoSchema(IgniteProductFeatures.class);
         withNoSchema(RollingUpgradeNodeData.class);

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java

@@ -97,11 +97,14 @@
 import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
 import org.apache.ignite.internal.processors.platform.message.PlatformMessageFilter;
 import org.apache.ignite.internal.processors.pool.PoolProcessor;
+import org.apache.ignite.internal.processors.security.IgniteSecurityProcessor;
+import org.apache.ignite.internal.processors.security.SecurityContextImpl;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
 import org.apache.ignite.internal.processors.tracing.MTC;
 import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
 import org.apache.ignite.internal.processors.tracing.Span;
 import org.apache.ignite.internal.processors.tracing.SpanTags;
+import org.apache.ignite.internal.thread.context.OperationContext;
 import org.apache.ignite.internal.thread.context.Scope;
 import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashSet;
 import org.apache.ignite.internal.util.IgniteUtils;
@@ -137,7 +140,6 @@
 import org.apache.ignite.spi.communication.tcp.internal.ConnectionRequestor;
 import org.apache.ignite.spi.communication.tcp.internal.TcpConnectionRequestDiscoveryMessage;
 import org.apache.ignite.spi.communication.tcp.internal.TcpInverseConnectionResponseMessage;
-import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -1317,7 +1319,7 @@ private void processP2PMessage(
 
                     assert obj != null;
 
-                    invokeListener(msg.policy(), lsnr, nodeId, obj, secSubjId(msg));
+                    invokeListener(msg.policy(), lsnr, nodeId, obj);
                 }
                 finally {
                     threadProcessingMessage(false, null);
@@ -1455,7 +1457,7 @@ private void processRegularMessage0(GridIoMessage msg, UUID nodeId) {
 
         assert obj != null;
 
-        invokeListener(msg.policy(), lsnr, nodeId, obj, secSubjId(msg));
+        invokeListener(msg.policy(), lsnr, nodeId, obj);
     }
 
     /**
@@ -1819,9 +1821,8 @@ private void unwindMessageSet(GridCommunicationMessageSet msgSet, GridMessageLis
      * @param lsnr Listener.
      * @param nodeId Node ID.
      * @param msg Message.
-     * @param secSubjId Security subject that will be used to open a security session.
      */
-    private void invokeListener(Byte plc, GridMessageListener lsnr, UUID nodeId, Object msg, UUID secSubjId) {
+    private void invokeListener(Byte plc, GridMessageListener lsnr, UUID nodeId, Object msg) {
         MTC.span().addLog(() -> "Invoke listener");
 
         Byte oldPlc = CUR_PLC.get();
@@ -1831,7 +1832,9 @@ private void invokeListener(Byte plc, GridMessageListener lsnr, UUID nodeId, Obj
         if (change)
             CUR_PLC.set(plc);
 
-        UUID newSecSubjId = secSubjId != null ? secSubjId : nodeId;
+        SecurityContextImpl secCtxMsg = OperationContext.get(IgniteSecurityProcessor.SEC_CTX_ATTR);
+
+        UUID newSecSubjId = secCtxMsg == null ? nodeId : secCtxMsg.subjId;
 
         try (Scope ignored = ctx.security().withContext(newSecSubjId)) {
             lsnr.onMessage(nodeId, msg, plc);
@@ -2029,11 +2032,8 @@ private long getInverseConnectionWaitTimeout() {
         return ctx.config().getFailureDetectionTimeout();
     }
 
-    /**
-     * @return One of two message wrappers. The first is {@link GridIoMessage}, the second is secured version {@link
-     * GridIoSecurityAwareMessage}.
-     */
-    private @NotNull GridIoMessage createGridIoMessage(
+    /** @return A {@link GridIoMessage} wrapper for {@code msg}. */
+    private GridIoMessage createGridIoMessage(
         Object topic,
         Message msg,
         byte plc,
@@ -2043,18 +2043,16 @@ private long getInverseConnectionWaitTimeout() {
     ) {
         GridIoMessage res;
 
-        if (ctx.security().enabled()) {
-            UUID secSubjId = null;
+        UUID secSubjId = ctx.security().enabled() && !ctx.security().isDefaultContext()
+            ? ctx.security().securityContext().subject().id()
+            : null;
 
-            if (!ctx.security().isDefaultContext())
-                secSubjId = ctx.security().securityContext().subject().id();
+        res = new GridIoMessage(plc, topic, msg, ordered, timeout, skipOnTimeout);
 
-            res = new GridIoSecurityAwareMessage(secSubjId, plc, topic, msg, ordered, timeout, skipOnTimeout);
+        try (Scope ignored = secSubjId == null ? Scope.NOOP_SCOPE
+            : OperationContext.set(IgniteSecurityProcessor.SEC_CTX_ATTR, new SecurityContextImpl(secSubjId))) {
+            res.opCtxMsg = ctx.operationContextDispatcher().collectDistributedAttributes();
         }
-        else
-            res = new GridIoMessage(plc, topic, msg, ordered, timeout, skipOnTimeout);
-
-        res.opCtxMsg = ctx.operationContextDispatcher().collectDistributedAttributes();
 
         return res;
     }
@@ -3812,7 +3810,7 @@ void unwind(GridMessageListener lsnr) {
 
                         MTC.span().addTag(SpanTags.MESSAGE, () -> traceName(fmc.message));
 
-                        invokeListener(plc, lsnr, nodeId, mc.message.message(), secSubjId(mc.message));
+                        invokeListener(plc, lsnr, nodeId, mc.message.message());
                     }
                     finally {
                         if (mc.closure != null)
@@ -4241,19 +4239,6 @@ public long binLatencyMcs() {
         }
     }
 
-    /**
-     * @return Security subject id.
-     */
-    private UUID secSubjId(GridIoMessage msg) {
-        if (ctx.security().enabled()) {
-            assert msg instanceof GridIoSecurityAwareMessage;
-
-            return ((GridIoSecurityAwareMessage)msg).securitySubjectId();
-        }
-
-        return null;
-    }
-
     /**
      * Responsible for handling network situation where server cannot open connection to client and
      * has to ask client to establish a connection to specific server.

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java

@@ -226,7 +226,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     };
 
     /** Discovery cached history size. */
-    private final int DISCOVERY_HISTORY_SIZE = getInteger(IGNITE_DISCOVERY_HISTORY_SIZE, DFLT_DISCOVERY_HISTORY_SIZE);
+    private static final int DISCOVERY_HISTORY_SIZE = getInteger(IGNITE_DISCOVERY_HISTORY_SIZE, DFLT_DISCOVERY_HISTORY_SIZE);
 
     /** */
     private final Object discoEvtMux = new Object();

modules/core/src/main/java/org/apache/ignite/internal/processors/authentication/IgniteAuthenticationProcessor.java

@@ -17,7 +17,6 @@
 
 package org.apache.ignite.internal.processors.authentication;
 
-import java.io.Serializable;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -57,12 +56,12 @@
 import org.apache.ignite.internal.processors.security.GridSecurityProcessor;
 import org.apache.ignite.internal.processors.security.IgniteSecurityProcessor;
 import org.apache.ignite.internal.processors.security.SecurityContext;
+import org.apache.ignite.internal.processors.security.SecurityContextImpl;
 import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.CU;
-import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.internal.util.worker.GridWorker;
 import org.apache.ignite.lang.IgniteFuture;
@@ -73,7 +72,6 @@
 import org.apache.ignite.plugin.security.SecurityException;
 import org.apache.ignite.plugin.security.SecurityPermission;
 import org.apache.ignite.plugin.security.SecuritySubject;
-import org.apache.ignite.plugin.security.SecuritySubjectType;
 import org.apache.ignite.spi.discovery.DiscoveryDataBag;
 import org.apache.ignite.spi.discovery.DiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
@@ -1302,7 +1300,7 @@ private RefreshUsersStorageWorker(ArrayList<User> usrs) {
         }
 
         /** {@inheritDoc} */
-        @Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
+        @Override protected void body() {
             if (ctx.clientNode())
                 return;
 
@@ -1331,74 +1329,4 @@ private RefreshUsersStorageWorker(ArrayList<User> usrs) {
             }
         }
     }
-
-    /** Represents {@link SecuritySubject} implementation. */
-    private static class SecuritySubjectImpl implements SecuritySubject {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** Security subject identifier. */
-        private final UUID id;
-
-        /** Security subject login.  */
-        private final String login;
-
-        /** Security subject type. */
-        private final SecuritySubjectType type;
-
-        /** Security subject address. */
-        private final InetSocketAddress addr;
-
-        /** */
-        public SecuritySubjectImpl(UUID id, String login, SecuritySubjectType type, InetSocketAddress addr) {
-            this.id = id;
-            this.login = login;
-            this.type = type;
-            this.addr = addr;
-        }
-
-        /** {@inheritDoc} */
-        @Override public UUID id() {
-            return id;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String login() {
-            return login;
-        }
-
-        /** {@inheritDoc} */
-        @Override public SecuritySubjectType type() {
-            return type;
-        }
-
-        /** {@inheritDoc} */
-        @Override public InetSocketAddress address() {
-            return addr;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(SecuritySubjectImpl.class, this);
-        }
-    }
-
-    /** Represents {@link SecurityContext} implementation that ignores any security permission checks. */
-    private static class SecurityContextImpl implements SecurityContext, Serializable {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private final SecuritySubject subj;
-
-        /** */
-        public SecurityContextImpl(UUID id, String login, SecuritySubjectType type, InetSocketAddress addr) {
-            subj = new SecuritySubjectImpl(id, login, type, addr);
-        }
-
-        /** {@inheritDoc} */
-        @Override public SecuritySubject subject() {
-            return subj;
-        }
-    }
 }