serverRoom/backend-csharp/amt-sdk-20-0-0-1/MPSModule/Src/MPS_Src/TunnelSupplier.cpp

//----------------------------------------------------------------------------
//
//  Copyright (C) Intel Corporation, 2006 - 2007.
//
//  File:       TunnelSupplier.cpp
//
//  Contents:   Handles incoming data on APF tunnel.	
//
//  Notes:      
//----------------------------------------------------------------------------
#include <ace/SOCK_Dgram.h>
#include <ace/Configuration.h>

#include "TunnelSupplier.h"
#include "TunnelHandler.h"
#include "TunnelConsumer.h"
#include "TunnelManager.h"
#include "TcpConsumer.h"
#include "ChannelConsumer.h"
#include "APF.h"
#include "connector.h"
#include "global.h"
#include "DevicePresence.h"

#define TEXT_UUID_SIZE		37
#define DELIMITER_HOST_PORT ":"
#define PORT_STR_LENGTH		5

// Utility function declaration
void StringFromUUID(const unsigned char binUUID[16], 
					unsigned char textUUID[TEXT_UUID_SIZE] );


STATUS AMT_Tunnel_Supplier::processReturnVal(STATUS rep)
{
	switch (rep)
	{
	case STATUS_SUCCESS:			// Success
	case STATUS_RECOVER_ERROR:		// some error we wish to keep this tunnel alive
	case STATUS_NETWORK_ERROR:		// We can't send disconnect in this case
	case STATUS_CONNECTION_CLOSED:	// We received disconnect message no need to reply
		return rep;
	case STATUS_AUTH_FAILURE:
		tunnel_handler_->_disconnect_data.setParams(APF_DISCONNECT_ILLEGAL_USER_NAME, true, PRIO_LOW);
		return STATUS_CONNECTION_CLOSED;
	}
	
	// All other status return from any process flow, is invalid and we are closing the channel
	tunnel_handler_->_disconnect_data.setParams(true, PRIO_HIGH);
	
	return STATUS_CONNECTION_CLOSED;
}

AMT_Tunnel_Supplier::AMT_Tunnel_Supplier(AMT_Tunnel_Handler *tunnel_handler):
tunnel_handler_(tunnel_handler), major_version_(), minor_version_(), trigger_reason_()
{
}

AMT_Tunnel_Supplier::~AMT_Tunnel_Supplier(void)
{
	ACE_DEBUG((MY_DEBUG 
		ACE_TEXT("--->AMT_Tunnel_Supplier dtor\n")));
}

// Get reference to this tunnel socket stream
ACE_SOCK_Stream& AMT_Tunnel_Supplier::getPeer(void) const
{
	return (ACE_SOCK_Stream &)tunnel_handler_->peer();
}

// Get reference to this tunnel consumer
AMT_Tunnel_Consumer* AMT_Tunnel_Supplier::consumer() const
{
	return tunnel_handler_->consumer();
}

// Get reference to this tunnel channel manager
Channel_Manager& AMT_Tunnel_Supplier::channel_manager() const
{
	return tunnel_handler_->channel_mgr();
}

void AMT_Tunnel_Supplier::setDisconnectReason(ACE_UINT32 disconnect_reason) 
{ 
	tunnel_handler_->_disconnect_data._disconnect_reason = disconnect_reason; 
}

ACE_UINT32 AMT_Tunnel_Supplier::getDisconnectReason() 
{ 
	return tunnel_handler_->_disconnect_data._disconnect_reason; 
}

// Set/Get this tunnel state
void AMT_Tunnel_Supplier::tunnelState (int state)		
{ 
	tunnel_handler_->tunnelState(state);
}
int AMT_Tunnel_Supplier::tunnelState (void) const		
{ 
	return tunnel_handler_->tunnelState();
}

// Receive and process incoming APF message.
STATUS AMT_Tunnel_Supplier::handle_input()
{ 
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::handle_input"));

	int		res;
	ACE_UINT8	messageType = 0;
	STATUS	status;


	// Get message Type
	if ((res = getPeer().recv_n(&messageType,1, &ACE_Time_Value(*getMaxTunnelTimeout()))) <=0)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier failed reading APF message type from network\n\tres= %d \terrno= %d\n"),
			res, errno));
		return STATUS_NETWORK_ERROR;
	}	

	switch (messageType)
	{
	case APF_DISCONNECT:
		status = process_disconnect();
		break;
	case APF_SERVICE_REQUEST:
		status = process_service_request();
		break;
	case APF_USERAUTH_REQUEST:
		status = process_userauth_request();
		break;
	case APF_GLOBAL_REQUEST:
		status = process_global_message();
		break;
	case APF_CHANNEL_OPEN:
		status = process_channel_open_direct();
		break;
	case APF_CHANNEL_OPEN_CONFIRMATION:
	case APF_CHANNEL_OPEN_FAILURE:
		status = process_channel_open_reply(messageType);
		break;
	case APF_CHANNEL_WINDOW_ADJUST:
		status = process_channel_window_adjust();
		break;
	case APF_CHANNEL_DATA:
		status = process_channel_data();
		break;
	case APF_CHANNEL_CLOSE:
		status = process_channel_close();
		break;
	case APF_PROTOCOLVERSION:
		status = process_init();
		break;
	case APF_KEEPALIVE_REQUEST:
		status = process_KeepAliveReq();
		break;
	case APF_KEEPALIVE_OPTIONS_REQ:
		status = process_KeepAliveOptionReq();
		break;
	// This tunnel can't receive any of the following message types:
	case APF_SERVICE_ACCEPT:
	case APF_USERAUTH_FAILURE:
	case APF_USERAUTH_SUCCESS:
	case APF_REQUEST_SUCCESS:
	case APF_REQUEST_FAILURE:
	default:
		ACE_DEBUG((MY_DEBUG
		ACE_TEXT("[%s] AMT_Tunnel_Supplier received illegal message type (%d) disconnecting\n"),identifier(),
		messageType));
		status = STATUS_ILLEGAL_APF_MESSAGE;
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		break;
	}

	return processReturnVal(status);
}

