SessionManager.java
/*
** Module : SessionManager.java
** Abstract : Abstract base class for session management
**
** Copyright (c) 2007-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ---------------------------Description-------------------------------
** 001 ECF 20071101 @35865 Created initial version. Abstract base class
** for session management. Used code from other
** classes as a base.
** 002 ECF 20071121 @35937 Fixed initialization sequence. initialize()
** now resets basic state, then calls init(),
** which can be overridden by subclasses.
** 003 ECF 20071127 @36047 Added getSecurityPlugin() method. Added
** certAlias parameter to a connectVirtual()
** variant.
** 004 ECF 20080213 @37066 Added isLocalNode(). Reports whether a given
** InetSocketAddress represents the local P2J
** node.
** 005 GES 20081024 @40331 Added support for establishing a session
** via an insecure socket (non-SSL).
** 006 SIY 20090512 @42141 Added support for waiting for shutdown.
** 007 GES 20090816 @43639 Minor signature change and formatting cleanup.
** 008 CA 20090825 @43756 Fixed a deadlock between Queue and SessionManager.
** This was discovered during remote DB connection
** closing.
** 009 CA 20090901 @43816 Renamed sessionsByContext. Created a map to hold
** control instances for each context. Now each of
** the per-context map holds the virtual control/
** session for the virtual context ID. The interrupt
** handler is no longer kept per each control
** instance; instead is set for the SessionManager
** singleton, and will be the same for the entire
** server.
** 010 CA 20090908 @43836 Fixes a NPE in processEcho (introduced by H009
** @43816).
** 011 CA 20090917 @43930 Fix abnormal connection end. In endSession, the TM
** will be notified to end the Conversation thread, as
** the session is being closed.
** 012 CA 20090918 @43934 H011 (@43930) should be applied only if the server
** is not shutting down as the TM call isn't executed
** within the correct context.
** 013 CA 20090918 @43948 Another case of H011 regression - TM.abortProcessing
** should be called only for DirectSession's.
** 014 GES 20091020 @44176 Make sure to interrupt long running server threads
** when the session ends. Previously this was not
** done if the queue was already dead, but that is
** too conservative since it will never get called
** due to how the protocol closes down.
** 015 GES 20091103 @44305 Added logging of some core connection parms.
** 016 CA 20091203 @44473 Virtual sessions must not interrupt the
** Conversation thread while shuting down.
** 017 CA 20100329 @44783 Added terminateVirtual, which allows sub-classes
** to provide virtual session termination code.
** 018 CA 20110105 Changes to allow virtual sessions from leaf nodes
** 019 CA 20110131 Changes to allow the direct sessions to authenti-
** cate using different credentials. Also, this way
** sessions can be created to different router nodes.
** 020 CA 20110202 Fixed a bug in H019, related to server-to-server
** connections.
** 021 LMR 20101210 Changed createQueue() to default
** net/connection/secure to false.
** 022 CA 20111209 Added getSessionContexts, to get the all active
** sessions which have invoked APIs in the given jar.
** Also, notify the ExportTracker when a session is
** finished, to be removed from all tracking data.
** 023 CS 20130204 Added java.net.Socket import to disambiguate from
** com.goldencode.p2j.util.Socket class
** 024 CA 20130529 Added appserver support.
** 025 CA 20130628 Context-local data for this class must not be reset by the agent.
** 026 CA 20130705 Exposed the bootstrap configuration.
** 027 CA 20130822 Fixed the net protocol to support async requests.
** 028 CA 20140206 Added a way of notifying when the server sockets are ready (needed by
** the scheduler, to know when to start launching processes).
** 029 CA 20140513 Added a weight for the context-local var, to ensure predetermined
** order during context reset.
** 030 OM 20160128 Added javadoc. Code cleanup & upgrade to Java7.
** 031 EVL 20160223 Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
** 032 CA 20170228 Added executeInContext.
** 033 CA 20170427 When a queue is stopped, context is set explicitly for each session
** associated with this queue, when notifications are sent.
** 034 CA 20180217 Virtual sessions can't cleanup their context concurrently, as they
** might perform remote calls (for e.g. database disconnection) - thus,
** they need to use a common lock.
** 035 GES 20190322 Added isConversationThread() helper.
** 036 IAS 20200722 Refactored to the new NetSocket API, added failed handshake handling.
** CA 20201003 Use an identity HashSet for the async threads.
** IAS 20210325 Added NIO configuration via BootstrapConfig
** IAS 20210505 Changed exception thrown on failed shutdown.
** IAS 20210827 Added allowed ciphers' filtering
** IAS 20220624 Added support for the configurable JCE/JSSE provider
** TJD 20220713 Synchronization and locking optimizations
** TJD 20220504 Upgrade do Java 11 minor changes
** 037 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 038 GBB 20230825 checkCallerAbort moved to SecurityUtil & calls updated.
** 039 TJD 20240123 Java 17 compatibility updates
** 040 ICP 20240704 Corrected SecurityUtil.checkCallerAbort call in getSessionContexts.
** 041 GBB 20240709 Hard-coded config name replaced by ConfigItem constant.
** 042 CA 20240910 'net:socket:nio' is enabled by default only for Java 8.
** 043 TJD 20231012 SessionManager.lock access optimizations
** 044 RNC 20250127 Added default methods 'setSessionLock' and 'getSessionLock'.
** 045 AL2 20250523 'net:socket:nio' is disabled by default.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.net;
import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;
import java.util.stream.*;
import javax.net.ssl.*;
import com.goldencode.p2j.cfg.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
import java.net.Socket;
import java.nio.channels.*;
import java.security.*;
/**
* This class is the central focal point of all queue and session management
* for the net package. It provides the public API for application code to
* manage sessions and, indirectly, their underlying message transport queues.
* The singleton instance of this class is created by the {@link
* SessionManagerFactory} and is retrieved with the static {@link #get()}
* method.
* <p>
* This class may not be instantiated directly. Much of the work is performed
* by concrete subclasses, though common functionality is implemented here.
* The API, while asymmetric in places depending upon whether the current node
* is a leaf or router, is provided here in its entirety to simplify use of
* this class from both within and outside of, the net package. Leaf methods
* that are inappropriately invoked from a router node will raise exceptions,
* and vice versa.
* <p>
* The session manager is responsible for:
* <ul>
* <li>Providing a mechanism for application code to bootstrap the runtime
* environment for either a leaf or router network node.
* <li>Creating, authenticating, registering and terminating direct and
* virtual (i.e., multiplexed) sessions with other network nodes.
* <li>Creating, managing, and destroying the underlying queues at the
* appropriate times.
* <li>Acting as both the requester and provider end-points in session
* establishment.
* <li>Interacting with sessions to dynamically set an appropriate security
* context for the current thread (router side only).
* <li>Reporting on the nature of the current network node (i.e., leaf vs
* router).
* </ul>
*/
public abstract class SessionManager
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(SessionManager.class);
/** Max number of SSL handshake attempts on connect */
protected static final int MAX_SSL_HANDSHAKE_ATTEMPTS = 3;
/** Singleton instance of this class */
private static SessionManager instance = null;
/** Topology database for this network node */
protected final Map<InetSocketAddress, Queue> topology = new HashMap<>();
/** Use count of the known addresses for this network node */
protected final Map<InetSocketAddress, Integer> addressUseCount= new HashMap<>();
/** Many-to-one map of queues to all of their associated sessions */
protected final Map<Queue, Set<BaseSession>> sessionsByQueue = new IdentityHashMap<>();
/** One-to-one map of context IDs to sessions */
protected final Map<Integer, BaseSession> virtualSessionsByContext = new HashMap<>();
/** One-to-one map of virtual locks to queues */
protected final Map<Queue, Object> virtualLocksByQueue = new ConcurrentHashMap<Queue, Object>();
/**
* One-to-one map of identity to queues. Used on leaf nodes and gets
* populated after a {@link Queue} was successfully created.
*/
protected final Map<Queue, String> identityByQueue = new HashMap<>();
/** Map of connected app servers; used on the P2J client-side. */
protected final Map<Queue, String> appServers = new HashMap<>();
/** One-to-one map of multiplexed context IDs to its associated Control instance. */
protected final Map<Integer, Control> virtualControlByContext = new HashMap<>();
/** Active session for the current client context */
protected final ContextLocal<BaseSession> activeSession = new ContextLocal<BaseSession>()
{
@Override
public WeightFactor getWeight()
{
return WeightFactor.LAST;
};
protected boolean isResetAllowed()
{
return false;
};
};
/** Object on which to synchronize access to session manager state */
protected final Object lock = new Object();
/** Security manager */
protected SecurityManager secMgr = null;
/** Bootstrap configuration for this network node */
protected BootstrapConfig config = null;
/** A set containing threads which always send async requests. */
private final Set<Thread> asyncThreads = Collections.newSetFromMap(new IdentityHashMap<>());
/** The handler used to process interrupt requests. */
private InterruptHandler interruptHandler;
/** Shutdown thread. */
private Thread stopper = new Thread(Dispatcher.dispThreads, new Stopper(), "queueStopper");
/** Mark start of shutdown process. */
private volatile boolean shutdownStarted = false;
/** Mark end of shutdown process. */
public volatile boolean shutdownDone = false;
/** Next ID to be assigned for a local security context. */
private int nextContextID = (new Random()).nextInt();
/**
* Constructor usable only by concrete subclasses. Enforces singleton
* pattern.
*
* @throws IllegalStateException
* if called more than once per JVM instance.
*/
protected SessionManager()
{
synchronized (SessionManager.class)
{
if (instance != null)
{
throw new IllegalStateException(
"There can be only one session manager");
}
instance = this;
stopper.setDaemon(true);
}
}
/**
* Report if the current threads is a conversation thread. A conversation thread is one
* which is dedicated to the processing for a single security session for its lifetime.
* Knowledge of this fact can enable optimizations since the thread is known to never
* change security contexts.
*
* @return {@code true} if the current thread is a conversion thread.
*/
public static boolean isConversationThread()
{
return Queue.isConversationThread();
}
/**
* Method used to access the concrete, singleton session manager instance
* for a network node.
*
* @return Singleton session manager, if it has been created; otherwise,
* <code>null</code>.
*/
public synchronized static SessionManager get()
{
return instance;
}
/**
* Register the given thread in the list of threads which always send async requests.
*
* @param thread
* An active thread instance.
*/
public final void registerAsyncThread(Thread thread)
{
synchronized (asyncThreads)
{
asyncThreads.add(thread);
}
}
/**
* DeRegister the given thread from the list of threads which always send async requests.
*
* @param thread
* An active thread instance.
*/
public final void deregisterAsyncThread(Thread thread)
{
synchronized (asyncThreads)
{
asyncThreads.remove(thread);
}
}
/**
* Check if the given thread is in the list of threads which always send async requests.
*
* @param thread
* An active thread instance.
*
* @return <code>true</code> if the thread is in the list of threads which always send async
* requests.
*/
public final boolean isAsyncThread(Thread thread)
{
synchronized (asyncThreads)
{
return asyncThreads.contains(thread);
}
}
/**
* Get the interrupt handler.
*
* @return The interrupt handler instance.
*/
public final InterruptHandler getInterruptHandler()
{
return interruptHandler;
}
/**
* Set the interrupt handler.
*
* @param handler
* The interrupt handler instance.
*/
public final void setInterruptHandler(InterruptHandler handler)
{
if (this.interruptHandler == null)
{
interruptHandler = handler;
}
}
/**
* Convenience method to detect whether this session manager operates on a
* leaf node.
*
* @return <code>true</code> if network node for this session manager is a
* leaf node; else <code>false</code>.
*/
public boolean isLeaf()
{
return false;
}
/**
* Expose the {@link BootstrapConfig config} used to start the P2J Client or Server.
*
* @return A copy of the {@link #config}.
*/
public BootstrapConfig config()
{
return new BootstrapConfig(config);
}
/**
* Convenience method to detect whether this session manager operates on a
* router node.
*
* @return <code>true</code> if network node for this session manager is a
* router node; else <code>false</code>.
*/
public boolean isRouter()
{
return false;
}
/**
* Getter for the session lock. By default, it does nothing.
*
* @return The value of the session lock.
*/
public boolean getSessionLock()
{
return false;
}
/**
* Setter for the session lock. By default, it does nothing.
*
* @param lockStatus
* The new status for the lock. {@code true} if the session launching will be stopped,
* {@code false} otherwise
*/
public boolean setSessionLock(boolean lockStatus)
{
return false;
}
/**
* Get the appserver associated with this P2j client session.
*
* @param session
* The P2J session.
*
* @return The appserver's name, or <code>null</code> if the client is not started as an
* appserver.
*/
public String getAppServer(Session session)
{
// this must be called only on the P2J client-side, where the app-server gets started.
synchronized (appServers)
{
return appServers.get(((BaseSession) session).getQueue());
}
}
/**
* Convenience method to detect whether the given socket address represents
* the current node. This is determined by comparing the port number
* with this node's port number, and by comparing the given host address
* with each address available on every network interface for the local
* machine. If the port numbers match and any address matches the given
* host address, the given address is considered to represent the current
* network node.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param address
* Socket address to check.
*
* @return <code>true</code> if an address match is detected, else
* <code>false</code>.
*
* @throws UnsupportedOperationException
* always.
*/
public boolean isLocalNode(InetSocketAddress address)
{
throw new UnsupportedOperationException();
}
/**
* Called by a requester to authenticate a virtual session. Invoked via
* remote call on the provider side of a connection to establish the
* provider's side of a virtual session.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param acctID
* Account identifier for the requesting party.
* @param contextID
* Value which uniquely identifies the security context of the
* requesting party. This will be used in communications from
* the provider side of the connection.
* @param queue
* Message transport queue for the provider side of the
* connection.
*
* @return Generated security context ID. This uniquely identifies the
* newly created provider-side security context for the virtual
* session. It is used by the requester to multiplex
* communications for this session over a shared queue. This ID
* will never be 0, which is reserved for direct sessions.
*
* @throws UnsupportedOperationException
* always.
* @throws IllegalAccessException
* if authentication is not successful.
*/
public int authenticateRemote(String acctID, int contextID, Queue queue)
throws IllegalAccessException
{
throw new UnsupportedOperationException();
}
/**
* Called by a requester to terminate a virtual session. Invoked via
* remote call on the provider side of a connection to terminate the
* provider's side of a virtual session.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param contextID
* Value which uniquely identifies the security context of the
* requesting party.
*
* @throws UnsupportedOperationException
* always.
*/
public void terminateRemote(int contextID)
{
throw new UnsupportedOperationException();
}
/**
* Called by a requester to initiate virtual session termination. This is
* invoked on the requester side of a connection to initiate termination of
* both the requester and provider sides of a virtual session.
*
* @param session
* The session which needs to be terminated.
*/
// Removed synchronized keyword as it does not need to be synchronized as all variables are synchronized
// before access
public void terminateVirtual(BaseSession session)
{
Queue queue = session.getQueue();
if (!queue.isRunning())
{
// queue already down, nothing can be done
return;
}
Object queueLock = null;
// this lock ensures synchronization per each queue
queueLock = virtualLocksByQueue.get(queue);
// if lock was removed do nothing
if (queueLock == null)
{
return;
}
synchronized (queueLock)
{
queueLock = virtualLocksByQueue.get(queue);
// if lock was removed in meantime do nothing
if (queueLock == null)
{
return;
}
callTerminateRemote(session);
endSession(session, false, true);
if (!queue.isRunning())
{
virtualLocksByQueue.remove(queue);
}
}
}
/**
* Make a synchronous method call to a remote router node to terminate the
* remote side of this virtual session. This must be done while the
* session's queue is still active, in the event this session's termination
* is to be followed by the termination of its queue.
* @param session
* session the be terminated
*/
private void callTerminateRemote(BaseSession session)
{
if (!session.isRunning())
{
// Can't call peer side if queue is already down.
return;
}
Message message = new Message(MessageTypes.REQUEST_SYNCH);
RoutingKey key = new RoutingKey(RoutingKey.TERMINATE_REMOTE);
key.setSourceAddress(session.getNodeAddress());
key.setDestinationAddress(session.getRemoteAddress());
message.setKey(key);
int contextID = session.getContextID();
message.setPayload(new Object[] { Integer.valueOf(contextID) });
try
{
session.transact(message, 0);
}
catch (ApplicationRequestedStop exc)
{
if (LOG.isLoggable(Level.FINE))
{
LOG.logp(Level.FINE,
"SessionManager.callTerminateRemote()",
"",
exc.getMessage());
}
}
catch (Exception exc)
{
// check if the cause is a InterruptedException. in this case, do
// not log it as a warning
Level logLvl = Level.WARNING;
String msg = "Error terminating virtual session peer.";
Throwable cause = exc;
while (cause != null)
{
if (cause instanceof InterruptedException)
{
logLvl = Level.FINE;
break;
}
if (cause instanceof SilentUnwindException)
{
msg = "SessionManager.callTerminateRemote()";
logLvl = Level.FINE;
break;
}
cause = cause.getCause();
}
if (LOG.isLoggable(logLvl))
{
LOG.log(logLvl, msg, exc);
}
}
}
/**
* Establish a direct connection to a network node using the bootstrap
* configuration stored during construction.
* <p>
* This implementation calls {@link #connectDirect(BootstrapConfig,
* InterruptHandler, SessionListener)}, with a <code>null</code> bootstrap
* config. So, the {@link #config} instance received during this manager's
* {@link SessionManagerFactory#createLeafNode(BootstrapConfig) creation}
* will be used instead.
*
* @param interruptHandler
* An object that is notified when the current thread is
* interrupted.
* @param notify
* Session listener to be notified of session events.
*
* @return Session established after the connection has been made.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public Session connectDirect(InterruptHandler interruptHandler,
SessionListener notify)
throws Exception
{
return connectDirect(null, interruptHandler, notify);
}
/**
* Establish a direct connection to a network node using the specified
* bootstrap configuration.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param config
* Bootstrap configuration information which contains at least
* enough information (host and port) to make a direct connection
* from this node to another.
* @param interruptHandler
* An object that is notified when the current thread is
* interrupted.
* @param notify
* Session listener to be notified of session events.
*
* @return Session established after the connection has been made.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public Session connectDirect(BootstrapConfig config,
InterruptHandler interruptHandler,
SessionListener notify)
throws Exception
{
throw new UnsupportedOperationException();
}
/**
* Establish a virtual (multiplexed) connection to a network node at the
* given address.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param address
* Socket address of remote network node.
* @param certAlias
* Alias of certificate to be read from trust store to enable
* certificate-based, server-to-server authentication. This
* should match the server ID of the remote node.
* @param notify
* Session listener to be notified of session events.
*
* @return Session established after the connection has been made.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public Session connectVirtual(InetSocketAddress address,
String certAlias,
SessionListener notify)
throws Exception
{
return connectVirtual(address, certAlias, notify, null);
}
/**
* Establish a virtual (multiplexed) connection to a network node at the
* given address.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param address
* Socket address of remote network node.
* @param certAlias
* Alias of certificate to be read from trust store to enable
* certificate-based, server-to-server authentication. This
* should match the server ID of the remote node.
* @param notify
* Session listener to be notified of session events.
* @param account
* Explicit account to authenticate on the remote side. May be <code>null</code>.
*
* @return Session established after the connection has been made.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public Session connectVirtual(InetSocketAddress address,
String certAlias,
SessionListener notify,
String account)
throws Exception
{
throw new UnsupportedOperationException();
}
/**
* Establish a virtual (multiplexed) connection to a network node at the
* address specified by the given bootstrap configuration.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param config
* Bootstrap configuration information which contains at least
* enough information (host and port) to make a virtual connection
* from this node to another.
* @param notify
* Session listener to be notified of session events.
*
* @return Session established after the connection has been made.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public Session connectVirtual(BootstrapConfig config,
SessionListener notify)
throws Exception
{
throw new UnsupportedOperationException();
}
/**
* Listen for incoming socket connections, authenticate each new
* connection, and establish a new queue for each authenticated connection.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param notify
* Session listener to be notified of session events.
*
* @throws UnsupportedOperationException
* always.
* @throws Exception
* not thrown.
*/
public void listen(SessionListener notify)
throws Exception
{
throw new UnsupportedOperationException();
}
/**
* Get the active session, if any. This is the last session whose security
* context was set.
*
* @return Active session, or <code>null</code>.
*/
public Session getSession()
{
return activeSession.get();
}
/**
* Check if the current session is a {@link VirtualSession}.
*
* @return See above.
*/
public boolean isVirtualSession()
{
return activeSession.get() instanceof VirtualSession;
}
/**
* Get the contexts for all active sessions which have invoked APIs in the
* given jar. If the jar name is <code>null</code>, list all active
* sessions.
* <p>
* <i><u>Restricted use</u></i>. This method checks the caller to be
* <code>com.goldencode.p2j.security.SecurityManager.getActiveSessions</code>.
*
* @param jar
* The jar name
* @param iface
* The interface for which we need the sessions;
* if <code>null</code>, search through the entire jar.
*
* @return See above.
*/
public Object[] getSessionContexts(String jar, String iface)
{
try
{
SecurityUtil.checkCallerAbort(
new String[]
{
"com.goldencode.p2j.security.SecurityManager.getActiveSessions",
"com.goldencode.p2j.security.SecurityAdmin.getActiveSessions"
},
"com.goldencode.p2j.net.SessionManager"
);
}
catch (RestrictedUseException e)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.log(Level.SEVERE,
"Unauthorized call of SessionManager.getSessionContexts",
e);
}
return new Object[0];
}
Set<Session> sessions = ExportTracker.listSessions(jar, iface);
List<Object> contexts = new ArrayList<>();
for(Session sess : sessions)
{
BaseSession base = (BaseSession) sess;
contexts.add(base.getContext());
}
return contexts.toArray();
}
/**
* Terminate all active sessions which have invoked the given APIs in the
* given jar. If the jar name is <code>null</code>, terminate all active
* tracked sessions.
* This method guarantees that all session termination listeners have been
* executed.
* <p>
* <i><u>Restricted use</u></i>. This method checks the caller to be
* <code>com.goldencode.p2j.security.SecurityManager.terminateSessions</code>.
*
* @param jar
* The jar name; if <code>null</code>, terminate all active tracked
* sessions.
* @param iface
* The interface for which we need to terminate the sessions; if
* <code>null</code>, terminate all sessions for the given jar.
*
* @throws RestrictedUseException
* If called improperly.
*/
public void terminateSessions(String jar, String iface)
throws RestrictedUseException
{
SecurityUtil.checkCallerAbort(
"com.goldencode.p2j.security.SecurityManager.terminateSessions",
"com.goldencode.p2j.net.SessionManager");
Set<Session> sessions = ExportTracker.listSessions(jar, iface);
for (Session session : sessions)
{
session.terminate();
}
}
/**
* Get the local node address of the active session, if any. This is the
* last session whose security context was set.
*
* @return Active session's local node address, or 0 if no active session
* exists.
*/
public int getNodeAddress()
{
Session session = getSession();
return (session != null ? session.getNodeAddress() : 0);
}
/**
* Get the remote node address of the active session, if any. This is the
* last session whose security context was set.
*
* @return Active session's remote node address, or 0 if no active session
* exists.
*/
public int getRemoteAddress()
{
Session session = getSession();
return (session != null ? session.getRemoteAddress() : 0);
}
/**
* Terminate ALL sessions managed by this session manager and shut down the
* network node itself. This involves terminating the connection listening
* loop (router node only), the node's primary transport queue, and the
* router queue (router node only).
*
* @throws AccessControlException
* if the caller's security context does not include sufficient
* rights to perform the shutdown.
*/
public void shutdown()
throws SecurityException
{
// Fail the call if not sufficient access rights.
if (!secMgr.checkShutdownAccess())
{
throw new SecurityException("Insufficient access rights for SHUTDOWN.");
}
synchronized (stopper)
{
if (!shutdownStarted)
{
shutdownStarted = true;
stopper.start();
}
}
}
/**
* Block the calling thread until the network portions of the server are fully initialized.
* At that point the server is ready to receive socket connections.
* @throws InterruptedException
* if the current thread was interrupted
*/
public void waitUntilReady()
throws InterruptedException
{
throw new UnsupportedOperationException();
}
/**
* Wait for end of shutdown thread. If shutdown thread was not started, it
* is started immediately.
*
* @throws AccessControlException
* if the caller's security context does not include sufficient
* rights to perform the shutdown.
*/
public void waitForShutdown()
throws SecurityException
{
// Fail the call if not sufficient access rights.
if (!secMgr.checkShutdownAccess())
{
throw new SecurityException("Insufficient access rights for SHUTDOWN.");
}
synchronized (stopper)
{
if (shutdownDone)
return;
if (!shutdownStarted)
{
shutdownStarted = true;
stopper.start();
}
while (!shutdownDone)
{
try
{
stopper.wait();
}
catch (Exception e)
{
}
}
}
}
/**
* Return a string representation of the internal state of this object,
* primarily for debugging purposes.
*
* @return String representation of this object.
*
* @see java.lang.Object#toString()
*/
public String toString()
{
return ("TOP: " + topology +
" SBQ: " + sessionsByQueue +
" SBC: " + virtualSessionsByContext +
" ACT: " + activeSession.get());
}
/**
* Execute the given task in the specified session's context.
* <p>
* Only a thread with the server's context is allowed to execute this.
*
* @param session
* The client session.
* @param task
* The task to execute.
*
* @throws RestrictedUseException
* If the current context is not a server context.
*/
public void executeInContext(Session session, Runnable task)
throws RestrictedUseException
{
if (!SecurityManager.getInstance().isServerAccount())
{
throw new RestrictedUseException("Only server accounts are allowed to use this API!");
}
try
{
setContext(((BaseSession) session).getQueue());
task.run();
}
finally
{
restoreContext();
}
}
/**
* Create a direct session for the given queue and register it with the
* session manager.
* <p>
* This implementation throws an exception. It must be overridden to
* provide the functionality described above.
*
* @param queue
* Underlying message transport for the new session.
*
* @return <code>true</code> if the session initialized successfully.
*/
protected boolean createDirectSession(Queue queue)
{
throw new UnsupportedOperationException();
}
/**
* Create a socket and the low level message transport queue, to be
* used to communicate with the remote node at the given socket address.
* Authenticate this network node as a valid requester of services from the
* remote node using the security manager.
* <p>
* The socket that is created is either secure or insecure, depending on
* the value of <code>net/connection/secure</code> bootstrap configuration
* flag.
*
* @param remoteAddr
* Address of the socket for the remote network node.
* @param config
* BootstrapConfig which should override that of the security
* manager for this session. If <code>null</code>, the security
* manager's default configuration is used.
*
* @return The newly created queue, if authentication was successful.
*
* @throws Exception
* if authentication fails.
*/
protected Queue createQueue(InetSocketAddress remoteAddr, BootstrapConfig config)
throws Exception
{
BootstrapConfig bc = (config == null) ? this.config : config;
int attempts = MAX_SSL_HANDSHAKE_ATTEMPTS;
NetSocket netSock = null;
while(attempts-- > 0)
{
try
{
netSock = connect(remoteAddr, bc);
break;
}
catch (SSLHandshakeException e)
{
if (attempts == 0)
{
throw e;
}
LOG.log(Level.WARNING, String.format("SSL handshake failed, %d attempts remaining", attempts));
continue;
}
}
// force authentication before any protocol ops can start
AuthData auth = secMgr.authenticateClient(netSock, bc);
if (auth == null)
{
throw new Exception("Authentication failed");
}
Queue queue = new Queue(netSock);
// testing for leaf node. The context should be irrelevant (as opposed to !isServerAccount)
if (secMgr.isClient())
{
synchronized (identityByQueue)
{
identityByQueue.put(queue, auth.getIdentity());
}
if (auth.isAppServer())
{
synchronized (appServers)
{
appServers.put(queue, auth.getAppServer());
}
}
}
return queue;
}
/**
* Create the low level socket to match the session's requirements and connect it to the peer. This
* supports secure and insecure sockets. For secure sockets, both NIO and the standard approach can be
* used.
*
* @param remoteAddr
* The address of the peer.
* @param bc
* The bootstrap configuration which defines the core connectivity.
*
* @return The low level socket resource.
*
* @throws IOException
* @throws SocketException
* @throws NoSuchAlgorithmException
* @throws KeyStoreException
* @throws UnrecoverableKeyException
* @throws KeyManagementException
* @throws ConfigurationException
*/
private NetSocket connect(InetSocketAddress remoteAddr, BootstrapConfig bc)
throws IOException,
SocketException,
NoSuchAlgorithmException,
KeyStoreException,
UnrecoverableKeyException,
KeyManagementException,
ConfigurationException,
NoSuchProviderException
{
// detect if we must use TLS (by default we don't use TLS)
boolean secure = bc.getBoolean("net", "connection", "secure", false);
boolean server = this.config.isServer();
Socket lowSocket;
SocketChannel channel = null;
boolean nio = bc.getBoolean(ConfigItem.SERVER_SOCKET_NIO, false);
if (nio)
{
channel = SocketChannel.open();
// channel.setOption(StandardSocketOptions.SO_REUSEADDR , true);
channel.setOption(StandardSocketOptions.TCP_NODELAY , true);
channel.connect(remoteAddr);
lowSocket = channel.socket();
}
else
{
lowSocket = new Socket();
lowSocket.setReuseAddress(true);
lowSocket.bind(null);
lowSocket.connect(remoteAddr);
}
Socket sock = lowSocket;
if (LOG.isLoggable(Level.FINER))
{
String msg = String.format("host = %s; port = %d; secure = %b;",
remoteAddr.getHostName(),
remoteAddr.getPort(),
secure);
LOG.logp(Level.FINER, "SessionManager.createQueue()", "", msg);
}
NetSocket netSock;
if (secure)
{
// if this is a server, always use the primordial configuration
// (used to initialize the SessionManager)
SSLContext ctx = secMgr.contextSm.getSecureSocketContext(server ? this.config : bc);
if (nio)
{
SSLEngine engine = ctx.createSSLEngine();
engine.setEnabledCipherSuites(
SecurityManager.getInstance().
filterCiphers(engine.getEnabledCipherSuites())
);
if (LOG.isLoggable(Level.CONFIG))
{
LOG.config(
Arrays.stream(
engine.getEnabledCipherSuites()).collect(Collectors.joining(","))
);
}
engine.setUseClientMode(true);
netSock = NetSocket.FACTORY.create(engine, channel);
}
else
{
SSLSocketFactory factory = ctx.getSocketFactory();
String host = remoteAddr.getHostName();
int port = remoteAddr.getPort();
// shift the low level socket into secure mode
sock = factory.createSocket(lowSocket, host, port, true);
sock.setTcpNoDelay(true);
netSock = NetSocket.FACTORY.create(sock);
}
}
else
{
netSock = channel != null ? NetSocket.FACTORY.create(channel) :
NetSocket.FACTORY.create(sock);
}
return netSock;
}
/**
* Terminate the socket connection listening loop for a router network
* node.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*/
protected void terminateInput()
{
}
/**
* End the specified network session. This involves:
* <ol>
* <li>optionally interrupting any currently running {@link Dispatcher}
* or {@link Conversation} thread or session;
* <li>deregistering the session from the session manager's database;
* <li>disabling the session's queue if no other sessions are still using
* it;
* <li>optionally notifying registered session listeners of the
* termination event;
* <li>stopping the session's queue transport if no other sessions are
* still using it;
* <li>terminating the session with the security manager.
* </ol>
*
* @param session
* Session that is ending.
* @param interrupt
* <code>true</code> to interrupt any long running service thread
* asynchronously.
* @param notify
* <code>true</code> to notify registered listeners of the
* termination. This should be <code>false</code> if the queue
* itself triggered the termination, since in that case the
* notification will already have occurred.
*/
protected void endSession(BaseSession session, boolean interrupt, boolean notify)
{
Queue queue = session.getQueue();
int contextID = session.getContextID();
boolean virtual;
// moved into synchronized section to prevent deadlocks
synchronized(lock)
{
virtual = virtualControlByContext.containsKey(contextID);
}
// if this is a virtual session, then the Conversation thread must not be interrupted.
if (!virtual && interrupt && isRouter())
{
// any further processing of the conversation thread needs to be ended
// so we notify it. On the next TM.honorStopCondition call, TM will
// intercept this flag and throw a SilentUnwindException.
if (!shutdownStarted && (session instanceof DirectSession))
{
TransactionManager.abortProcessing();
}
// Interrupt any long running server thread since otherwise that
// thread may continue for an arbitrary amount of time even though
// the session is ended. This must happen after the abortProcessing()
// call above so that if the interrupt() is handled by calling code,
// the flag is set to allow the fatal nature of this condition to
// be known (since the session must exit).
Control.interrupt();
}
// Remove the session from the session manager's database.
boolean queueInUse = deregisterSession(session);
// Determine if session's queue needs to be stopped (i.e., it is no
// longer needed by any other session and it is currently running).
boolean stopQueue = (!queueInUse && queue.isRunning());
// Disable the session's queue if necessary.
if (stopQueue)
{
queue.disable(null);
}
// Notify listeners of the termination.
if (notify)
{
session.sendTerminationEvent();
}
// Stop the session's queue if necessary.
if (stopQueue)
{
queue.stopTransport(null);
// remove mappings created in createQueue()
if (secMgr.isClient())
{
synchronized (identityByQueue)
{
identityByQueue.remove(queue);
}
synchronized (appServers)
{
appServers.remove(queue);
}
}
}
// End the session.
session.end();
}
/**
* Register a new session with the session manager. This updates the
* session manager's database, which includes:
* <ul>
* <li>a topology mapping of socket addresses to queues;
* <li>a mapping of local context IDs to sessions;
* <li>a mapping of queues to sets of sessions which use them.
* </ul>
*
* @param contextID
* An integer which uniquely identifies a virtual session on the
* local node. This ID is used by the peer side of the session to
* multiplex communications with multiple virtual sessions over a
* single queue.
* @param remoteAddr
* Socket address of the network node on the remote side of the
* session.
* @param session
* Session being registered.
*/
protected void registerSession(int contextID,
InetSocketAddress remoteAddr,
BaseSession session)
{
Queue queue = session.getQueue();
if (remoteAddr == null)
{
remoteAddr = queue.getRemoteInetAddress();
}
synchronized (lock)
{
int n = addressUseCount.containsKey(remoteAddr)
? addressUseCount.get(remoteAddr) : 0;
addressUseCount.put(remoteAddr, ++n);
topology.put(remoteAddr, queue);
if (contextID != 0)
{
virtualSessionsByContext.put(contextID, session);
}
// virtual sessions require to cleanup on the same lock... because they use the same
// queue for transport, and might require to perform disconnect - which can't be done
// concurrently
secMgr.setCleanupLock(session.getContext(), queue.getCleanupLock());
Set<BaseSession> set = sessionsByQueue.get(queue);
if (set == null)
{
set = new HashSet<BaseSession>();
sessionsByQueue.put(queue, set);
}
set.add(session);
}
// notify the export tracker when the session ends
session.addSessionListener(ExportTracker.getTracker());
}
/**
* Deregister a session with the session manager. This updates the session
* manager's database, which includes:
* <ul>
* <li>a topology mapping of socket addresses to queues;
* <li>a mapping of local context IDs to sessions;
* <li>a mapping of queues to sets of sessions which use them.
* </ul>
* <p>
* If the session being deregistered is the last session to use its queue,
* this fact is indicated by the method's return value.
* <p>
* This method is tolerant of being called multiple times for the same
* session instance, since there are several reasons this method may be
* called at session termination:
* <ul>
* <li>explicit termination by a call to {@link Session#terminate()};
* <li>the queue going down, due either to a connectivity error, or to
* the connection being disrupted by the termination (normal or
* abnormal) of the queue's peer.
* </ul>
*
* @param session
* Session being deregistered.
*
* @return <code>true</code> if the session's underlying queue is still
* needed by other sessions; <code>false</code> if this was the
* last session using its queue, and the queue can now be stopped.
*/
protected boolean deregisterSession(BaseSession session)
{
assert (session != null);
boolean queueInUse = true;
Queue queue = session.getQueue();
synchronized (queue)
{
synchronized (lock)
{
int contextID = session.getContextID();
if (contextID != 0)
{
virtualSessionsByContext.remove(contextID);
virtualControlByContext.remove(contextID);
}
Set<BaseSession> set = sessionsByQueue.get(queue);
if (set == null)
{
// This session already was deregistered, as the queue is no
// longer in use.
return false;
}
set.remove(session);
InetSocketAddress remoteAddr = queue.getRemoteInetAddress();
int n = addressUseCount.get(remoteAddr) - 1;
if (n == 0)
{
topology.remove(remoteAddr);
addressUseCount.remove(remoteAddr);
}
else
{
addressUseCount.put(remoteAddr, n);
}
if (set.isEmpty())
{
sessionsByQueue.remove(queue);
queueInUse = false;
}
}
}
return queueInUse;
}
/**
* Set the security context for the current thread, using the session
* identified by the given routing key or transport queue.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*
* @param key
* Routing key for an incoming message.
* @param queue
* Message transport queue.
*/
protected void setContext(RoutingKey key, Queue queue)
{
}
/**
* Set the security context for the current thread, using the session
* identified by the given queue. It is assumed that there is one and only
* one session using the queue.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*
* @param queue
* Queue which uniquely identifies the session whose security
* context is to be used.
*/
protected void setContext(Queue queue)
{
}
/**
* Set the security context for the current thread, using the security
* context stored in the given session. Mark this session as the active
* session for this context.
*
* @param session
* Session whose security context is to be associated with the
* current thread.
*/
protected void setContext(BaseSession session)
{
}
/**
* Restore the current thread's security context to whatever it was before
* it was last explicitly set.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*/
protected void restoreContext()
{
}
/**
* Initialize security context switching for the calling thread.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*/
protected void setInitialContext()
{
}
/**
* Release the initial security context assigned to the calling thread.
* <p>
* This implementation is a NOP; it must be overridden to provide the
* above functionality.
*/
protected void dropInitialContext()
{
}
/**
* Initialize the session manager beyond what is done by the {@link
* #initialize(BootstrapConfig)} method.
* <p>
* This implementation does nothing. It should be overridden by subclasses
* with special initialization requirements.
*
* @throws ConfigurationException
* if there is any configuration error.
*/
protected void init()
throws ConfigurationException
{
}
/**
* Generate a security context ID which uniquely identifies the security
* context of a virtual session. This ID will never be 0, which is
* reserved for direct connections.
*
* @return Unique security context ID.
*/
protected int nextContextID()
{
int id;
synchronized (lock)
{
// Advance to the next local context ID not already in use; note
// that 0 is reserved for direct sessions.
id = nextContextID;
while (++id == 0 || virtualSessionsByContext.containsKey(id));
nextContextID = id;
// Store placeholder to reserve this identifier.
virtualSessionsByContext.put(id, null);
}
return id;
}
/**
* Make a synchronous method call to a remote router node to authenticate
* the specified account and retrieve the remote node's context ID for the
* newly authenticated, virtual session. Also provide this node's context
* ID for its side of the virtual session, so that the remote node can
* initiate transactions with the proper virtual session on this side.
*
* @param acctID
* User or process account identifier; must be known to the
* remote node.
* @param contextID
* Value which uniquely identifies the security context of this
* side of the virtual session.
* @param queue
* Queue used to transact the authentication method call.
*
* @return Value which uniquely identifies the security context of the
* remote node's side of the virtual session.
*
* @throws Exception
* if an exception is thrown by the remote method invoked.
*/
protected int callAuthenticateRemote(String acctID,
int contextID,
Queue queue)
throws Exception
{
Message message = new Message(MessageTypes.REQUEST_SYNCH);
RoutingKey key = new RoutingKey(RoutingKey.AUTHENTICATE_REMOTE);
key.setSourceAddress(queue.getNodeAddress());
key.setDestinationAddress(queue.getRemoteAddress());
message.setKey(key);
Object[] payload = new Object[]
{
acctID,
contextID
};
message.setPayload(payload);
Integer reply = (Integer) queue.transact(message, 0);
return reply;
}
/**
* Initialize the session manager. This method is called by the {@link
* SessionManagerFactory} after the session manager instance has been
* constructed or retrieved for reuse.
*
* @param config
* Bootstrap configuration for this network node.
*
* @throws ConfigurationException
* if there is any configuration error.
*/
final void initialize(BootstrapConfig config)
throws ConfigurationException
{
if (config == null)
{
throw new ConfigurationException(
"Bootstrap configuration is required");
}
this.config = config;
secMgr = SecurityManager.getInstance();
assert (secMgr != null);
synchronized (lock)
{
topology.clear();
addressUseCount.clear();
sessionsByQueue.clear();
virtualSessionsByContext.clear();
virtualControlByContext.clear();
activeSession.set(null);
// Give subclasses a chance to do additional initialization.
init();
}
}
/**
* Get the security plugin which is responsible for the resources of the
* specified type. Delegates to the <code>SecurityManager</code>. If
* the current node is a leaf node, <code>null</code> is returned.
* <p>
* Important Note: a security context should be set before this method is
* invoked. Otherwise, <code>null</code> will be returned.
*
* @param resourceTypeName
* Type of resource managed by the security plugin.
*
* @return <code>AbstractResource</code> object, or <code>null</code>.
*/
AbstractResource getSecurityPlugin(String resourceTypeName)
{
return secMgr.getPluginInstance(resourceTypeName);
}
/**
* For each session associated with the given queue, notify all of the
* session's session listeners of the termination event. At the time this
* method is invoked, the queue has been disabled for new messages, but the
* underlying transport mechanism is still active.
*
* @param queue
* Queue which is terminating.
*
* @see #queueStopped(Queue)
*/
void notifyTermination(Queue queue)
{
Set<BaseSession> set = null;
synchronized (lock)
{
set = sessionsByQueue.get(queue);
}
if (set != null)
{
for (BaseSession next : set)
{
setContext(next);
try
{
next.sendTerminationEvent();
}
finally
{
restoreContext();
}
}
}
// queue is terminating, remove the virtual lock (if any)
virtualLocksByQueue.remove(queue);
}
/**
* End all sessions associated with the given queue, since the queue has
* been stopped. By the time this is called, the low level message
* transport is no longer running. This method is about cleaning up higher
* level session resources, including security contexts. All session
* listeners already have been notified of the termination.
*
* @param queue
* Queue which has been stopped.
*
* @see #notifyTermination(Queue)
*/
void queueStopped(Queue queue)
{
synchronized (queue)
{
Set<BaseSession> set = null;
synchronized (lock)
{
// Iterate through all sessions for the given queue, stopping each
// one. We make a copy of the set of sessions to avoid a concurrent
// modification exception when the session is deregistered from the
// session manager's database.
set = sessionsByQueue.get(queue);
if (set == null)
{
return;
}
}
// moved out internal synchronized section to prevent deadlocks
Set<BaseSession> copy = new HashSet<BaseSession>(set);
Iterator<BaseSession> iter = copy.iterator();
boolean interrupt = true;
while (iter.hasNext())
{
BaseSession session = iter.next();
endSession(session, interrupt, false);
interrupt = false;
}
}
}
/**
* Process the reply to an inbound echo request, if a session can be found
* for the message's context.
*
* @param queue
* The queue which received the message.
* @param msg
* Echo request message.
*/
void processEcho(Queue queue, Message msg)
{
int id = msg.getKey().getContextID();
BaseSession session = null;
synchronized (lock)
{
Set<BaseSession> set = sessionsByQueue.get(queue);
for (BaseSession s : set)
{
if (id == s.getContextID())
{
session = s;
break;
}
}
}
if (session != null)
{
session.echoReply(msg);
}
}
// Clean up all remaining sessions and tear down network environment,
private class Stopper
implements Runnable
{
public void run()
{
// Wait a while before cleaning up to give remote shutdown call a
// chance to return before tearing down the transport.
try
{
Thread.sleep(1000);
}
catch (InterruptedException exc)
{
}
try
{
// Collect all sessions in an independent set whose iteration will
// not be disrupted by endSession() processing.
Set<BaseSession> copy = null;
synchronized (lock)
{
copy = new HashSet<>();
for (Queue key : sessionsByQueue.keySet())
{
copy.addAll(sessionsByQueue.get(key));
}
}
// End all sessions.
boolean interrupt = true;
Iterator<BaseSession> iter = copy.iterator();
while (iter.hasNext())
{
endSession(iter.next(), interrupt, true);
interrupt = false;
}
boolean isRoutingNode = isRouter();
if (isRoutingNode)
{
terminateInput();
}
Queue.terminateQueue();
if (isRoutingNode)
{
Router.terminateQueue();
}
}
finally
{
synchronized (stopper)
{
shutdownDone = true;
}
}
}
};
}