[public/netxms.git] / include / nxcpapi.h

/*
** NetXMS - Network Management System
** NXCP API
** Copyright (C) 2003-2010 Victor Kirhenshtein
**
** This program is free software; you can redistribute it and/or modify
** it under the terms of the GNU Lesser 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 General Public License for more details.
**
** You should have received a copy of the GNU Lesser General Public License
** along with this program; if not, write to the Free Software
** Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
**
** File: nxcpapi.h
**
**/

#ifndef _nxcpapi_h_
#define _nxcpapi_h_

#include <nms_util.h>
#include <nms_threads.h>
#include <uuid.h>

#ifdef _WIN32
#include <wincrypt.h>
#endif

/**
 * Temporary buffer structure for RecvNXCPMessage() function
 */
typedef struct
{
   UINT32 bufferSize;
   UINT32 bufferPos;
   char buffer[NXCP_TEMP_BUF_SIZE];
} NXCP_BUFFER;


#ifdef __cplusplus

struct MessageField;

/**
 * Parsed NXCP message
 */
class LIBNETXMS_EXPORTABLE NXCPMessage
{
private:
   UINT16 m_code;
   UINT16 m_flags;
   UINT32 m_id;
   MessageField *m_fields; // Message fields
   int m_version;          // Protocol version
   BYTE *m_data;           // binary data
   size_t m_dataSize;      // binary data size

   void *set(UINT32 fieldId, BYTE type, const void *value, bool isSigned = false, size_t size = 0);
   void *get(UINT32 fieldId, BYTE requiredType, BYTE *fieldType = NULL);
   NXCP_MESSAGE_FIELD *find(UINT32 fieldId);

public:
   NXCPMessage(int version = NXCP_VERSION);
   NXCPMessage(NXCPMessage *msg);
   NXCPMessage(NXCP_MESSAGE *rawMag, int version = NXCP_VERSION);
   ~NXCPMessage();

   NXCP_MESSAGE *createMessage();

   UINT16 getCode() { return m_code; }
   void setCode(UINT16 code) { m_code = code; }

   UINT32 getId() { return m_id; }
   void setId(UINT32 id) { m_id = id; }

   bool isEndOfFile() { return (m_flags & MF_END_OF_FILE) ? true : false; }
   bool isEndOfSequence() { return (m_flags & MF_END_OF_SEQUENCE) ? true : false; }
   bool isReverseOrder() { return (m_flags & MF_REVERSE_ORDER) ? true : false; }
   bool isBinary() { return (m_flags & MF_BINARY) ? true : false; }
   bool isControl() { return (m_flags & MF_CONTROL) ? true : false; }
   bool isCompressed() { return (m_flags & MF_COMPRESSED) ? true : false; }

   BYTE *getBinaryData() { return m_data; }
   size_t getBinaryDataSize() { return m_dataSize; }

   bool isFieldExist(UINT32 fieldId) { return find(fieldId) != NULL; }
   int getFieldType(UINT32 fieldId);

   void setField(UINT32 fieldId, INT16 value) { set(fieldId, NXCP_DT_INT16, &value, true); }
   void setField(UINT32 fieldId, UINT16 value) { set(fieldId, NXCP_DT_INT16, &value, false); }
   void setField(UINT32 fieldId, INT32 value) { set(fieldId, NXCP_DT_INT32, &value, true); }
   void setField(UINT32 fieldId, UINT32 value) { set(fieldId, NXCP_DT_INT32, &value, false); }
   void setField(UINT32 fieldId, INT64 value) { set(fieldId, NXCP_DT_INT64, &value, true); }
   void setField(UINT32 fieldId, UINT64 value) { set(fieldId, NXCP_DT_INT64, &value, false); }
   void setField(UINT32 fieldId, double value) { set(fieldId, NXCP_DT_FLOAT, &value); }
   void setField(UINT32 fieldId, bool value) { INT16 v = value ? 1 : 0; set(fieldId, NXCP_DT_INT16, &v, true); }
   void setField(UINT32 fieldId, const TCHAR *value) { if (value != NULL) set(fieldId, NXCP_DT_STRING, value); }
   void setField(UINT32 fieldId, const TCHAR *value, size_t maxLen) { if (value != NULL) set(fieldId, NXCP_DT_STRING, value, false, maxLen); }
   void setField(UINT32 fieldId, const BYTE *value, size_t size) { set(fieldId, NXCP_DT_BINARY, value, false, size); }
   void setField(UINT32 fieldId, const InetAddress& value) { set(fieldId, NXCP_DT_INETADDR, &value); }
   void setField(UINT32 fieldId, const uuid& value) { set(fieldId, NXCP_DT_BINARY, value.getValue(), false, UUID_LENGTH); }
#ifdef UNICODE
   void setFieldFromMBString(UINT32 fieldId, const char *value);
#else
   void setFieldFromMBString(UINT32 fieldId, const char *value) { set(fieldId, NXCP_DT_STRING, value); }
#endif
   void setFieldFromTime(UINT32 fieldId, time_t value) { UINT64 t = (UINT64)value; set(fieldId, NXCP_DT_INT64, &t); }
   void setFieldFromInt32Array(UINT32 fieldId, size_t numElements, const UINT32 *elements);
   void setFieldFromInt32Array(UINT32 fieldId, IntegerArray<UINT32> *data);
   bool setFieldFromFile(UINT32 fieldId, const TCHAR *pszFileName);