// negotiate protocol version
STATUS AMT_Tunnel_Supplier::process_init()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_init"));
	STATUS rep = STATUS_SUCCESS;
	
	// Check tunnel state
	if (tunnelState() != AMT_Tunnel_Handler::SESSION_INIT)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received protocol version in invalid state (%d)\n"),
			tunnelState()));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	APF_ProtocolVersion his_version;
	// First read AMT version message
	if ((rep = his_version.read(getPeer())) != STATUS_SUCCESS){
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed reading AMT version message\n")));
		return rep;
	}
	
	// Validate the version message format.
	if (his_version.triggerReason  < USER_INITIATED_REQUEST ||
		his_version.triggerReason  > PERIODIC_REQUEST)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received invalid version message\n")));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		return STATUS_ILLEGAL_APF_MESSAGE;
	}

	// Save trigger reason and UUID
	trigger_reason_ = his_version.triggerReason;
	for (int i=0; i < 16; i++)
		UUID_[i] = his_version.systemId[i];

	// Send version reply to AMT
	APF_ProtocolVersion my_version;
	my_version.majorVersion = (ACE_UINT32)APF_PROTOCOL_VERSION_MAJOR;
	my_version.minorVersion = (ACE_UINT32)APF_PROTOCOL_VERSION_MINOR;
	
	// for debug
	my_version.triggerReason = 0;
	for (int i=0; i < 16; i++)
	{
		my_version.systemId[i] = (ACE_UINT8)i;
		my_version.reserved[i] = (ACE_UINT32)i;
	}

	if ((rep = consumer()->send_APF_message(my_version)) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to send version message\n")));
		return rep;
	}

	// Decide the protocol version as the smallest
	major_version_ = my_version.majorVersion;
	minor_version_ = my_version.minorVersion;

	// This implementation only supports one protocol version
	// If AMTs protocol is lower reject this connection
	if ((his_version.majorVersion < my_version.majorVersion) || 
		((his_version.majorVersion == my_version.majorVersion && 
		  his_version.minorVersion < my_version.minorVersion))) 
	{
		ACE_DEBUG((MY_WARNING
			ACE_TEXT("[%s] Unsupported APF version (%d.%d)\n"),identifier(),
			his_version.majorVersion, his_version.minorVersion));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_VERSION_NOT_SUPPORTED);
		return STATUS_ILLEGAL_APF_MESSAGE;
	}

	// once agreed upon, we completely ignore the protocol version for now.
	tunnelState(AMT_Tunnel_Handler::SESSION_OPENED);

	unsigned char textUUID[TEXT_UUID_SIZE];
	StringFromUUID(UUID_, textUUID);
	tunnel_handler_->set_identifier(ACE_CString((char*)textUUID,TEXT_UUID_SIZE));
		
	ACE_DEBUG ((MY_INFO 
		ACE_TEXT ("Intel remote client %s is now connected.\n"),
		identifier()));

	return STATUS_SUCCESS;
}

// handle disconnect message
STATUS AMT_Tunnel_Supplier::process_disconnect()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_disconnect"));

	STATUS rep;
	APF_Disconnect msg;
	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS){
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT("AMT_Tunnel_Supplier: failed reading disconnect message\n")));
			return STATUS_NETWORK_ERROR;
	}
	
	
	ACE_DEBUG ((MY_DEBUG 
		ACE_TEXT ("Tunnel disconnected (reason code:%d)\n"),
		msg.reason));

	//close socket for write:
	getPeer().close_writer();
	
	//set disconnect params - so later on, on handle_close, we will inform consumer to close itself.
	tunnel_handler_->_disconnect_data.setParams(false, PRIO_HIGH);
	
	// Remove this supplier from the reactor
	return STATUS_CONNECTION_CLOSED;
}

// handle service request
STATUS AMT_Tunnel_Supplier::process_service_request()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_service_request"));

	STATUS rep = STATUS_SUCCESS;
	APF_ServiceRequest request;
	if ((rep = request.read(getPeer())) !=  STATUS_SUCCESS) {
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed reading service request message\n")));
		return rep;
	}


	ACE_DEBUG ((MY_DEBUG 
		ACE_TEXT ("Service request: %s\n"),
		request.serviceName.c_str()));

	// Handle auth SERVICE request
	if (request.serviceName.compare(APF_SERVICE_AUTH) == 0)
	{
		if (tunnelState() != AMT_Tunnel_Handler::SESSION_OPENED)
		{
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT("AMT_Tunnel_Supplier: received APF auth service request in invalid state (%d)\n"),
				tunnelState()));
			return STATUS_AUTH_FAILURE;
		}
		tunnelState(AMT_Tunnel_Handler::AUTH_PENDING);
		APF_ServiceAccept accept;
		accept.serviceName = request.serviceName;
		accept.serviceNameLength = request.serviceNameLength;
		rep = consumer()->send_APF_message(accept);
		CHECK_STATUS_REP(rep, ACE_TEXT("failed to send reply for APF_SERVICE_AUTH request"));
		return STATUS_SUCCESS;
	}

	// Handle port forwarding service request
	if (request.serviceName.compare(APF_SERVICE_PFWD) == 0)
	{
		// We allow port forwarding without authentication for now
		// Here we should check whether an authentication is required
		const bool* amtNeedAuthentication = getAmtNeedAuthentication();
		if (amtNeedAuthentication == NULL)
		{
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT("Failed to get if AMT authentication is required.")));
			return STATUS_FAILURE;
		}
		if ((*amtNeedAuthentication) &&
			(tunnelState() != AMT_Tunnel_Handler::SESSION_AUTHENTICATED))
		{
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT("AMT_Tunnel_Supplier: received APF pfwd service request in invalid state (%d)\n"),
				tunnelState()));
			tunnel_handler_->_disconnect_data.setParams(APF_DISCONNECT_ILLEGAL_USER_NAME, true, PRIO_LOW);
			return STATUS_INVALID_TUNNEL_STATE;
		}
		tunnelState(AMT_Tunnel_Handler::PFWD_OPEN);

		APF_ServiceAccept accept;
		accept.serviceName = request.serviceName;
		accept.serviceNameLength = request.serviceNameLength;
		rep = consumer()->send_APF_message(accept);
		CHECK_STATUS_REP(rep, ACE_TEXT("failed to send reply for APF_SERVICE_PFWD request"));
		return STATUS_SUCCESS;
	}

	ACE_DEBUG((MY_DEBUG
		ACE_TEXT("AMT_Tunnel_Supplier: received APF service request with illegal service name (%s)\n"),
		request.serviceName.c_str()));
	setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
	return STATUS_ILLEGAL_APF_MESSAGE;
}

