plugins/OStatus/classes/HubSub.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2010, 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('GNUSOCIAL')) { exit(1); }
  21
  22 /**
  23  * WebSub (previously PuSH) feed subscription record
  24  * @package Hub
  25  * @author Brion Vibber <brion@status.net>
  26  */
  27 class HubSub extends Managed_DataObject
  28 {
  29     public $__table = 'hubsub';
  30
  31     public $hashkey; // sha1(topic . '|' . $callback); (topic, callback) key is too long for myisam in utf8
  32     public $topic;      // varchar(191)   not 255 because utf8mb4 takes more space
  33     public $callback;   // varchar(191)   not 255 because utf8mb4 takes more space
  34     public $secret;
  35     public $sub_start;
  36     public $sub_end;
  37     public $created;
  38     public $modified;
  39
  40     static function hashkey($topic, $callback)
  41     {
  42         return sha1($topic . '|' . $callback);
  43     }
  44
  45     public static function getByHashkey($topic, $callback)
  46     {
  47         return self::getKV('hashkey', self::hashkey($topic, $callback));
  48     }
  49
  50     public static function schemaDef()
  51     {
  52         return array(
  53             'fields' => array(
  54                 'hashkey' => array('type' => 'char', 'not null' => true, 'length' => 40, 'description' => 'HubSub hashkey'),
  55                 'topic' => array('type' => 'varchar', 'not null' => true, 'length' => 191, 'description' => 'HubSub topic'),
  56                 'callback' => array('type' => 'varchar', 'not null' => true, 'length' => 191, 'description' => 'HubSub callback'),
  57                 'secret' => array('type' => 'text', 'description' => 'HubSub stored secret'),
  58                 'sub_start' => array('type' => 'datetime', 'description' => 'subscription start'),
  59                 'sub_end' => array('type' => 'datetime', 'description' => 'subscription end'),
  60                 'errors' => array('type' => 'integer', 'not null' => true, 'default' => 0, 'description' => 'Queue handling error count, is reset on success.'),
  61                 'error_start' => array('type' => 'datetime', 'default' => null, 'description' => 'time of first error since latest success, should be null if no errors have been counted'),
  62                 'last_error' => array('type' => 'datetime', 'default' => null, 'description' => 'time of last failure, if ever'),
  63                 'last_error_msg' => array('type' => 'text', 'default' => null, 'description' => 'Last error _message_'),
  64                 'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
  65                 'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
  66             ),
  67             'primary key' => array('hashkey'),
  68             'indexes' => array(
  69                 'hubsub_callback_idx' => array('callback'),
  70                 'hubsub_topic_idx' => array('topic'),
  71             ),
  72         );
  73     }
  74
  75     function getErrors()
  76     {
  77         return intval($this->errors);
  78     }
  79
  80     // $msg is only set if $error_count is 0
  81     function setErrors($error_count, $msg=null)
  82     {
  83         assert(is_int($error_count));
  84         if (!is_int($error_count) || $error_count < 0) {
  85             common_log(LOG_ERR, 'HubSub->setErrors was given a bad value: '._ve($error_count));
  86             throw new ServerException('HubSub error count must be an integer higher or equal to 0.');
  87         }
  88
  89         $orig = clone($this);
  90         $now = common_sql_now();
  91
  92         if ($error_count === 1) {
  93             // Record when the errors started
  94             $this->error_start = $now;
  95         }
  96         if ($error_count > 0) {
  97             // Record this error's occurrence in time
  98             $this->last_error = $now;
  99             $this->last_error_msg = $msg;
 100         } else {
 101             $this->error_start = null;
 102             $this->last_error = null;
 103             $this->last_error_msg = null;
 104         }
 105
 106         $this->errors = $error_count;
 107         $this->update($orig);
 108     }
 109
 110     function resetErrors()
 111     {
 112         return $this->setErrors(0);
 113     }
 114
 115     function incrementErrors($msg=null)
 116     {
 117         return $this->setErrors($this->getErrors()+1, $msg);
 118     }
 119
 120     /**
 121      * Validates a requested lease length, sets length plus
 122      * subscription start & end dates.
 123      *
 124      * Does not save to database -- use before insert() or update().
 125      *
 126      * @param int $length in seconds
 127      */
 128     function setLease($length)
 129     {
 130         common_debug('WebSub hub got requested lease_seconds=='._ve($length));
 131         assert(is_int($length));
 132
 133         $min = 86400;   // 3600*24 (one day)
 134         $max = 86400 * 30;
 135
 136         if ($length == 0) {
 137             // We want to garbage collect dead subscriptions!
 138             $length = $max;
 139         } elseif( $length < $min) {
 140             $length = $min;
 141         } else if ($length > $max) {
 142             $length = $max;
 143         }
 144
 145         common_debug('WebSub hub after sanitation: lease_seconds=='._ve($length));
 146         $this->sub_start = common_sql_now();
 147         $this->sub_end = common_sql_date(time() + $length);
 148     }
 149
 150     function getLeaseTime()
 151     {
 152         if (empty($this->sub_start) || empty($this->sub_end)) {
 153             return null;
 154         }
 155         $length = strtotime($this->sub_end) - strtotime($this->sub_start);
 156         assert($length > 0);
 157         return $length;
 158     }
 159
 160     function getLeaseRemaining()
 161     {
 162         if (empty($this->sub_end)) {
 163             return null;
 164         }
 165         return strtotime($this->sub_end) - time();
 166     }
 167
 168     /**
 169      * Schedule a future verification ping to the subscriber.
 170      * If queues are disabled, will be immediate.
 171      *
 172      * @param string $mode 'subscribe' or 'unsubscribe'
 173      * @param string $token hub.verify_token value, if provided by client
 174      */
 175     function scheduleVerify($mode, $token=null, $retries=null)
 176     {
 177         if ($retries === null) {
 178             $retries = intval(common_config('ostatus', 'hub_retries'));
 179         }
 180         $data = array('sub' => clone($this),
 181                       'mode' => $mode,
 182                       'token' => $token,    // let's put it in there if remote uses PuSH <0.4
 183                       'retries' => $retries);
 184         $qm = QueueManager::get();
 185         $qm->enqueue($data, 'hubconf');
 186     }
 187
 188     public function getTopic()
 189     {
 190         return $this->topic;
 191     }
 192
 193     /**
 194      * Send a verification ping to subscriber, and if confirmed apply the changes.
 195      * This may create, update, or delete the database record.
 196      *
 197      * @param string $mode 'subscribe' or 'unsubscribe'
 198      * @param string $token hub.verify_token value, if provided by client
 199      * @throws ClientException on failure
 200      */
 201     function verify($mode, $token=null)
 202     {
 203         assert($mode == 'subscribe' || $mode == 'unsubscribe');
 204
 205         $challenge = common_random_hexstr(32);
 206         $params = array('hub.mode' => $mode,
 207                         'hub.topic' => $this->getTopic(),
 208                         'hub.challenge' => $challenge);
 209         if ($mode == 'subscribe') {
 210             $params['hub.lease_seconds'] = $this->getLeaseTime();
 211         }
 212         if ($token !== null) {  // TODO: deprecated in PuSH 0.4
 213             $params['hub.verify_token'] = $token;   // let's put it in there if remote uses PuSH <0.4
 214         }
 215
 216         // Any existing query string parameters must be preserved
 217         $url = $this->callback;
 218         if (strpos($url, '?') !== false) {
 219             $url .= '&';
 220         } else {
 221             $url .= '?';
 222         }
 223         $url .= http_build_query($params, '', '&');
 224
 225         $request = new HTTPClient();
 226         $response = $request->get($url);
 227         $status = $response->getStatus();
 228
 229         if ($status >= 200 && $status < 300) {
 230             common_log(LOG_INFO, "Verified {$mode} of {$this->callback}:{$this->getTopic()}");
 231         } else {
 232             // TRANS: Client exception. %s is a HTTP status code.
 233             throw new ClientException(sprintf(_m('Hub subscriber verification returned HTTP %s.'),$status));
 234         }
 235
 236         $old = HubSub::getByHashkey($this->getTopic(), $this->callback);
 237         if ($mode == 'subscribe') {
 238             if ($old instanceof HubSub) {
 239                 $this->update($old);
 240             } else {
 241                 $ok = $this->insert();
 242             }
 243         } else if ($mode == 'unsubscribe') {
 244             if ($old instanceof HubSub) {
 245                 $old->delete();
 246             } else {
 247                 // That's ok, we're already unsubscribed.
 248             }
 249         }
 250     }
 251
 252     // set the hashkey automagically on insert
 253     protected function onInsert()
 254     {
 255         $this->setHashkey();
 256         $this->created = common_sql_now();
 257         $this->modified = common_sql_now();
 258     }
 259
 260     // update the hashkey automagically if needed
 261     protected function onUpdateKeys(Managed_DataObject $orig)
 262     {
 263         if ($this->topic !== $orig->topic || $this->callback !== $orig->callback) {
 264             $this->setHashkey();
 265         }
 266     }
 267
 268     protected function setHashkey()
 269     {
 270         $this->hashkey = self::hashkey($this->topic, $this->callback);
 271     }
 272
 273     /**
 274      * Schedule delivery of a 'fat ping' to the subscriber's callback
 275      * endpoint. If queues are disabled, this will run immediately.
 276      *
 277      * @param string $atom well-formed Atom feed
 278      * @param int $retries optional count of retries if POST fails; defaults to hub_retries from config or 0 if unset
 279      */
 280     function distribute($atom, $retries=null)
 281     {
 282         if ($retries === null) {
 283             $retries = intval(common_config('ostatus', 'hub_retries'));
 284         }
 285
 286         $data = array('topic' => $this->getTopic(),
 287                       'callback' => $this->callback,
 288                       'atom' => $atom,
 289                       'retries' => $retries);
 290         common_log(LOG_INFO, sprintf('Queuing WebSub: %s to %s', _ve($data['topic']), _ve($data['callback'])));
 291         $qm = QueueManager::get();
 292         $qm->enqueue($data, 'hubout');
 293     }
 294
 295     /**
 296      * Queue up a large batch of pushes to multiple subscribers
 297      * for this same topic update.
 298      *
 299      * If queues are disabled, this will run immediately.
 300      *
 301      * @param string $atom well-formed Atom feed
 302      * @param array $pushCallbacks list of callback URLs
 303      */
 304     function bulkDistribute($atom, array $pushCallbacks)
 305     {
 306         if (empty($pushCallbacks)) {
 307             common_log(LOG_ERR, 'Callback list empty for bulkDistribute.');
 308             return false;
 309         }
 310         $data = array('atom' => $atom,
 311                       'topic' => $this->getTopic(),
 312                       'pushCallbacks' => $pushCallbacks);
 313         common_log(LOG_INFO, "Queuing WebSub batch: {$this->getTopic()} to ".count($pushCallbacks)." sites");
 314         $qm = QueueManager::get();
 315         $qm->enqueue($data, 'hubprep');
 316         return true;
 317     }
 318
 319     /**
 320      * @return  boolean     true/false for HTTP response
 321      * @throws  Exception   for lower-than-HTTP errors (such as NS lookup failure, connection refused...)
 322      */
 323     public static function pushAtom($topic, $callback, $atom, $secret=null, $hashalg='sha1')
 324     {
 325         $headers = array('Content-Type: application/atom+xml');
 326         if ($secret) {
 327             $hmac = hash_hmac($hashalg, $atom, $secret);
 328             $headers[] = "X-Hub-Signature: {$hashalg}={$hmac}";
 329         } else {
 330             $hmac = '(none)';
 331         }
 332         common_log(LOG_INFO, sprintf('About to WebSub-push feed to %s for %s, HMAC %s', _ve($callback), _ve($topic), _ve($hmac)));
 333
 334         $request = new HTTPClient();
 335         $request->setConfig(array('follow_redirects' => false));
 336         $request->setBody($atom);
 337
 338         // This will throw exception on non-HTTP failures
 339         try {
 340             $response = $request->post($callback, $headers);
 341         } catch (Exception $e) {
 342             common_debug(sprintf('WebSub callback to %s for %s failed with exception %s: %s', _ve($callback), _ve($topic), _ve(get_class($e)), _ve($e->getMessage())));
 343             throw $e;
 344         }
 345
 346         return $response->isOk();
 347     }
 348
 349     /**
 350      * Send a 'fat ping' to the subscriber's callback endpoint
 351      * containing the given Atom feed chunk.
 352      *
 353      * Determination of which feed items to send should be done at
 354      * a higher level; don't just shove in a complete feed!
 355      *
 356      * FIXME: Add 'failed' incremental count.
 357      *
 358      * @param string $atom well-formed Atom feed
 359      * @return  boolean     Whether the PuSH was accepted or not.
 360      * @throws Exception (HTTP or general)
 361      */
 362     function push($atom)
 363     {
 364         try {
 365             $success = self::pushAtom($this->getTopic(), $this->callback, $atom, $this->secret);
 366             if ($success) {
 367                 return true;
 368             } elseif ('https' === parse_url($this->callback, PHP_URL_SCHEME)) {
 369                 // Already HTTPS, no need to check remote http/https migration issues
 370                 return false;
 371             }
 372             // if pushAtom returned false and we didn't try an HTTPS endpoint,
 373             // let's try HTTPS too (assuming only http:// and https:// are used ;))
 374
 375         } catch (Exception $e) {
 376             if ('https' === parse_url($this->callback, PHP_URL_SCHEME)) {
 377                 // Already HTTPS, no need to check remote http/https migration issues
 378                 throw $e;
 379             }
 380         }
 381
 382
 383         // We failed the WebSub push, but it might be that the remote site has changed their configuration to HTTPS
 384         common_debug('WebSub HTTPSFIX: push failed, so we need to see if it can be the remote http->https migration issue.');
 385
 386         // XXX: DO NOT trust a Location header here, _especially_ from 'http' protocols,
 387         // but not 'https' either at least if we don't do proper CA verification. Trust that
 388         // the most common change here is simply switching 'http' to 'https' and we will
 389         // solve 99% of all of these issues for now. There should be a proper mechanism
 390         // if we want to change the callback URLs, preferrably just manual resubscriptions
 391         // from the remote side, combined with implemented WebSub subscription timeouts.
 392
 393         // Test if the feed callback for this node has already been migrated to HTTPS in our database
 394         // (otherwise we'd get collisions when inserting it further down)
 395         $httpscallback = preg_replace('/^http/', 'https', $this->callback, 1);
 396         $alreadyreplaced = self::getByHashKey($this->getTopic(), $httpscallback);
 397         if ($alreadyreplaced instanceof HubSub) {
 398             // Let's remove the old HTTP callback object.
 399             $this->delete();
 400
 401             // XXX: I think this means we might lose a message or two when
 402             //      remote side migrates to HTTPS because we only try _once_
 403             //      for _one_ WebSub push. The rest of the posts already
 404             //      stored in our queue (if any) will not find a HubSub
 405             //      object. This could maybe be fixed by handling migration
 406             //      in HubOutQueueHandler while handling the item there.
 407             common_debug('WebSub HTTPSFIX: Pushing Atom to HTTPS callback instead of HTTP, because of switch to HTTPS since enrolled in queue.');
 408             return self::pushAtom($this->getTopic(), $httpscallback, $atom, $this->secret);
 409         }
 410
 411         common_debug('WebSub HTTPSFIX: callback to '._ve($this->callback).' for '._ve($this->getTopic()).' trying HTTPS callback: '._ve($httpscallback));
 412         $success = self::pushAtom($this->getTopic(), $httpscallback, $atom, $this->secret);
 413         if ($success) {
 414             // Yay, we made a successful push to https://, let's remember this in the future!
 415             $orig = clone($this);
 416             $this->callback = $httpscallback;
 417             // NOTE: hashkey will be set in $this->onUpdateKeys($orig) through updateWithKeys
 418             $this->updateWithKeys($orig);
 419             return true;
 420         }
 421
 422         // If there have been any exceptions thrown before, they're handled
 423         // higher up. This function's return value is just whether the WebSub
 424         // push was accepted or not.
 425         return $success;
 426     }
 427 }