   INT16 getFieldAsInt16(UINT32 fieldId);
   UINT16 getFieldAsUInt16(UINT32 fieldId);
   INT32 getFieldAsInt32(UINT32 fieldId);
   UINT32 getFieldAsUInt32(UINT32 fieldId);
   INT64 getFieldAsInt64(UINT32 fieldId);
   UINT64 getFieldAsUInt64(UINT32 fieldId);
   double getFieldAsDouble(UINT32 fieldId);
   bool getFieldAsBoolean(UINT32 fieldId);
   time_t getFieldAsTime(UINT32 fieldId);
   UINT32 getFieldAsInt32Array(UINT32 fieldId, UINT32 numElements, UINT32 *buffer);
   UINT32 getFieldAsInt32Array(UINT32 fieldId, IntegerArray<UINT32> *data);
   BYTE *getBinaryFieldPtr(UINT32 fieldId, size_t *size);
   TCHAR *getFieldAsString(UINT32 fieldId, TCHAR *buffer = NULL, size_t bufferSize = 0);
	char *getFieldAsMBString(UINT32 fieldId, char *buffer = NULL, size_t bufferSize = 0);
	char *getFieldAsUtf8String(UINT32 fieldId, char *buffer = NULL, size_t bufferSize = 0);
   UINT32 getFieldAsBinary(UINT32 fieldId, BYTE *buffer, size_t bufferSize);
   InetAddress getFieldAsInetAddress(UINT32 fieldId);
   uuid getFieldAsGUID(UINT32 fieldId);

   void deleteAllFields();

   void disableEncryption() { m_flags |= MF_DONT_ENCRYPT; }
   void setEndOfSequence() { m_flags |= MF_END_OF_SEQUENCE; }
   void setReverseOrderFlag() { m_flags |= MF_REVERSE_ORDER; }

   static String dump(NXCP_MESSAGE *msg, int version);
};

/**
 * Message waiting queue element structure
 */
typedef struct
{
   void *msg;         // Pointer to message, either to NXCPMessage object or raw message
   UINT64 sequence;   // Sequence number
   UINT32 id;         // Message ID
   UINT32 ttl;        // Message time-to-live in milliseconds
   UINT16 code;       // Message code
   UINT16 isBinary;   // 1 for binary (raw) messages
} WAIT_QUEUE_ELEMENT;

/**
 * Max number of waiting threads in message queue
 */
#define MAX_MSGQUEUE_WAITERS     32

/**
 * Message waiting queue class
 */
class LIBNETXMS_EXPORTABLE MsgWaitQueue
{
private:
#if defined(_WIN32)
   CRITICAL_SECTION m_mutex;
   HANDLE m_wakeupEvents[MAX_MSGQUEUE_WAITERS];
   BYTE m_waiters[MAX_MSGQUEUE_WAITERS];
#elif defined(_USE_GNU_PTH)
   pth_mutex_t m_mutex;
   pth_cond_t m_wakeupCondition;
#else
   pthread_mutex_t m_mutex;
   pthread_cond_t m_wakeupCondition;
#endif
   UINT32 m_holdTime;
   int m_size;
   int m_allocated;
   WAIT_QUEUE_ELEMENT *m_elements;
   UINT64 m_sequence;

   void *waitForMessageInternal(UINT16 isBinary, UINT16 code, UINT32 id, UINT32 timeout);

   void lock()
   {
#ifdef _WIN32
      EnterCriticalSection(&m_mutex);
#elif defined(_USE_GNU_PTH)
      pth_mutex_acquire(&m_mutex, FALSE, NULL);
#else
      pthread_mutex_lock(&m_mutex);
#endif
   }

   void unlock()
   {
#ifdef _WIN32
      LeaveCriticalSection(&m_mutex);
#elif defined(_USE_GNU_PTH)
      pth_mutex_release(&m_mutex);
#else
      pthread_mutex_unlock(&m_mutex);
#endif
   }

   void housekeeperRun();

   static MUTEX m_housekeeperLock;
   static HashMap<UINT64, MsgWaitQueue> *m_activeQueues;
   static CONDITION m_shutdownCondition;
   static THREAD m_housekeeperThread;
   static EnumerationCallbackResult houseKeeperCallback(const void *key, const void *object, void *arg);
   static THREAD_RESULT THREAD_CALL housekeeperThread(void *);
   static EnumerationCallbackResult diagInfoCallback(const void *key, const void *object, void *arg);

public:
   MsgWaitQueue();
   ~MsgWaitQueue();

   void put(NXCPMessage *pMsg);
   void put(NXCP_MESSAGE *pMsg);
   NXCPMessage *waitForMessage(WORD wCode, UINT32 dwId, UINT32 dwTimeOut)
   {
      return (NXCPMessage *)waitForMessageInternal(0, wCode, dwId, dwTimeOut);
   }
   NXCP_MESSAGE *waitForRawMessage(WORD wCode, UINT32 dwId, UINT32 dwTimeOut)
   {
      return (NXCP_MESSAGE *)waitForMessageInternal(1, wCode, dwId, dwTimeOut);
   }

   void clear();
   void setHoldTime(UINT32 holdTime) { m_holdTime = holdTime; }

   static void shutdown();
   static String getDiagInfo();
};

/**
 * NXCP encryption context
 */
class LIBNETXMS_EXPORTABLE NXCPEncryptionContext : public RefCountObject
{
private:
	int m_cipher;
	BYTE *m_sessionKey;
	int m_keyLength;
	BYTE m_iv[EVP_MAX_IV_LENGTH];
#ifdef _WITH_ENCRYPTION
   EVP_CIPHER_CTX m_encryptor;
   EVP_CIPHER_CTX m_decryptor;
   MUTEX m_encryptorLock;
#endif

	NXCPEncryptionContext();
   bool initCipher(int cipher);

public:
	static NXCPEncryptionContext *create(NXCPMessage *msg, RSA *privateKey);
	static NXCPEncryptionContext *create(UINT32 ciphers);

	virtual ~NXCPEncryptionContext();

   NXCP_ENCRYPTED_MESSAGE *encryptMessage(NXCP_MESSAGE *msg);
   bool decryptMessage(NXCP_ENCRYPTED_MESSAGE *msg, BYTE *decryptionBuffer);

	int getCipher() { return m_cipher; }
	BYTE *getSessionKey() { return m_sessionKey; }
	int getKeyLength() { return m_keyLength; }
	BYTE *getIV() { return m_iv; }
};

/**
 * Message receiver result codes
 */
enum MessageReceiverResult
{
   MSGRECV_SUCCESS = 0,
   MSGRECV_CLOSED = 1,
   MSGRECV_TIMEOUT = 2,
   MSGRECV_COMM_FAILURE = 3,
   MSGRECV_DECRYPTION_FAILURE = 4
};