// handle user authentication message
STATUS AMT_Tunnel_Supplier::process_userauth_request()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_userauth_request"));

	STATUS rep = STATUS_SUCCESS;

	// Check tunnel state
	if (tunnelState() != AMT_Tunnel_Handler::AUTH_PENDING)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("received user authentication request in invalid state (%d)\n"),
			tunnelState()));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	APF_UserauthRequest request;
	if ((rep = request.read(getPeer())) != STATUS_SUCCESS){
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("failed to read user authentication request\n")));
		return rep;
	}

	// Authentication parameters seem to be fine.
	// Rest of auth depends on dll and parameters, in case of auth failure, a disconnect is issued anyway.
	// ==> Can cancel the timeout on tunnel opening .
	ACE_DEBUG((MY_DEBUG	ACE_TEXT("Got user authentication request, canceling timeout.\n")));
	tunnel_handler_->reactor()->cancel_timer(tunnel_handler_);

	ACE_DEBUG ((MY_DEBUG 
		ACE_TEXT ("User authentication request\n\tUsername:\t%s\n"),
		request.username.c_str()));

	// Check method name
	if (request.methodName.compare(APF_USERAUTH_PASSWORD) == 0)
	{
		//verify the user authorized:

	// Not currently supported in Linux
	#ifdef ACE_WIN32 
		char error[MAX_DLL_ERR_LEN];
   
		const bool* amtNeedAuthentication = getAmtNeedAuthentication();
		const ACE_TString* amtDllName = getAmtDllName();
		const ACE_TString* amtDllParameters = getAmtDllParameters();

		if (amtNeedAuthentication == NULL || amtDllName == NULL || amtDllParameters == NULL)
		{
			ACE_DEBUG((MY_INFO ACE_TEXT("Failed to get AMT authentication details.\n"),
				identifier(),
				error));
			return STATUS_AUTH_FAILURE;
		}

		Authentication_Param amt_auth_params(*amtNeedAuthentication,
											 *amtDllName,
											 *amtDllParameters);
	
		if ((amt_auth_params._authenticate)&&
			(_amt_auth_func != NULL) &&
			(_amt_auth_func (request.username.c_str(),
							request.password.c_str(),
							amt_auth_params._dllParams.c_str(), 
							&(error[0]), MAX_DLL_ERR_LEN ) == 0))
		{
			ACE_DEBUG((MY_INFO ACE_TEXT("[%s] Failed to authenticate Intel remote client: %s\n"), 
				identifier(),
				error));

#if defined (ACE_WIN32) && !defined (ACE_LACKS_WIN32_SERVICES) && defined (_SERVICE)
			ACE_CString errStr(error);
			if (errStr.find("Parse connection string error") != ACE_CString::npos)
			{
				m_pLogger.LogErrorEvent(MPS_ERROR, MPS_AUTH_BAD_USAGE);
			} 
			else if (errStr.find("Error opening file.") != ACE_CString::npos)
			{
				m_pLogger.LogErrorEvent(MPS_ERROR, MPS_AUTH_BAD_FILE);
			}
			else if (errStr.find("SOAP failure") != ACE_CString::npos)
			{
				m_pLogger.LogErrorEvent(MPS_ERROR, MPS_AUTH_SOAP_FAILURE);
			}
#endif  /* ACE_WIN32 && !(ACE_LACKS_WIN32_SERVICES) && (_SERVICE)*/
			
			//prepare reply message:
			APF_UserauthFailure auth_failure;
			auth_failure.methodNameListLength = (ACE_UINT32)ACE_OS::strlen(APF_USERAUTH_FAILURE_STR);
			auth_failure.methodNameList = APF_USERAUTH_FAILURE_STR;
	
			//send the msg:
			rep =  consumer()->send_APF_message(auth_failure,PRIO_HIGH);
			CHECK_STATUS_REP(rep, ACE_TEXT("failed to send reply for APF_USERAUTH_FAILURE request\n"));
			return STATUS_AUTH_FAILURE;
		}
	#endif /* ACE_WIN32 */

		ACE_DEBUG((MY_DEBUG ACE_TEXT("authentication SUCCESS\n")));
		
	}
	else if(request.methodName.compare(APF_USERAUTH_NONE) == 0)
	{
		const bool* amtNeedAuthentication = getAmtNeedAuthentication();
		if (amtNeedAuthentication == NULL)
		{
			ACE_DEBUG((MY_INFO ACE_TEXT("Failed to get if AMT authentication is required.")));
			return STATUS_AUTH_FAILURE;
		}
		if(*amtNeedAuthentication)
		{
			ACE_DEBUG((MY_INFO ACE_TEXT("[%s] Failed to authenticate Intel remote client, no authentication provided\n"), identifier()));
			//prepare reply message:
			APF_UserauthFailure auth_failure;
			auth_failure.methodNameListLength = (ACE_UINT32)ACE_OS::strlen(APF_USERAUTH_FAILURE_STR);
			auth_failure.methodNameList = APF_USERAUTH_FAILURE_STR;
	
			//send the msg:
			rep =  consumer()->send_APF_message(auth_failure,PRIO_HIGH);
			CHECK_STATUS_REP(rep, ACE_TEXT("failed to send reply for APF_USERAUTH_FAILURE request\n"));
			return STATUS_AUTH_FAILURE;
		}
		else 
		{
			// Password is not needed - let him in.
			ACE_DEBUG ((MY_DEBUG 
				ACE_TEXT ("\tNo password required\n")));
		}
	}
	else
	{
		ACE_DEBUG((MY_DEBUG
		ACE_TEXT("AMT_Tunnel_Supplier: invalid method name (%s) in user authentication request\n"),
		request.methodName.c_str())); 
		return STATUS_ILLEGAL_APF_MESSAGE;
	}
	tunnelState(AMT_Tunnel_Handler::SESSION_AUTHENTICATED);

	// For now we always accept the user
	rep =  consumer()->send_APF_message((ACE_UINT8)APF_USERAUTH_SUCCESS);
	CHECK_STATUS_REP(rep, ACE_TEXT("failed to send reply for APF_USERAUTH_SUCCESS request\n"));

	return STATUS_SUCCESS;
}

