RouterSessionManager.java
/*
** Module : RouterSessionManager.java
** Abstract : SessionManager implementation for a router network node
**
** Copyright (c) 2007-2025, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- --------------------------Description-------------------------------
** 001 ECF 20071101 @35867 Created initial version. SessionManager
** implementation for a router network node.
** Note: reused code from other source files.
** 002 ECF 20071121 @35940 Fixed initialization sequence. initialize()
** is now init(), which is called from the
** parent's initialize() method.
** 003 ECF 20071127 @36049 Added certAlias parameter to connectVirtual()
** variant. If alias is null, requester-side
** validation of remote server's certificate is
** not performed.
** 004 ECF 20071207 @36275 Removed try-catch in Incoming constructor.
** Only IOException can be thrown, which was
** already in the throws clause of the c'tor. We
** were losing information about the error,
** since only the error message was being stored
** in the new IOException being thrown.
** 005 ECF 20080213 @37067 Added isLocalNode(). Reports whether a given
** InetSocketAddress represents the local P2J
** node.
** 006 GES 20081024 @40330 Insecure socket support. Refactored to allow
** multiple listening loops.
** 007 JJC 20090208 @41405 Added getSSLSocket() method.
** 008 ECF 20090205 @42103 Changed to accommodate updated Queue API.
** Queue.stop() method now accepts a boolean
** parameter to optionally switch security
** context to the session's user's context.
** 009 SVL 20090715 @43196 Added OOME protection into listenWorker() and
** Incoming.run().
** 010 SVL 20090717 @43227 Server socket is restarted into listenWorker() if
** it has been closed because of error.
** 011 GES 20090722 @43329 Mark the insecure listening thread as daemon. Exit
** the listenWorker() in the case of a clean shutdown.
** This was regressed by H010.
** 012 GES 20090722 @43337 Better thread names to enhance debugging.
** 013 GES 20090816 @43638 Protect createDirectSession() from executing if
** the queue is stopped before the create can occur.
** Added a cleanup method that clears a leak in the
** incoming map in that same race condition.
** 014 NVS 20090816 @43645 Logging non-failure type early client exits during
** authentication.
** 015 NVS 20090820 @43709 With recent changes to the NET package, the
** connection requestor now has responsibility
** to mark the queue as initialized for session.
** 016 CA 20090824 @43720 In Incoming.run, the logging level for
** "authentication failure" message was changed from
** WARNING to FINER.
** 017 CA 20090825 @43755 Fixed a deadlock between Queue and SessionManager.
** This was discovered during remote DB connection
** closing.
** 018 CA 20090901 @43815 Changes related to sessionsByContext rename. When
** the context is set for a virtual session, it will
** create and set in the context a control for this
** virtual session. Fixed a leak related to the
** virtual session deregistration (they were not
** removed from the virtualSessionsByContext map, as
** their context ID is the peer's ID, not the local
** one).
** 019 CA 20091118 @44422 Fixed a deadlock encountered when the client
** disconnects while the router builds the direct
** session.
** 020 CA 20091202 @44472 Rolled back H019 (@44422). The real deadlock was
** with RouterSessionManager.connectVirtual. All
** nested locks should sync on Queue instance first
** and on SessionManager.lock second.
** 021 CA 20100329 @44782 Added terminateVirtual implementation, which
** ensures proper synchronization is done using a
** distinct lock per each virtual queue, before
** terminating the specified session.
** 022 CA 20110105 Changes to allow virtual sessions from leaf nodes
** 023 LMR 20101210 Changed init(), connectVirtual() and listen() to
** learn the ports to open from net/server/port,
** net/server/insecure_port, net/server/secure_port
** in the right order, depending on
** net/connection/secure and
** net/connection/insecure bootstrap config entries
** or command line options (net/connection/insecure
** only applies to the server). Secure and insecure
** environments are set up independently; if not
** set, net/connection/secure defaults to false and
** net/connection/insecure to true and they can't
** be both set to false; a ConfigurationException
** is thrown if proper port options are found to be
** missing.
** 024 CA 20110609 Fixes a bug which deadlocked the socket listener
** thread if a misconfigured client was sending the
** wrong data to the socket (i.e. secure to an
** insecure port and vice versa).
** 025 CA 20110802 Fixed a deadlock which occurs if the user closes
** the terminal window right after authentication
** process has started (i.e. right after ENTER was
** pressed to send the username and password).
** 026 CS 20130204 Added java.net.Socket import to disambiguate from
** com.goldencode.p2j.util.Socket class.
** 027 CA 20130529 Allow virtual connections to authenticate using an explicit account;
** needed by appserver implementation.
** 028 CA 20130704 Changed "security:account:name" to the "access:subject:id", when
** specifying an explicit account in connectVirtual.
** Start appservers automatically, if needed.
** 029 OM 20131008 Added listeners to the new session created by createDirectSession().
** 030 CA 20140206 Added a way of notifying when the server sockets are ready (needed by
** the scheduler, to know when to start launching processes).
** 031 EVL 20160225 Javadoc fixes to make compatible with Oracle Java 8 for Solaris 10.
** 032 IAS 20160331 Moved non-null assertion to a proper place
** 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 20190613 Fixed a deadlock related to allHosts - this must be populated in a
** lock.
** 035 CA 20190902 The context for the requester session must be dedicated to it and not
** inherited, as the session is on its own queue - and when the queue is
** terminating, it will switch the activeSession, which will leak to
** the inherited context's session (if setup like this).
** 036 IAS 20200722 Refactored to the new NetSocket API, added handshake error handling
** IAS 20210325 Added NIO configuration via BootstrapConfig
** IAS 20210827 Added allowed ciphers' filtering
** IAS 20220311 Correct processing of unknown host
** TJD 20220713 Synchronization and locking optimizations
** 037 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 038 GBB 20230825 SecurityManager context & virtual connection method calls updated.
** 039 GBB 20240709 Hard-coded config names replaced by ConfigItem constants.
** 040 CA 20240910 'net:socket:nio' is enabled by default only for Java 8.
** 041 TJD 20231012 SessionManager.lock access optimizations
** 042 RNC 20250113 Added locking for session launching.
** 043 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.net.Socket;
import java.nio.*;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
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.ErrorManager;
import com.goldencode.p2j.util.logging.*;
/**
* Concrete implementation of a session manager for a router network node.
* This class overrides all of the methods of <code>SessionManager</code>
* which have behavior specific to a router node, such as security context
* management, virtual session multiplexing and management, integration with
* the node's router, and server-side socket connection handling.
* <p>
* An inner class manages new, incoming socket connections, creating a new
* message transport queue for each. These resources are likewise cleaned up
* when each connection is finished.
* <p>
* This class is not instantiated directly. A singleton instance is created
* using the {@link SessionManagerFactory#createRouterNode(BootstrapConfig)}
* factory method. This instance is later retrieved with {@link
* SessionManager#get()}.
*/
public final class RouterSessionManager
extends SessionManager
{
/** Logger. */
private static final CentralLogger LOG = CentralLogger.get(RouterSessionManager.class);
/** Thread group for the connection threads. */
private static final ThreadGroup connThreads = new ThreadGroup(
Thread.currentThread().getThreadGroup(), "connection");
/** Map of queues to contexts pending direct session creation. */
private static final Map<Queue, Incoming> incomingMap =
Collections.synchronizedMap(new HashMap<Queue, Incoming>());
/** Server socket upon which all secure client connections arrive. */
private ServerSocket secure = null;
/** Server socket upon which all INSECURE client connections arrive. */
private ServerSocket insecure = null;
/** All host addresses on all network interfaces for the local machine. */
private Set<InetAddress> allHosts = null;
/** Shutdown flag. */
private boolean shutdown = false;
/** Signals when the network portions of the server are ready. */
private CountDownLatch ready = new CountDownLatch(1);
private static final AtomicInteger INPUT_COUNT = new AtomicInteger(0);
/** Lock used to stop the launching of new sessions. If {@code true}, all incoming socket connections
* are ignored. */
private boolean sessionLock = false;
/**
* Constructor which enforces the singleton pattern.
*
* @throws IllegalStateException
* if called more than once per JVM instance.
* @throws NullPointerException
* if <code>config</code> is <code>null</code>.
*/
RouterSessionManager()
{
super();
}
/**
* Expose the SSLServerSocket so the port and other parameters
* used by the socket can be exposed to logic outside this class
*
* @return <code>SSLServerSocket</code>.
*/
public ServerSocket getSSLSocket()
{
return secure;
}
/**
* Convenience method to detect whether this session manager operates on a
* router node.
*
* @return <code>true</code>.
*/
public boolean isRouter()
{
return true;
}
/**
* 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.
*
* @param address
* Socket address to check.
*
* @return <code>true</code> if an address match is detected, else
* <code>false</code>.
*/
public boolean isLocalNode(InetSocketAddress address)
{
InetAddress host = address.getAddress();
if (host == null)
{
ErrorManager.recordOrThrowError(
new int[] {5193},
new String[] {
String.format(
"Unable to find host %s in file HOSTS or HOSTS file not found in expected location",
address.getHostString()
)
},
false, false, true
);
}
int port = address.getPort();
int securePort = -1;
int insecurePort = -1;
if (secure == null)
{
// Not yet listening; pull value from bootstrap config.
securePort = config.getInt(ConfigItem.SERVER_PORT, -1);
insecurePort = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
}
else
{
// Get port from server socket(s).
securePort = secure.getLocalPort();
if (insecure != null)
{
insecurePort = insecure.getLocalPort();
}
}
if (port != securePort && port != insecurePort)
{
return false;
}
// as there is a dedicated lock object, we dont synchronize on "this" but on that lock
boolean result = false;
synchronized (lock)
{
if (allHosts == null)
{
allHosts = new HashSet<InetAddress>();
try
{
Enumeration<NetworkInterface> e1 =
NetworkInterface.getNetworkInterfaces();
while (e1.hasMoreElements())
{
NetworkInterface ni = e1.nextElement();
Enumeration<InetAddress> e2 = ni.getInetAddresses();
while (e2.hasMoreElements())
{
InetAddress ia = e2.nextElement();
allHosts.add(ia);
}
}
}
catch (SocketException exc)
{
throw new RuntimeException("Network error", exc);
}
}
result = allHosts.contains(host);
}
return result;
}
/**
* Establish a virtual (multiplexed) connection to a network node at the
* given address.
* <p>
* This method is called by the requester side to establish a virtual
* session. The first time it must connect to a particular destination, a
* message transport queue is established. This requires authentication
* between two router nodes, which is the first level of a two-level
* authentication process. Thereafter, requests for additional sessions
* with that endpoint use the existing queue, and communications are
* multiplexed by context ID.
* <p>
* Once a transport is available, the current user's account identifier is
* used to authenticate at the remote node. This user authentication
* represents the second level of authentication. On the remote side, if
* authentication is successful (the account must be registered under the
* same ID on the remote side, and have sufficient rights to authenticate),
* the other "half" of the virtual session is established on the remote
* side. A unique context ID is assigned to that session and is returned
* to the requester. This context ID is used to establish a virtual
* session ({@link RequesterSession}) on this side. This ID is later used
* in outbound messages to multiplex communications over the shared queue.
* <p>
* This method creates an in-memory <code>BootstrapConfig</code> object to
* enable the first level of authentication, which is certificate-based.
* Configuration entries are populated as follows (note that some values
* are retrieved from the <code>BootstrapConfig</code> instance set by
* {@link SessionManager#initialize(BootstrapConfig)}; this version of the
* configuration is referred to below as the <i>default config</i>):
* <ul>
* <li>server mode is set to false; even though the requester is a
* router node, it is treated as a client from the security manager's
* perspective when requesting authentication
* <li>net/server/host - taken from <code>address</code> parameter
* <li>net/server/port - taken from <code>address</code> parameter
* <li>security/certificate/validate - <code>true</code> only if
* <code>certAlias</code> is non-<code>null</code>, otherwise is left
* unset, which defaults to <code>false</code> with the security
* manager; if <code>true</code>, forces certificate-based
* validation of the providing server
* <li>security/truststore/alias - taken from <code>certAlias</code>
* parameter, if that parameter is non-<code>null</code>, otherwise
* is left unset; this is necessary to allow the security manager to
* check the correct truststore for the providing server, if
* certification-based validation of the providing server is desired
* </ul>
*
* @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. May be
* <code>null</code>, which indicates requesting server does not
* validate providing server's certificate.
* @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 Exception
* if there are any errors reading configuration, creating a
* queue, or authenticating the user session remotely.
*/
public Session connectVirtual(InetSocketAddress address,
String certAlias,
SessionListener notify,
String account)
throws Exception
{
String host = address.getHostName();
int port = address.getPort();
BootstrapConfig config = new BootstrapConfig();
config.setServer(false);
// Network category.
config.setConfigItem(ConfigItem.SERVER_HOST, host);
config.setConfigItem(ConfigItem.SERVER_SECURE_PORT, String.valueOf(port));
// virtual connections are always made over secure sockets
config.setConfigItem(ConfigItem.SECURE, String.valueOf(true));
// Security category.
if (certAlias != null)
{
// Enable requester-side validation of remote server's certificate.
config.setConfigItem("security", "certificate", "validate", "true");
config.setConfigItem("security", "truststore", "alias", certAlias);
}
if (account != null)
{
config.setConfigItem("access", "subject", "id", account);
}
return connectVirtual(config, notify);
}
/**
* Establish a virtual (multiplexed) connection to a network node at the
* given address.
* <p>
* This method is called by the requester side to establish a virtual
* session. The first time it must connect to a particular destination, a
* message transport queue is established. This requires authentication
* between two router nodes, which is the first level of a two-level
* authentication process. Thereafter, requests for additional sessions
* with that endpoint use the existing queue, and communications are
* multiplexed by context ID.
* <p>
* Once a transport is available, the current user's account identifier is
* used to authenticate at the remote node. This user authentication
* represents the second level of authentication. On the remote side, if
* authentication is successful (the account must be registered under the
* same ID on the remote side, and have sufficient rights to authenticate),
* the other "half" of the virtual session is established on the remote
* side. A unique context ID is assigned to that session and is returned
* to the requester. This context ID is used to establish a virtual
* session ({@link RequesterSession}) on this side. This ID is later used
* in outbound messages to multiplex communications over the shared queue.
* <p>
* This method is synchronized so, if multiple clients are calling this
* method, only one of them will create the connection.
*
* @param config
* Bootstrap configuration information which contains at least
* enough information to make a virtual connection from this node
* to another. See {@link #connectVirtual(BootstrapConfig, SessionListener)}
* for details.
* @param notify
* Session listener to be notified of session events.
*
* @return Session established after the connection has been made.
*
* @throws Exception
* if there are any errors reading configuration, creating a
* queue, or authenticating the user session remotely.
*/
public Session connectVirtual(BootstrapConfig config,
SessionListener notify)
throws Exception
{
// removed synchronized keyword as there is a lock object to synchronize on
// net/server/secure_port is only read if net/connection/secure==true
if (!config.getBoolean(ConfigItem.SECURE, false))
{
throw new ConfigurationException("net/connection/secure is false");
}
int port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
String host = config.getString(ConfigItem.SERVER_HOST, null);
InetSocketAddress remoteAddr = new InetSocketAddress(host, port);
Queue queue = null;
boolean createdQueue = false;
synchronized (lock)
{
// during queue creation there is no lock on SM.lock, so its safe to
// sync on SM.lock here
queue = topology.get(remoteAddr);
if (queue == null)
{
queue = this.createQueue(remoteAddr, config);
createdQueue = true;
virtualLocksByQueue.put(queue, new Object());
}
}
if (createdQueue)
{
// no synchronization on SM.lock is needed for queue start
int nodeAddr = this.config.getInt("net",
"server",
"nodeAddr",
0x00010000);
queue.setNodeAddress(nodeAddr);
// Begin protocol operations.
int timeout = config.getInt("net", "server", "timeout", 600000);
queue.start(true, timeout, false, false);
}
BaseSession session = null;
try
{
// no synchronization on SM.lock is needed for authentication
// first check if an explicit account is not set
String identity = config.getConfigItem("access", "subject", "id");
if (identity == null || identity.equals(""))
{
Object context = secMgr.virtualSm.getEffectiveContext();
// use the account for the current context, if not explicitly set
identity = secMgr.virtualSm.getIdentity(context);
}
assert (!"".equals(identity));
int contextID = nextContextID();
int peerID = callAuthenticateRemote(identity, contextID, queue);
// registerSession sets a lock on SM.lock, so a lock on queue must be
// acquired first
synchronized (queue)
{
Object virtualContext = secMgr.virtualSm.createSecurityContext(identity);
if (virtualContext == null)
{
throw new IllegalAccessException("Could not create virtual session.");
}
session = new RequesterSession(queue, virtualContext, peerID, contextID);
queue.sessionInitialized();
if (notify != null)
{
session.addSessionListener(notify);
}
// Update session manager database.
registerSession(contextID, remoteAddr, session);
}
// At this moment the session is almost ready for use, let the listeners the
// initialization has ended
session.sendInitializationEvent();
}
catch (IllegalAccessException exc)
{
if (createdQueue)
{
// remote authentication was denied; if we created the queue
// and no one else is using it, tear it down
Set<BaseSession> sessions = null;
synchronized (lock)
{
sessions = sessionsByQueue.get(queue);
}
// moved out synchronized section to prevent deadlocks
if (sessions == null)
{
queue.stop(exc, false);
}
}
exc.fillInStackTrace();
throw exc;
}
return session;
}
/**
* 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 calling thread was interrupted
*/
public void waitUntilReady()
throws InterruptedException
{
ready.await();
}
/**
* Listen for incoming socket connections, authenticate each new
* connection, and establish a new queue for each authenticated connection.
* <p>
* Depending on which modes (secure/insecure) are enabled, this method
* will start one or two socket listening loops. The insecure socket
* can be used to obtain better performance but must only be used in cases
* where the network transport is known to be secure. This is notoriously
* difficult to ensure and so such usage should be avoided if at all
* possible.
* <p>
* If both insecure and secure sockets are used, a second thread will be
* launched to implement the secure listening loop.
*
* @param notify
* Session listener to be notified of session events.
*
* @throws Exception
* if there is any configuration error, or error creating a server
* socket.
*/
public void listen(final SessionListener notify)
throws Exception
{
// see which ports have to be opened: secure and/or non-secure
// (init() checks to make sure that (isSecure || isInsecure)==true)
boolean isSecure = this.config.getBoolean(ConfigItem.SECURE, false);
boolean isInsecure = this.config.getBoolean("net", "connection", "insecure", true);
//optionally init the non-secure environment
if (isInsecure)
{
int insecurePort = config.getInt(ConfigItem.SERVER_PORT, -1);
if (insecurePort == -1)
{
insecurePort = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
}
createServerSocket(false, insecurePort);
if (LOG.isLoggable(Level.WARNING))
{
LOG.logp(Level.WARNING,
"SessionManager.listen()",
"",
"INSECURE sockets in use!");
}
}
// optionally init the secure environment
if (isSecure)
{
int securePort = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
if (securePort != -1)
{
createServerSocket(true, securePort);
// start the secondary listening thread if needed
if (isInsecure)
{
Thread t = new Thread(connThreads, new Runnable()
{
public void run()
{
listenWorker(notify, secure, Mode.SSL);
}
});
t.setDaemon(true);
t.setName("Secure Socket Listener");
t.start();
}
}
// if the insecure mode is not enabled the current thread listens on
// the secure socket
}
// log this before the current thread starts its listening loop
if (LOG.isLoggable(Level.INFO))
{
LOG.logp(Level.INFO,
"SessionManager.listen()",
"",
"Server ready");
}
ready.countDown();
// start listening loop in the current thread (secure or non-secure):
if (isInsecure)
{
// if enabled, the secure listening loop has been already started in
// a separate thread
listenWorker(notify, insecure, Mode.INSECURE);
}
else
{
// secure mode must've been enabled (we tested for this in init())
listenWorker(notify, secure, Mode.SSL);
}
}
/**
* 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 (super.toString() + " PENDING: " + incomingMap);
}
/**
* Setter for the {@link #sessionLock} variable.
*
* @param lockStatus
* The new status for the lock. {@code true} if the session launching will be stopped,
* {@code false} otherwise
*
* @return The new value of the {@link #sessionLock} variable.
*/
@Override
public synchronized boolean setSessionLock(boolean lockStatus)
{
sessionLock = lockStatus;
return sessionLock;
}
/**
* Getter for the {@link #sessionLock} variable.
*
* @return The value of the {@link #sessionLock} variable.
*/
@Override
public synchronized boolean getSessionLock()
{
return sessionLock;
}
/**
* 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.
*
* @param identity
* Account identifier for the requesting party.
* @param peerID
* 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 IllegalAccessException
* if authentication is not successful.
*/
public int authenticateRemote(String identity, int peerID, Queue queue)
throws IllegalAccessException
{
incomingMap.remove(queue);
Object context = null;
try
{
context = secMgr.authenticateRemote(identity);
}
catch (RestrictedUseException exc)
{
// safe calling location; should not be thrown
}
if (context == null)
{
throw new IllegalAccessException(
"Remote access for " + identity + " is not allowed");
}
int contextID = nextContextID();
VirtualSession session = new VirtualSession(queue,
context,
peerID,
contextID);
registerSession(contextID, null, session);
// At this moment the session is almost ready for use, let the listeners the
// initialization has ended
session.sendInitializationEvent();
return contextID;
}
/**
* 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.
*
* @param contextID
* Value which uniquely identifies the security context of the
* requesting party.
*/
public void terminateRemote(int contextID)
{
Queue queue = null;
// determine queue first, so locking by queue is possible
synchronized (lock)
{
BaseSession session = virtualSessionsByContext.get(contextID);
assert (session != null);
queue = session.getQueue();
}
synchronized (queue)
{
BaseSession session = null;
synchronized (lock)
{
session = virtualSessionsByContext.get(contextID);
assert (session != null);
}
// moved out synchronized section to prevent deadlocks
// endSession can be called out of SM.lock
endSession(session, true, true);
}
}
/**
* Perform any special initialization which is not handled by the default
* (common) logic. This includes checking the bootstrap configuration for
* valid values, initializing the dispatcher, and initializing the router.
*
* @throws IllegalStateException
* if the router node is already listening for connections.
* @throws ConfigurationException
* if any configuration error occurs during initialization.
*/
protected void init()
throws ConfigurationException
{
if (secure != null || insecure != null)
{
throw new IllegalStateException("The server is already running.");
}
boolean isSecure = this.config.getBoolean(ConfigItem.SECURE, false);
boolean isInsecure = this.config.getBoolean("net", "connection", "insecure", true);
if (!isSecure && !isInsecure)
{
throw new ConfigurationException("net/connection/secure and " +
"net/connection/insecure are both false");
}
incomingMap.clear();
String serverId = config.getString("security", "server", "id", null);
if (serverId == null || serverId.length() == 0)
{
throw new ConfigurationException("Missing server ID.");
}
int nodeAddr = config.getInt("net", "server", "nodeAddr", 0x00010000);
if (nodeAddr == 0 || (nodeAddr & AddressServer.NODE_MASK) != 0)
{
throw new ConfigurationException("Missing or invalid node address.");
}
int port;
if (isInsecure)
{
port = config.getInt(ConfigItem.SERVER_PORT, -1);
if (port <= 0)
{
port = config.getInt(ConfigItem.SERVER_INSECURE_PORT, -1);
}
if (port <= 0)
{
throw new ConfigurationException("Missing or invalid insecure port number.");
}
}
if (isSecure)
{
port = config.getInt(ConfigItem.SERVER_SECURE_PORT, -1);
if (port <= 0)
{
throw new ConfigurationException("Missing or invalid secure port number.");
}
}
// initialize the protocol's dispatcher
int threads = config.getInt("net", "dispatcher", "threads", 2);
Dispatcher.initialize(threads);
// init router
int rthreads = config.getInt("net", "router", "threads", 1);
Router.initRouter(nodeAddr, rthreads);
}
/**
* Create a direct session for the given queue and register it with the
* session manager. This enables the deferred creation of a direct
* session on the provider side of the connection, during the requester's
* queue's handshaking with the router. This deferral is necessary
* because there is not enough information at the time the connection is
* established and the queue is created, to determine what type of session
* is needed. The security context, which is temporarily stored during
* connection time (when the requester is authenticated), is now associated
* with the new session and is removed from its temporary storage.
*
* @param queue
* Underlying message transport for the new session.
*
* @return <code>true</code> if the session initialized successfully.
*/
protected boolean createDirectSession(Queue queue)
{
synchronized (queue)
{
// removed internal synchronized section to prevent deadlocks
// if the session was aborted before we could initialize, there is
// nothing to do
if (!queue.isRunning())
return false;
Incoming incoming = incomingMap.remove(queue);
assert (incoming != null);
DirectSession session = new DirectSession(queue, incoming.context);
if (incoming.notify != null)
{
session.addSessionListener(incoming.notify);
}
registerSession(0, null, session);
// add saved callbacks for initialization and termination of session
for (SessionListener sl : incoming.socket.getSessionListeners())
{
session.addSessionListener(sl);
}
}
return true;
}
/**
* Close the server's socket if it was ever initialized.
*/
protected void terminateInput()
{
try
{
synchronized (lock)
{
shutdown = true;
}
if (secure != null)
{
secure.close();
}
if (insecure != null)
{
insecure.close();
}
// wait a while after terminating
Thread.sleep(1000);
}
catch (InterruptedException exc)
{
}
catch (IOException exc)
{
// log the fact that some form of failure has occurred
if (LOG.isLoggable(Level.WARNING))
{
LOG.logp(Level.WARNING,
"RouterSessionManager.terminateInput()",
"",
"socket close failure",
exc);
}
}
}
/**
* 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>
* The parent's implementation is overridden here to permit additional
* cleanup necessary for this implementation. However, this implementation
* first invokes the parent's.
*
* @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)
{
boolean queueInUse = super.deregisterSession(session);
Queue queue = session.getQueue();
assert (queue != null);
synchronized (queue)
{
synchronized (lock)
{
int contextID = session.getContextID();
// if this is a virtual session, we must get the ID from the local
// peer ID
if (session instanceof VirtualSession)
{
contextID = ((VirtualSession) session).getLocalContextID();
virtualSessionsByContext.remove(contextID);
virtualControlByContext.remove(contextID);
}
}
}
if (!queueInUse)
{
incomingMap.remove(queue);
}
return queueInUse;
}
/**
* Set the security context for the current thread, using the session
* identified by the given routing key or transport queue. If the routing
* key has a security context ID of 0, the context of the default session
* for the given queue is used. If the routing key's context ID is
* non-zero, the appropriate session is retrieved for the ID and its
* context is used.
*
* @param key
* Routing key for an incoming message.
* @param queue
* Message transport queue.
*/
protected void setContext(RoutingKey key, Queue queue)
{
int contextID = key.getContextID();
if (contextID == 0)
{
// No demultiplexing required; there is only one session for the
// queue.
setContext(queue);
}
else
{
// Demultiplex the session by context ID from among the sessions
// sharing the queue.
BaseSession session = null;
synchronized (lock)
{
session = virtualSessionsByContext.get(contextID);
}
setContext(session);
synchronized (lock)
{
// get the control for this context
Control ct = Control.locate(contextID);
if (!ct.isReady())
{
ct.initControl(session);
}
Control.setContext(ct);
}
}
}
/**
* 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.
*
* @param queue
* Queue which uniquely identifies the session whose security
* context is to be used.
*
* @throws IllegalArgumentException
* if there is not exactly one session for the given queue.
*/
protected void setContext(Queue queue)
{
BaseSession session = null;
synchronized (queue)
{
synchronized (lock)
{
Set<BaseSession> set = sessionsByQueue.get(queue);
// this is a special case of setting the context after the queue is
// stopped: if the requester side terminated abnormally and it had a
// chance to establish more than one virtual session, the remote side
// will "see" more than one session for this queue; thus, in this
// case, we will use the associated DirectSession as the source of
// the context, and fail if there is more than one DirectSession for
// this queue
if (!queue.isRunning() && set != null)
{
Iterator<BaseSession> iter = set.iterator();
boolean found = false;
while (iter.hasNext())
{
BaseSession sess = iter.next();
if (sess instanceof DirectSession)
{
if (!found)
{
found = true;
session = sess;
}
else
{
throw new IllegalArgumentException(
"No default session found for queue");
}
}
}
}
if (session == null)
{
if (set == null || set.size() != 1)
{
throw new IllegalArgumentException(
"No default session found for queue");
}
session = set.iterator().next();
}
}
}
setContext(session);
}
/**
* 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.
*/
@Override
protected void setContext(BaseSession session)
{
// Set the session's current context.
session.setContext();
// Remember the current session.
activeSession.set(session);
}
/**
* Restore the current thread's security context to whatever it was before
* it was last explicitly set.
*/
protected void restoreContext()
{
try
{
secMgr.contextSm.popAndRestoreSecurityContext();
}
catch (RestrictedUseException exc)
{
throw new RuntimeException("Invalid security context switch");
}
}
/**
* Initialize security context switching for the calling thread.
*/
protected void setInitialContext()
{
try
{
secMgr.contextSm.setInitialSecurityContext();
}
catch (RestrictedUseException exc)
{
throw new RuntimeException("Invalid security context switch");
}
}
/**
* Release the initial security context assigned to the calling thread.
*/
protected void dropInitialContext()
{
secMgr.contextSm.dropInitialSecurityContext();
}
/**
* Cleanup session state before initialization.
*
* @param queue
* The queue that represents the partially initialized session
* to be cleaned up.
*
* @return Security manager context to that still requires cleanup.
*/
Object abortSessionInit(Queue queue)
{
Incoming incoming = incomingMap.remove(queue);
return incoming == null ? null : incoming.context;
}
/**
* Core listening loop for incoming socket connections.
*
* @param notify
* Session listener to be notified of session events.
* @param input
* The socket on which to listen.
* @param mode
* The connection mode.
*/
private void listenWorker(SessionListener notify, ServerSocket input, Mode mode)
{
String transientName = String.format("Incoming %s Connector", mode);
String logtxt = String.format("SessionManager.listenWorker(%s)", mode);
// main socket listening loop
while (true)
{
Socket sock = null;
SocketChannel channel = null;
ServerSocketChannel serverSocketChannel = input.getChannel();
try
{
Incoming incoming;
if (serverSocketChannel != null)
{
channel = serverSocketChannel.accept();
channel.setOption(StandardSocketOptions.TCP_NODELAY , true);
incoming = new Incoming(channel, mode, notify);
}
else
{
sock = input.accept();
sock.setTcpNoDelay(true);
incoming = new Incoming(sock, mode, notify);
}
// we need to authenticate this incoming connection before
// starting the queue, since this is a potentially long-running
// task we create a thread that does this and then that thread
// exits (this way this listening loop is not blocked)
Thread t = new Thread(connThreads, incoming);
// make sure this temporary thread doesn't stop the server from
// exiting
t.setDaemon(true);
t.setName(transientName + "#" + INPUT_COUNT.incrementAndGet());
t.start();
}
catch (Throwable thr)
{
// log the fact that some form of failure has occurred
try
{
boolean clean = false;
synchronized (lock)
{
clean = shutdown;
}
if (clean)
{
if (LOG.isLoggable(Level.INFO))
{
LOG.logp(Level.INFO,
logtxt,
"",
"Server shutting down.");
}
// exit the listener
return;
}
else
{
if (LOG.isLoggable(Level.WARNING))
{
LOG.logp(Level.WARNING,
logtxt,
"",
"Failure in listening loop.",
thr);
}
}
// release the connection with the recently connected client
if (sock != null)
{
try
{
sock.close();
}
catch (IOException e)
{
// Ignore.
}
}
if (thr instanceof OutOfMemoryError)
{
try
{
// Some delay to let OOME dissapear.
Thread.sleep(5000);
}
catch (InterruptedException e)
{
// Ignore.
}
}
// if the server socket is closed then we should restart it
if (input.isClosed())
{
try
{
boolean isSecure = input instanceof SSLServerSocket;
input = createServerSocket(isSecure,
input.getLocalPort());
}
catch (Exception e)
{
if (LOG.isLoggable(Level.SEVERE))
{
LOG.logp(Level.SEVERE,
logtxt,
"",
"error while recreating server socket",
e);
}
break;
}
}
}
catch (OutOfMemoryError e)
{
// Ignore.
}
}
}
if (LOG.isLoggable(Level.INFO))
{
LOG.logp(Level.INFO,
logtxt,
"",
"Server complete");
}
}
/**
* Create required server socket and assign it to the corresponding
* field-level variable: {@link #secure} or {@link #insecure}.
*
* @param isSecure
* Determines whether the created socket should be secure.
* @param port
* The port which the socket should listen to.
*
* @return created server socket.
*
* @throws Exception
* if an error has occurred while creating server socket.
*/
private ServerSocket createServerSocket(boolean isSecure, int port)
throws Exception
{
InetSocketAddress endpoint = new InetSocketAddress(port);
if (config.getBoolean(ConfigItem.SERVER_SOCKET_NIO, false))
{
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.bind(endpoint);
ServerSocket socket = serverSocketChannel.socket();
if (isSecure)
{
secure = socket;
}
else
{
insecure = socket;
}
return socket;
}
if (isSecure)
{
SSLContext ctx = secMgr.contextSm.getSecureSocketContext();
SSLServerSocketFactory factory = ctx.getServerSocketFactory();
secure = factory.createServerSocket(port);
((SSLServerSocket) secure).setWantClientAuth(true);
}
else
{
insecure = new ServerSocket();
insecure.bind(endpoint);
}
return isSecure ? secure : insecure;
}
/**
* Connection mode
*/
private static enum Mode
{
INSECURE,
SSL
}
/**
* This class performs all long running tasks with an incoming connection,
* such as authentication and queue initialization.
*/
private class Incoming
implements Runnable
{
/** Listener for session events. */
private final SessionListener notify;
/** Client connection socket. */
private NetSocket socket = null;
/** Incoming connection socket. */
private final Socket inputSocket;
/** The input channel. */
private final SocketChannel inputChannel;
/** Security context */
private Object context = null;
/** The connection mode. */
private Mode mode;
/**
* Constructs an instance.
*
* @param socket
* Incoming session's transport.
* @param mode
* The connection mode.
* @param notify
* Listener to receive session event notifications. May be
* <code>null</code>.
*/
Incoming(Socket socket, Mode mode, SessionListener notify)
{
this.inputChannel = null;
this.inputSocket = socket;
this.notify = notify;
this.mode = mode;
}
/**
* Constructs an instance.
*
* @param channel
* Incoming session's channel.
* @param mode
* The connection mode.
* @param notify
* Listener to receive session event notifications. May be
* <code>null</code>.
*/
Incoming(SocketChannel channel, Mode mode, SessionListener notify)
{
this.inputChannel = channel;
this.inputSocket = channel.socket();
this.notify = notify;
this.mode = mode;
}
/**
* Main processing thread. It does the following:
* <ul>
* <li>initializes security context</li>
* <li>authenticates client</li>
* <li>starts queue</li>
* </ul>
*/
public void run()
{
SessionManager sessMgr = SessionManager.get();
Queue queue = null;
boolean inContext = false;
LOG.log(Level.FINE, "New incoming connection: " + Thread.currentThread().getName());
try
{
if (inputChannel == null)
{
this.socket = NetSocket.FACTORY.create(inputSocket);
}
else if (mode == Mode.INSECURE)
{
this.socket = NetSocket.FACTORY.create(inputChannel);
}
else
{
SSLContext ctx = secMgr.contextSm.getSecureSocketContext();
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(false);
engine.setNeedClientAuth(false);
engine.setWantClientAuth(true);
this.socket = NetSocket.FACTORY.create(engine, inputChannel);
}
if (getSessionLock())
{
try
{
if (socket != null && !socket.isClosed())
{
// first read the AUTHTYPE sent by the client
socket.readInt();
// then send the locking status
socket.writeInt(SecurityConstants.AUTH_MODE_NONE);
socket.writeInt(SecurityConstants.AUTH_ACTION_LOCKED);
socket.flush();
socket.close();
}
}
catch (IOException e)
{
// The socket was most probably closed.
}
return;
}
// obtain an initial security context
sessMgr.setInitialContext();
inContext = true;
// authenticate client
context = sessMgr.secMgr.authenticateLocal(socket);
if (context != null)
{
if (context instanceof String)
{
// log the fact that some form of client exit has occurred
if (LOG.isLoggable(Level.FINE))
{
LOG.logp(Level.WARNING,
"Incoming.run()",
"",
(String) context);
}
}
else
{
queue = new Queue(socket);
incomingMap.put(queue, this);
queue.start(false, 0, false, false);
}
}
else
{
// log the fact that some form of failure has occurred
if (LOG.isLoggable(Level.FINER))
{
LOG.logp(Level.FINER,
"Incoming.run()",
"",
"authentication failure");
}
}
}
catch (Throwable thr)
{
// log the fact that some form of failure has occurred
try
{
if (LOG.isLoggable(Level.WARNING))
{
if (thr instanceof SSLHandshakeException)
{
LOG.log(Level.WARNING, thr.getMessage());
}
else
{
LOG.logp(Level.WARNING,
"Incoming.run()",
"",
"initialization failure",
thr);
}
}
if (queue != null)
{
incomingMap.remove(queue);
queue.stop(thr instanceof Exception ? (Exception) thr :
new Exception(thr),
false);
}
if (context != null && !socket.isClosed())
{
socket.close();
}
}
catch (OutOfMemoryError e)
{
// ignore
}
}
finally
{
if (inContext)
{
sessMgr.dropInitialContext();
}
LOG.log(Level.FINE, "Incoming connection OK: " + Thread.currentThread().getName());
}
}
}
}