/**
 * Message receiver - abstract base class
 */
class LIBNETXMS_EXPORTABLE AbstractMessageReceiver
{
private:
   BYTE *m_buffer;
   BYTE *m_decryptionBuffer;
   NXCPEncryptionContext *m_encryptionContext;
   size_t m_initialSize;
   size_t m_size;
   size_t m_maxSize;
   size_t m_dataSize;
   size_t m_bytesToSkip;

   NXCPMessage *getMessageFromBuffer();

protected:
   virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout) = 0;

public:
   AbstractMessageReceiver(size_t initialSize, size_t maxSize);
   virtual ~AbstractMessageReceiver();

   void setEncryptionContext(NXCPEncryptionContext *ctx) { m_encryptionContext = ctx; }

   NXCPMessage *readMessage(UINT32 timeout, MessageReceiverResult *result);
   NXCP_MESSAGE *getRawMessageBuffer() { return (NXCP_MESSAGE *)m_buffer; }

   static const TCHAR *resultToText(MessageReceiverResult result);
};

/**
 * Message receiver - socket implementation
 */
class LIBNETXMS_EXPORTABLE SocketMessageReceiver : public AbstractMessageReceiver
{
private:
   SOCKET m_socket;

protected:
   virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout);

public:
   SocketMessageReceiver(SOCKET socket, size_t initialSize, size_t maxSize);
   virtual ~SocketMessageReceiver();
};

/**
 * Message receiver - UNIX socket/named pipe implementation
 */
class LIBNETXMS_EXPORTABLE PipeMessageReceiver : public AbstractMessageReceiver
{
private:
   HPIPE m_pipe;
#ifdef _WIN32
	HANDLE m_readEvent;
#endif

protected:
   virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout);

public:
   PipeMessageReceiver(HPIPE pipe, size_t initialSize, size_t maxSize);
   virtual ~PipeMessageReceiver();
};

/**
 * NXCP compression methods
 */
enum NXCPCompressionMethod
{
   NXCP_COMPRESSION_NONE = 0,
   NXCP_COMPRESSION_LZ4 = 1
};

/**
 * Abstract stream compressor
 */
class LIBNETXMS_EXPORTABLE StreamCompressor
{
public:
   virtual ~StreamCompressor();

   virtual size_t compress(const BYTE *in, size_t inSize, BYTE *out, size_t maxOutSize) = 0;
   virtual size_t decompress(const BYTE *in, size_t inSize, const BYTE **out) = 0;
   virtual size_t compressBufferSize(size_t dataSize) = 0;

   static StreamCompressor *create(NXCPCompressionMethod method, bool compress, size_t maxBlockSize);
};

/**
 * Dummy stream compressor
 */
class LIBNETXMS_EXPORTABLE DummyStreamCompressor : public StreamCompressor
{
public:
   virtual ~DummyStreamCompressor();

   virtual size_t compress(const BYTE *in, size_t inSize, BYTE *out, size_t maxOutSize);
   virtual size_t decompress(const BYTE *in, size_t inSize, const BYTE **out);
   virtual size_t compressBufferSize(size_t dataSize);
};

struct __LZ4_stream_t;
struct __LZ4_streamDecode_t;

/**
 * LZ4 stream compressor
 */
class LIBNETXMS_EXPORTABLE LZ4StreamCompressor : public StreamCompressor
{
private:
   union
   {
      __LZ4_stream_t *encoder;
      __LZ4_streamDecode_t *decoder;
   } m_stream;
   char *m_buffer;
   size_t m_maxBlockSize;
   size_t m_bufferSize;
   size_t m_bufferPos;
   bool m_compress;

public:
   LZ4StreamCompressor(bool compress, size_t maxBlockSize);
   virtual ~LZ4StreamCompressor();

   virtual size_t compress(const BYTE *in, size_t inSize, BYTE *out, size_t maxOutSize);
   virtual size_t decompress(const BYTE *in, size_t inSize, const BYTE **out);
   virtual size_t compressBufferSize(size_t dataSize);
};

#if 0
/**
 * NXCP message consumer interface
 */
class LIBNETXMS_EXPORTABLE MessageConsumer
{
public:
   virtual SOCKET getSocket() = 0;
   virtual void processMessage(NXCPMessage *msg) = 0;
};

/**
 * Socket receiver - manages receiving NXCP messages from multiple sockets
 */
class LIBNETXMS_EXPORTABLE SocketReceiver
{
private:
   THREAD m_thread;
   HashMap<SOCKET, MessageConsumer> *m_consumers;

   static int m_maxSocketsPerThread;
   static ObjectArray<SocketReceiver> *m_receivers;

public:
   static void start();
   static void shutdown();

   static void addConsumer(MessageConsumer *mc);
   static void removeConsumer(MessageConsumer *mc);

   static String getDiagInfo();
};
#endif

#else    /* __cplusplus */

typedef void NXCPMessage;
typedef void NXCPEncryptionContext;

#endif


//
// Functions
//

#ifdef __cplusplus

int LIBNETXMS_EXPORTABLE RecvNXCPMessage(SOCKET hSocket, NXCP_MESSAGE *pMsg,
                                         NXCP_BUFFER *pBuffer, UINT32 dwMaxMsgSize,
                                         NXCPEncryptionContext **ppCtx,
                                         BYTE *pDecryptionBuffer, UINT32 dwTimeout);
int LIBNETXMS_EXPORTABLE RecvNXCPMessageEx(SOCKET hSocket, NXCP_MESSAGE **msgBuffer,
                                           NXCP_BUFFER *nxcpBuffer, UINT32 *bufferSize,
                                           NXCPEncryptionContext **ppCtx,
                                           BYTE **decryptionBuffer, UINT32 dwTimeout,
														 UINT32 maxMsgSize);
NXCP_MESSAGE LIBNETXMS_EXPORTABLE *CreateRawNXCPMessage(WORD wCode, UINT32 dwId, WORD flags,
                                                        UINT32 dwDataSize, void *pData,
                                                        NXCP_MESSAGE *pBuffer);