// read a global message from the socket and process it.
STATUS AMT_Tunnel_Supplier::process_global_message()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_global_message\n"));

	STATUS					rep = STATUS_SUCCESS;
	APF_GlobalRequestHeader header;
	if ((rep = header.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read global message\n")));
		return rep;
	}

	// No need for PFWD in UDP
	if (header.requestString.compare(APF_UDP_SENDTO) == 0)
	{
		return process_udp_send_to();
	}

	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received global message in invalid state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	if (header.requestString.compare(APF_TCP_FORWARD_REQUEST) == 0)
	{
		return process_tcp_forward();
	}
	if (header.requestString.compare(APF_TCP_FORWARD_CANCEL) == 0)
	{
		return process_tcp_forward_cancel();
	}

	ACE_DEBUG((MY_DEBUG
		ACE_TEXT("AMT_Tunnel_Supplier: invalid global message string\n\t%s\n"),
		header.requestString.c_str()));
	setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
	return STATUS_ILLEGAL_APF_MESSAGE;
}

// read a tcp forward request from the socket and process it.
STATUS AMT_Tunnel_Supplier::process_tcp_forward()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_tcp_forward\n"));
	APF_TcpForwardRequest request;
	APF_TcpForwardReply reply;
	STATUS rep;
	if ((rep = request.read(getPeer())) != STATUS_SUCCESS){
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read tcp forward request\n")));
		return rep;
	}

	//removes '[' ']' if exists (legal in ipv6)
//	ACE_CString addr_to_bind = request.addressToBind;
	if (request.addressToBind.find("[") != ACE_CString::npos)
	{
		request.addressToBind = request.addressToBind.substr(request.addressToBind.find('[')+1, request.addressToBind.find(']')-1);
		request.addressToBindLength = request.addressToBind.length();
	}
	{
		ACE_WRITE_GUARD_RETURN(	ACE_RW_Thread_Mutex,
			lock,
			Tunnel_Manager::instance().guard(),
			STATUS_FAILURE);

		// Check if this address and port already exists in the tunnel manager.
		AMT_Tunnel_Handler* old_tunnel = NULL;
		if ((old_tunnel = Tunnel_Manager::instance().find_tunnel(request.addressToBind, request.port)) != NULL)
		{
			if (old_tunnel == tunnel_handler_)
			{
				// Tunnel already exists
				rep = consumer()->send_APF_message((ACE_UINT8)APF_REQUEST_FAILURE);
				CHECK_STATUS_REP(rep, ACE_TEXT("AMT_Tunnel_Supplier: failed to send tcp forward request failure\n"));
				return STATUS_SUCCESS;
			}
			//if such tunnel exist - close the old one.
			ACE_DEBUG((MY_DEBUG 
				ACE_TEXT("AMT_Tunnel_Supplier: received tcp forward request on existed address and port\n\t%s:%d\n"),
				request.addressToBind.c_str(), request.port));

			//remove from tunnel mangers map:
			tunnel_handler_->removeFromTunnelMgr();
			lock.release();
			tunnel_handler_->reactor()->remove_handler(old_tunnel, ACE_Event_Handler::READ_MASK);
		}
		if (request.port == 0)
		{
			ACE_DEBUG((MY_DEBUG 
				ACE_TEXT("AMT_Tunnel_Supplier: received tcp forward request with illegal port = 0\n")));
			rep = consumer()->send_APF_message((ACE_UINT8)APF_REQUEST_FAILURE);
			CHECK_STATUS_REP(rep, ACE_TEXT("AMT_Tunnel_Supplier: failed to send tcp forward request failure\n"));
			return STATUS_SUCCESS;
		}
		
		(tunnel_handler_)->set_identifier(request.addressToBind);
		
		ACE_DEBUG ((MY_INFO 
			ACE_TEXT ("Intel remote client %s started port forwarding to address %s:%d\n"),
			identifier(), request.addressToBind.c_str(),
			request.port));

		//save this connection details:
		(tunnel_handler_->_tcp_forward_connections).push_back(Port_Address(request.addressToBind, request.port));
		int tmp = 0;
		if ((tmp = Tunnel_Manager::instance().add_tunnel(tunnel_handler_,
			request.addressToBind, request.port)) != 0)
		{
			reply.status = APF_REQUEST_FAILURE;
		}
		else
		{
			// Send the reply back to AMT
			reply.status = APF_REQUEST_SUCCESS;
			reply.portBound = request.port;
		}
	}
	if (sendAlertToSubscribMCList(	mps__ConnectionStateTypeDefinition__CONNECTED,
		request.addressToBind,
		request.port) == STATUS_FAILURE)
	{
		ACE_ERROR_RETURN ((MY_WARNING
			ACE_TEXT("[%s] Failed to send SOAP notification upon new port forwarding\n"),identifier()),STATUS_FAILURE);
	}
	
	if ((rep = consumer()->send_APF_message(reply)) != STATUS_SUCCESS)
	{
		ACE_DEBUG ((MY_DEBUG 
			ACE_TEXT ("failed to send APF message:TcpForwardReplay\n")));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		return rep;
	}
	return STATUS_SUCCESS;
}

