lib/xmppmanager.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2008, 2009, StatusNet, Inc.
   5  *
   6  * This program is free software: you can redistribute it and/or modify
   7  * it under the terms of the GNU Affero 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 Affero General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Affero General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }
  21
  22 /**
  23  * XMPP background connection manager for XMPP-using queue handlers,
  24  * allowing them to send outgoing messages on the right connection.
  25  *
  26  * Input is handled during socket select loop, keepalive pings during idle.
  27  * Any incoming messages will be forwarded to the main XmppDaemon process,
  28  * which handles direct user interaction.
  29  *
  30  * In a multi-site queuedaemon.php run, one connection will be instantiated
  31  * for each site being handled by the current process that has XMPP enabled.
  32  */
  33 class XmppManager extends IoManager
  34 {
  35     protected $site = null;
  36     protected $pingid = 0;
  37     protected $lastping = null;
  38     protected $conn = null;
  39
  40     static protected $singletons = array();
  41
  42     const PING_INTERVAL = 120;
  43
  44     /**
  45      * Fetch the singleton XmppManager for the current site.
  46      * @return mixed XmppManager, or false if unneeded
  47      */
  48     public static function get()
  49     {
  50         if (common_config('xmpp', 'enabled')) {
  51             $site = StatusNet::currentSite();
  52             if (empty(self::$singletons[$site])) {
  53                 self::$singletons[$site] = new XmppManager();
  54             }
  55             return self::$singletons[$site];
  56         } else {
  57             return false;
  58         }
  59     }
  60
  61     /**
  62      * Tell the i/o master we need one instance for each supporting site
  63      * being handled in this process.
  64      */
  65     public static function multiSite()
  66     {
  67         return IoManager::INSTANCE_PER_SITE;
  68     }
  69
  70     function __construct()
  71     {
  72         $this->site = StatusNet::currentSite();
  73         $this->resource = common_config('xmpp', 'resource') . 'daemon';
  74     }
  75
  76     /**
  77      * Initialize connection to server.
  78      * @return boolean true on success
  79      */
  80     public function start($master)
  81     {
  82         parent::start($master);
  83         $this->switchSite();
  84
  85         require_once INSTALLDIR . "/lib/jabber.php";
  86
  87         # Low priority; we don't want to receive messages
  88
  89         common_log(LOG_INFO, "INITIALIZE");
  90         $this->conn = jabber_connect($this->resource);
  91
  92         if (empty($this->conn)) {
  93             common_log(LOG_ERR, "Couldn't connect to server.");
  94             return false;
  95         }
  96
  97         $this->log(LOG_DEBUG, "Initializing stanza handlers.");
  98
  99         $this->conn->addEventHandler('message', 'handle_message', $this);
 100         $this->conn->addEventHandler('presence', 'handle_presence', $this);
 101         $this->conn->addEventHandler('reconnect', 'handle_reconnect', $this);
 102
 103         $this->conn->setReconnectTimeout(600);
 104         // @todo Needs i18n?
 105         jabber_send_presence("Send me a message to post a notice", 'available', null, 'available', 100);
 106
 107         return !is_null($this->conn);
 108     }
 109
 110     /**
 111      * Message pump is triggered on socket input, so we only need an idle()
 112      * call often enough to trigger our outgoing pings.
 113      */
 114     function timeout()
 115     {
 116         return self::PING_INTERVAL;
 117     }
 118
 119     /**
 120      * Lists the XMPP connection socket to allow i/o master to wake
 121      * when input comes in here as well as from the queue source.
 122      *
 123      * @return array of resources
 124      */
 125     public function getSockets()
 126     {
 127         if ($this->conn) {
 128             return array($this->conn->getSocket());
 129         } else {
 130             return array();
 131         }
 132     }
 133
 134     /**
 135      * Process XMPP events that have come in over the wire.
 136      * Side effects: may switch site configuration
 137      * @fixme may kill process on XMPP error
 138      * @param resource $socket
 139      */
 140     public function handleInput($socket)
 141     {
 142         $this->switchSite();
 143
 144         # Process the queue for as long as needed
 145         try {
 146             if ($this->conn) {
 147                 assert($socket === $this->conn->getSocket());
 148
 149                 common_log(LOG_DEBUG, "Servicing the XMPP queue.");
 150                 $this->stats('xmpp_process');
 151                 $this->conn->processTime(0);
 152             }
 153         } catch (XMPPHP_Exception $e) {
 154             common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 155             die($e->getMessage());
 156         }
 157     }
 158
 159     /**
 160      * Idle processing for io manager's execution loop.
 161      * Send keepalive pings to server.
 162      *
 163      * Side effect: kills process on exception from XMPP library.
 164      *
 165      * @fixme non-dying error handling
 166      */
 167     public function idle($timeout=0)
 168     {
 169         if ($this->conn) {
 170             $now = time();
 171             if (empty($this->lastping) || $now - $this->lastping > self::PING_INTERVAL) {
 172                 $this->switchSite();
 173                 try {
 174                     $this->sendPing();
 175                     $this->lastping = $now;
 176                 } catch (XMPPHP_Exception $e) {
 177                     common_log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
 178                     die($e->getMessage());
 179                 }
 180             }
 181         }
 182     }
 183
 184     /**
 185      * For queue handlers to pass us a message to push out,
 186      * if we're active.
 187      *
 188      * @fixme should this be blocking etc?
 189      *
 190      * @param string $msg XML stanza to send
 191      * @return boolean success
 192      */
 193     public function send($msg)
 194     {
 195         if ($this->conn && !$this->conn->isDisconnected()) {
 196             $bytes = $this->conn->send($msg);
 197             if ($bytes > 0) {
 198                 $this->conn->processTime(0);
 199                 return true;
 200             } else {
 201                 return false;
 202             }
 203         } else {
 204             // Can't send right now...
 205             return false;
 206         }
 207     }
 208
 209     /**
 210      * Send a keepalive ping to the XMPP server.
 211      */
 212     protected function sendPing()
 213     {
 214         $jid = jabber_daemon_address().'/'.$this->resource;
 215         $server = common_config('xmpp', 'server');
 216
 217         if (!isset($this->pingid)) {
 218             $this->pingid = 0;
 219         } else {
 220             $this->pingid++;
 221         }
 222
 223         common_log(LOG_DEBUG, "Sending ping #{$this->pingid}");
 224
 225         $this->conn->send("<iq from='{$jid}' to='{$server}' id='ping_{$this->pingid}' type='get'><ping xmlns='urn:xmpp:ping'/></iq>");
 226     }
 227
 228     /**
 229      * Callback for Jabber reconnect event
 230      * @param $pl
 231      */
 232     function handle_reconnect(&$pl)
 233     {
 234         common_log(LOG_NOTICE, 'XMPP reconnected');
 235
 236         $this->conn->processUntil('session_start');
 237         $this->conn->presence(null, 'available', null, 'available', 100);
 238     }
 239
 240
 241     function get_user($from)
 242     {
 243         $user = User::staticGet('jabber', jabber_normalize_jid($from));
 244         return $user;
 245     }
 246
 247     /**
 248      * XMPP callback for handling message input...
 249      * @param array $pl XMPP payload
 250      */
 251     function handle_message(&$pl)
 252     {
 253         $from = jabber_normalize_jid($pl['from']);
 254
 255         if ($pl['type'] != 'chat') {
 256             $this->log(LOG_WARNING, "Ignoring message of type ".$pl['type']." from $from: " . $pl['xml']->toString());
 257             return;
 258         }
 259
 260         if (mb_strlen($pl['body']) == 0) {
 261             $this->log(LOG_WARNING, "Ignoring message with empty body from $from: "  . $pl['xml']->toString());
 262             return;
 263         }
 264
 265         // Forwarded from another daemon for us to handle; this shouldn't
 266         // happen any more but we might get some legacy items.
 267         if ($this->is_self($from)) {
 268             $this->log(LOG_INFO, "Got forwarded notice from self ($from).");
 269             $from = $this->get_ofrom($pl);
 270             $this->log(LOG_INFO, "Originally sent by $from.");
 271             if (is_null($from) || $this->is_self($from)) {
 272                 $this->log(LOG_INFO, "Ignoring notice originally sent by $from.");
 273                 return;
 274             }
 275         }
 276
 277         $user = $this->get_user($from);
 278
 279         // For common_current_user to work
 280         global $_cur;
 281         $_cur = $user;
 282
 283         if (!$user) {
 284             // TRANS: %s is the URL to the StatusNet site's Instant Messaging settings.
 285             $this->from_site($from, sprintf(_('Unknown user. Go to %s ' .
 286                              'to add your address to your account'),common_local_url('imsettings')));
 287             $this->log(LOG_WARNING, 'Message from unknown user ' . $from);
 288             return;
 289         }
 290         if ($this->handle_command($user, $pl['body'])) {
 291             $this->log(LOG_INFO, "Command message by $from handled.");
 292             return;
 293         } else if ($this->is_autoreply($pl['body'])) {
 294             $this->log(LOG_INFO, 'Ignoring auto reply from ' . $from);
 295             return;
 296         } else if ($this->is_otr($pl['body'])) {
 297             $this->log(LOG_INFO, 'Ignoring OTR from ' . $from);
 298             return;
 299         } else {
 300
 301             $this->log(LOG_INFO, 'Posting a notice from ' . $user->nickname);
 302
 303             $this->add_notice($user, $pl);
 304         }
 305
 306         $user->free();
 307         unset($user);
 308         unset($_cur);
 309
 310         unset($pl['xml']);
 311         $pl['xml'] = null;
 312
 313         $pl = null;
 314         unset($pl);
 315     }
 316
 317     function is_self($from)
 318     {
 319         return preg_match('/^'.strtolower(jabber_daemon_address()).'/', strtolower($from));
 320     }
 321
 322     function get_ofrom($pl)
 323     {
 324         $xml = $pl['xml'];
 325         $addresses = $xml->sub('addresses');
 326         if (!$addresses) {
 327             $this->log(LOG_WARNING, 'Forwarded message without addresses');
 328             return null;
 329         }
 330         $address = $addresses->sub('address');
 331         if (!$address) {
 332             $this->log(LOG_WARNING, 'Forwarded message without address');
 333             return null;
 334         }
 335         if (!array_key_exists('type', $address->attrs)) {
 336             $this->log(LOG_WARNING, 'No type for forwarded message');
 337             return null;
 338         }
 339         $type = $address->attrs['type'];
 340         if ($type != 'ofrom') {
 341             $this->log(LOG_WARNING, 'Type of forwarded message is not ofrom');
 342             return null;
 343         }
 344         if (!array_key_exists('jid', $address->attrs)) {
 345             $this->log(LOG_WARNING, 'No jid for forwarded message');
 346             return null;
 347         }
 348         $jid = $address->attrs['jid'];
 349         if (!$jid) {
 350             $this->log(LOG_WARNING, 'Could not get jid from address');
 351             return null;
 352         }
 353         $this->log(LOG_DEBUG, 'Got message forwarded from jid ' . $jid);
 354         return $jid;
 355     }
 356
 357     function is_autoreply($txt)
 358     {
 359         if (preg_match('/[\[\(]?[Aa]uto[-\s]?[Rr]e(ply|sponse)[\]\)]/', $txt)) {
 360             return true;
 361         } else if (preg_match('/^System: Message wasn\'t delivered. Offline storage size was exceeded.$/', $txt)) {
 362             return true;
 363         } else {
 364             return false;
 365         }
 366     }
 367
 368     function is_otr($txt)
 369     {
 370         if (preg_match('/^\?OTR/', $txt)) {
 371             return true;
 372         } else {
 373             return false;
 374         }
 375     }
 376
 377     function from_site($address, $msg)
 378     {
 379         $text = '['.common_config('site', 'name') . '] ' . $msg;
 380         jabber_send_message($address, $text);
 381     }
 382
 383     function handle_command($user, $body)
 384     {
 385         $inter = new CommandInterpreter();
 386         $cmd = $inter->handle_command($user, $body);
 387         if ($cmd) {
 388             $chan = new XMPPChannel($this->conn);
 389             $cmd->execute($chan);
 390             return true;
 391         } else {
 392             return false;
 393         }
 394     }
 395
 396     function add_notice(&$user, &$pl)
 397     {
 398         $body = trim($pl['body']);
 399         $content_shortened = common_shorten_links($body);
 400         if (Notice::contentTooLong($content_shortened)) {
 401           $from = jabber_normalize_jid($pl['from']);
 402           // TRANS: Response to XMPP source when it sent too long a message.
 403           // TRANS: %1$d the maximum number of allowed characters (used for plural), %2$d is the sent number.
 404           $this->from_site($from, sprintf(_m('Message too long. Maximum is %1$d character, you sent %2$d.',
 405                                              'Message too long. Maximum is %1$d characters, you sent %2$d.',
 406                                              Notice::maxContent()),
 407                                           Notice::maxContent(),
 408                                           mb_strlen($content_shortened)));
 409           return;
 410         }
 411
 412         try {
 413             $notice = Notice::saveNew($user->id, $content_shortened, 'xmpp');
 414         } catch (Exception $e) {
 415             $this->log(LOG_ERR, $e->getMessage());
 416             $this->from_site($user->jabber, $e->getMessage());
 417             return;
 418         }
 419
 420         common_broadcast_notice($notice);
 421         $this->log(LOG_INFO,
 422                    'Added notice ' . $notice->id . ' from user ' . $user->nickname);
 423         $notice->free();
 424         unset($notice);
 425     }
 426
 427     function handle_presence(&$pl)
 428     {
 429         $from = jabber_normalize_jid($pl['from']);
 430         switch ($pl['type']) {
 431          case 'subscribe':
 432             # We let anyone subscribe
 433             $this->subscribed($from);
 434             $this->log(LOG_INFO,
 435                        'Accepted subscription from ' . $from);
 436             break;
 437          case 'subscribed':
 438          case 'unsubscribed':
 439          case 'unsubscribe':
 440             $this->log(LOG_INFO,
 441                        'Ignoring  "' . $pl['type'] . '" from ' . $from);
 442             break;
 443          default:
 444             if (!$pl['type']) {
 445                 $user = User::staticGet('jabber', $from);
 446                 if (!$user) {
 447                     $this->log(LOG_WARNING, 'Presence from unknown user ' . $from);
 448                     return;
 449                 }
 450                 if ($user->updatefrompresence) {
 451                     $this->log(LOG_INFO, 'Updating ' . $user->nickname .
 452                                ' status from presence.');
 453                     $this->add_notice($user, $pl);
 454                 }
 455                 $user->free();
 456                 unset($user);
 457             }
 458             break;
 459         }
 460         unset($pl['xml']);
 461         $pl['xml'] = null;
 462
 463         $pl = null;
 464         unset($pl);
 465     }
 466
 467     function log($level, $msg)
 468     {
 469         $text = 'XMPPDaemon('.$this->resource.'): '.$msg;
 470         common_log($level, $text);
 471     }
 472
 473     function subscribed($to)
 474     {
 475         jabber_special_presence('subscribed', $to);
 476     }
 477
 478     /**
 479      * Make sure we're on the right site configuration
 480      */
 481     protected function switchSite()
 482     {
 483         if ($this->site != StatusNet::currentSite()) {
 484             common_log(LOG_DEBUG, __METHOD__ . ": switching to site $this->site");
 485             $this->stats('switch');
 486             StatusNet::switchSite($this->site);
 487         }
 488     }
 489 }