TCHAR LIBNETXMS_EXPORTABLE *NXCPMessageCodeName(WORD wCode, TCHAR *buffer);
BOOL LIBNETXMS_EXPORTABLE SendFileOverNXCP(SOCKET hSocket, UINT32 dwId, const TCHAR *pszFile,
                                           NXCPEncryptionContext *pCtx, long offset,
														 void (* progressCallback)(INT64, void *), void *cbArg,
														 MUTEX mutex, NXCPCompressionMethod compressionMethod = NXCP_COMPRESSION_NONE);
BOOL LIBNETXMS_EXPORTABLE NXCPGetPeerProtocolVersion(SOCKET hSocket, int *pnVersion, MUTEX mutex);

bool LIBNETXMS_EXPORTABLE InitCryptoLib(UINT32 dwEnabledCiphers, void (*debugCallback)(int, const TCHAR *, va_list args));
UINT32 LIBNETXMS_EXPORTABLE NXCPGetSupportedCiphers();
String LIBNETXMS_EXPORTABLE NXCPGetSupportedCiphersAsText();
NXCP_ENCRYPTED_MESSAGE LIBNETXMS_EXPORTABLE *NXCPEncryptMessage(NXCPEncryptionContext *pCtx, NXCP_MESSAGE *pMsg);
bool LIBNETXMS_EXPORTABLE NXCPDecryptMessage(NXCPEncryptionContext *pCtx,
                                             NXCP_ENCRYPTED_MESSAGE *pMsg,
                                             BYTE *pDecryptionBuffer);
UINT32 LIBNETXMS_EXPORTABLE SetupEncryptionContext(NXCPMessage *pMsg,
                                                  NXCPEncryptionContext **ppCtx,
                                                  NXCPMessage **ppResponse,
                                                  RSA *pPrivateKey, int nNXCPVersion);
void LIBNETXMS_EXPORTABLE PrepareKeyRequestMsg(NXCPMessage *pMsg, RSA *pServerKey, bool useX509Format);
RSA LIBNETXMS_EXPORTABLE *LoadRSAKeys(const TCHAR *pszKeyFile);

#ifdef _WIN32
BOOL LIBNETXMS_EXPORTABLE SignMessageWithCAPI(BYTE *pMsg, UINT32 dwMsgLen, const CERT_CONTEXT *pCert,
												          BYTE *pBuffer, size_t bufferSize, UINT32 *pdwSigLen);
#endif

#endif

#endif   /* _nxcpapi_h_ */
Commit	Line	Data
9fa031cd	1	/*
0d9c1b96	2	** NetXMS - Network Management System
9f2ad16c	3	** NXCP API
68f384ea	4	** Copyright (C) 2003-2010 Victor Kirhenshtein
0d9c1b96 VK	5	**
0d9c1b96 VK	6	** This program is free software; you can redistribute it and/or modify
65d2c384 VK	7	** it under the terms of the GNU Lesser General Public License as published by
65d2c384 VK	8	** the Free Software Foundation; either version 3 of the License, or
0d9c1b96 VK	9	** (at your option) any later version.
	10	**
	11	** This program is distributed in the hope that it will be useful,
	12	** but WITHOUT ANY WARRANTY; without even the implied warranty of
	13	** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	14	** GNU General Public License for more details.
	15	**
68f384ea	16	** You should have received a copy of the GNU Lesser General Public License
0d9c1b96 VK	17	** along with this program; if not, write to the Free Software
	18	** Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
	19	**
9f2ad16c	20	** File: nxcpapi.h
0d9c1b96 VK	21	**
	22	**/
	23
9f2ad16c VK	24	#ifndef _nxcpapi_h_
9f2ad16c VK	25	#define _nxcpapi_h_
0d9c1b96	26
9f2ad16c	27	#include <nms_util.h>
bf6fb6c3	28	#include <nms_threads.h>
de4af576	29	#include <uuid.h>
8c290af0	30
bb8136f7 VK	31	#ifdef _WIN32
	32	#include <wincrypt.h>
	33	#endif
	34
4af351c7	35	/**
b368969c	36	* Temporary buffer structure for RecvNXCPMessage() function
4af351c7	37	*/
0d9c1b96 VK	38	typedef struct
0d9c1b96 VK	39	{
b368969c VK	40	UINT32 bufferSize;
b368969c VK	41	UINT32 bufferPos;
43a6f3ca	42	char buffer[NXCP_TEMP_BUF_SIZE];
b368969c	43	} NXCP_BUFFER;
0d9c1b96 VK	44
0d9c1b96 VK	45
9f2ad16c VK	46	#ifdef __cplusplus
9f2ad16c VK	47
5c44534b VK	48	struct MessageField;
5c44534b VK	49
4af351c7 VK	50	/**
	51	* Parsed NXCP message
	52	*/