// read a tcp forward cancel request from the socket and process it.
STATUS AMT_Tunnel_Supplier::process_tcp_forward_cancel()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_tcp_forward_cancel"));
	STATUS rep = STATUS_SUCCESS;
	APF_TcpForwardCancelRequest request;
	if ((rep = request.read(getPeer()))!= STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read tcp forward cancel request\n")));
		return rep;
	}

	ACE_DEBUG ((MY_INFO 
		ACE_TEXT ("Intel remote client %s canceled port forwarding to address %s:%d\n"),
		identifier(),
		request.addressToBind.c_str(),
		request.port));

	{
		ACE_WRITE_GUARD_RETURN(	ACE_RW_Thread_Mutex,
			lock,
			Tunnel_Manager::instance().guard(),
			STATUS_FAILURE);

		// Check if this address and port already exists in the tunnel manager.
		if (!Tunnel_Manager::instance ().tunnel_exists(request.addressToBind, request.port))
		{
			ACE_DEBUG ((MY_DEBUG 
				ACE_TEXT ("Tunnel doesn't exists in map\n")));
			rep = consumer()->send_APF_message((ACE_UINT8)APF_REQUEST_FAILURE);
			CHECK_STATUS_REP(rep, ACE_TEXT("failed to send APF message:TcpForwardCancel\n"));
			return STATUS_SUCCESS;
		}

		Port_Address port_add(request.addressToBind, request.port);
		tunnel_handler_->removeTcpForwardConnection(port_add);
	}
	// Send alert to MC
	if (sendAlertToSubscribMCList(	mps__ConnectionStateTypeDefinition__DISCONNECTED,
									request.addressToBind,
									request.port) == -1)
	{
		ACE_DEBUG ((MY_WARNING 
			ACE_TEXT("[%s] Failed to send SOAP notification on port forwarding cancellation\n"),identifier()));
	}

	rep = consumer()->send_APF_message((ACE_UINT8)APF_REQUEST_SUCCESS);
	CHECK_STATUS_REP(rep, ACE_TEXT ("failed to send APF message:TcpForwardCancelReplay\n"));
	return STATUS_SUCCESS;
}

// read udp send to request from the socket and process it.
STATUS AMT_Tunnel_Supplier::process_udp_send_to()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_udp_send_to"));
	STATUS rep;

	//read UDP request:
	APF_UdpSendTo request;

	if ((rep = request.read(getPeer())) != STATUS_SUCCESS) 
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read UDP send to message\n")));
		return rep;
	}
	
	// Check if can send a message to that server using the filter hash 
	// (specified in the configuration) 
	const bool* authServersListNeeded = getAuthServersListNeeded();
	if (authServersListNeeded == NULL)
	{
		ACE_DEBUG((MY_INFO
			ACE_TEXT("Failed to get if servers list is required\n")));
		return STATUS_FAILURE;
	}
	if ((*authServersListNeeded) && !checkInFilter(request.hostStr, request.port))
	{
		// Not found MC in filter hash - don't continue to send UDP message.
		ACE_DEBUG((MY_INFO
			ACE_TEXT("[%s] Did not send UDP to %s:%d - not in filter list\n"),
			identifier(),
			request.hostStr.c_str(),
			request.port));
		// Filtering is still a success.
		return STATUS_SUCCESS;
	}

	//send UDP data
	ACE_SOCK_Dgram cli_dgram;
	ACE_INET_Addr mc_addr;
	mc_addr.set(request.port, request.hostStr.c_str());

	if (cli_dgram.open(ACE_Addr::sap_any, mc_addr.get_type ()) == -1)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("protocol %d, SOCK_Dgram open FAILURE\n"),
			mc_addr.get_type()));
		// We are not closing the tunnel if the udp send failed	
		return STATUS_RECOVER_ERROR;
	}
	else if (cli_dgram.send (request.data->base(), request.dataLength, mc_addr) == -1)
	{
		ACE_DEBUG((MY_WARNING
			ACE_TEXT("[%s] Intel remote client failed to send a UDP message to %s:%d\n"),
			identifier(),request.hostStr.c_str(),request.port));
		// We are not closing the tunnel if the udp send failed
		cli_dgram.close();
		return STATUS_RECOVER_ERROR; 
	}
	cli_dgram.close();
	ACE_DEBUG((MY_INFO 
		ACE_TEXT("[%s] Intel remote client send a UDP message to %s:%d\n"),
		identifier(),
		request.hostStr.c_str(),
		request.port));
	
	return STATUS_SUCCESS;
}

bool AMT_Tunnel_Supplier::checkInFilter(const ACE_CString& host, const ACE_UINT32& port) 
{
	// this check is a sanity check so no error will occur later. 
	if (port > USHRT_MAX) 
		return false;
	char bufPort[PORT_STR_LENGTH + 1];
	ACE_OS::sprintf(bufPort, "%d", port);
	ACE_TString addressStr = host + DELIMITER_HOST_PORT + bufPort;

	AuthServerHash &filterHash = *getChangableAutherizedServersHash();
	if (&filterHash == NULL)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("Failed to get changeable authorized server hash.\n")));
		return false;
	}

	ACE_READ_GUARD_RETURN(ACE_RW_Thread_Mutex,
		lock,
		filterHash.guard(),
		false);

	return (filterHash.find(addressStr) != filterHash.end());
}

