otsdaq-utilities  3.09.00
ChatSupervisor.cc
1 #include "otsdaq-utilities/Chat/ChatSupervisor.h"
2 #include "otsdaq/CgiDataUtilities/CgiDataUtilities.h"
3 #include "otsdaq/Macros/CoutMacros.h"
4 #include "otsdaq/Macros/StringMacros.h"
5 #include "otsdaq/MessageFacility/MessageFacility.h"
6 #include "otsdaq/XmlUtilities/HttpXmlDocument.h"
7 
8 #include <xdaq/NamespaceURI.h>
9 
10 #include <fcntl.h>
11 #include <signal.h>
12 #include <sys/file.h>
13 #include <sys/wait.h>
14 #include <unistd.h>
15 
16 #include <fstream>
17 #include <iostream>
18 #include <sstream>
19 
20 using namespace ots;
21 
22 #undef __MF_SUBJECT__
23 #define __MF_SUBJECT__ "Chat"
24 
25 XDAQ_INSTANTIATOR_IMPL(ChatSupervisor)
26 
27 //==============================================================================
28 ChatSupervisor::ChatSupervisor(xdaq::ApplicationStub* stub) : CoreSupervisorBase(stub)
29 {
30  INIT_MF("." /*directory used is USER_DATA/LOG/.*/);
31 
32  ChatLastUpdateIndex = 1; // skip 0
33  slackDaemonPid_ = -1;
34 
35  // run ots_setup_slack.sh to enable OTS_EN_SLACK environment variable
36  enableSlackChat = (std::getenv("OTS_EN_SLACK") != nullptr &&
37  std::string(std::getenv("OTS_EN_SLACK")) == "1");
38  if(enableSlackChat)
39  {
40  const char* env = std::getenv("OTSDAQ_UTILITIES_DIR");
41  chatSupervisorToolsPath_ = env ? env : "";
42 
43  if(chatSupervisorToolsPath_.empty())
44  enableSlackChat = false;
45  if(chatSupervisorToolsPath_.back() != '/')
46  chatSupervisorToolsPath_ += '/';
47  chatSupervisorToolsPath_ += "tools/";
48 
49  const char* userData = std::getenv("USER_DATA");
50  if(userData)
51  slackInboxPath_ = std::string(userData) + "/ChatSlackInbox.txt";
52  else
53  slackInboxPath_ =
54  "/tmp/ots_slack_inbox_uid" + std::to_string(getuid()) + ".txt";
55 
56  // Require Slack configuration before enabling daemon/send-to-Slack behavior.
57  if(std::getenv("SLACK_BOT_TOKEN") == nullptr ||
58  std::getenv("SLACK_CHANNEL") == nullptr ||
59  std::getenv("SLACK_CHANNEL_ID") == nullptr)
60  enableSlackChat = false;
61 
62  __COUT__ << "ChatSupervisor: Slack chat "
63  << (enableSlackChat ? "enabled" : "disabled") << __E__;
64  __COUT__ << "ChatSupervisor path: " << chatSupervisorToolsPath_ << __E__;
65 
66  if(enableSlackChat)
67  startSlackDaemon();
68  }
69 }
70 
71 //==============================================================================
72 ChatSupervisor::~ChatSupervisor(void) { destroy(); }
73 
74 //==============================================================================
75 void ChatSupervisor::destroy(void) { stopSlackDaemon(); }
76 
77 //==============================================================================
78 void ChatSupervisor::defaultPage(xgi::Input* /* cgiIn */, xgi::Output* out)
79 {
80  out->getHTTPResponseHeader().addHeader("Access-Control-Allow-Origin", "*");
81  out->getHTTPResponseHeader().addHeader("Pragma", "no-cache");
82 
83  *out << "<!DOCTYPE HTML><html lang='en'><frameset col='100%' row='100%'><frame "
84  "src='/WebPath/html/Chat.html?urn="
85  << this->getApplicationDescriptor()->getLocalId() << "'></frameset></html>";
86 } //end defaultPage()
87 
88 //==============================================================================
91 void ChatSupervisor::forceSupervisorPropertyValues()
92 {
93  CorePropertySupervisorBase::setSupervisorProperty(
94  CorePropertySupervisorBase::SUPERVISOR_PROPERTIES.AutomatedRequestTypes,
95  "RefreshChat");
96 }
97 
98 //==============================================================================
102 void ChatSupervisor::request(const std::string& requestType,
103  cgicc::Cgicc& cgiIn,
104  HttpXmlDocument& xmlOut,
105  const WebUsers::RequestUserInfo& /*userInfo*/)
106 {
107  __COUTVS__(40, requestType);
108 
109  // Commands:
110  // RefreshChat
111  // RefreshUsers
112  // SendChat
113 
114  cleanupExpiredChats();
115 
116  if(requestType == "RefreshChat")
117  {
118  receiveFromSlack();
119 
120  std::string lastUpdateIndexString =
121  CgiDataUtilities::postData(cgiIn, "lastUpdateIndex");
122  std::string user = CgiDataUtilities::postData(cgiIn, "user");
123  uint64_t lastUpdateIndex;
124  sscanf(lastUpdateIndexString.c_str(), "%lu", &lastUpdateIndex);
125 
126  insertChatRefresh(&xmlOut, lastUpdateIndex, user);
127  }
128  else if(requestType == "RefreshUsers")
129  {
130  insertActiveUsers(&xmlOut);
131  }
132  else if(requestType == "SendChat")
133  {
134  std::string chat = CgiDataUtilities::postData(cgiIn, "chat");
135  std::string user = CgiDataUtilities::postData(cgiIn, "user");
136 
137  escapeChat(chat);
138 
139  newChat(chat, user);
140  }
141  else if(requestType == "PageUser")
142  {
143  std::string topage = CgiDataUtilities::postData(cgiIn, "topage");
144  unsigned int topageId = CgiDataUtilities::postDataAsInt(cgiIn, "topageId");
145  std::string user = CgiDataUtilities::postData(cgiIn, "user");
146 
147  __COUT__ << "Paging = " << topage.substr(0, 10)
148  << "... from user = " << user.substr(0, 10) << std::endl;
149 
150  __COUTV__(topageId);
151 
152  theRemoteWebUsers_.sendSystemMessage(topage,
153  user + " is paging you to come chat.");
154  }
155  else
156  {
157  __SUP_SS__ << "requestType Request, " << requestType
158  << ", not recognized by the Chat Editor Supervisor (was it intended "
159  "for another Supervisor?)."
160  << __E__;
161  __SUP_SS_THROW__;
162  }
163 
164 } // end request()
165 
166 //==============================================================================
170 void ChatSupervisor::escapeChat(std::string& /*chat*/)
171 {
172  // char reserved[] = {'"','\'','&','<','>'};
173  // std::string replace[] = {"&#34;","&#39;","&#38;","&#60;","&#62;"};
174  // for(uint64_t i=0;i<chat.size();++i)
175  // for(uint64_t j=0;j<chat.size();++j)
176  // if(chat[i] ==
177 } // end escapeChat()
178 
179 //==============================================================================
181 void ChatSupervisor::insertActiveUsers(HttpXmlDocument* xmlOut)
182 {
183  xmlOut->addTextElementToData("active_users", theRemoteWebUsers_.getActiveUserList());
184 } // end insertActiveUsers()
185 
186 //==============================================================================
193 void ChatSupervisor::insertChatRefresh(HttpXmlDocument* xmlOut,
194  uint64_t lastUpdateIndex,
195  const std::string& user)
196 {
197  newUser(user);
198 
199  if(!isLastUpdateIndexStale(lastUpdateIndex))
200  return; // if lastUpdateIndex is current, return nothing
201 
202  // return new update index, full chat user list, and new chats!
203 
204  char tempStr[50];
205  sprintf(tempStr, "%lu", ChatLastUpdateIndex);
206  xmlOut->addTextElementToData("last_update_index", tempStr);
207 
208  // get all users
209  xmlOut->addTextElementToData("chat_users", "");
210  for(uint64_t i = 0; i < ChatUsers_.size(); ++i)
211  xmlOut->addTextElementToParent("chat_user", ChatUsers_[i], "chat_users");
212 
213  //if lastUpdateIndex == 0, first request, so give give full history!
214 
215  // get all accounts
216  xmlOut->addTextElementToData("chat_history", "");
217  for(uint64_t i = 0; i < ChatHistoryEntry_.size(); ++i) // output oldest to new
218  {
219  __COUTT__ << "Chat[" << i << "]: " << ChatHistoryIndex_[i] << " vs "
220  << lastUpdateIndex << __E__;
221  if(isChatOld(ChatHistoryIndex_[i], lastUpdateIndex))
222  continue;
223 
224  xmlOut->addTextElementToParent(
225  "chat_entry", ChatHistoryEntry_[i], "chat_history");
226  xmlOut->addTextElementToParent(
227  "chat_author", ChatHistoryAuthor_[i], "chat_history");
228  sprintf(tempStr, "%lu", ChatHistoryTime_[i]);
229  xmlOut->addTextElementToParent("chat_time", tempStr, "chat_history");
230  }
231 } // end insertChatRefresh()
232 
233 //==============================================================================
236 void ChatSupervisor::newUser(const std::string& user)
237 {
238  for(uint64_t i = 0; i < ChatUsers_.size(); ++i)
239  if(ChatUsers_[i] == user)
240  {
241  ChatUsersTime_[i] = time(0); // update time
242  return; // do not add new if found
243  }
244 
245  __COUT__ << "New user: " << user << std::endl;
246  // add and increment
247  ChatUsers_.push_back(user);
248  ChatUsersTime_.push_back(time(0));
249  newChat(user + " joined the chat.",
250  "ots"); // add status message to chat, increment update
251 } // end newUser()
252 
253 //==============================================================================
256 void ChatSupervisor::newChat(const std::string& chat,
257  const std::string& user,
258  bool fromSlack)
259 {
260  ChatHistoryEntry_.push_back(chat);
261  ChatHistoryAuthor_.push_back(user);
262  ChatHistoryTime_.push_back(time(0));
263  ChatHistoryIndex_.push_back(incrementAndGetLastUpdate());
264  if(enableSlackChat && !fromSlack)
265  sendToSlack(user, chat);
266 }
267 
268 //==============================================================================
271 bool ChatSupervisor::isChatOld(uint64_t chatIndex, uint64_t last)
272 {
273  return (last - chatIndex < (uint64_t(1) << 62));
274 }
275 
276 //==============================================================================
278 bool ChatSupervisor::isLastUpdateIndexStale(uint64_t last)
279 {
280  return ChatLastUpdateIndex != last;
281 }
282 
283 //==============================================================================
285 uint64_t ChatSupervisor::incrementAndGetLastUpdate()
286 {
287  if(!++ChatLastUpdateIndex)
288  ++ChatLastUpdateIndex; // skip 0
289  return ChatLastUpdateIndex;
290 }
291 
292 //==============================================================================
295 void ChatSupervisor::cleanupExpiredChats()
296 {
297  for(uint64_t i = 0; i < ChatHistoryEntry_.size(); ++i)
298  if(i >= CHAT_HISTORY_MAX_ENTRIES ||
299  ChatHistoryTime_[i] + CHAT_HISTORY_EXPIRATION_TIME < time(0)) // expired
300  {
301  removeChatHistoryEntry(i);
302  --i; // rewind loop
303  }
304  else
305  break; // chronological order, so first encountered that is still valid exit
306  // loop
307 
308  for(uint64_t i = 0; i < ChatUsers_.size(); ++i)
309  if(ChatUsersTime_[i] + CHAT_HISTORY_EXPIRATION_TIME < time(0)) // expired
310  {
311  removeChatUserEntry(i);
312  --i; // rewind loop
313  }
314  else
315  break; // chronological order, so first encountered that is still valid exit
316  // loop
317 } // end cleanupExpiredChats()
318 
319 //==============================================================================
321 void ChatSupervisor::removeChatHistoryEntry(uint64_t i)
322 {
323  ChatHistoryEntry_.erase(ChatHistoryEntry_.begin() + i);
324  ChatHistoryTime_.erase(ChatHistoryTime_.begin() + i);
325  ChatHistoryAuthor_.erase(ChatHistoryAuthor_.begin() + i);
326  ChatHistoryIndex_.erase(ChatHistoryIndex_.begin() + i);
327 } // end removeChatHistoryEntry()
328 
329 //==============================================================================
331 void ChatSupervisor::removeChatUserEntry(uint64_t i)
332 {
333  newChat(ChatUsers_[i] + " left the chat.",
334  "ots"); // add status message to chat, increment update
335  ChatUsers_.erase(ChatUsers_.begin() + i);
336  ChatUsersTime_.erase(ChatUsersTime_.begin() + i);
337 } // end removeChatUserEntry()
338 
339 //==============================================================================
341 void ChatSupervisor::sendToSlack(const std::string& user, const std::string& message)
342 {
343  // URL-encode message and user so that UTF-8 emoji bytes, quotes, and any
344  // other shell-special characters survive the shell command line intact.
345  // SendSlackChat.py calls urllib.parse.unquote() on both args to decode.
346  std::string command = "python3 " + chatSupervisorToolsPath_ + "SendSlackChat.py " +
347  "--message " + StringMacros::encodeURIComponent(message) +
348  " --user " + StringMacros::encodeURIComponent(user);
349  __COUT__ << "Executing command: " << command << __E__;
350 
351  try
352  {
353  auto result = StringMacros::exec(command.c_str());
354 
355  if(!result.empty() && result.find("Error:") != std::string::npos)
356  __COUT__ << "Error from SendSlackChat.py: " << result << __E__;
357  else if(!result.empty())
358  __COUT__ << "Response from SendSlackChat.py: " << result << __E__;
359  }
360  catch(const std::exception& e)
361  {
362  __COUT__ << "Exception while executing command: " << e.what() << __E__;
363  }
364 } // end sendToSlack()
365 
366 //==============================================================================
369 void ChatSupervisor::startSlackDaemon()
370 {
371  if(slackDaemonPid_ > 0)
372  return;
373 
374  std::string script = chatSupervisorToolsPath_ + "ReceiveSlackChat.py";
375  const char* intervalEnv = std::getenv("OTS_SLACK_POLL_INTERVAL");
376  std::string interval = intervalEnv ? intervalEnv : "30";
377  __COUT__ << "Starting Slack receive daemon: " << script << " -> " << slackInboxPath_
378  << " (interval=" << interval << "s)" << __E__;
379 
380  pid_t pid = fork();
381  if(pid < 0)
382  {
383  __COUT__ << "Failed to fork Slack receive daemon" << __E__;
384  return;
385  }
386  if(pid == 0)
387  {
388  execlp("python3",
389  "python3",
390  script.c_str(),
391  "--output",
392  slackInboxPath_.c_str(),
393  "--interval",
394  interval.c_str(),
395  (char*)nullptr);
396  _exit(1);
397  }
398 
399  slackDaemonPid_ = pid;
400  __COUT__ << "Slack receive daemon started with PID " << slackDaemonPid_ << __E__;
401 } // end startSlackDaemon()
402 
403 //==============================================================================
406 void ChatSupervisor::stopSlackDaemon()
407 {
408  if(slackDaemonPid_ <= 0)
409  return;
410 
411  int status = 0;
412  pid_t probe_pid = waitpid(slackDaemonPid_, &status, WNOHANG);
413  if(probe_pid < 0)
414  {
415  __COUT__ << "Failed to probe Slack receive daemon PID " << slackDaemonPid_
416  << "; leaving daemon state and inbox files intact" << __E__;
417  return;
418  }
419  bool can_manage = probe_pid != slackDaemonPid_;
420 
421  if(can_manage)
422  {
423  __COUT__ << "Stopping Slack receive daemon PID " << slackDaemonPid_ << __E__;
424  kill(slackDaemonPid_, SIGTERM);
425  }
426  bool exited = !can_manage;
427  for(int i = 0; i < 50 && !exited; ++i) // ~5s total
428  {
429  pid_t r = waitpid(slackDaemonPid_, &status, WNOHANG);
430  if(r == slackDaemonPid_)
431  {
432  exited = true;
433  break;
434  }
435  if(r < 0)
436  {
437  exited = true; // already reaped / not our child
438  break;
439  }
440  usleep(100000);
441  }
442  if(!exited)
443  {
444  __COUT__ << "Slack receive daemon did not exit after SIGTERM; sending SIGKILL"
445  << __E__;
446  kill(slackDaemonPid_, SIGKILL);
447  waitpid(slackDaemonPid_, &status, 0);
448  }
449  else
450  {
451  (void)waitpid(
452  slackDaemonPid_, &status, WNOHANG); // ensure reaped if it exited quickly
453  }
454 
455  slackDaemonPid_ = -1;
456 
457  // Clean up the inbox file
458  unlink(slackInboxPath_.c_str());
459  unlink((slackInboxPath_ + ".lock").c_str());
460 } // end stopSlackDaemon()
461 
462 //==============================================================================
466 void ChatSupervisor::receiveFromSlack()
467 {
468  if(!enableSlackChat || slackInboxPath_.empty())
469  return;
470 
471  std::string lockPath = slackInboxPath_ + ".lock";
472  int lockFd = open(lockPath.c_str(), O_WRONLY | O_CREAT, 0644);
473  if(lockFd < 0)
474  {
475  __COUT__ << "receiveFromSlack: cannot open lock file " << lockPath << __E__;
476  return;
477  }
478 
479  if(flock(lockFd, LOCK_EX | LOCK_NB) != 0)
480  {
481  close(lockFd);
482  return; // daemon is writing, try next cycle
483  }
484 
485  std::vector<std::string> lines;
486  {
487  std::ifstream infile(slackInboxPath_);
488  if(infile.is_open())
489  {
490  std::string line;
491  while(std::getline(infile, line))
492  if(!line.empty())
493  lines.push_back(line);
494  }
495  }
496 
497  // Truncate the file after reading
498  if(!lines.empty())
499  {
500  __COUT__ << "receiveFromSlack: read " << lines.size() << " line(s) from "
501  << slackInboxPath_ << __E__;
502  std::ofstream clearFile(slackInboxPath_, std::ios::trunc);
503  }
504 
505  flock(lockFd, LOCK_UN);
506  close(lockFd);
507 
508  for(const auto& line : lines)
509  {
510  size_t firstTab = line.find('\t');
511  if(firstTab == std::string::npos)
512  {
513  __COUT__ << "receiveFromSlack: skipping malformed line (no tab): " << line
514  << __E__;
515  continue;
516  }
517 
518  std::string tag = line.substr(0, firstTab);
519  if(tag != "MSG")
520  {
521  __COUT__ << "receiveFromSlack: skipping unknown tag '" << tag << "'" << __E__;
522  continue;
523  }
524 
525  size_t secondTab = line.find('\t', firstTab + 1);
526  if(secondTab == std::string::npos)
527  {
528  __COUT__ << "receiveFromSlack: skipping malformed MSG line: " << line
529  << __E__;
530  continue;
531  }
532 
533  std::string user = line.substr(firstTab + 1, secondTab - firstTab - 1);
534  std::string message = line.substr(secondTab + 1);
535 
536  auto replaceAll =
537  [](std::string& s, const std::string& from, const std::string& to) {
538  size_t pos = 0;
539  while((pos = s.find(from, pos)) != std::string::npos)
540  {
541  s.replace(pos, from.size(), to);
542  pos += to.size();
543  }
544  };
545 
546  // Keep OTS chat percent-encoding expected by WebGUI/html/Chat.html convertForClient().
547  replaceAll(user, "&", "%26");
548  replaceAll(user, "<", "%3C");
549  replaceAll(user, ">", "%3E");
550  replaceAll(user, "\"", "%22");
551  replaceAll(user, "'", "%27");
552 
553  // Reverse ReceiveSlackChat.py escaping: "\\\\" -> "\\" and "\\n" -> newline marker.
554  std::string decoded;
555  decoded.reserve(message.size());
556  for(size_t i = 0; i < message.size(); ++i)
557  {
558  if(message[i] == '\\' && i + 1 < message.size())
559  {
560  if(message[i + 1] == '\\')
561  {
562  decoded.push_back('\\');
563  ++i;
564  continue;
565  }
566  if(message[i + 1] == 'n')
567  {
568  decoded += "%0A%0D";
569  ++i;
570  continue;
571  }
572  }
573  decoded.push_back(message[i]);
574  }
575  message.swap(decoded);
576 
577  replaceAll(message, "&", "%26");
578  replaceAll(message, "<", "%3C");
579  replaceAll(message, ">", "%3E");
580  replaceAll(message, "\"", "%22");
581  replaceAll(message, "'", "%27");
582  replaceAll(message, " ", "%20%20");
583 
584  __COUT__ << "receiveFromSlack: injecting message from user '" << user << "' ("
585  << message.size() << " bytes)" << __E__;
586  newChat(message, "[slack] " + user, /*fromSlack=*/true);
587  }
588 } // end receiveFromSlack()
ots::CgiDataUtilities::postData
static std::string postData(cgicc::Cgicc &cgi, const std::string &needle)
ots::ChatSupervisor::request
virtual void request(const std::string &requestType, cgicc::Cgicc &cgiIn, HttpXmlDocument &xmlOut, const WebUsers::RequestUserInfo &userInfo) override
end forceSupervisorPropertyValues()
Definition: ChatSupervisor.cc:102
ots::ChatSupervisor::forceSupervisorPropertyValues
virtual void forceSupervisorPropertyValues(void) override
override to force supervisor property values (and ignore user settings)
Definition: ChatSupervisor.cc:91
ots::XmlDocument::addTextElementToParent
xercesc::DOMElement * addTextElementToParent(const std::string &childName, const std::string &childText, xercesc::DOMElement *parent)
ots::INIT_MF
void INIT_MF(const char *name)
ots::StringMacros::exec
static std::string exec(const char *cmd)