b368969c	53	class LIBNETXMS_EXPORTABLE NXCPMessage
0d9c1b96 VK	54	{
0d9c1b96 VK	55	private:
b368969c VK	56	UINT16 m_code;
b368969c VK	57	UINT16 m_flags;
5c44534b	58	UINT32 m_id;
6be0a20b VK	59	MessageField *m_fields; // Message fields
	60	int m_version; // Protocol version
	61	BYTE *m_data; // binary data
	62	size_t m_dataSize; // binary data size
0d9c1b96	63
712b2760	64	void set(UINT32 fieldId, BYTE type, const void value, bool isSigned = false, size_t size = 0);
8f238fd7	65	void get(UINT32 fieldId, BYTE requiredType, BYTE fieldType = NULL);
b368969c	66	NXCP_MESSAGE_FIELD *find(UINT32 fieldId);
0d9c1b96 VK	67
0d9c1b96 VK	68	public:
b368969c VK	69	NXCPMessage(int version = NXCP_VERSION);
	70	NXCPMessage(NXCPMessage *msg);
	71	NXCPMessage(NXCP_MESSAGE *rawMag, int version = NXCP_VERSION);
	72	~NXCPMessage();
0d9c1b96	73
b368969c	74	NXCP_MESSAGE *createMessage();
0d9c1b96	75
b368969c VK	76	UINT16 getCode() { return m_code; }
b368969c VK	77	void setCode(UINT16 code) { m_code = code; }
0d9c1b96	78
b368969c VK	79	UINT32 getId() { return m_id; }
b368969c VK	80	void setId(UINT32 id) { m_id = id; }
0d9c1b96	81
6be0a20b	82	bool isEndOfFile() { return (m_flags & MF_END_OF_FILE) ? true : false; }
5c44534b VK	83	bool isEndOfSequence() { return (m_flags & MF_END_OF_SEQUENCE) ? true : false; }
5c44534b VK	84	bool isReverseOrder() { return (m_flags & MF_REVERSE_ORDER) ? true : false; }
6be0a20b	85	bool isBinary() { return (m_flags & MF_BINARY) ? true : false; }
81222795 VK	86	bool isControl() { return (m_flags & MF_CONTROL) ? true : false; }
81222795 VK	87	bool isCompressed() { return (m_flags & MF_COMPRESSED) ? true : false; }
6be0a20b VK	88
	89	BYTE *getBinaryData() { return m_data; }
	90	size_t getBinaryDataSize() { return m_dataSize; }
4af351c7	91
4101571e VK	92	bool isFieldExist(UINT32 fieldId) { return find(fieldId) != NULL; }
	93	int getFieldType(UINT32 fieldId);
	94
712b2760 VK	95	void setField(UINT32 fieldId, INT16 value) { set(fieldId, NXCP_DT_INT16, &value, true); }
	96	void setField(UINT32 fieldId, UINT16 value) { set(fieldId, NXCP_DT_INT16, &value, false); }
	97	void setField(UINT32 fieldId, INT32 value) { set(fieldId, NXCP_DT_INT32, &value, true); }
	98	void setField(UINT32 fieldId, UINT32 value) { set(fieldId, NXCP_DT_INT32, &value, false); }
	99	void setField(UINT32 fieldId, INT64 value) { set(fieldId, NXCP_DT_INT64, &value, true); }
	100	void setField(UINT32 fieldId, UINT64 value) { set(fieldId, NXCP_DT_INT64, &value, false); }
b368969c	101	void setField(UINT32 fieldId, double value) { set(fieldId, NXCP_DT_FLOAT, &value); }
19a2cd39	102	void setField(UINT32 fieldId, bool value) { INT16 v = value ? 1 : 0; set(fieldId, NXCP_DT_INT16, &v, true); }
b368969c	103	void setField(UINT32 fieldId, const TCHAR *value) { if (value != NULL) set(fieldId, NXCP_DT_STRING, value); }
712b2760	104	void setField(UINT32 fieldId, const TCHAR *value, size_t maxLen) { if (value != NULL) set(fieldId, NXCP_DT_STRING, value, false, maxLen); }
5f573844	105	void setField(UINT32 fieldId, const BYTE *value, size_t size) { set(fieldId, NXCP_DT_BINARY, value, false, size); }
de4af576 VK	106	void setField(UINT32 fieldId, const InetAddress& value) { set(fieldId, NXCP_DT_INETADDR, &value); }
de4af576 VK	107	void setField(UINT32 fieldId, const uuid& value) { set(fieldId, NXCP_DT_BINARY, value.getValue(), false, UUID_LENGTH); }
35f836fe	108	#ifdef UNICODE
b368969c	109	void setFieldFromMBString(UINT32 fieldId, const char *value);
35f836fe	110	#else
b368969c	111	void setFieldFromMBString(UINT32 fieldId, const char *value) { set(fieldId, NXCP_DT_STRING, value); }
35f836fe	112	#endif
b368969c	113	void setFieldFromTime(UINT32 fieldId, time_t value) { UINT64 t = (UINT64)value; set(fieldId, NXCP_DT_INT64, &t); }
14149881	114	void setFieldFromInt32Array(UINT32 fieldId, size_t numElements, const UINT32 *elements);
b368969c	115	void setFieldFromInt32Array(UINT32 fieldId, IntegerArray<UINT32> *data);
8f238fd7	116	bool setFieldFromFile(UINT32 fieldId, const TCHAR *pszFileName);
967893bb	117
4101571e	118	INT16 getFieldAsInt16(UINT32 fieldId);
b368969c	119	UINT16 getFieldAsUInt16(UINT32 fieldId);
4101571e	120	INT32 getFieldAsInt32(UINT32 fieldId);
b368969c	121	UINT32 getFieldAsUInt32(UINT32 fieldId);
4101571e	122	INT64 getFieldAsInt64(UINT32 fieldId);
b368969c	123	UINT64 getFieldAsUInt64(UINT32 fieldId);
4101571e	124	double getFieldAsDouble(UINT32 fieldId);
d6bbfa4e	125	bool getFieldAsBoolean(UINT32 fieldId);
8f238fd7	126	time_t getFieldAsTime(UINT32 fieldId);
9a68ca24 VK	127	UINT32 getFieldAsInt32Array(UINT32 fieldId, UINT32 numElements, UINT32 *buffer);
9a68ca24 VK	128	UINT32 getFieldAsInt32Array(UINT32 fieldId, IntegerArray<UINT32> *data);
ac14e3e6	129	BYTE getBinaryFieldPtr(UINT32 fieldId, size_t size);
b368969c VK	130	TCHAR getFieldAsString(UINT32 fieldId, TCHAR buffer = NULL, size_t bufferSize = 0);
	131	char getFieldAsMBString(UINT32 fieldId, char buffer = NULL, size_t bufferSize = 0);
	132	char getFieldAsUtf8String(UINT32 fieldId, char buffer = NULL, size_t bufferSize = 0);
	133	UINT32 getFieldAsBinary(UINT32 fieldId, BYTE *buffer, size_t bufferSize);
712b2760	134	InetAddress getFieldAsInetAddress(UINT32 fieldId);
de4af576	135	uuid getFieldAsGUID(UINT32 fieldId);
4101571e	136
b368969c	137	void deleteAllFields();
e44ac467	138
5c44534b VK	139	void disableEncryption() { m_flags \|= MF_DONT_ENCRYPT; }
	140	void setEndOfSequence() { m_flags \|= MF_END_OF_SEQUENCE; }
	141	void setReverseOrderFlag() { m_flags \|= MF_REVERSE_ORDER; }
934f53da	142
b368969c	143	static String dump(NXCP_MESSAGE *msg, int version);
0d9c1b96 VK	144	};
0d9c1b96 VK	145
4af351c7 VK	146	/**
	147	* Message waiting queue element structure
	148	*/