// Handle channel open reply
STATUS AMT_Tunnel_Supplier::process_channel_open_reply(ACE_UINT8 status)
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_channel_open_reply"));
	
	// Check tunnel state
	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT ("AMT_Tunnel_Supplier: channel open direct request with illegal state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}
	STATUS rep;
	APF_ChannelOpenReply reply;
	if ((rep = reply.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read channel open reply message\n")));
		return rep;
	}
	

	ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, lock, channel_manager().guard(),STATUS_FAILURE);

	// Get the channel from map
	AMT_Channel *ch = NULL;
	ch = channel_manager().find_channel(reply.recipientChannel, 
		Channel_Manager::MPS_ENDPOINT);
	if (ch == NULL || ch->getState() != AMT_Channel::PENDING_OPEN)
	{
		// Channel doesn't exists - error
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier::channel_open_reply Channel doesn't exists\n")));
		
		return STATUS_INVALID_CHANNEL;
	}

	// If confirmation
	if (status == APF_CHANNEL_OPEN_CONFIRMATION)
	{
		ACE_DEBUG((MY_INFO
			ACE_TEXT("Channel between Intel remote client %s and management console %s opened. Channel ID [%d:%d]\n"),
			identifier(),
			((SocksConsumer*)(ch->getTcpConsumer()))->identifier(),
			reply.recipientChannel,reply.senderChannel));

		ch->setAMTid(reply.senderChannel);
		if ((channel_manager().add_amt_id(reply.recipientChannel, 
										 reply.senderChannel)) != 0)
		{
			ACE_DEBUG((MY_DEBUG
				ACE_TEXT("AMT_Tunnel_Supplier::channel_open_reply invalid senderChannel\n")));
			setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
			return STATUS_ILLEGAL_APF_MESSAGE;
		}
		ch->setState(AMT_Channel::OPEN);

		// Send reply to consumers
		// We don't check the return code of the following, if a failure in the
		// tcp consumer/channel consumer occurred we expect the channel to be closed
		((SocksConsumer*)(ch->getTcpConsumer()))->openRep(Tcp_Consumer::SUCCESS);
		ch->getChannelConsumer()->activateChannel(reply.senderChannel,reply.initialWindowSize);
	}
	else if (status == APF_CHANNEL_OPEN_FAILURE)
	{
		ACE_DEBUG((MY_WARNING
			ACE_TEXT("[%s] Intel remote client rejected connection request\n"),identifier()));

		// Delete channel from map
		channel_manager().remove_channel(reply.recipientChannel, 
			Channel_Manager::MPS_ENDPOINT);

		ch->getChannelConsumer()->channelOpenFailure();
		((SocksConsumer*)ch->getTcpConsumer())->openRep(Tcp_Consumer::FAILURE);
		ch->getTcpConsumer()->connectionClose();
		
		delete ch;
	}
	else 
	{
		// impossible. sanity check
		ACE_ERROR((MY_DEBUG 
			ACE_TEXT(" channel open reply failure. unknown status = %d\n"),
			status));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		return STATUS_ILLEGAL_APF_MESSAGE;
	}

	return STATUS_SUCCESS;
}

//-------------------------------------------------
// Process a channel open request from AMT.
//-------------------------------------------------
STATUS AMT_Tunnel_Supplier::process_channel_open_direct()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_channel_open_direct"));
	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT ("AMT_Tunnel_Supplier channel open direct request with illegal state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	STATUS rep = STATUS_FAILURE;

	//read message:
	APF_ChannelOpenDirectRequest request;
	if ((rep = request.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read channel open direct request\n")));
		return rep;
	}


	// Verify APF message
	if (request.channelTypeString.compare(APF_CHANNEL_OPEN_DIRECT_FORWARD) != 0)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received channel open direct request with invalid type (%s)\n"),
			request.channelTypeString.c_str()));
		return STATUS_ILLEGAL_APF_MESSAGE;
	}

	ACE_DEBUG((MY_DEBUG  
		ACE_TEXT ("Received Channel Open Direct Request (%s:%d)\n"),
		request.targetHostString.c_str(),
		request.targetPort));

	const bool* authServersListNeeded = getAuthServersListNeeded();
	if (authServersListNeeded == NULL)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("Failed to get if servers list is required.\n")));
		return STATUS_FAILURE;
	}

	if ((*authServersListNeeded) && !checkInFilter(request.targetHostString, request.targetPort))
	{
		// Not found MC in filter hash - don't continue to send TCP message.
		ACE_DEBUG((MY_INFO 
			ACE_TEXT("[%s] Did not send TCP to %s:%d - not in filter list\n"),
			identifier(),
			request.targetHostString.c_str(),
			request.targetPort));

		//send failure reply:
		consumer()->channelOpenDirectRep(false,request.senderChannel, NULL, NULL);

		// Filtering is still a success.
		return STATUS_SUCCESS;
	}
	ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, lock, channel_manager().guard(),STATUS_FAILURE);

	// Create a new pending channel
	AMT_Channel *ch = channel_manager().create_channel(request.senderChannel);
	if (ch == NULL)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT("AMT_Tunnel_Supplier denied open direct request, invalid sender channel or no more channels available!!!\n")));

		consumer()->channelOpenDirectRep(false,request.senderChannel, NULL, NULL);
		
		return STATUS_SUCCESS;		
	}
	
	// Update the channel details
	ch->setWindowSize(*getMaximumWindowSize());
	ch->setState(AMT_Channel::PENDING_OPEN);

	//create tcp:
	Tcp_Svc_Handler* tcp; 
	ACE_NEW_RETURN (tcp, Tcp_Svc_Handler(request.initialWindowSize, ch->getAMTid(), consumer()), STATUS_MALLOC_FAILURE);

	//set tcp handlers addresses:
	ACE_INET_Addr remote_addr(request.targetPort, request.targetHostString.c_str(),AF_INET);
	tcp->remote_addr(remote_addr);

	ACE_INET_Addr local_addr(request.targetPort, request.originatorIpAddress.c_str(),AF_INET);
	tcp->local_addr(local_addr);

	
	//connect TCP handler:
	Connection_Handler_Connector* connector = &Connection_Handler_Connector::instance();
	//NOTE:
	// the connection performed in sync way (blocking until connection will complete)
	if (connector->initiate_connection(tcp) == -1)
	{
		//send failure reply:
		consumer()->channelOpenDirectRep(false,ch->getAMTid(), NULL, NULL, ch->getMPSid());
		channel_manager().remove_channel(ch->getMPSid(), Channel_Manager::MPS_ENDPOINT);
		delete ch;
	}
	
	return STATUS_SUCCESS;
}

