"Fossies" - the Fresh Open Source Software Archive 
Member "netxms-3.1.300/src/server/libnxsrv/agent.cpp" (7 Jan 2020, 71280 Bytes) of package /linux/misc/netxms-3.1.300.tar.gz:
As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) C and C++ source code syntax highlighting (style:
standard) with prefixed line numbers and
code folding option.
Alternatively you can here
view or
download the uninterpreted source code file.
For more information about "agent.cpp" see the
Fossies "Dox" file reference documentation and the last
Fossies "Diffs" side-by-side code changes report:
3.0.2357_vs_3.1.241.
1 /*
2 ** NetXMS - Network Management System
3 ** Server Library
4 ** Copyright (C) 2003-2019 Victor Kirhenshtein
5 **
6 ** This program is free software; you can redistribute it and/or modify
7 ** it under the terms of the GNU Lesser General Public License as published by
8 ** the Free Software Foundation; either version 3 of the License, or
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 **
16 ** You should have received a copy of the GNU Lesser General Public License
17 ** along with this program; if not, write to the Free Software
18 ** Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
19 **
20 ** File: agent.cpp
21 **
22 **/
23
24 #include "libnxsrv.h"
25 #include <stdarg.h>
26 #include <nxstat.h>
27
28 #ifndef _WIN32
29 #define _tell(f) lseek(f,0,SEEK_CUR)
30 #endif
31
32 #define DEBUG_TAG _T("agent.conn")
33
34 /**
35 * Constants
36 */
37 #define MAX_MSG_SIZE 268435456
38
39 /**
40 * Agent connection thread pool
41 */
42 LIBNXSRV_EXPORTABLE_VAR(ThreadPool *g_agentConnectionThreadPool) = NULL;
43
44 /**
45 * Unique connection ID
46 */
47 static VolatileCounter s_connectionId = 0;
48
49 /**
50 * Static data
51 */
52 #ifdef _WITH_ENCRYPTION
53 static int m_iDefaultEncryptionPolicy = ENCRYPTION_ALLOWED;
54 #else
55 static int m_iDefaultEncryptionPolicy = ENCRYPTION_DISABLED;
56 #endif
57
58 /**
59 * Set default encryption policy for agent communication
60 */
61 void LIBNXSRV_EXPORTABLE SetAgentDEP(int iPolicy)
62 {
63 #ifdef _WITH_ENCRYPTION
64 m_iDefaultEncryptionPolicy = iPolicy;
65 #endif
66 }
67
68 /**
69 * Receiver thread starter
70 */
71 THREAD_RESULT THREAD_CALL AgentConnection::receiverThreadStarter(void *pArg)
72 {
73 ThreadSetName("AgentReceiver");
74 ((AgentConnection *)pArg)->receiverThread();
75 ((AgentConnection *)pArg)->decInternalRefCount();
76 return THREAD_OK;
77 }
78
79 /**
80 * Constructor for AgentConnection
81 */
82 AgentConnection::AgentConnection(const InetAddress& addr, WORD port, int authMethod, const TCHAR *secret, bool allowCompression)
83 {
84 m_internalRefCount = 1;
85 m_userRefCount = 1;
86 m_debugId = InterlockedIncrement(&s_connectionId);
87 m_addr = addr;
88 m_wPort = port;
89 m_iAuthMethod = authMethod;
90 if (secret != NULL)
91 {
92 #ifdef UNICODE
93 WideCharToMultiByte(CP_ACP, WC_COMPOSITECHECK | WC_DEFAULTCHAR, secret, -1, m_szSecret, MAX_SECRET_LENGTH, NULL, NULL);
94 m_szSecret[MAX_SECRET_LENGTH - 1] = 0;
95 #else
96 strlcpy(m_szSecret, secret, MAX_SECRET_LENGTH);
97 #endif
98 }
99 else
100 {
101 m_szSecret[0] = 0;
102 }
103 m_allowCompression = allowCompression;
104 m_channel = NULL;
105 m_tLastCommandTime = 0;
106 m_pMsgWaitQueue = new MsgWaitQueue;
107 m_requestId = 0;
108 m_connectionTimeout = 5000; // 5 seconds
109 m_dwCommandTimeout = 5000; // Default timeout 5 seconds
110 m_isConnected = false;
111 m_mutexDataLock = MutexCreate();
112 m_mutexSocketWrite = MutexCreate();
113 m_hReceiverThread = INVALID_THREAD_HANDLE;
114 m_pCtx = NULL;
115 m_iEncryptionPolicy = m_iDefaultEncryptionPolicy;
116 m_useProxy = false;
117 m_iProxyAuth = AUTH_NONE;
118 m_wProxyPort = 4700;
119 m_dwRecvTimeout = 420000; // 7 minutes
120 m_nProtocolVersion = NXCP_VERSION;
121 m_hCurrFile = -1;
122 m_deleteFileOnDownloadFailure = true;
123 m_condFileDownload = ConditionCreate(TRUE);
124 m_fileDownloadSucceeded = false;
125 m_fileUploadInProgress = false;
126 m_sendToClientMessageCallback = NULL;
127 m_dwDownloadRequestId = 0;
128 m_downloadProgressCallback = NULL;
129 m_downloadProgressCallbackArg = NULL;
130 m_bulkDataProcessing = 0;
131 m_controlServer = false;
132 m_masterServer = false;
133 }
134
135 /**
136 * Destructor
137 */
138 AgentConnection::~AgentConnection()
139 {
140 debugPrintf(7, _T("AgentConnection destructor called (this=%p, thread=%p)"), this, (void *)m_hReceiverThread);
141
142 ThreadDetach(m_hReceiverThread);
143
144 delete m_pMsgWaitQueue;
145 if (m_pCtx != NULL)
146 m_pCtx->decRefCount();
147
148 if (m_hCurrFile != -1)
149 {
150 _close(m_hCurrFile);
151 onFileDownload(false);
152 }
153 else if (m_sendToClientMessageCallback != NULL)
154 {
155 onFileDownload(false);
156 }
157
158 if (m_channel != NULL)
159 m_channel->decRefCount();
160
161 MutexDestroy(m_mutexDataLock);
162 MutexDestroy(m_mutexSocketWrite);
163 ConditionDestroy(m_condFileDownload);
164 }
165
166 /**
167 * Write debug output
168 */
169 void AgentConnection::debugPrintf(int level, const TCHAR *format, ...)
170 {
171 va_list args;
172 va_start(args, format);
173 nxlog_debug_tag_object2(DEBUG_TAG, m_debugId, level, format, args);
174 va_end(args);
175 }
176
177 /**
178 * Receiver thread
179 */
180 void AgentConnection::receiverThread()
181 {
182 AbstractCommChannel *channel = m_channel;
183 UINT32 msgBufferSize = 1024;
184
185 // Initialize raw message receiving function
186 NXCP_BUFFER *msgBuffer = (NXCP_BUFFER *)MemAlloc(sizeof(NXCP_BUFFER));
187 NXCPInitBuffer(msgBuffer);
188
189 // Allocate space for raw message
190 NXCP_MESSAGE *rawMsg = (NXCP_MESSAGE *)MemAlloc(msgBufferSize);
191 #ifdef _WITH_ENCRYPTION
192 BYTE *decryptionBuffer = (BYTE *)MemAlloc(msgBufferSize);
193 #else
194 BYTE *decryptionBuffer = NULL;
195 #endif
196
197 while(true)
198 {
199 // Shrink buffer after receiving large message
200 if (msgBufferSize > 131072)
201 {
202 msgBufferSize = 131072;
203 rawMsg = MemRealloc(rawMsg, msgBufferSize);
204 if (decryptionBuffer != NULL)
205 decryptionBuffer = MemRealloc(decryptionBuffer, msgBufferSize);
206 }
207
208 // Receive raw message
209 int rc = RecvNXCPMessageEx(channel, &rawMsg, msgBuffer, &msgBufferSize,
210 &m_pCtx, (decryptionBuffer != NULL) ? &decryptionBuffer : NULL,
211 m_dwRecvTimeout, MAX_MSG_SIZE);
212 if (rc <= 0)
213 {
214 if ((rc != 0) && (WSAGetLastError() != WSAESHUTDOWN))
215 debugPrintf(6, _T("AgentConnection::ReceiverThread(): RecvNXCPMessage() failed: error=%d, socket_error=%d"), rc, WSAGetLastError());
216 else
217 debugPrintf(6, _T("AgentConnection::ReceiverThread(): communication channel shutdown"));
218 break;
219 }
220
221 if (IsShutdownInProgress())
222 {
223 debugPrintf(6, _T("AgentConnection::ReceiverThread(): process shutdown"));
224 break;
225 }
226
227 // Check if we get too large message
228 if (rc == 1)
229 {
230 TCHAR buffer[64];
231 debugPrintf(6, _T("Received too large message %s (%d bytes)"), NXCPMessageCodeName(ntohs(rawMsg->code), buffer), ntohl(rawMsg->size));
232 continue;
233 }
234
235 // Check if we are unable to decrypt message
236 if (rc == 2)
237 {
238 debugPrintf(6, _T("Unable to decrypt received message"));
239 continue;
240 }
241
242 // Check for timeout
243 if (rc == 3)
244 {
245 if (m_fileUploadInProgress)
246 continue; // Receive timeout may occur when uploading large files via slow links
247 debugPrintf(6, _T("Timed out waiting for message"));
248 break;
249 }
250
251 // Check that actual received packet size is equal to encoded in packet
252 if ((int)ntohl(rawMsg->size) != rc)
253 {
254 debugPrintf(6, _T("RecvMsg: Bad packet length [size=%d ActualSize=%d]"), ntohl(rawMsg->size), rc);
255 continue; // Bad packet, wait for next
256 }
257
258 if (ntohs(rawMsg->flags) & MF_BINARY)
259 {
260 // Convert message header to host format
261 rawMsg->id = ntohl(rawMsg->id);
262 rawMsg->code = ntohs(rawMsg->code);
263 rawMsg->numFields = ntohl(rawMsg->numFields);
264 if (nxlog_get_debug_level_tag_object(DEBUG_TAG, m_debugId) >= 6)
265 {
266 TCHAR buffer[64];
267 debugPrintf(6, _T("Received raw message %s (%d) from agent at %s"),
268 NXCPMessageCodeName(rawMsg->code, buffer), rawMsg->id, (const TCHAR *)m_addr.toString());
269 }
270
271 if ((rawMsg->code == CMD_FILE_DATA) && (rawMsg->id == m_dwDownloadRequestId))
272 {
273 if (m_sendToClientMessageCallback != NULL)
274 {
275 rawMsg->code = ntohs(rawMsg->code);
276 rawMsg->numFields = ntohl(rawMsg->numFields);
277 m_sendToClientMessageCallback(rawMsg, m_downloadProgressCallbackArg);
278
279 if (ntohs(rawMsg->flags) & MF_END_OF_FILE)
280 {
281 m_sendToClientMessageCallback = NULL;
282 onFileDownload(true);
283 }
284 else
285 {
286 if (m_downloadProgressCallback != NULL)
287 {
288 m_downloadProgressCallback(rawMsg->size - (NXCP_HEADER_SIZE + 8), m_downloadProgressCallbackArg);
289 }
290 }
291 }
292 else
293 {
294 if (m_hCurrFile != -1)
295 {
296 if (_write(m_hCurrFile, rawMsg->fields, rawMsg->numFields) == (int)rawMsg->numFields)
297 {
298 if (ntohs(rawMsg->flags) & MF_END_OF_FILE)
299 {
300 _close(m_hCurrFile);
301 m_hCurrFile = -1;
302
303 onFileDownload(true);
304 }
305 else
306 {
307 if (m_downloadProgressCallback != NULL)
308 {
309 m_downloadProgressCallback(_tell(m_hCurrFile), m_downloadProgressCallbackArg);
310 }
311 }
312 }
313 }
314 else
315 {
316 // I/O error
317 _close(m_hCurrFile);
318 m_hCurrFile = -1;
319
320 onFileDownload(false);
321 }
322 }
323 }
324 else if ((rawMsg->code == CMD_ABORT_FILE_TRANSFER) && (rawMsg->id == m_dwDownloadRequestId))
325 {
326 if (m_sendToClientMessageCallback != NULL)
327 {
328 rawMsg->code = ntohs(rawMsg->code);
329 rawMsg->numFields = ntohl(rawMsg->numFields);
330 m_sendToClientMessageCallback(rawMsg, m_downloadProgressCallbackArg);
331 m_sendToClientMessageCallback = NULL;
332
333 onFileDownload(false);
334 }
335 else
336 {
337 //error on agent side
338 _close(m_hCurrFile);
339 m_hCurrFile = -1;
340
341 onFileDownload(false);
342 }
343 }
344 else if (rawMsg->code == CMD_TCP_PROXY_DATA)
345 {
346 processTcpProxyData(rawMsg->id, rawMsg->fields, rawMsg->numFields);
347 }
348 }
349 else if (ntohs(rawMsg->flags) & MF_CONTROL)
350 {
351 // Convert message header to host format
352 rawMsg->id = ntohl(rawMsg->id);
353 rawMsg->code = ntohs(rawMsg->code);
354 rawMsg->flags = ntohs(rawMsg->flags);
355 rawMsg->numFields = ntohl(rawMsg->numFields);
356 if (nxlog_get_debug_level_tag_object(DEBUG_TAG, m_debugId) >= 6)
357 {
358 TCHAR buffer[64];
359 debugPrintf(6, _T("Received control message %s from agent at %s"),
360 NXCPMessageCodeName(rawMsg->code, buffer), (const TCHAR *)m_addr.toString());
361 }
362 m_pMsgWaitQueue->put((NXCP_MESSAGE *)nx_memdup(rawMsg, ntohl(rawMsg->size)));
363 }
364 else
365 {
366 // Create message object from raw message
367 NXCPMessage *msg = NXCPMessage::deserialize(rawMsg, m_nProtocolVersion);
368 if (msg != NULL)
369 {
370 if (nxlog_get_debug_level_tag_object(DEBUG_TAG, m_debugId) >= 6)
371 {
372 TCHAR buffer[64];
373 debugPrintf(6, _T("Received message %s (%d) from agent at %s"),
374 NXCPMessageCodeName(msg->getCode(), buffer), msg->getId(), (const TCHAR *)m_addr.toString());
375 }
376 switch(msg->getCode())
377 {
378 case CMD_REQUEST_COMPLETED:
379 case CMD_SESSION_KEY:
380 m_pMsgWaitQueue->put(msg);
381 break;
382 case CMD_TRAP:
383 if (g_agentConnectionThreadPool != NULL)
384 {
385 incInternalRefCount();
386 ThreadPoolExecute(g_agentConnectionThreadPool, this, &AgentConnection::onTrapCallback, msg);
387 }
388 else
389 {
390 delete msg;
391 }
392 break;
393 case CMD_SYSLOG_RECORDS:
394 if (g_agentConnectionThreadPool != NULL)
395 {
396 incInternalRefCount();
397 ThreadPoolExecute(g_agentConnectionThreadPool, this, &AgentConnection::onSyslogMessageCallback, msg);
398 }
399 else
400 {
401 delete msg;
402 }
403 break;
404 case CMD_PUSH_DCI_DATA:
405 if (g_agentConnectionThreadPool != NULL)
406 {
407 incInternalRefCount();
408 ThreadPoolExecute(g_agentConnectionThreadPool, this, &AgentConnection::onDataPushCallback, msg);
409 }
410 else
411 {
412 delete msg;
413 }
414 break;
415 case CMD_DCI_DATA:
416 if (g_agentConnectionThreadPool != NULL)
417 {
418 incInternalRefCount();
419 ThreadPoolExecute(g_agentConnectionThreadPool, this, &AgentConnection::processCollectedDataCallback, msg);
420 }
421 else
422 {
423 NXCPMessage response(CMD_REQUEST_COMPLETED, msg->getId(), m_nProtocolVersion);
424 response.setField(VID_RCC, ERR_INTERNAL_ERROR);
425 sendMessage(&response);
426 delete msg;
427 }
428 break;
429 case CMD_FILE_MONITORING:
430 onFileMonitoringData(msg);
431 delete msg;
432 break;
433 case CMD_SNMP_TRAP:
434 if (g_agentConnectionThreadPool != NULL)
435 {
436 incInternalRefCount();
437 ThreadPoolExecute(g_agentConnectionThreadPool, this, &AgentConnection::onSnmpTrapCallback, msg);
438 }
439 else
440 {
441 delete msg;
442 }
443 break;
444 case CMD_CLOSE_TCP_PROXY:
445 processTcpProxyData(msg->getFieldAsUInt32(VID_CHANNEL_ID), NULL, 0);
446 delete msg;
447 break;
448 default:
449 if (processCustomMessage(msg))
450 delete msg;
451 else
452 m_pMsgWaitQueue->put(msg);
453 break;
454 }
455 }
456 else
457 {
458 debugPrintf(6, _T("RecvMsg: message deserialization error"));
459 }
460 }
461 }
462 debugPrintf(6, _T("Receiver loop terminated"));
463
464 // Close socket and mark connection as disconnected
465 lock();
466 if (m_hCurrFile != -1)
467 {
468 _close(m_hCurrFile);
469 m_hCurrFile = -1;
470 onFileDownload(false);
471 }
472 else if (m_sendToClientMessageCallback != NULL)
473 {
474 m_sendToClientMessageCallback = NULL;
475 onFileDownload(false);
476 }
477
478 debugPrintf(6, _T("Closing communication channel"));
479 channel->close();
480 channel->decRefCount();
481 if (m_pCtx != NULL)
482 {
483 m_pCtx->decRefCount();
484 m_pCtx = NULL;
485 }
486 m_isConnected = false;
487 unlock();
488
489 MemFree(rawMsg);
490 MemFree(msgBuffer);
491 #ifdef _WITH_ENCRYPTION
492 MemFree(decryptionBuffer);
493 #endif
494
495 debugPrintf(6, _T("Receiver thread stopped"));
496 }
497
498 /**
499 * Create channel. Default implementation creates socket channel.
500 */
501 AbstractCommChannel *AgentConnection::createChannel()
502 {
503 SOCKET s = m_useProxy ?
504 ConnectToHost(m_proxyAddr, m_wProxyPort, m_connectionTimeout) :
505 ConnectToHost(m_addr, m_wPort, m_connectionTimeout);
506
507 // Connect to server
508 if (s == INVALID_SOCKET)
509 {
510 TCHAR buffer[64];
511 debugPrintf(5, _T("Cannot establish connection with agent at %s:%d"),
512 m_useProxy ? m_proxyAddr.toString(buffer) : m_addr.toString(buffer),
513 (int)(m_useProxy ? m_wProxyPort : m_wPort));
514 return NULL;
515 }
516
517 return new SocketCommChannel(s);
518 }
519
520 /**
521 * Acquire communication channel. Caller must call decRefCount to release channel.
522 */
523 AbstractCommChannel *AgentConnection::acquireChannel()
524 {
525 lock();
526 AbstractCommChannel *channel = m_channel;
527 if (channel != NULL)
528 channel->incRefCount();
529 unlock();
530 return channel;
531 }
532
533 /**
534 * Connect to agent
535 */
536 bool AgentConnection::connect(RSA *pServerKey, UINT32 *pdwError, UINT32 *pdwSocketError, UINT64 serverId)
537 {
538 TCHAR szBuffer[256];
539 bool success = false;
540 bool forceEncryption = false;
541 bool secondPass = false;
542 UINT32 dwError = 0;
543
544 if (pdwError != NULL)
545 *pdwError = ERR_INTERNAL_ERROR;
546
547 if (pdwSocketError != NULL)
548 *pdwSocketError = 0;
549
550 // Check if already connected
551 if (m_isConnected)
552 return false;
553
554 // Wait for receiver thread from previous connection, if any
555 ThreadJoin(m_hReceiverThread);
556 m_hReceiverThread = INVALID_THREAD_HANDLE;
557
558 // Check if we need to close existing channel
559 if (m_channel != NULL)
560 {
561 m_channel->decRefCount();
562 m_channel = NULL;
563 }
564
565 m_channel = createChannel();
566 if (m_channel == NULL)
567 {
568 debugPrintf(6, _T("Cannot create communication channel"));
569 dwError = ERR_CONNECT_FAILED;
570 goto connect_cleanup;
571 }
572
573 if (!NXCPGetPeerProtocolVersion(m_channel, &m_nProtocolVersion, m_mutexSocketWrite))
574 {
575 debugPrintf(6, _T("Protocol version negotiation failed"));
576 dwError = ERR_INTERNAL_ERROR;
577 goto connect_cleanup;
578 }
579 debugPrintf(6, _T("Using NXCP version %d"), m_nProtocolVersion);
580
581 // Start receiver thread
582 incInternalRefCount();
583 m_channel->incRefCount(); // for receiver thread
584 m_hReceiverThread = ThreadCreateEx(receiverThreadStarter, 0, this);
585 if (m_hReceiverThread == INVALID_THREAD_HANDLE)
586 {
587 debugPrintf(3, _T("Cannot start receiver thread"));
588 dwError = ERR_INTERNAL_ERROR;
589 m_channel->decRefCount();
590 goto connect_cleanup;
591 }
592
593 // Setup encryption
594 setup_encryption:
595 if ((m_iEncryptionPolicy == ENCRYPTION_PREFERRED) ||
596 (m_iEncryptionPolicy == ENCRYPTION_REQUIRED) ||
597 forceEncryption) // Agent require encryption
598 {
599 if (pServerKey != NULL)
600 {
601 dwError = setupEncryption(pServerKey);
602 if ((dwError != ERR_SUCCESS) &&
603 ((m_iEncryptionPolicy == ENCRYPTION_REQUIRED) || forceEncryption))
604 goto connect_cleanup;
605 }
606 else
607 {
608 if ((m_iEncryptionPolicy == ENCRYPTION_REQUIRED) || forceEncryption)
609 {
610 dwError = ERR_ENCRYPTION_REQUIRED;
611 goto connect_cleanup;
612 }
613 }
614 }
615
616 // Authenticate itself to agent
617 if ((dwError = authenticate(m_useProxy && !secondPass)) != ERR_SUCCESS)
618 {
619 if ((dwError == ERR_ENCRYPTION_REQUIRED) &&
620 (m_iEncryptionPolicy != ENCRYPTION_DISABLED))
621 {
622 forceEncryption = true;
623 goto setup_encryption;
624 }
625 debugPrintf(5, _T("Authentication to agent %s failed (%s)"), m_addr.toString(szBuffer),
626 AgentErrorCodeToText(dwError));
627 goto connect_cleanup;
628 }
629
630 // Test connectivity and inform agent about server capabilities
631 if ((dwError = setServerCapabilities()) != ERR_SUCCESS)
632 {
633 if ((dwError == ERR_ENCRYPTION_REQUIRED) &&
634 (m_iEncryptionPolicy != ENCRYPTION_DISABLED))
635 {
636 forceEncryption = true;
637 goto setup_encryption;
638 }
639 if (dwError != ERR_UNKNOWN_COMMAND) // Older agents may not support enable IPv6 command
640 {
641 debugPrintf(5, _T("Communication with agent %s failed (%s)"), m_addr.toString(szBuffer), AgentErrorCodeToText(dwError));
642 goto connect_cleanup;
643 }
644 }
645
646 if (m_useProxy && !secondPass)
647 {
648 dwError = setupProxyConnection();
649 if (dwError != ERR_SUCCESS)
650 goto connect_cleanup;
651 lock();
652 if (m_pCtx != NULL)
653 {
654 m_pCtx->decRefCount();
655 m_pCtx = NULL;
656 }
657 unlock();
658
659 debugPrintf(6, _T("Proxy connection established"));
660
661 // Renegotiate NXCP version with actual target agent
662 NXCP_MESSAGE msg;
663 msg.id = 0;
664 msg.numFields = 0;
665 msg.size = htonl(NXCP_HEADER_SIZE);
666 msg.code = htons(CMD_GET_NXCP_CAPS);
667 msg.flags = htons(MF_CONTROL);
668 if (m_channel->send(&msg, NXCP_HEADER_SIZE, m_mutexSocketWrite) == NXCP_HEADER_SIZE)
669 {
670 NXCP_MESSAGE *rsp = m_pMsgWaitQueue->waitForRawMessage(CMD_NXCP_CAPS, 0, m_dwCommandTimeout);
671 if (rsp != NULL)
672 {
673 m_nProtocolVersion = rsp->numFields >> 24;
674 free(rsp);
675 }
676 else
677 {
678 // assume that peer doesn't understand CMD_GET_NXCP_CAPS message
679 // and set version number to 1
680 m_nProtocolVersion = 1;
681 }
682 debugPrintf(6, _T("Using NXCP version %d after re-negotioation"), m_nProtocolVersion);
683 }
684 else
685 {
686 debugPrintf(6, _T("Protocol version re-negotiation failed - cannot send CMD_GET_NXCP_CAPS message"));
687 dwError = ERR_CONNECTION_BROKEN;
688 goto connect_cleanup;
689 }
690
691 secondPass = true;
692 forceEncryption = false;
693 goto setup_encryption;
694 }
695
696 if (serverId != 0)
697 setServerId(serverId);
698
699 success = true;
700 dwError = ERR_SUCCESS;
701
702 connect_cleanup:
703 if (!success)
704 {
705 if (pdwSocketError != NULL)
706 *pdwSocketError = (UINT32)WSAGetLastError();
707
708 lock();
709 if (m_channel != NULL)
710 m_channel->shutdown();
711 unlock();
712 ThreadJoin(m_hReceiverThread);
713 m_hReceiverThread = INVALID_THREAD_HANDLE;
714
715 lock();
716 if (m_channel != NULL)
717 {
718 m_channel->close();
719 m_channel->decRefCount();
720 m_channel = NULL;
721 }
722
723 if (m_pCtx != NULL)
724 {
725 m_pCtx->decRefCount();
726 m_pCtx = NULL;
727 }
728
729 unlock();
730 }
731 m_isConnected = success;
732 if (pdwError != NULL)
733 *pdwError = dwError;
734 return success;
735 }
736
737 /**
738 * Disconnect from agent
739 */
740 void AgentConnection::disconnect()
741 {
742 debugPrintf(6, _T("disconnect() called"));
743 lock();
744 if (m_hCurrFile != -1)
745 {
746 _close(m_hCurrFile);
747 m_hCurrFile = -1;
748 onFileDownload(false);
749 }
750 else if (m_sendToClientMessageCallback != NULL)
751 {
752 m_sendToClientMessageCallback = NULL;
753 onFileDownload(false);
754 }
755
756 if (m_channel != NULL)
757 {
758 m_channel->shutdown();
759 m_channel->decRefCount();
760 m_channel = NULL;
761 }
762 m_isConnected = false;
763 unlock();
764 debugPrintf(6, _T("Disconnect completed"));
765 }
766
767 /**
768 * Set authentication data
769 */
770 void AgentConnection::setAuthData(int method, const TCHAR *secret)
771 {
772 m_iAuthMethod = method;
773 #ifdef UNICODE
774 WideCharToMultiByte(CP_ACP, WC_COMPOSITECHECK | WC_DEFAULTCHAR, secret, -1, m_szSecret, MAX_SECRET_LENGTH, NULL, NULL);
775 m_szSecret[MAX_SECRET_LENGTH - 1] = 0;
776 #else
777 strlcpy(m_szSecret, secret, MAX_SECRET_LENGTH);
778 #endif
779 }
780
781 /**
782 * Get interface list from agent
783 */
784 InterfaceList *AgentConnection::getInterfaceList()
785 {
786 StringList *data;
787 if (getList(_T("Net.InterfaceList"), &data) != ERR_SUCCESS)
788 return NULL;
789
790 InterfaceList *pIfList = new InterfaceList(data->size());
791
792 // Parse result set. Each line should have the following format:
793 // index ip_address/mask_bits iftype mac_address name
794 // index ip_address/mask_bits iftype(mtu) mac_address name
795 for(int i = 0; i < data->size(); i++)
796 {
797 TCHAR *line = MemCopyString(data->get(i));
798 TCHAR *pBuf = line;
799 UINT32 ifIndex = 0;
800
801 // Index
802 TCHAR *pChar = _tcschr(pBuf, ' ');
803 if (pChar != NULL)
804 {
805 *pChar = 0;
806 ifIndex = _tcstoul(pBuf, NULL, 10);
807 pBuf = pChar + 1;
808 }
809
810 bool newInterface = false;
811 InterfaceInfo *iface = pIfList->findByIfIndex(ifIndex);
812 if (iface == NULL)
813 {
814 iface = new InterfaceInfo(ifIndex);
815 newInterface = true;
816 }
817
818 // Address and mask
819 pChar = _tcschr(pBuf, _T(' '));
820 if (pChar != NULL)
821 {
822 TCHAR *pSlash;
823 static TCHAR defaultMask[] = _T("24");
824
825 *pChar = 0;
826 pSlash = _tcschr(pBuf, _T('/'));
827 if (pSlash != NULL)
828 {
829 *pSlash = 0;
830 pSlash++;
831 }
832 else // Just a paranoia protection, should'n happen if agent working correctly
833 {
834 pSlash = defaultMask;
835 }
836 InetAddress addr = InetAddress::parse(pBuf);
837 if (addr.isValid())
838 {
839 addr.setMaskBits(_tcstol(pSlash, NULL, 10));
840 // Agent may return 0.0.0.0/0 for interfaces without IP address
841 if ((addr.getFamily() != AF_INET) || (addr.getAddressV4() != 0))
842 iface->ipAddrList.add(addr);
843 }
844 pBuf = pChar + 1;
845 }
846
847 if (newInterface)
848 {
849 // Interface type
850 pChar = _tcschr(pBuf, ' ');
851 if (pChar != NULL)
852 {
853 *pChar = 0;
854
855 TCHAR *eptr;
856 iface->type = _tcstoul(pBuf, &eptr, 10);
857
858 // newer agents can return if_type(mtu)
859 if (*eptr == _T('('))
860 {
861 pBuf = eptr + 1;
862 eptr = _tcschr(pBuf, _T(')'));
863 if (eptr != NULL)
864 {
865 *eptr = 0;
866 iface->mtu = _tcstol(pBuf, NULL, 10);
867 }
868 }
869
870 pBuf = pChar + 1;
871 }
872
873 // MAC address
874 pChar = _tcschr(pBuf, ' ');
875 if (pChar != NULL)
876 {
877 *pChar = 0;
878 StrToBin(pBuf, iface->macAddr, MAC_ADDR_LENGTH);
879 pBuf = pChar + 1;
880 }
881
882 // Name (set description to name)
883 _tcslcpy(iface->name, pBuf, MAX_DB_STRING);
884 _tcslcpy(iface->description, pBuf, MAX_DB_STRING);
885
886 pIfList->add(iface);
887 }
888 MemFree(line);
889 }
890
891 delete data;
892 return pIfList;
893 }
894
895 /**
896 * Get parameter value
897 */
898 UINT32 AgentConnection::getParameter(const TCHAR *pszParam, UINT32 dwBufSize, TCHAR *pszBuffer)
899 {
900 if (!m_isConnected)
901 return ERR_NOT_CONNECTED;
902
903 NXCPMessage msg(m_nProtocolVersion);
904 UINT32 dwRqId = generateRequestId();
905 msg.setCode(CMD_GET_PARAMETER);
906 msg.setId(dwRqId);
907 msg.setField(VID_PARAMETER, pszParam);
908
909 UINT32 dwRetCode;
910 if (sendMessage(&msg))
911 {
912 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
913 if (response != NULL)
914 {
915 dwRetCode = response->getFieldAsUInt32(VID_RCC);
916 if (dwRetCode == ERR_SUCCESS)
917 {
918 if (response->isFieldExist(VID_VALUE))
919 {
920 response->getFieldAsString(VID_VALUE, pszBuffer, dwBufSize);
921 }
922 else
923 {
924 dwRetCode = ERR_MALFORMED_RESPONSE;
925 debugPrintf(3, _T("Malformed response to CMD_GET_PARAMETER"));
926 }
927 }
928 delete response;
929 }
930 else
931 {
932 dwRetCode = ERR_REQUEST_TIMEOUT;
933 }
934 }
935 else
936 {
937 dwRetCode = ERR_CONNECTION_BROKEN;
938 }
939 return dwRetCode;
940 }
941
942 /**
943 * Get ARP cache
944 */
945 ArpCache *AgentConnection::getArpCache()
946 {
947 StringList *data;
948 if (getList(_T("Net.ArpCache"), &data) != ERR_SUCCESS)
949 return NULL;
950
951 // Create empty structure
952 ArpCache *arpCache = new ArpCache();
953
954 TCHAR szByte[4], *pBuf, *pChar;
955 szByte[2] = 0;
956
957 // Parse data lines
958 // Each line has form of XXXXXXXXXXXX a.b.c.d n
959 // where XXXXXXXXXXXX is a MAC address (12 hexadecimal digits)
960 // a.b.c.d is an IP address in decimal dotted notation
961 // n is an interface index
962 for(int i = 0; i < data->size(); i++)
963 {
964 TCHAR *line = MemCopyString(data->get(i));
965 pBuf = line;
966 if (_tcslen(pBuf) < 20) // Invalid line
967 {
968 debugPrintf(7, _T("AgentConnection::getArpCache(): invalid line received from agent (\"%s\")"), line);
969 free(line);
970 continue;
971 }
972
973 // MAC address
974 BYTE macAddr[6];
975 for(int j = 0; j < 6; j++)
976 {
977 memcpy(szByte, pBuf, sizeof(TCHAR) * 2);
978 macAddr[j] = (BYTE)_tcstol(szByte, NULL, 16);
979 pBuf += 2;
980 }
981
982 // IP address
983 while(*pBuf == ' ')
984 pBuf++;
985 pChar = _tcschr(pBuf, _T(' '));
986 if (pChar != NULL)
987 *pChar = 0;
988 InetAddress ipAddr = InetAddress::parse(pBuf);
989
990 // Interface index
991 UINT32 ifIndex = (pChar != NULL) ? _tcstoul(pChar + 1, NULL, 10) : 0;
992
993 arpCache->addEntry(ipAddr, MacAddress(macAddr, 6), ifIndex);
994
995 free(line);
996 }
997
998 delete data;
999 return arpCache;
1000 }
1001
1002 /**
1003 * Send dummy command to agent (can be used for keepalive)
1004 */
1005 UINT32 AgentConnection::nop()
1006 {
1007 if (!m_isConnected)
1008 return ERR_CONNECTION_BROKEN;
1009
1010 NXCPMessage msg(m_nProtocolVersion);
1011 UINT32 dwRqId;
1012
1013 dwRqId = generateRequestId();
1014 msg.setCode(CMD_KEEPALIVE);
1015 msg.setId(dwRqId);
1016 if (sendMessage(&msg))
1017 return waitForRCC(dwRqId, m_dwCommandTimeout);
1018 else
1019 return ERR_CONNECTION_BROKEN;
1020 }
1021
1022 /**
1023 * inform agent about server capabilities
1024 */
1025 UINT32 AgentConnection::setServerCapabilities()
1026 {
1027 NXCPMessage msg(m_nProtocolVersion);
1028 UINT32 dwRqId = generateRequestId();
1029 msg.setCode(CMD_SET_SERVER_CAPABILITIES);
1030 msg.setField(VID_ENABLED, (INT16)1); // Enables IPv6 on pre-2.0 agents
1031 msg.setField(VID_IPV6_SUPPORT, (INT16)1);
1032 msg.setField(VID_BULK_RECONCILIATION, (INT16)1);
1033 msg.setField(VID_ENABLE_COMPRESSION, (INT16)(m_allowCompression ? 1 : 0));
1034 msg.setId(dwRqId);
1035 if (!sendMessage(&msg))
1036 return ERR_CONNECTION_BROKEN;
1037
1038 NXCPMessage *response = m_pMsgWaitQueue->waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
1039 if (response == NULL)
1040 return ERR_REQUEST_TIMEOUT;
1041
1042 UINT32 rcc = response->getFieldAsUInt32(VID_RCC);
1043 if (rcc == ERR_SUCCESS)
1044 {
1045 if (response->isFieldExist(VID_FLAGS))
1046 {
1047 UINT16 flags = response->getFieldAsUInt16(VID_FLAGS);
1048 if (flags & 0x01)
1049 m_controlServer = true;
1050 if (flags & 0x02)
1051 m_masterServer = true;
1052 }
1053 else
1054 {
1055 // Agents before 2.2.13 do not return access flags, assume this server has full access
1056 m_controlServer = true;
1057 m_masterServer = true;
1058 }
1059 }
1060 delete response;
1061 return rcc;
1062 }
1063
1064 /**
1065 * Set server ID
1066 */
1067 UINT32 AgentConnection::setServerId(UINT64 serverId)
1068 {
1069 NXCPMessage msg(m_nProtocolVersion);
1070 UINT32 dwRqId = generateRequestId();
1071 msg.setCode(CMD_SET_SERVER_ID);
1072 msg.setField(VID_SERVER_ID, serverId);
1073 msg.setId(dwRqId);
1074 if (sendMessage(&msg))
1075 return waitForRCC(dwRqId, m_dwCommandTimeout);
1076 else
1077 return ERR_CONNECTION_BROKEN;
1078 }
1079
1080 /**
1081 * Wait for request completion code
1082 */
1083 UINT32 AgentConnection::waitForRCC(UINT32 dwRqId, UINT32 dwTimeOut)
1084 {
1085 UINT32 rcc;
1086 NXCPMessage *response = m_pMsgWaitQueue->waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, dwTimeOut);
1087 if (response != NULL)
1088 {
1089 rcc = response->getFieldAsUInt32(VID_RCC);
1090 delete response;
1091 }
1092 else
1093 {
1094 rcc = ERR_REQUEST_TIMEOUT;
1095 }
1096 return rcc;
1097 }
1098
1099 /**
1100 * Send message to agent
1101 */
1102 bool AgentConnection::sendMessage(NXCPMessage *pMsg)
1103 {
1104 AbstractCommChannel *channel = acquireChannel();
1105 if (channel == NULL)
1106 return false;
1107
1108 if (nxlog_get_debug_level_tag_object(DEBUG_TAG, m_debugId) >= 6)
1109 {
1110 TCHAR buffer[64];
1111 debugPrintf(6, _T("Sending message %s (%d) to agent at %s"),
1112 NXCPMessageCodeName(pMsg->getCode(), buffer), pMsg->getId(), (const TCHAR *)m_addr.toString());
1113 }
1114
1115 bool success;
1116 NXCP_MESSAGE *rawMsg = pMsg->serialize(m_allowCompression);
1117 NXCPEncryptionContext *pCtx = acquireEncryptionContext();
1118 if (pCtx != NULL)
1119 {
1120 NXCP_ENCRYPTED_MESSAGE *pEnMsg = pCtx->encryptMessage(rawMsg);
1121 if (pEnMsg != NULL)
1122 {
1123 success = (channel->send(pEnMsg, ntohl(pEnMsg->size), m_mutexSocketWrite) == (int)ntohl(pEnMsg->size));
1124 free(pEnMsg);
1125 }
1126 else
1127 {
1128 success = false;
1129 }
1130 pCtx->decRefCount();
1131 }
1132 else
1133 {
1134 success = (channel->send(rawMsg, ntohl(rawMsg->size), m_mutexSocketWrite) == (int)ntohl(rawMsg->size));
1135 }
1136 free(rawMsg);
1137 channel->decRefCount();
1138 return success;
1139 }
1140
1141 /**
1142 * Send raw message to agent
1143 */
1144 bool AgentConnection::sendRawMessage(NXCP_MESSAGE *pMsg)
1145 {
1146 AbstractCommChannel *channel = acquireChannel();
1147 if (channel == NULL)
1148 return false;
1149
1150 bool success;
1151 NXCP_MESSAGE *rawMsg = pMsg;
1152 NXCPEncryptionContext *pCtx = acquireEncryptionContext();
1153 if (pCtx != NULL)
1154 {
1155 NXCP_ENCRYPTED_MESSAGE *pEnMsg = pCtx->encryptMessage(rawMsg);
1156 if (pEnMsg != NULL)
1157 {
1158 success = (channel->send(pEnMsg, ntohl(pEnMsg->size), m_mutexSocketWrite) == (int)ntohl(pEnMsg->size));
1159 free(pEnMsg);
1160 }
1161 else
1162 {
1163 success = false;
1164 }
1165 pCtx->decRefCount();
1166 }
1167 else
1168 {
1169 success = (channel->send(rawMsg, ntohl(rawMsg->size), m_mutexSocketWrite) == (int)ntohl(rawMsg->size));
1170 }
1171 channel->decRefCount();
1172 return success;
1173 }
1174
1175 /**
1176 * Callback for sending raw NXCP message in background
1177 */
1178 void AgentConnection::postRawMessageCallback(NXCP_MESSAGE *msg)
1179 {
1180 sendRawMessage(msg);
1181 free(msg);
1182 decInternalRefCount();
1183 }
1184
1185 /**
1186 * Send raw NXCP message in background. Provided message will be destroyed after sending.
1187 */
1188 void AgentConnection::postRawMessage(NXCP_MESSAGE *msg)
1189 {
1190 incInternalRefCount();
1191 TCHAR key[64];
1192 _sntprintf(key, 64, _T("PostMessage_%p"), this);
1193 ThreadPoolExecuteSerialized(g_agentConnectionThreadPool, key, this, &AgentConnection::postRawMessageCallback, msg);
1194 }
1195
1196 /**
1197 * Callback for processing incoming event on separate thread
1198 */
1199 void AgentConnection::onTrapCallback(NXCPMessage *msg)
1200 {
1201 onTrap(msg);
1202 delete msg;
1203 decInternalRefCount();
1204 }
1205
1206 /**
1207 * Trap handler. Should be overriden in derived classes to implement
1208 * actual trap processing. Default implementation do nothing.
1209 */
1210 void AgentConnection::onTrap(NXCPMessage *pMsg)
1211 {
1212 }
1213
1214 /**
1215 * Callback for processing incoming syslog message on separate thread
1216 */
1217 void AgentConnection::onSyslogMessageCallback(NXCPMessage *msg)
1218 {
1219 onSyslogMessage(msg);
1220 delete msg;
1221 decInternalRefCount();
1222 }
1223
1224 /**
1225 * Syslog message handler. Should be overriden in derived classes to implement
1226 * actual message processing. Default implementation do nothing.
1227 */
1228 void AgentConnection::onSyslogMessage(NXCPMessage *pMsg)
1229 {
1230 }
1231
1232 /**
1233 * Callback for processing data push on separate thread
1234 */
1235 void AgentConnection::onDataPushCallback(NXCPMessage *msg)
1236 {
1237 onDataPush(msg);
1238 delete msg;
1239 decInternalRefCount();
1240 }
1241
1242 /**
1243 * Data push handler. Should be overriden in derived classes to implement
1244 * actual data push processing. Default implementation do nothing.
1245 */
1246 void AgentConnection::onDataPush(NXCPMessage *pMsg)
1247 {
1248 }
1249
1250 /**
1251 * Monitoring data handler. Should be overriden in derived classes to implement
1252 * actual monitoring data processing. Default implementation do nothing.
1253 */
1254 void AgentConnection::onFileMonitoringData(NXCPMessage *pMsg)
1255 {
1256 }
1257
1258 /**
1259 * Callback for processing data push on separate thread
1260 */
1261 void AgentConnection::onSnmpTrapCallback(NXCPMessage *msg)
1262 {
1263 onSnmpTrap(msg);
1264 delete msg;
1265 decInternalRefCount();
1266 }
1267
1268 /**
1269 * SNMP trap handler. Should be overriden in derived classes to implement
1270 * actual SNMP trap processing. Default implementation do nothing.
1271 */
1272 void AgentConnection::onSnmpTrap(NXCPMessage *pMsg)
1273 {
1274 }
1275
1276 /**
1277 * Custom message handler
1278 * If returns true, message considered as processed and will not be placed in wait queue
1279 */
1280 bool AgentConnection::processCustomMessage(NXCPMessage *pMsg)
1281 {
1282 return false;
1283 }
1284
1285 /**
1286 * Get list of values
1287 */
1288 UINT32 AgentConnection::getList(const TCHAR *param, StringList **list)
1289 {
1290 UINT32 rcc;
1291 *list = NULL;
1292 if (m_isConnected)
1293 {
1294 NXCPMessage msg(CMD_GET_LIST, generateRequestId(), m_nProtocolVersion);
1295 msg.setField(VID_PARAMETER, param);
1296 if (sendMessage(&msg))
1297 {
1298 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, msg.getId(), m_dwCommandTimeout);
1299 if (response != NULL)
1300 {
1301 rcc = response->getFieldAsUInt32(VID_RCC);
1302 if (rcc == ERR_SUCCESS)
1303 {
1304 *list = new StringList();
1305 int count = response->getFieldAsInt32(VID_NUM_STRINGS);
1306 for(int i = 0; i < count; i++)
1307 (*list)->addPreallocated(response->getFieldAsString(VID_ENUM_VALUE_BASE + i));
1308 }
1309 delete response;
1310 }
1311 else
1312 {
1313 rcc = ERR_REQUEST_TIMEOUT;
1314 }
1315 }
1316 else
1317 {
1318 rcc = ERR_CONNECTION_BROKEN;
1319 }
1320 }
1321 else
1322 {
1323 rcc = ERR_NOT_CONNECTED;
1324 }
1325
1326 return rcc;
1327 }
1328
1329 /**
1330 * Get table
1331 */
1332 UINT32 AgentConnection::getTable(const TCHAR *pszParam, Table **table)
1333 {
1334 NXCPMessage msg(m_nProtocolVersion), *pResponse;
1335 UINT32 dwRqId, dwRetCode;
1336
1337 *table = NULL;
1338 if (m_isConnected)
1339 {
1340 dwRqId = generateRequestId();
1341 msg.setCode(CMD_GET_TABLE);
1342 msg.setId(dwRqId);
1343 msg.setField(VID_PARAMETER, pszParam);
1344 if (sendMessage(&msg))
1345 {
1346 pResponse = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
1347 if (pResponse != NULL)
1348 {
1349 dwRetCode = pResponse->getFieldAsUInt32(VID_RCC);
1350 if (dwRetCode == ERR_SUCCESS)
1351 {
1352 *table = new Table(pResponse);
1353 }
1354 delete pResponse;
1355 }
1356 else
1357 {
1358 dwRetCode = ERR_REQUEST_TIMEOUT;
1359 }
1360 }
1361 else
1362 {
1363 dwRetCode = ERR_CONNECTION_BROKEN;
1364 }
1365 }
1366 else
1367 {
1368 dwRetCode = ERR_NOT_CONNECTED;
1369 }
1370
1371 return dwRetCode;
1372 }
1373
1374 /**
1375 * Authenticate to agent
1376 */
1377 UINT32 AgentConnection::authenticate(BOOL bProxyData)
1378 {
1379 NXCPMessage msg(m_nProtocolVersion);
1380 UINT32 dwRqId;
1381 BYTE hash[32];
1382 int iAuthMethod = bProxyData ? m_iProxyAuth : m_iAuthMethod;
1383 const char *pszSecret = bProxyData ? m_szProxySecret : m_szSecret;
1384 #ifdef UNICODE
1385 WCHAR szBuffer[MAX_SECRET_LENGTH];
1386 #endif
1387
1388 if (iAuthMethod == AUTH_NONE)
1389 return ERR_SUCCESS; // No authentication required
1390
1391 dwRqId = generateRequestId();
1392 msg.setCode(CMD_AUTHENTICATE);
1393 msg.setId(dwRqId);
1394 msg.setField(VID_AUTH_METHOD, (WORD)iAuthMethod);
1395 switch(iAuthMethod)
1396 {
1397 case AUTH_PLAINTEXT:
1398 #ifdef UNICODE
1399 MultiByteToWideChar(CP_ACP, MB_PRECOMPOSED, pszSecret, -1, szBuffer, MAX_SECRET_LENGTH);
1400 msg.setField(VID_SHARED_SECRET, szBuffer);
1401 #else
1402 msg.setField(VID_SHARED_SECRET, pszSecret);
1403 #endif
1404 break;
1405 case AUTH_MD5_HASH:
1406 CalculateMD5Hash((BYTE *)pszSecret, (int)strlen(pszSecret), hash);
1407 msg.setField(VID_SHARED_SECRET, hash, MD5_DIGEST_SIZE);
1408 break;
1409 case AUTH_SHA1_HASH:
1410 CalculateSHA1Hash((BYTE *)pszSecret, (int)strlen(pszSecret), hash);
1411 msg.setField(VID_SHARED_SECRET, hash, SHA1_DIGEST_SIZE);
1412 break;
1413 default:
1414 break;
1415 }
1416 if (sendMessage(&msg))
1417 return waitForRCC(dwRqId, m_dwCommandTimeout);
1418 else
1419 return ERR_CONNECTION_BROKEN;
1420 }
1421
1422 /**
1423 * Execute action on agent
1424 */
1425 UINT32 AgentConnection::execAction(const TCHAR *action, const StringList &list,
1426 bool withOutput, void (* outputCallback)(ActionCallbackEvent, const TCHAR *, void *), void *cbData)
1427 {
1428 NXCPMessage msg(m_nProtocolVersion);
1429 UINT32 dwRqId;
1430 int i;
1431
1432 if (!m_isConnected)
1433 return ERR_NOT_CONNECTED;
1434
1435 dwRqId = generateRequestId();
1436 msg.setCode(CMD_ACTION);
1437 msg.setId(dwRqId);
1438 msg.setField(VID_ACTION_NAME, action);
1439 msg.setField(VID_RECEIVE_OUTPUT, (UINT16)(withOutput ? 1 : 0));
1440 msg.setField(VID_NUM_ARGS, (UINT32)list.size());
1441 for(i = 0; i < list.size(); i++)
1442 msg.setField(VID_ACTION_ARG_BASE + i, list.get(i));
1443
1444 if (sendMessage(&msg))
1445 {
1446 if (withOutput)
1447 {
1448 UINT32 rcc = waitForRCC(dwRqId, m_dwCommandTimeout);
1449 if (rcc == ERR_SUCCESS)
1450 {
1451 outputCallback(ACE_CONNECTED, NULL, cbData); // Indicate successful start
1452 bool eos = false;
1453 while(!eos)
1454 {
1455 NXCPMessage *response = waitForMessage(CMD_COMMAND_OUTPUT, dwRqId, m_dwCommandTimeout * 10);
1456 if (response != NULL)
1457 {
1458 eos = response->isEndOfSequence();
1459 if (response->isFieldExist(VID_MESSAGE))
1460 {
1461 TCHAR line[4096];
1462 response->getFieldAsString(VID_MESSAGE, line, 4096);
1463 outputCallback(ACE_DATA, line, cbData);
1464 }
1465 delete response;
1466 }
1467 else
1468 {
1469 return ERR_REQUEST_TIMEOUT;
1470 }
1471 }
1472 outputCallback(ACE_DISCONNECTED, NULL, cbData);
1473 return ERR_SUCCESS;
1474 }
1475 else
1476 {
1477 return rcc;
1478 }
1479 }
1480 else
1481 {
1482 return waitForRCC(dwRqId, m_dwCommandTimeout);
1483 }
1484 }
1485 else
1486 {
1487 return ERR_CONNECTION_BROKEN;
1488 }
1489 }
1490
1491 /**
1492 * Upload file to agent
1493 */
1494 UINT32 AgentConnection::uploadFile(const TCHAR *localFile, const TCHAR *destinationFile, void (* progressCallback)(INT64, void *), void *cbArg, NXCPStreamCompressionMethod compMethod)
1495 {
1496 UINT32 dwRqId, dwResult;
1497 NXCPMessage msg(m_nProtocolVersion);
1498
1499 // Disable compression if it is disabled on connection level or if agent do not support it
1500 if (!m_allowCompression || (m_nProtocolVersion < 4))
1501 compMethod = NXCP_STREAM_COMPRESSION_NONE;
1502
1503 if (!m_isConnected)
1504 return ERR_NOT_CONNECTED;
1505
1506 dwRqId = generateRequestId();
1507 msg.setId(dwRqId);
1508
1509 time_t lastModTime = 0;
1510 NX_STAT_STRUCT st;
1511 if (CALL_STAT(localFile, &st) == 0)
1512 {
1513 lastModTime = st.st_mtime;
1514 }
1515
1516 // Use core agent if destination file name is not set and file manager subagent otherwise
1517 if ((destinationFile == NULL) || (*destinationFile == 0))
1518 {
1519 msg.setCode(CMD_TRANSFER_FILE);
1520 int i;
1521 for(i = (int)_tcslen(localFile) - 1;
1522 (i >= 0) && (localFile[i] != '\\') && (localFile[i] != '/'); i--);
1523 msg.setField(VID_FILE_NAME, &localFile[i + 1]);
1524 }
1525 else
1526 {
1527 msg.setCode(CMD_FILEMGR_UPLOAD);
1528 msg.setField(VID_OVERWRITE, true);
1529 msg.setField(VID_FILE_NAME, destinationFile);
1530 }
1531 msg.setFieldFromTime(VID_MODIFICATION_TIME, lastModTime);
1532
1533 if (sendMessage(&msg))
1534 {
1535 dwResult = waitForRCC(dwRqId, m_dwCommandTimeout);
1536 }
1537 else
1538 {
1539 dwResult = ERR_CONNECTION_BROKEN;
1540 }
1541
1542 if (dwResult == ERR_SUCCESS)
1543 {
1544 AbstractCommChannel *channel = acquireChannel();
1545 if (channel != NULL)
1546 {
1547 debugPrintf(5, _T("Sending file \"%s\" to agent %s compression"),
1548 localFile, (compMethod == NXCP_STREAM_COMPRESSION_NONE) ? _T("without") : _T("with"));
1549 m_fileUploadInProgress = true;
1550 NXCPEncryptionContext *ctx = acquireEncryptionContext();
1551 if (SendFileOverNXCP(channel, dwRqId, localFile, ctx, 0, progressCallback, cbArg, m_mutexSocketWrite, compMethod))
1552 dwResult = waitForRCC(dwRqId, m_dwCommandTimeout);
1553 else
1554 dwResult = ERR_IO_FAILURE;
1555 m_fileUploadInProgress = false;
1556 if (ctx != NULL)
1557 ctx->decRefCount();
1558 channel->decRefCount();
1559 }
1560 else
1561 {
1562 dwResult = ERR_CONNECTION_BROKEN;
1563 }
1564 }
1565
1566 return dwResult;
1567 }
1568
1569 /**
1570 * Send upgrade command
1571 */
1572 UINT32 AgentConnection::startUpgrade(const TCHAR *pszPkgName)
1573 {
1574 UINT32 dwRqId, dwResult;
1575 NXCPMessage msg(m_nProtocolVersion);
1576 int i;
1577
1578 if (!m_isConnected)
1579 return ERR_NOT_CONNECTED;
1580
1581 dwRqId = generateRequestId();
1582
1583 msg.setCode(CMD_UPGRADE_AGENT);
1584 msg.setId(dwRqId);
1585 for(i = (int)_tcslen(pszPkgName) - 1;
1586 (i >= 0) && (pszPkgName[i] != '\\') && (pszPkgName[i] != '/'); i--);
1587 msg.setField(VID_FILE_NAME, &pszPkgName[i + 1]);
1588
1589 if (sendMessage(&msg))
1590 {
1591 dwResult = waitForRCC(dwRqId, m_dwCommandTimeout);
1592 }
1593 else
1594 {
1595 dwResult = ERR_CONNECTION_BROKEN;
1596 }
1597
1598 return dwResult;
1599 }
1600
1601 /**
1602 * Check status of network service via agent
1603 */
1604 UINT32 AgentConnection::checkNetworkService(UINT32 *pdwStatus, const InetAddress& addr, int iServiceType,
1605 WORD wPort, WORD wProto, const TCHAR *pszRequest,
1606 const TCHAR *pszResponse, UINT32 *responseTime)
1607 {
1608 UINT32 dwRqId, dwResult;
1609 NXCPMessage msg(m_nProtocolVersion), *pResponse;
1610 static WORD m_wDefaultPort[] = { 7, 22, 110, 25, 21, 80, 443, 23 };
1611
1612 if (!m_isConnected)
1613 return ERR_NOT_CONNECTED;
1614
1615 dwRqId = generateRequestId();
1616
1617 msg.setCode(CMD_CHECK_NETWORK_SERVICE);
1618 msg.setId(dwRqId);
1619 msg.setField(VID_IP_ADDRESS, addr);
1620 msg.setField(VID_SERVICE_TYPE, (WORD)iServiceType);
1621 msg.setField(VID_IP_PORT,
1622 (wPort != 0) ? wPort :
1623 m_wDefaultPort[((iServiceType >= NETSRV_CUSTOM) &&
1624 (iServiceType <= NETSRV_TELNET)) ? iServiceType : 0]);
1625 msg.setField(VID_IP_PROTO, (wProto != 0) ? wProto : (WORD)IPPROTO_TCP);
1626 msg.setField(VID_SERVICE_REQUEST, pszRequest);
1627 msg.setField(VID_SERVICE_RESPONSE, pszResponse);
1628
1629 if (sendMessage(&msg))
1630 {
1631 // Wait up to 90 seconds for results
1632 pResponse = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, 90000);
1633 if (pResponse != NULL)
1634 {
1635 dwResult = pResponse->getFieldAsUInt32(VID_RCC);
1636 if (dwResult == ERR_SUCCESS)
1637 {
1638 *pdwStatus = pResponse->getFieldAsUInt32(VID_SERVICE_STATUS);
1639 if (responseTime != NULL)
1640 {
1641 *responseTime = pResponse->getFieldAsUInt32(VID_RESPONSE_TIME);
1642 }
1643 }
1644 delete pResponse;
1645 }
1646 else
1647 {
1648 dwResult = ERR_REQUEST_TIMEOUT;
1649 }
1650 }
1651 else
1652 {
1653 dwResult = ERR_CONNECTION_BROKEN;
1654 }
1655
1656 return dwResult;
1657 }
1658
1659 /**
1660 * Get list of supported parameters from agent
1661 */
1662 UINT32 AgentConnection::getSupportedParameters(ObjectArray<AgentParameterDefinition> **paramList, ObjectArray<AgentTableDefinition> **tableList)
1663 {
1664 UINT32 dwRqId, dwResult;
1665 NXCPMessage msg(m_nProtocolVersion), *pResponse;
1666
1667 *paramList = NULL;
1668 *tableList = NULL;
1669
1670 if (!m_isConnected)
1671 return ERR_NOT_CONNECTED;
1672
1673 dwRqId = generateRequestId();
1674
1675 msg.setCode(CMD_GET_PARAMETER_LIST);
1676 msg.setId(dwRqId);
1677
1678 if (sendMessage(&msg))
1679 {
1680 pResponse = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
1681 if (pResponse != NULL)
1682 {
1683 dwResult = pResponse->getFieldAsUInt32(VID_RCC);
1684 DbgPrintf(6, _T("AgentConnection::getSupportedParameters(): RCC=%d"), dwResult);
1685 if (dwResult == ERR_SUCCESS)
1686 {
1687 UINT32 count = pResponse->getFieldAsUInt32(VID_NUM_PARAMETERS);
1688 ObjectArray<AgentParameterDefinition> *plist = new ObjectArray<AgentParameterDefinition>(count, 16, true);
1689 for(UINT32 i = 0, id = VID_PARAM_LIST_BASE; i < count; i++)
1690 {
1691 plist->add(new AgentParameterDefinition(pResponse, id));
1692 id += 3;
1693 }
1694 *paramList = plist;
1695 DbgPrintf(6, _T("AgentConnection::getSupportedParameters(): %d parameters received from agent"), count);
1696
1697 count = pResponse->getFieldAsUInt32(VID_NUM_TABLES);
1698 ObjectArray<AgentTableDefinition> *tlist = new ObjectArray<AgentTableDefinition>(count, 16, true);
1699 for(UINT32 i = 0, id = VID_TABLE_LIST_BASE; i < count; i++)
1700 {
1701 tlist->add(new AgentTableDefinition(pResponse, id));
1702 id += 3;
1703 }
1704 *tableList = tlist;
1705 DbgPrintf(6, _T("AgentConnection::getSupportedParameters(): %d tables received from agent"), count);
1706 }
1707 delete pResponse;
1708 }
1709 else
1710 {
1711 dwResult = ERR_REQUEST_TIMEOUT;
1712 }
1713 }
1714 else
1715 {
1716 dwResult = ERR_CONNECTION_BROKEN;
1717 }
1718
1719 return dwResult;
1720 }
1721
1722 /**
1723 * Setup encryption
1724 */
1725 UINT32 AgentConnection::setupEncryption(RSA *pServerKey)
1726 {
1727 #ifdef _WITH_ENCRYPTION
1728 NXCPMessage msg(m_nProtocolVersion), *pResp;
1729 UINT32 dwRqId, dwError, dwResult;
1730
1731 dwRqId = generateRequestId();
1732
1733 PrepareKeyRequestMsg(&msg, pServerKey, false);
1734 msg.setId(dwRqId);
1735 if (sendMessage(&msg))
1736 {
1737 pResp = waitForMessage(CMD_SESSION_KEY, dwRqId, m_dwCommandTimeout);
1738 if (pResp != NULL)
1739 {
1740 dwResult = SetupEncryptionContext(pResp, &m_pCtx, NULL, pServerKey, m_nProtocolVersion);
1741 switch(dwResult)
1742 {
1743 case RCC_SUCCESS:
1744 dwError = ERR_SUCCESS;
1745 break;
1746 case RCC_NO_CIPHERS:
1747 dwError = ERR_NO_CIPHERS;
1748 break;
1749 case RCC_INVALID_PUBLIC_KEY:
1750 dwError = ERR_INVALID_PUBLIC_KEY;
1751 break;
1752 case RCC_INVALID_SESSION_KEY:
1753 dwError = ERR_INVALID_SESSION_KEY;
1754 break;
1755 default:
1756 dwError = ERR_INTERNAL_ERROR;
1757 break;
1758 }
1759 delete pResp;
1760 }
1761 else
1762 {
1763 dwError = ERR_REQUEST_TIMEOUT;
1764 }
1765 }
1766 else
1767 {
1768 dwError = ERR_CONNECTION_BROKEN;
1769 }
1770
1771 return dwError;
1772 #else
1773 return ERR_NOT_IMPLEMENTED;
1774 #endif
1775 }
1776
1777 /**
1778 * Get configuration file from agent
1779 */
1780 UINT32 AgentConnection::getConfigFile(TCHAR **ppszConfig, UINT32 *pdwSize)
1781 {
1782 *ppszConfig = NULL;
1783 *pdwSize = 0;
1784
1785 if (!m_isConnected)
1786 return ERR_NOT_CONNECTED;
1787
1788 UINT32 dwResult;
1789 UINT32 dwRqId = generateRequestId();
1790
1791 NXCPMessage msg(m_nProtocolVersion);
1792 msg.setCode(CMD_GET_AGENT_CONFIG);
1793 msg.setId(dwRqId);
1794
1795 if (sendMessage(&msg))
1796 {
1797 NXCPMessage *pResponse = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
1798 if (pResponse != NULL)
1799 {
1800 dwResult = pResponse->getFieldAsUInt32(VID_RCC);
1801 if (dwResult == ERR_SUCCESS)
1802 {
1803 size_t size = pResponse->getFieldAsBinary(VID_CONFIG_FILE, NULL, 0);
1804 BYTE *utf8Text = (BYTE *)malloc(size + 1);
1805 pResponse->getFieldAsBinary(VID_CONFIG_FILE, (BYTE *)utf8Text, size);
1806
1807 // We expect text file, so replace all non-printable characters with spaces
1808 for(size_t i = 0; i < size; i++)
1809 if ((utf8Text[i] < ' ') &&
1810 (utf8Text[i] != '\t') &&
1811 (utf8Text[i] != '\r') &&
1812 (utf8Text[i] != '\n'))
1813 utf8Text[i] = ' ';
1814 utf8Text[size] = 0;
1815
1816 #ifdef UNICODE
1817 *ppszConfig = WideStringFromUTF8String((char *)utf8Text);
1818 #else
1819 *ppszConfig = MBStringFromUTF8String((char *)utf8Text);
1820 #endif
1821 free(utf8Text);
1822 *pdwSize = (UINT32)_tcslen(*ppszConfig);
1823 }
1824 delete pResponse;
1825 }
1826 else
1827 {
1828 dwResult = ERR_REQUEST_TIMEOUT;
1829 }
1830 }
1831 else
1832 {
1833 dwResult = ERR_CONNECTION_BROKEN;
1834 }
1835
1836 return dwResult;
1837 }
1838
1839 /**
1840 * Update configuration file on agent
1841 */
1842 UINT32 AgentConnection::updateConfigFile(const TCHAR *pszConfig)
1843 {
1844 UINT32 dwRqId, dwResult;
1845 NXCPMessage msg(m_nProtocolVersion);
1846 #ifdef UNICODE
1847 int nChars;
1848 BYTE *pBuffer;
1849 #endif
1850
1851 if (!m_isConnected)
1852 return ERR_NOT_CONNECTED;
1853
1854 dwRqId = generateRequestId();
1855
1856 msg.setCode(CMD_UPDATE_AGENT_CONFIG);
1857 msg.setId(dwRqId);
1858 #ifdef UNICODE
1859 nChars = (int)_tcslen(pszConfig);
1860 pBuffer = (BYTE *)malloc(nChars + 1);
1861 WideCharToMultiByte(CP_ACP, WC_COMPOSITECHECK | WC_DEFAULTCHAR,
1862 pszConfig, nChars, (char *)pBuffer, nChars + 1, NULL, NULL);
1863 msg.setField(VID_CONFIG_FILE, pBuffer, nChars);
1864 free(pBuffer);
1865 #else
1866 msg.setField(VID_CONFIG_FILE, (BYTE *)pszConfig, (UINT32)strlen(pszConfig));
1867 #endif
1868
1869 if (sendMessage(&msg))
1870 {
1871 dwResult = waitForRCC(dwRqId, m_dwCommandTimeout);
1872 }
1873 else
1874 {
1875 dwResult = ERR_CONNECTION_BROKEN;
1876 }
1877
1878 return dwResult;
1879 }
1880
1881 /**
1882 * Get routing table from agent
1883 */
1884 ROUTING_TABLE *AgentConnection::getRoutingTable()
1885 {
1886 StringList *data;
1887 if (getList(_T("Net.IP.RoutingTable"), &data) != ERR_SUCCESS)
1888 return NULL;
1889
1890 ROUTING_TABLE *pRT = (ROUTING_TABLE *)malloc(sizeof(ROUTING_TABLE));
1891 pRT->iNumEntries = data->size();
1892 pRT->pRoutes = (ROUTE *)calloc(data->size(), sizeof(ROUTE));
1893 for(int i = 0; i < data->size(); i++)
1894 {
1895 TCHAR *line = MemCopyString(data->get(i));
1896 TCHAR *pBuf = line;
1897
1898 // Destination address and mask
1899 TCHAR *pChar = _tcschr(pBuf, _T(' '));
1900 if (pChar != NULL)
1901 {
1902 TCHAR *pSlash;
1903 static TCHAR defaultMask[] = _T("24");
1904
1905 *pChar = 0;
1906 pSlash = _tcschr(pBuf, _T('/'));
1907 if (pSlash != NULL)
1908 {
1909 *pSlash = 0;
1910 pSlash++;
1911 }
1912 else // Just a paranoia protection, should'n happen if agent working correctly
1913 {
1914 pSlash = defaultMask;
1915 }
1916 pRT->pRoutes[i].dwDestAddr = ntohl(_t_inet_addr(pBuf));
1917 UINT32 dwBits = _tcstoul(pSlash, NULL, 10);
1918 pRT->pRoutes[i].dwDestMask = (dwBits == 32) ? 0xFFFFFFFF : (~(0xFFFFFFFF >> dwBits));
1919 pBuf = pChar + 1;
1920 }
1921
1922 // Next hop address
1923 pChar = _tcschr(pBuf, _T(' '));
1924 if (pChar != NULL)
1925 {
1926 *pChar = 0;
1927 pRT->pRoutes[i].dwNextHop = ntohl(_t_inet_addr(pBuf));
1928 pBuf = pChar + 1;
1929 }
1930
1931 // Interface index
1932 pChar = _tcschr(pBuf, ' ');
1933 if (pChar != NULL)
1934 {
1935 *pChar = 0;
1936 pRT->pRoutes[i].dwIfIndex = _tcstoul(pBuf, NULL, 10);
1937 pBuf = pChar + 1;
1938 }
1939
1940 // Route type
1941 pRT->pRoutes[i].dwRouteType = _tcstoul(pBuf, NULL, 10);
1942
1943 free(line);
1944 }
1945
1946 delete data;
1947 return pRT;
1948 }
1949
1950 /**
1951 * Set proxy information
1952 */
1953 void AgentConnection::setProxy(InetAddress addr, WORD wPort, int iAuthMethod, const TCHAR *pszSecret)
1954 {
1955 m_proxyAddr = addr;
1956 m_wProxyPort = wPort;
1957 m_iProxyAuth = iAuthMethod;
1958 if (pszSecret != NULL)
1959 {
1960 #ifdef UNICODE
1961 WideCharToMultiByte(CP_ACP, WC_COMPOSITECHECK | WC_DEFAULTCHAR,
1962 pszSecret, -1, m_szProxySecret, MAX_SECRET_LENGTH, NULL, NULL);
1963 #else
1964 nx_strncpy(m_szProxySecret, pszSecret, MAX_SECRET_LENGTH);
1965 #endif
1966 }
1967 else
1968 {
1969 m_szProxySecret[0] = 0;
1970 }
1971 m_useProxy = true;
1972 }
1973
1974 /**
1975 * Setup proxy connection
1976 */
1977 UINT32 AgentConnection::setupProxyConnection()
1978 {
1979 NXCPMessage msg(m_nProtocolVersion);
1980 UINT32 dwRqId;
1981
1982 dwRqId = generateRequestId();
1983 msg.setCode(CMD_SETUP_PROXY_CONNECTION);
1984 msg.setId(dwRqId);
1985 msg.setField(VID_IP_ADDRESS, m_addr.getAddressV4()); // For compatibility with agents < 2.2.7
1986 msg.setField(VID_DESTINATION_ADDRESS, m_addr);
1987 msg.setField(VID_AGENT_PORT, m_wPort);
1988 if (sendMessage(&msg))
1989 return waitForRCC(dwRqId, 60000); // Wait 60 seconds for remote connect
1990 else
1991 return ERR_CONNECTION_BROKEN;
1992 }
1993
1994 /**
1995 * Enable trap receiving on connection
1996 */
1997 UINT32 AgentConnection::enableTraps()
1998 {
1999 NXCPMessage msg(m_nProtocolVersion);
2000 UINT32 dwRqId;
2001
2002 dwRqId = generateRequestId();
2003 msg.setCode(CMD_ENABLE_AGENT_TRAPS);
2004 msg.setId(dwRqId);
2005 if (sendMessage(&msg))
2006 return waitForRCC(dwRqId, m_dwCommandTimeout);
2007 else
2008 return ERR_CONNECTION_BROKEN;
2009 }
2010
2011 /**
2012 * Enable trap receiving on connection
2013 */
2014 UINT32 AgentConnection::enableFileUpdates()
2015 {
2016 NXCPMessage msg(m_nProtocolVersion);
2017 UINT32 dwRqId;
2018
2019 dwRqId = generateRequestId();
2020 msg.setCode(CMD_ENABLE_FILE_UPDATES);
2021 msg.setId(dwRqId);
2022 if (sendMessage(&msg))
2023 {
2024 return waitForRCC(dwRqId, m_dwCommandTimeout);
2025 }
2026 else
2027 return ERR_CONNECTION_BROKEN;
2028 }
2029
2030 /**
2031 * Take screenshot from remote system
2032 */
2033 UINT32 AgentConnection::takeScreenshot(const TCHAR *sessionName, BYTE **data, size_t *size)
2034 {
2035 NXCPMessage msg(m_nProtocolVersion);
2036 UINT32 dwRqId;
2037
2038 dwRqId = generateRequestId();
2039 msg.setCode(CMD_TAKE_SCREENSHOT);
2040 msg.setId(dwRqId);
2041 msg.setField(VID_NAME, sessionName);
2042 if (sendMessage(&msg))
2043 {
2044 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
2045 if (response != NULL)
2046 {
2047 UINT32 rcc = response->getFieldAsUInt32(VID_RCC);
2048 if (rcc == ERR_SUCCESS)
2049 {
2050 const BYTE *p = response->getBinaryFieldPtr(VID_FILE_DATA, size);
2051 if (p != NULL)
2052 {
2053 *data = (BYTE *)malloc(*size);
2054 memcpy(*data, p, *size);
2055 }
2056 else
2057 {
2058 *data = NULL;
2059 }
2060 }
2061 delete response;
2062 return rcc;
2063 }
2064 else
2065 {
2066 return ERR_REQUEST_TIMEOUT;
2067 }
2068 }
2069 else
2070 {
2071 return ERR_CONNECTION_BROKEN;
2072 }
2073 }
2074
2075 /**
2076 * Resolve hostname by IP address in local network
2077 */
2078 TCHAR *AgentConnection::getHostByAddr(const InetAddress& ipAddr, TCHAR *buffer, size_t bufLen)
2079 {
2080 NXCPMessage msg(m_nProtocolVersion);
2081 UINT32 dwRqId = generateRequestId();
2082 msg.setCode(CMD_GET_HOSTNAME_BY_IPADDR);
2083 msg.setId(dwRqId);
2084 msg.setField(VID_IP_ADDRESS, ipAddr);
2085 TCHAR *result = NULL;
2086 if (sendMessage(&msg))
2087 {
2088 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
2089 if (response != NULL)
2090 {
2091 if (response->getFieldAsUInt32(VID_RCC) == ERR_SUCCESS)
2092 {
2093 result = response->getFieldAsString(VID_NAME, buffer, bufLen);
2094 if ((result != NULL) && (*result == 0))
2095 {
2096 // Agents before 2.2.16 can return empty string instead of error if IP cannot be resolved
2097 if (buffer == NULL)
2098 MemFree(result);
2099 result = NULL;
2100 }
2101 }
2102 delete response;
2103 }
2104 }
2105 return result;
2106 }
2107
2108 /**
2109 * Send custom request to agent
2110 */
2111 NXCPMessage *AgentConnection::customRequest(NXCPMessage *pRequest, const TCHAR *recvFile, bool append,
2112 void (*downloadProgressCallback)(size_t, void *), void (*fileResendCallback)(NXCP_MESSAGE *, void *), void *cbArg)
2113 {
2114 UINT32 dwRqId, rcc;
2115 NXCPMessage *msg = NULL;
2116
2117 dwRqId = generateRequestId();
2118 pRequest->setId(dwRqId);
2119 if (recvFile != NULL)
2120 {
2121 rcc = prepareFileDownload(recvFile, dwRqId, append, downloadProgressCallback, fileResendCallback,cbArg);
2122 if (rcc != ERR_SUCCESS)
2123 {
2124 // Create fake response message
2125 msg = new NXCPMessage;
2126 msg->setCode(CMD_REQUEST_COMPLETED);
2127 msg->setId(dwRqId);
2128 msg->setField(VID_RCC, rcc);
2129 }
2130 }
2131
2132 if (msg == NULL)
2133 {
2134 if (sendMessage(pRequest))
2135 {
2136 msg = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
2137 if ((msg != NULL) && (recvFile != NULL))
2138 {
2139 if (msg->getFieldAsUInt32(VID_RCC) == ERR_SUCCESS)
2140 {
2141 if (ConditionWait(m_condFileDownload, 1800000)) // 30 min timeout
2142 {
2143 if (!m_fileDownloadSucceeded)
2144 {
2145 msg->setField(VID_RCC, ERR_IO_FAILURE);
2146 if (m_deleteFileOnDownloadFailure)
2147 _tremove(recvFile);
2148 }
2149 }
2150 else
2151 {
2152 msg->setField(VID_RCC, ERR_REQUEST_TIMEOUT);
2153 }
2154 }
2155 else
2156 {
2157 if (fileResendCallback != NULL)
2158 {
2159 _close(m_hCurrFile);
2160 m_hCurrFile = -1;
2161 _tremove(recvFile);
2162 }
2163 }
2164 }
2165
2166 }
2167 }
2168
2169 return msg;
2170 }
2171
2172 /**
2173 * Cancel file download
2174 */
2175 UINT32 AgentConnection::cancelFileDownload()
2176 {
2177 NXCPMessage msg(CMD_CANCEL_FILE_DOWNLOAD, generateRequestId(), getProtocolVersion());
2178 msg.setField(VID_REQUEST_ID, m_dwDownloadRequestId);
2179
2180 UINT32 rcc;
2181 if (sendMessage(&msg))
2182 {
2183 NXCPMessage *result = waitForMessage(CMD_REQUEST_COMPLETED, msg.getId(), m_dwCommandTimeout);
2184 if (result != NULL)
2185 {
2186 rcc = result->getFieldAsUInt32(VID_RCC);
2187 delete result;
2188 }
2189 else
2190 {
2191 rcc = ERR_REQUEST_TIMEOUT;
2192 }
2193 }
2194 else
2195 {
2196 rcc = ERR_CONNECTION_BROKEN;
2197 }
2198 return rcc;
2199 }
2200
2201 /**
2202 * Prepare for file download
2203 */
2204 UINT32 AgentConnection::prepareFileDownload(const TCHAR *fileName, UINT32 rqId, bool append, void (*downloadProgressCallback)(size_t, void *),
2205 void (* fileResendCallback)(NXCP_MESSAGE *, void *), void *cbArg)
2206 {
2207 if (fileResendCallback == NULL)
2208 {
2209 if (m_hCurrFile != -1)
2210 return ERR_RESOURCE_BUSY;
2211
2212 nx_strncpy(m_currentFileName, fileName, MAX_PATH);
2213 ConditionReset(m_condFileDownload);
2214 m_hCurrFile = _topen(fileName, (append ? 0 : (O_CREAT | O_TRUNC)) | O_RDWR | O_BINARY, S_IREAD | S_IWRITE);
2215 if (m_hCurrFile == -1)
2216 {
2217 DbgPrintf(4, _T("AgentConnection::PrepareFileDownload(): cannot open file %s (%s); append=%d rqId=%d"),
2218 fileName, _tcserror(errno), append, rqId);
2219 }
2220 else
2221 {
2222 if (append)
2223 _lseek(m_hCurrFile, 0, SEEK_END);
2224 }
2225
2226 m_dwDownloadRequestId = rqId;
2227 m_downloadProgressCallback = downloadProgressCallback;
2228 m_downloadProgressCallbackArg = cbArg;
2229
2230 m_sendToClientMessageCallback = NULL;
2231
2232 return (m_hCurrFile != -1) ? ERR_SUCCESS : ERR_FILE_OPEN_ERROR;
2233 }
2234 else
2235 {
2236 ConditionReset(m_condFileDownload);
2237
2238 m_dwDownloadRequestId = rqId;
2239 m_downloadProgressCallback = downloadProgressCallback;
2240 m_downloadProgressCallbackArg = cbArg;
2241
2242 m_sendToClientMessageCallback = fileResendCallback;
2243
2244 return ERR_SUCCESS;
2245 }
2246 }
2247
2248 /**
2249 * File upload completion handler
2250 */
2251 void AgentConnection::onFileDownload(bool success)
2252 {
2253 if (!success && m_deleteFileOnDownloadFailure)
2254 _tremove(m_currentFileName);
2255 m_fileDownloadSucceeded = success;
2256 ConditionSet(m_condFileDownload);
2257 }
2258
2259 /**
2260 * Enable trap receiving on connection
2261 */
2262 UINT32 AgentConnection::getPolicyInventory(AgentPolicyInfo **info)
2263 {
2264 NXCPMessage msg(m_nProtocolVersion);
2265 UINT32 dwRqId, rcc;
2266
2267 *info = NULL;
2268 dwRqId = generateRequestId();
2269 msg.setCode(CMD_GET_POLICY_INVENTORY);
2270 msg.setId(dwRqId);
2271 if (sendMessage(&msg))
2272 {
2273 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, dwRqId, m_dwCommandTimeout);
2274 if (response != NULL)
2275 {
2276 rcc = response->getFieldAsUInt32(VID_RCC);
2277 if (rcc == ERR_SUCCESS)
2278 *info = new AgentPolicyInfo(response);
2279 delete response;
2280 }
2281 else
2282 {
2283 rcc = ERR_REQUEST_TIMEOUT;
2284 }
2285 }
2286 else
2287 {
2288 rcc = ERR_CONNECTION_BROKEN;
2289 }
2290 return rcc;
2291 }
2292
2293 /**
2294 * Uninstall policy by GUID
2295 */
2296 UINT32 AgentConnection::uninstallPolicy(const uuid& guid)
2297 {
2298 UINT32 rqId, rcc;
2299 NXCPMessage msg(m_nProtocolVersion);
2300
2301 rqId = generateRequestId();
2302 msg.setId(rqId);
2303 msg.setCode(CMD_UNINSTALL_AGENT_POLICY);
2304 msg.setField(VID_GUID, guid);
2305 if (sendMessage(&msg))
2306 {
2307 rcc = waitForRCC(rqId, m_dwCommandTimeout);
2308 }
2309 else
2310 {
2311 rcc = ERR_CONNECTION_BROKEN;
2312 }
2313 return rcc;
2314 }
2315
2316 /**
2317 * Acquire encryption context
2318 */
2319 NXCPEncryptionContext *AgentConnection::acquireEncryptionContext()
2320 {
2321 lock();
2322 NXCPEncryptionContext *ctx = m_pCtx;
2323 if (ctx != NULL)
2324 ctx->incRefCount();
2325 unlock();
2326 return ctx;
2327 }
2328
2329 /**
2330 * Callback for processing collected data on separate thread
2331 */
2332 void AgentConnection::processCollectedDataCallback(NXCPMessage *msg)
2333 {
2334 NXCPMessage response(CMD_REQUEST_COMPLETED, msg->getId(), m_nProtocolVersion);
2335
2336 if (msg->getFieldAsBoolean(VID_BULK_RECONCILIATION))
2337 {
2338 // Check that only one bulk data processor is running
2339 if (InterlockedIncrement(&m_bulkDataProcessing) == 1)
2340 {
2341 response.setField(VID_RCC, processBulkCollectedData(msg, &response));
2342 }
2343 else
2344 {
2345 response.setField(VID_RCC, ERR_RESOURCE_BUSY);
2346 }
2347 InterlockedDecrement(&m_bulkDataProcessing);
2348 }
2349 else
2350 {
2351 UINT32 rcc = processCollectedData(msg);
2352 response.setField(VID_RCC, rcc);
2353 }
2354
2355 sendMessage(&response);
2356 delete msg;
2357 decInternalRefCount();
2358 }
2359
2360 /**
2361 * Process collected data information (for DCI with agent-side cache)
2362 */
2363 UINT32 AgentConnection::processCollectedData(NXCPMessage *msg)
2364 {
2365 return ERR_NOT_IMPLEMENTED;
2366 }
2367
2368 /**
2369 * Process collected data information in bulk mode (for DCI with agent-side cache)
2370 */
2371 UINT32 AgentConnection::processBulkCollectedData(NXCPMessage *request, NXCPMessage *response)
2372 {
2373 return ERR_NOT_IMPLEMENTED;
2374 }
2375
2376 /**
2377 * Setup TCP proxy
2378 */
2379 UINT32 AgentConnection::setupTcpProxy(const InetAddress& ipAddr, UINT16 port, UINT32 *channelId)
2380 {
2381 UINT32 requestId = generateRequestId();
2382 NXCPMessage msg(CMD_SETUP_TCP_PROXY, requestId, m_nProtocolVersion);
2383 msg.setField(VID_IP_ADDRESS, ipAddr);
2384 msg.setField(VID_PORT, port);
2385
2386 UINT32 rcc;
2387 if (sendMessage(&msg))
2388 {
2389 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, requestId, m_dwCommandTimeout);
2390 if (response != NULL)
2391 {
2392 rcc = response->getFieldAsUInt32(VID_RCC);
2393 if (rcc == ERR_SUCCESS)
2394 {
2395 *channelId = response->getFieldAsUInt32(VID_CHANNEL_ID);
2396 }
2397 delete response;
2398 }
2399 else
2400 {
2401 rcc = ERR_REQUEST_TIMEOUT;
2402 }
2403 }
2404 else
2405 {
2406 rcc = ERR_CONNECTION_BROKEN;
2407 }
2408 return rcc;
2409 }
2410
2411 /**
2412 * Close TCP proxy
2413 */
2414 UINT32 AgentConnection::closeTcpProxy(UINT32 channelId)
2415 {
2416 UINT32 requestId = generateRequestId();
2417 NXCPMessage msg(CMD_CLOSE_TCP_PROXY, requestId, m_nProtocolVersion);
2418 msg.setField(VID_CHANNEL_ID, channelId);
2419 UINT32 rcc;
2420 if (sendMessage(&msg))
2421 {
2422 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, requestId, m_dwCommandTimeout);
2423 if (response != NULL)
2424 {
2425 rcc = response->getFieldAsUInt32(VID_RCC);
2426 delete response;
2427 }
2428 else
2429 {
2430 rcc = ERR_REQUEST_TIMEOUT;
2431 }
2432 }
2433 else
2434 {
2435 rcc = ERR_CONNECTION_BROKEN;
2436 }
2437 return rcc;
2438 }
2439
2440 /**
2441 * Process data received from TCP proxy
2442 */
2443 void AgentConnection::processTcpProxyData(UINT32 channelId, const void *data, size_t size)
2444 {
2445 }
2446
2447 /**
2448 * Get file set information
2449 */
2450 UINT32 AgentConnection::getFileSetInfo(const StringList &fileSet, ObjectArray<RemoteFileInfo> **fileSetInfo)
2451 {
2452 *fileSetInfo = NULL;
2453 UINT32 requestId = generateRequestId();
2454 NXCPMessage msg(CMD_GET_FILE_SET_DETAILS, requestId, m_nProtocolVersion);
2455 fileSet.fillMessage(&msg, VID_ELEMENT_LIST_BASE, VID_NUM_ELEMENTS);
2456 UINT32 rcc;
2457 if (sendMessage(&msg))
2458 {
2459 NXCPMessage *response = waitForMessage(CMD_REQUEST_COMPLETED, requestId, m_dwCommandTimeout);
2460 if (response != NULL)
2461 {
2462 rcc = response->getFieldAsUInt32(VID_RCC);
2463 if (rcc == ERR_SUCCESS)
2464 {
2465 int count = response->getFieldAsInt32(VID_NUM_ELEMENTS);
2466 if (count == fileSet.size())
2467 {
2468 auto info = new ObjectArray<RemoteFileInfo>(count, 16, true);
2469 UINT32 fieldId = VID_ELEMENT_LIST_BASE;
2470 for(int i = 0; i < count; i++)
2471 {
2472 info->add(new RemoteFileInfo(response, fieldId, fileSet.get(i)));
2473 fieldId += 10;
2474 }
2475 *fileSetInfo = info;
2476 }
2477 else
2478 {
2479 rcc = ERR_INTERNAL_ERROR;
2480 }
2481 }
2482 delete response;
2483 }
2484 else
2485 {
2486 rcc = ERR_REQUEST_TIMEOUT;
2487 }
2488 }
2489 else
2490 {
2491 rcc = ERR_CONNECTION_BROKEN;
2492 }
2493 return rcc;
2494 }
2495
2496 /**
2497 * Create new agent parameter definition from NXCP message
2498 */
2499 AgentParameterDefinition::AgentParameterDefinition(NXCPMessage *msg, UINT32 baseId)
2500 {
2501 m_name = msg->getFieldAsString(baseId);
2502 m_description = msg->getFieldAsString(baseId + 1);
2503 m_dataType = (int)msg->getFieldAsUInt16(baseId + 2);
2504 }
2505
2506 /**
2507 * Create new agent parameter definition from another definition object
2508 */
2509 AgentParameterDefinition::AgentParameterDefinition(AgentParameterDefinition *src)
2510 {
2511 m_name = MemCopyString(src->m_name);
2512 m_description = MemCopyString(src->m_description);
2513 m_dataType = src->m_dataType;
2514 }
2515
2516 /**
2517 * Create new agent parameter definition from scratch
2518 */
2519 AgentParameterDefinition::AgentParameterDefinition(const TCHAR *name, const TCHAR *description, int dataType)
2520 {
2521 m_name = MemCopyString(name);
2522 m_description = MemCopyString(description);
2523 m_dataType = dataType;
2524 }
2525
2526 /**
2527 * Destructor for agent parameter definition
2528 */
2529 AgentParameterDefinition::~AgentParameterDefinition()
2530 {
2531 MemFree(m_name);
2532 MemFree(m_description);
2533 }
2534
2535 /**
2536 * Fill NXCP message
2537 */
2538 UINT32 AgentParameterDefinition::fillMessage(NXCPMessage *msg, UINT32 baseId)
2539 {
2540 msg->setField(baseId, m_name);
2541 msg->setField(baseId + 1, m_description);
2542 msg->setField(baseId + 2, (WORD)m_dataType);
2543 return 3;
2544 }
2545
2546 /**
2547 * Create new agent table definition from NXCP message
2548 */
2549 AgentTableDefinition::AgentTableDefinition(NXCPMessage *msg, UINT32 baseId)
2550 {
2551 m_name = msg->getFieldAsString(baseId);
2552 m_description = msg->getFieldAsString(baseId + 2);
2553
2554 TCHAR *instanceColumns = msg->getFieldAsString(baseId + 1);
2555 if (instanceColumns != NULL)
2556 {
2557 m_instanceColumns = new StringList(instanceColumns, _T("|"));
2558 free(instanceColumns);
2559 }
2560 else
2561 {
2562 m_instanceColumns = new StringList;
2563 }
2564
2565 m_columns = new ObjectArray<AgentTableColumnDefinition>(16, 16, true);
2566 }
2567
2568 /**
2569 * Create new agent table definition from another definition object
2570 */
2571 AgentTableDefinition::AgentTableDefinition(AgentTableDefinition *src)
2572 {
2573 m_name = MemCopyString(src->m_name);
2574 m_description = MemCopyString(src->m_description);
2575 m_instanceColumns = new StringList(src->m_instanceColumns);
2576 m_columns = new ObjectArray<AgentTableColumnDefinition>(16, 16, true);
2577 for(int i = 0; i < src->m_columns->size(); i++)
2578 {
2579 m_columns->add(new AgentTableColumnDefinition(src->m_columns->get(i)));
2580 }
2581 }
2582 /**
2583 * Destructor for agent table definition
2584 */
2585 AgentTableDefinition::~AgentTableDefinition()
2586 {
2587 MemFree(m_name);
2588 MemFree(m_description);
2589 delete m_instanceColumns;
2590 delete m_columns;
2591 }
2592
2593 /**
2594 * Fill NXCP message
2595 */
2596 UINT32 AgentTableDefinition::fillMessage(NXCPMessage *msg, UINT32 baseId)
2597 {
2598 msg->setField(baseId + 1, m_name);
2599 msg->setField(baseId + 2, m_description);
2600
2601 TCHAR *instanceColumns = m_instanceColumns->join(_T("|"));
2602 msg->setField(baseId + 3, instanceColumns);
2603 free(instanceColumns);
2604
2605 UINT32 varId = baseId + 4;
2606 for(int i = 0; i < m_columns->size(); i++)
2607 {
2608 msg->setField(varId++, m_columns->get(i)->m_name);
2609 msg->setField(varId++, (WORD)m_columns->get(i)->m_dataType);
2610 }
2611
2612 msg->setField(baseId, varId - baseId);
2613 return varId - baseId;
2614 }
2615
2616 /**
2617 * Create remote file info object
2618 */
2619 RemoteFileInfo::RemoteFileInfo(NXCPMessage *msg, UINT32 baseId, const TCHAR *name)
2620 {
2621 m_name = MemCopyString(name);
2622 m_status = msg->getFieldAsUInt32(baseId);
2623 if (m_status == ERR_SUCCESS)
2624 {
2625 m_size = msg->getFieldAsUInt64(baseId + 1);
2626 m_mtime = msg->getFieldAsTime(baseId + 2);
2627 msg->getFieldAsBinary(baseId + 3, m_hash, MD5_DIGEST_SIZE);
2628 }
2629 else
2630 {
2631 m_size = 0;
2632 m_mtime = 0;
2633 memset(m_hash, 0, MD5_DIGEST_SIZE);
2634 }
2635 }
2636
2637 /**
2638 * Destroy remote file info object
2639 */
2640 RemoteFileInfo::~RemoteFileInfo()
2641 {
2642 MemFree(m_name);
2643 }