deaa4491 VK	149	typedef struct
deaa4491 VK	150	{
b368969c	151	void *msg; // Pointer to message, either to NXCPMessage object or raw message
14149881	152	UINT64 sequence; // Sequence number
f128c07c VK	153	UINT32 id; // Message ID
	154	UINT32 ttl; // Message time-to-live in milliseconds
	155	UINT16 code; // Message code
	156	UINT16 isBinary; // 1 for binary (raw) messages
deaa4491 VK	157	} WAIT_QUEUE_ELEMENT;
deaa4491 VK	158
4af351c7	159	/**
f128c07c VK	160	* Max number of waiting threads in message queue
	161	*/
	162	#define MAX_MSGQUEUE_WAITERS 32
	163
	164	/**
4af351c7 VK	165	* Message waiting queue class
4af351c7 VK	166	*/
9f2ad16c	167	class LIBNETXMS_EXPORTABLE MsgWaitQueue
deaa4491	168	{
deaa4491	169	private:
ef8a3c32	170	#if defined(_WIN32)
f128c07c VK	171	CRITICAL_SECTION m_mutex;
	172	HANDLE m_wakeupEvents[MAX_MSGQUEUE_WAITERS];
	173	BYTE m_waiters[MAX_MSGQUEUE_WAITERS];
ef8a3c32 VK	174	#elif defined(_USE_GNU_PTH)
	175	pth_mutex_t m_mutex;
	176	pth_cond_t m_wakeupCondition;
f128c07c VK	177	#else
	178	pthread_mutex_t m_mutex;
	179	pthread_cond_t m_wakeupCondition;
	180	#endif
f128c07c VK	181	UINT32 m_holdTime;
	182	int m_size;
	183	int m_allocated;
	184	WAIT_QUEUE_ELEMENT *m_elements;
14149881	185	UINT64 m_sequence;
deaa4491	186
f128c07c VK	187	void *waitForMessageInternal(UINT16 isBinary, UINT16 code, UINT32 id, UINT32 timeout);
	188
	189	void lock()
	190	{
	191	#ifdef _WIN32
	192	EnterCriticalSection(&m_mutex);
ef8a3c32 VK	193	#elif defined(_USE_GNU_PTH)
ef8a3c32 VK	194	pth_mutex_acquire(&m_mutex, FALSE, NULL);
f128c07c VK	195	#else
	196	pthread_mutex_lock(&m_mutex);
	197	#endif
	198	}
	199
	200	void unlock()
	201	{
	202	#ifdef _WIN32
	203	LeaveCriticalSection(&m_mutex);
ef8a3c32 VK	204	#elif defined(_USE_GNU_PTH)
ef8a3c32 VK	205	pth_mutex_release(&m_mutex);
f128c07c VK	206	#else
	207	pthread_mutex_unlock(&m_mutex);
	208	#endif
	209	}
cc022855	210
d87ddcc2 VK	211	void housekeeperRun();
	212
	213	static MUTEX m_housekeeperLock;
	214	static HashMap<UINT64, MsgWaitQueue> *m_activeQueues;
	215	static CONDITION m_shutdownCondition;
	216	static THREAD m_housekeeperThread;
	217	static EnumerationCallbackResult houseKeeperCallback(const void key, const void object, void *arg);
	218	static THREAD_RESULT THREAD_CALL housekeeperThread(void *);
	219	static EnumerationCallbackResult diagInfoCallback(const void key, const void object, void *arg);
deaa4491 VK	220
	221	public:
	222	MsgWaitQueue();
	223	~MsgWaitQueue();
	224
b368969c VK	225	void put(NXCPMessage *pMsg);
	226	void put(NXCP_MESSAGE *pMsg);
	227	NXCPMessage *waitForMessage(WORD wCode, UINT32 dwId, UINT32 dwTimeOut)
deaa4491	228	{
b368969c	229	return (NXCPMessage *)waitForMessageInternal(0, wCode, dwId, dwTimeOut);
deaa4491	230	}
b368969c	231	NXCP_MESSAGE *waitForRawMessage(WORD wCode, UINT32 dwId, UINT32 dwTimeOut)
deaa4491	232	{
b368969c	233	return (NXCP_MESSAGE *)waitForMessageInternal(1, wCode, dwId, dwTimeOut);
deaa4491	234	}
cc022855	235
c17f6cbc	236	void clear();
f128c07c	237	void setHoldTime(UINT32 holdTime) { m_holdTime = holdTime; }
d87ddcc2 VK	238
	239	static void shutdown();
	240	static String getDiagInfo();
deaa4491 VK	241	};
deaa4491 VK	242
b549a0f8 VK	243	/**
	244	* NXCP encryption context
	245	*/