// Process channel windows adjust message
STATUS AMT_Tunnel_Supplier::process_channel_window_adjust()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_channel_window_adjust"));

	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT ("AMT_Tunnel_Supplier: received channel window adjust with illegal state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	STATUS rep;
	APF_ChannelWindowAdjust msg;
	AMT_Channel *ch = NULL;
	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read channel window adjust message\n")));
		return rep;
	}
	ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, lock, channel_manager().guard(),STATUS_FAILURE);

	// Get the channel from the map
	ch = channel_manager().find_channel(msg.recipientChannel, 
		Channel_Manager::MPS_ENDPOINT);
	if (ch == NULL)
	{
		// Channel doesn't exist - according do APF protocol  
		return STATUS_SUCCESS;
	}
	ACE_DEBUG ((MY_DEBUG 
		ACE_TEXT ("Channel window adjust: AMT(%d) => MC(%d), size = %d\n"),
		ch->getAMTid(),
		ch->getMPSid(),
		msg.bytesToAdd));

	if (ch->getChannelConsumer() == NULL)
	{
		return STATUS_FATAL_ERROR;
	}

	// Here we give CPU time to the channel consumer (to put messages into our consumer)
	if (ch->getChannelConsumer()->setWinSize(msg.bytesToAdd) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("failed to set window size in process_channel_window_adjust")));
	}

	return STATUS_SUCCESS;
}

// process channel data
STATUS AMT_Tunnel_Supplier::process_channel_data()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_channel_data"));
	ACE_DEBUG((MY_TRACE
			ACE_TEXT("[%s] AMT_Tunnel_Supplier::process_channel_data\n"), identifier())); 

	APF_ChannelData msg;
	STATUS			rep = STATUS_SUCCESS;

	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT ("AMT_Tunnel_Supplier: received channel data with illegal state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("[%s] AMT_Tunnel_Supplier: failed to read channel data message\n"), identifier())); 
		return rep;
	}
	ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex, lock, channel_manager().guard(),STATUS_FAILURE);
	// Get the channel from map
	AMT_Channel *ch = NULL;
	ch = channel_manager().find_channel(msg.recipientChannel, 
		Channel_Manager::MPS_ENDPOINT);
	if (ch == NULL)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("received data from AMT but failed to find the match channel\n")));
		// Channel doesn't exists - error
		return STATUS_FAILURE;
	}
	else if ((ch->getState() != AMT_Channel::OPEN) && (ch->getState() != AMT_Channel::MC_CLOSED))
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("received data from AMT but channel not in OPEN state\n")));
		return STATUS_FAILURE;
	}

	ACE_DEBUG ((MY_DEBUG 
		ACE_TEXT ("[%s] Channel data: AMT(%d) => MC(%d), size = %d\n"),
		identifier(),
		ch->getAMTid(),
		ch->getMPSid(),
		msg.dataLen));

	rep = ch->getTcpConsumer()->sendData(msg.data);
	if (rep != STATUS_SUCCESS)
	{
		ACE_DEBUG ((MY_DEBUG
			ACE_TEXT("failed to send data to MC\n")));
	}

	if (ch->getWindowSize() <  msg.dataLen)
	{
		ACE_DEBUG ((MY_DEBUG ACE_TEXT("received more data to send then window size\n")));
		ch->setWindowSize(0);
	}
	else
	{
		ch->setWindowSize(ch->getWindowSize() - msg.dataLen);
	}
		
	if (ch->getWindowSize() <= *getMinimumWindowSize())
	{
		APF_ChannelWindowAdjust win_adj;
		win_adj.recipientChannel = ch->getAMTid();
		win_adj.bytesToAdd = *getMaximumWindowSize() - ch->getWindowSize();

		rep = this->consumer()->send_APF_message(win_adj);
		CHECK_STATUS_REP(rep, ACE_TEXT("failed to send APF:Window Adjust message\n"));

		ch->setWindowSize(*getMaximumWindowSize());
		ACE_DEBUG ((MY_DEBUG 
			ACE_TEXT ("Channel window adjust: MC(%d) => AMT(%d), size = %d\n"),
			ch->getMPSid(),
			ch->getAMTid(),
			win_adj.bytesToAdd));
	}
	
	return rep;
}

// Process channel close message
STATUS AMT_Tunnel_Supplier::process_channel_close()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_channel_close"));
	ACE_DEBUG((MY_DEBUG ACE_TEXT("AMT_Tunnel_Supplier::process_channel_close\n")));
	APF_ChannelClose	msg;
	STATUS				rep = STATUS_SUCCESS;

	if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	{
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT ("AMT_Tunnel_Supplier: received channel close with illegal state\n")));
		return STATUS_INVALID_TUNNEL_STATE;
	}

	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read channel close message\n")));
		return rep;
	}
	

	
	// Get lock on channel manager
	ACE_GUARD_RETURN(	ACE_Recursive_Thread_Mutex, 
						lock, 
						channel_manager().guard(),
						STATUS_FAILURE);

	// Get the channel from map
	AMT_Channel *ch = NULL;
	ch = channel_manager().find_channel(msg.recipientChannel, 
		Channel_Manager::MPS_ENDPOINT);
	if (ch == NULL)
	{
		// Channel doesn't exists - error
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received channel close on invalid channel\n")));
		return STATUS_FAILURE;
	}

	
	Tcp_Consumer* tcp_consumer = ch->getTcpConsumer(); 
	ch->setTcpConsumer(NULL); 
	// The MC send close request first, we can delete this channel
	if (ch->getState() == AMT_Channel::MC_CLOSED)
	{
		ACE_DEBUG((MY_INFO 
			ACE_TEXT("Channel between Intel remote client %s and management console %s closed. Channel ID [%d:%d]\n"),
			identifier(), tcp_consumer->identifier(),ch->getAMTid(),ch->getMPSid()));
		
		if (ch->getChannelConsumer() != NULL)
		{
			ch->getChannelConsumer()->connectionUnbind();
		}
		// Remove this channel from both maps
		channel_manager().remove_channel(msg.recipientChannel, Channel_Manager::MPS_ENDPOINT);
		delete ch;
	}

	// Check if AMT requested to close this channel
	else if (ch->getState() == AMT_Channel::OPEN) {
		ch->setState(AMT_Channel::TUNNEL_CLOSED);
		// Notify this channel consumer that we are closed
		ACE_DEBUG((MY_DEBUG 
			ACE_TEXT("AMT_Tunnel_Supplier: Channel close MC(%d)\n"),
			msg.recipientChannel));
		ch->getChannelConsumer()->channelClose();
	}

	else {
		// received close channel on illegal state
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: received close channel on illegal state\n")));
		return STATUS_FAILURE;

	}
	lock.release();

	// Notify consumer to close
	if (tcp_consumer!= NULL)
	{
		tcp_consumer->connectionClose();
	}

	return STATUS_SUCCESS;
}

// Close this supplier, called from the reactor after read mask close
STATUS AMT_Tunnel_Supplier::handle_close()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::close"));
	return STATUS_CONNECTION_CLOSED;	
}

STATUS AMT_Tunnel_Supplier::process_KeepAliveReq()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_KeepAliveReq"));

	APF_KeepAliveReq	msg;
	STATUS				rep = STATUS_SUCCESS;

	//if (tunnelState() != AMT_Tunnel_Handler::PFWD_OPEN)
	//{
	//	ACE_DEBUG((MY_ERROR 
	//		ACE_TEXT ("AMT_Tunnel_Supplier: received channel close with illegal state\n")));
	//	return STATUS_INVALID_TUNNEL_STATE;
	//}

	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read keepAlive message\n")));
		return rep;
	}
	APF_KeepAliveReply reply;

	reply._cookie = msg._cookie;
	if ((rep = consumer()->send_APF_message(reply)) != STATUS_SUCCESS)
	{
		ACE_DEBUG ((MY_DEBUG 
			ACE_TEXT ("failed to send APF message:KeepAliveReply\n")));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		return rep;
	}
	return STATUS_SUCCESS;
}

STATUS AMT_Tunnel_Supplier::process_KeepAliveOptionReq()
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::process_KeepAliveOptionReq"));

	APF_KeepAliveOptionReq	msg;
	STATUS					rep = STATUS_SUCCESS;


	if ((rep = msg.read(getPeer())) != STATUS_SUCCESS)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("AMT_Tunnel_Supplier: failed to read keepAlive message\n")));
		return rep;
	}
	APF_KeepAliveOptionReply reply;

	reply._keepalive_interval	= 0;
	reply._read_timeout			= 0;

	if ((rep = consumer()->send_APF_message(reply)) != STATUS_SUCCESS)
	{
		ACE_DEBUG ((MY_DEBUG 
			ACE_TEXT ("failed to send APF message:KeepAliveReply\n")));
		setDisconnectReason(APF_DISCONNECT_PROTOCOL_ERROR);
		return rep;
	}
	return STATUS_SUCCESS;
}
//-----------------------------------------
// Parses 16 bytes of UUID sent from AMT
// device in hello message to UUID string format.
//
// Arguments:
//		@binUUID        - UUID to convert
//		@textUUID       - [out] UUID string in correct format
// 
//-----------------------------------------
void StringFromUUID(const unsigned char binUUID[16], 
					unsigned char textUUID[TEXT_UUID_SIZE])
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::StringFromUUID"));

	unsigned char *hexString = (unsigned char *)binUUID;
	int res = sprintf_s((char *)textUUID, TEXT_UUID_SIZE, "%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X",
		*(unsigned long *) (hexString),
		*(unsigned short *)(hexString+4),
		*(unsigned short *)(hexString+4+2),
		*(unsigned char *) (hexString+4+2+2),
		*(unsigned char *) (hexString+4+2+2+1),
		*(unsigned char *) (hexString+4+2+2+2),
		*(unsigned char *) (hexString+4+2+2+3),
		*(unsigned char *) (hexString+4+2+2+4),
		*(unsigned char *) (hexString+4+2+2+5),
		*(unsigned char *) (hexString+4+2+2+6),
		*(unsigned char *) (hexString+4+2+2+7)
		);
	if (res == -1) {
		ACE_ERROR((MY_ERROR
			ACE_TEXT(" Failed to parse UUID sent from AMT to string\n")));
	}
}
STATUS AMT_Tunnel_Supplier::sendAlertToSubscribMCList(	mps__ConnectionStateTypeDefinition state, 
													  ACE_CString &fqdn, 
													  unsigned short port)
{
	ACE_TRACE(ACE_TEXT("AMT_Tunnel_Supplier::sendAlertToSubscribMCList"));
	ACE_CString str_state;
	if (state == mps__ConnectionStateTypeDefinition__CONNECTED)
		str_state = "CONNECT";
	else
		str_state = "DISCONNECT";

	unsigned char textUUID[TEXT_UUID_SIZE];
	StringFromUUID(UUID_, textUUID);
	ACE_CString fqdn_upper_case=toUpper(fqdn);
	ACE_DEBUG((MY_DEBUG 
		ACE_TEXT("Sending notification to management console\n\tDevice = %s:%d - %s\n"),
		fqdn_upper_case.c_str(), port, str_state.c_str()));

	
	string temp_fqdn(fqdn_upper_case.c_str());
	string temp_uuid((char*)textUUID);
	string temp_hostname((*getMCSocksListenIP()).c_str()); 
	
	const unsigned short* mcHttpListenPort = getMCHttpListenPort();
	const unsigned short* mcSocksListenPort = getMCSocksListenPort();
	if (mcHttpListenPort == NULL || mcSocksListenPort == NULL)
	{
		ACE_DEBUG((MY_DEBUG
			ACE_TEXT("Failed to get MC authentication details.\n")));
		return STATUS_FAILURE;
	}
	DevicePresence::instance().sendEvent(
		temp_fqdn,
		port,
		temp_uuid, 
		temp_hostname,
		*mcHttpListenPort,
		*mcSocksListenPort,
		state);

	return STATUS_SUCCESS;
}




const char* AMT_Tunnel_Supplier::identifier()
{ 
	return tunnel_handler_->identifier();
}