98abc9f1 VK	246	class LIBNETXMS_EXPORTABLE NXCPEncryptionContext : public RefCountObject
	247	{
	248	private:
	249	int m_cipher;
	250	BYTE *m_sessionKey;
	251	int m_keyLength;
	252	BYTE m_iv[EVP_MAX_IV_LENGTH];
e4e091f0	253	#ifdef _WITH_ENCRYPTION
e6336a90 VK	254	EVP_CIPHER_CTX m_encryptor;
e6336a90 VK	255	EVP_CIPHER_CTX m_decryptor;
db05c2af	256	MUTEX m_encryptorLock;
e4e091f0	257	#endif
98abc9f1 VK	258
98abc9f1 VK	259	NXCPEncryptionContext();
e6336a90	260	bool initCipher(int cipher);
98abc9f1 VK	261
98abc9f1 VK	262	public:
b368969c	263	static NXCPEncryptionContext create(NXCPMessage msg, RSA *privateKey);
967893bb	264	static NXCPEncryptionContext *create(UINT32 ciphers);
98abc9f1 VK	265
	266	virtual ~NXCPEncryptionContext();
	267
b368969c	268	NXCP_ENCRYPTED_MESSAGE encryptMessage(NXCP_MESSAGE msg);
6be0a20b	269	bool decryptMessage(NXCP_ENCRYPTED_MESSAGE msg, BYTE decryptionBuffer);
e6336a90	270
98abc9f1 VK	271	int getCipher() { return m_cipher; }
	272	BYTE *getSessionKey() { return m_sessionKey; }
	273	int getKeyLength() { return m_keyLength; }
	274	BYTE *getIV() { return m_iv; }
	275	};
	276
6be0a20b VK	277	/**
	278	* Message receiver result codes
	279	*/
	280	enum MessageReceiverResult
	281	{
	282	MSGRECV_SUCCESS = 0,
	283	MSGRECV_CLOSED = 1,
	284	MSGRECV_TIMEOUT = 2,
	285	MSGRECV_COMM_FAILURE = 3,
	286	MSGRECV_DECRYPTION_FAILURE = 4
	287	};
	288
	289	/**
	290	* Message receiver - abstract base class
	291	*/
	292	class LIBNETXMS_EXPORTABLE AbstractMessageReceiver
	293	{
	294	private:
	295	BYTE *m_buffer;
	296	BYTE *m_decryptionBuffer;
	297	NXCPEncryptionContext *m_encryptionContext;
	298	size_t m_initialSize;
	299	size_t m_size;
	300	size_t m_maxSize;
	301	size_t m_dataSize;
	302	size_t m_bytesToSkip;
	303
b368969c	304	NXCPMessage *getMessageFromBuffer();
6be0a20b VK	305
	306	protected:
	307	virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout) = 0;
	308
	309	public:
	310	AbstractMessageReceiver(size_t initialSize, size_t maxSize);
	311	virtual ~AbstractMessageReceiver();
	312
	313	void setEncryptionContext(NXCPEncryptionContext *ctx) { m_encryptionContext = ctx; }
	314
b368969c VK	315	NXCPMessage readMessage(UINT32 timeout, MessageReceiverResult result);
	316	NXCP_MESSAGE getRawMessageBuffer() { return (NXCP_MESSAGE )m_buffer; }
	317
	318	static const TCHAR *resultToText(MessageReceiverResult result);
6be0a20b VK	319	};
	320
	321	/**
	322	* Message receiver - socket implementation
	323	*/
	324	class LIBNETXMS_EXPORTABLE SocketMessageReceiver : public AbstractMessageReceiver
	325	{
	326	private:
	327	SOCKET m_socket;
	328
	329	protected:
	330	virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout);
	331
	332	public:
	333	SocketMessageReceiver(SOCKET socket, size_t initialSize, size_t maxSize);
	334	virtual ~SocketMessageReceiver();
	335	};
	336
	337	/**
	338	* Message receiver - UNIX socket/named pipe implementation
	339	*/
	340	class LIBNETXMS_EXPORTABLE PipeMessageReceiver : public AbstractMessageReceiver
	341	{
	342	private:
	343	HPIPE m_pipe;
	344	#ifdef _WIN32
	345	HANDLE m_readEvent;
	346	#endif
	347
	348	protected:
	349	virtual int readBytes(BYTE *buffer, size_t size, UINT32 timeout);
	350
	351	public:
	352	PipeMessageReceiver(HPIPE pipe, size_t initialSize, size_t maxSize);
	353	virtual ~PipeMessageReceiver();
	354	};
	355
503da871 VK	356	/**
	357	* NXCP compression methods
	358	*/
	359	enum NXCPCompressionMethod
	360	{
	361	NXCP_COMPRESSION_NONE = 0,
	362	NXCP_COMPRESSION_LZ4 = 1
	363	};
	364
	365	/**
	366	* Abstract stream compressor
	367	*/
	368	class LIBNETXMS_EXPORTABLE StreamCompressor
	369	{
	370	public:
9757ca46 VK	371	virtual ~StreamCompressor();
9757ca46 VK	372
503da871 VK	373	virtual size_t compress(const BYTE in, size_t inSize, BYTE out, size_t maxOutSize) = 0;
	374	virtual size_t decompress(const BYTE in, size_t inSize, const BYTE *out) = 0;
	375	virtual size_t compressBufferSize(size_t dataSize) = 0;
	376
	377	static StreamCompressor *create(NXCPCompressionMethod method, bool compress, size_t maxBlockSize);
	378	};
	379
	380	/**
	381	* Dummy stream compressor
	382	*/
	383	class LIBNETXMS_EXPORTABLE DummyStreamCompressor : public StreamCompressor
	384	{
	385	public:
9757ca46 VK	386	virtual ~DummyStreamCompressor();
9757ca46 VK	387
503da871 VK	388	virtual size_t compress(const BYTE in, size_t inSize, BYTE out, size_t maxOutSize);
	389	virtual size_t decompress(const BYTE in, size_t inSize, const BYTE *out);
	390	virtual size_t compressBufferSize(size_t dataSize);
	391	};
	392
	393	struct __LZ4_stream_t;
	394	struct __LZ4_streamDecode_t;
	395
	396	/**
	397	* LZ4 stream compressor
	398	*/
	399	class LIBNETXMS_EXPORTABLE LZ4StreamCompressor : public StreamCompressor
	400	{
	401	private:
	402	union
	403	{
	404	__LZ4_stream_t *encoder;
	405	__LZ4_streamDecode_t *decoder;
	406	} m_stream;
	407	char *m_buffer;
	408	size_t m_maxBlockSize;
	409	size_t m_bufferSize;
	410	size_t m_bufferPos;
	411	bool m_compress;
	412
	413	public:
	414	LZ4StreamCompressor(bool compress, size_t maxBlockSize);
	415	virtual ~LZ4StreamCompressor();
	416
	417	virtual size_t compress(const BYTE in, size_t inSize, BYTE out, size_t maxOutSize);
	418	virtual size_t decompress(const BYTE in, size_t inSize, const BYTE *out);
	419	virtual size_t compressBufferSize(size_t dataSize);
	420	};
	421
208d7427 VK	422	#if 0
	423	/**
	424	* NXCP message consumer interface
	425	*/
	426	class LIBNETXMS_EXPORTABLE MessageConsumer
	427	{
	428	public:
	429	virtual SOCKET getSocket() = 0;
	430	virtual void processMessage(NXCPMessage *msg) = 0;
	431	};
	432
	433	/**
	434	* Socket receiver - manages receiving NXCP messages from multiple sockets
	435	*/
	436	class LIBNETXMS_EXPORTABLE SocketReceiver
	437	{
	438	private:
	439	THREAD m_thread;
	440	HashMap<SOCKET, MessageConsumer> *m_consumers;
	441
	442	static int m_maxSocketsPerThread;
	443	static ObjectArray<SocketReceiver> *m_receivers;
	444
	445	public:
	446	static void start();
	447	static void shutdown();
	448
	449	static void addConsumer(MessageConsumer *mc);
	450	static void removeConsumer(MessageConsumer *mc);
	451
	452	static String getDiagInfo();
	453	};
	454	#endif
	455
9f2ad16c VK	456	#else /* __cplusplus */
9f2ad16c VK	457
b368969c	458	typedef void NXCPMessage;
98abc9f1	459	typedef void NXCPEncryptionContext;
9f2ad16c VK	460
	461	#endif
	462
deaa4491 VK	463
deaa4491 VK	464	//
0d9c1b96 VK	465	// Functions
	466	//
	467
	468	#ifdef __cplusplus
0d9c1b96	469
b368969c VK	470	int LIBNETXMS_EXPORTABLE RecvNXCPMessage(SOCKET hSocket, NXCP_MESSAGE *pMsg,
b368969c VK	471	NXCP_BUFFER *pBuffer, UINT32 dwMaxMsgSize,
98abc9f1	472	NXCPEncryptionContext **ppCtx,
967893bb	473	BYTE *pDecryptionBuffer, UINT32 dwTimeout);
b368969c VK	474	int LIBNETXMS_EXPORTABLE RecvNXCPMessageEx(SOCKET hSocket, NXCP_MESSAGE **msgBuffer,
b368969c VK	475	NXCP_BUFFER nxcpBuffer, UINT32 bufferSize,
cc022855	476	NXCPEncryptionContext **ppCtx,
967893bb VK	477	BYTE **decryptionBuffer, UINT32 dwTimeout,
967893bb VK	478	UINT32 maxMsgSize);
b368969c	479	NXCP_MESSAGE LIBNETXMS_EXPORTABLE *CreateRawNXCPMessage(WORD wCode, UINT32 dwId, WORD flags,
967893bb	480	UINT32 dwDataSize, void *pData,
b368969c VK	481	NXCP_MESSAGE *pBuffer);
b368969c VK	482	TCHAR LIBNETXMS_EXPORTABLE NXCPMessageCodeName(WORD wCode, TCHAR buffer);
967893bb	483	BOOL LIBNETXMS_EXPORTABLE SendFileOverNXCP(SOCKET hSocket, UINT32 dwId, const TCHAR *pszFile,
cc022855	484	NXCPEncryptionContext *pCtx, long offset,
7b8b337e	485	void (* progressCallback)(INT64, void ), void cbArg,
503da871	486	MUTEX mutex, NXCPCompressionMethod compressionMethod = NXCP_COMPRESSION_NONE);
7b8b337e	487	BOOL LIBNETXMS_EXPORTABLE NXCPGetPeerProtocolVersion(SOCKET hSocket, int *pnVersion, MUTEX mutex);
cc022855	488
bf6fb6c3	489	bool LIBNETXMS_EXPORTABLE InitCryptoLib(UINT32 dwEnabledCiphers, void (debugCallback)(int, const TCHAR , va_list args));
e5390fb5 VK	490	UINT32 LIBNETXMS_EXPORTABLE NXCPGetSupportedCiphers();
e5390fb5 VK	491	String LIBNETXMS_EXPORTABLE NXCPGetSupportedCiphersAsText();
b368969c	492	NXCP_ENCRYPTED_MESSAGE LIBNETXMS_EXPORTABLE NXCPEncryptMessage(NXCPEncryptionContext pCtx, NXCP_MESSAGE *pMsg);
bf6fb6c3	493	bool LIBNETXMS_EXPORTABLE NXCPDecryptMessage(NXCPEncryptionContext *pCtx,
6be0a20b	494	NXCP_ENCRYPTED_MESSAGE *pMsg,
e44ac467	495	BYTE *pDecryptionBuffer);
b368969c	496	UINT32 LIBNETXMS_EXPORTABLE SetupEncryptionContext(NXCPMessage *pMsg,
98abc9f1	497	NXCPEncryptionContext **ppCtx,
b368969c	498	NXCPMessage **ppResponse,
4bd3d6c5	499	RSA *pPrivateKey, int nNXCPVersion);
b368969c	500	void LIBNETXMS_EXPORTABLE PrepareKeyRequestMsg(NXCPMessage pMsg, RSA pServerKey, bool useX509Format);
3aa0061c	501	RSA LIBNETXMS_EXPORTABLE LoadRSAKeys(const TCHAR pszKeyFile);
15cf5acc	502
bb8136f7	503	#ifdef _WIN32
967893bb	504	BOOL LIBNETXMS_EXPORTABLE SignMessageWithCAPI(BYTE pMsg, UINT32 dwMsgLen, const CERT_CONTEXT pCert,
b368969c	505	BYTE pBuffer, size_t bufferSize, UINT32 pdwSigLen);
bb8136f7 VK	506	#endif
bb8136f7 VK	507
0d9c1b96 VK	508	#endif
0d9c1b96 VK	509
9f2ad16c	510	#endif /* _nxcpapi_h_ */