plugins/OStatus/classes/FeedSub.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2009-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 /**
  21  * @package OStatusPlugin
  22  * @maintainer Brion Vibber <brion@status.net>
  23  */
  24
  25 /*
  26 PuSH subscription flow:
  27
  28     $profile->subscribe()
  29         generate random verification token
  30             save to verify_token
  31         sends a sub request to the hub...
  32
  33     main/push/callback
  34         hub sends confirmation back to us via GET
  35         We verify the request, then echo back the challenge.
  36         On our end, we save the time we subscribed and the lease expiration
  37
  38     main/push/callback
  39         hub sends us updates via POST
  40
  41 */
  42 class FeedDBException extends FeedSubException
  43 {
  44     public $obj;
  45
  46     function __construct($obj)
  47     {
  48         parent::__construct('Database insert failure');
  49         $this->obj = $obj;
  50     }
  51 }
  52
  53 /**
  54  * FeedSub handles low-level PubHubSubbub (PuSH) subscriptions.
  55  * Higher-level behavior building OStatus stuff on top is handled
  56  * under Ostatus_profile.
  57  */
  58 class FeedSub extends Memcached_DataObject
  59 {
  60     public $__table = 'feedsub';
  61
  62     public $id;
  63     public $uri;
  64
  65     // PuSH subscription data
  66     public $huburi;
  67     public $secret;
  68     public $verify_token;
  69     public $sub_state; // subscribe, active, unsubscribe, inactive
  70     public $sub_start;
  71     public $sub_end;
  72     public $last_update;
  73
  74     public $created;
  75     public $modified;
  76
  77     public /*static*/ function staticGet($k, $v=null)
  78     {
  79         return parent::staticGet(__CLASS__, $k, $v);
  80     }
  81
  82     /**
  83      * return table definition for DB_DataObject
  84      *
  85      * DB_DataObject needs to know something about the table to manipulate
  86      * instances. This method provides all the DB_DataObject needs to know.
  87      *
  88      * @return array array of column definitions
  89      */
  90     function table()
  91     {
  92         return array('id' => DB_DATAOBJECT_INT + DB_DATAOBJECT_NOTNULL,
  93                      'uri' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
  94                      'huburi' =>  DB_DATAOBJECT_STR,
  95                      'secret' => DB_DATAOBJECT_STR,
  96                      'verify_token' => DB_DATAOBJECT_STR,
  97                      'sub_state' => DB_DATAOBJECT_STR + DB_DATAOBJECT_NOTNULL,
  98                      'sub_start' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
  99                      'sub_end' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
 100                      'last_update' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME,
 101                      'created' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL,
 102                      'modified' => DB_DATAOBJECT_STR + DB_DATAOBJECT_DATE + DB_DATAOBJECT_TIME + DB_DATAOBJECT_NOTNULL);
 103     }
 104
 105     static function schemaDef()
 106     {
 107         return array(new ColumnDef('id', 'integer',
 108                                    /*size*/ null,
 109                                    /*nullable*/ false,
 110                                    /*key*/ 'PRI',
 111                                    /*default*/ null,
 112                                    /*extra*/ null,
 113                                    /*auto_increment*/ true),
 114                      new ColumnDef('uri', 'varchar',
 115                                    255, false, 'UNI'),
 116                      new ColumnDef('huburi', 'text',
 117                                    null, true),
 118                      new ColumnDef('verify_token', 'text',
 119                                    null, true),
 120                      new ColumnDef('secret', 'text',
 121                                    null, true),
 122                      new ColumnDef('sub_state', "enum('subscribe','active','unsubscribe','inactive')",
 123                                    null, false),
 124                      new ColumnDef('sub_start', 'datetime',
 125                                    null, true),
 126                      new ColumnDef('sub_end', 'datetime',
 127                                    null, true),
 128                      new ColumnDef('last_update', 'datetime',
 129                                    null, false),
 130                      new ColumnDef('created', 'datetime',
 131                                    null, false),
 132                      new ColumnDef('modified', 'datetime',
 133                                    null, false));
 134     }
 135
 136     /**
 137      * return key definitions for DB_DataObject
 138      *
 139      * DB_DataObject needs to know about keys that the table has; this function
 140      * defines them.
 141      *
 142      * @return array key definitions
 143      */
 144     function keys()
 145     {
 146         return array_keys($this->keyTypes());
 147     }
 148
 149     /**
 150      * return key definitions for Memcached_DataObject
 151      *
 152      * Our caching system uses the same key definitions, but uses a different
 153      * method to get them.
 154      *
 155      * @return array key definitions
 156      */
 157     function keyTypes()
 158     {
 159         return array('id' => 'K', 'uri' => 'U');
 160     }
 161
 162     function sequenceKey()
 163     {
 164         return array('id', true, false);
 165     }
 166
 167     /**
 168      * Fetch the StatusNet-side profile for this feed
 169      * @return Profile
 170      */
 171     public function localProfile()
 172     {
 173         if ($this->profile_id) {
 174             return Profile::staticGet('id', $this->profile_id);
 175         }
 176         return null;
 177     }
 178
 179     /**
 180      * Fetch the StatusNet-side profile for this feed
 181      * @return Profile
 182      */
 183     public function localGroup()
 184     {
 185         if ($this->group_id) {
 186             return User_group::staticGet('id', $this->group_id);
 187         }
 188         return null;
 189     }
 190
 191     /**
 192      * @param string $feeduri
 193      * @return FeedSub
 194      * @throws FeedSubException if feed is invalid or lacks PuSH setup
 195      */
 196     public static function ensureFeed($feeduri)
 197     {
 198         $current = self::staticGet('uri', $feeduri);
 199         if ($current) {
 200             return $current;
 201         }
 202
 203         $discover = new FeedDiscovery();
 204         $discover->discoverFromFeedURL($feeduri);
 205
 206         $huburi = $discover->getHubLink();
 207         if (!$huburi && !common_config('feedsub', 'fallback_hub')) {
 208             throw new FeedSubNoHubException();
 209         }
 210
 211         $feedsub = new FeedSub();
 212         $feedsub->uri = $feeduri;
 213         $feedsub->huburi = $huburi;
 214         $feedsub->sub_state = 'inactive';
 215
 216         $feedsub->created = common_sql_now();
 217         $feedsub->modified = common_sql_now();
 218
 219         $result = $feedsub->insert();
 220         if (empty($result)) {
 221             throw new FeedDBException($feedsub);
 222         }
 223
 224         return $feedsub;
 225     }
 226
 227     /**
 228      * Send a subscription request to the hub for this feed.
 229      * The hub will later send us a confirmation POST to /main/push/callback.
 230      *
 231      * @return bool true on success, false on failure
 232      * @throws ServerException if feed state is not valid
 233      */
 234     public function subscribe($mode='subscribe')
 235     {
 236         if ($this->sub_state && $this->sub_state != 'inactive') {
 237             common_log(LOG_WARNING, "Attempting to (re)start PuSH subscription to $this->uri in unexpected state $this->sub_state");
 238         }
 239         if (empty($this->huburi)) {
 240             if (common_config('feedsub', 'fallback_hub')) {
 241                 // No native hub on this feed?
 242                 // Use our fallback hub, which handles polling on our behalf.
 243             } else if (common_config('feedsub', 'nohub')) {
 244                 // Fake it! We're just testing remote feeds w/o hubs.
 245                 // We'll never actually get updates in this mode.
 246                 return true;
 247             } else {
 248                 throw new ServerException(_m('Attempting to start PuSH subscription for feed with no hub.'));
 249             }
 250         }
 251
 252         return $this->doSubscribe('subscribe');
 253     }
 254
 255     /**
 256      * Send a PuSH unsubscription request to the hub for this feed.
 257      * The hub will later send us a confirmation POST to /main/push/callback.
 258      * Warning: this will cancel the subscription even if someone else in
 259      * the system is using it. Most callers will want garbageCollect() instead,
 260      * which confirms there's no uses left.
 261      *
 262      * @return bool true on success, false on failure
 263      * @throws ServerException if feed state is not valid
 264      */
 265     public function unsubscribe() {
 266         if ($this->sub_state != 'active') {
 267             common_log(LOG_WARNING, "Attempting to (re)end PuSH subscription to $this->uri in unexpected state $this->sub_state");
 268         }
 269         if (empty($this->huburi)) {
 270             if (common_config('feedsub', 'fallback_hub')) {
 271                 // No native hub on this feed?
 272                 // Use our fallback hub, which handles polling on our behalf.
 273             } else if (common_config('feedsub', 'nohub')) {
 274                 // Fake it! We're just testing remote feeds w/o hubs.
 275                 // We'll never actually get updates in this mode.
 276                 return true;
 277             } else {
 278                 throw new ServerException(_m('Attempting to end PuSH subscription for feed with no hub.'));
 279             }
 280         }
 281
 282         return $this->doSubscribe('unsubscribe');
 283     }
 284
 285     /**
 286      * Check if there are any active local uses of this feed, and if not then
 287      * make sure it's inactive, unsubscribing if necessary.
 288      *
 289      * @return boolean true if the subscription is now inactive, false if still active.
 290      */
 291     public function garbageCollect()
 292     {
 293         if ($this->sub_state == '' || $this->sub_state == 'inactive') {
 294             // No active PuSH subscription, we can just leave it be.
 295             return true;
 296         } else {
 297             // PuSH subscription is either active or in an indeterminate state.
 298             // Check if we're out of subscribers, and if so send an unsubscribe.
 299             $count = 0;
 300             Event::handle('FeedSubSubscriberCount', array($this, &$count));
 301
 302             if ($count) {
 303                 common_log(LOG_INFO, __METHOD__ . ': ok, ' . $count . ' user(s) left for ' . $this->uri);
 304                 return false;
 305             } else {
 306                 common_log(LOG_INFO, __METHOD__ . ': unsubscribing, no users left for ' . $this->uri);
 307                 return $this->unsubscribe();
 308             }
 309         }
 310     }
 311
 312     protected function doSubscribe($mode)
 313     {
 314         $orig = clone($this);
 315         $this->verify_token = common_good_rand(16);
 316         if ($mode == 'subscribe') {
 317             $this->secret = common_good_rand(32);
 318         }
 319         $this->sub_state = $mode;
 320         $this->update($orig);
 321         unset($orig);
 322
 323         try {
 324             $callback = common_local_url('pushcallback', array('feed' => $this->id));
 325             $headers = array('Content-Type: application/x-www-form-urlencoded');
 326             $post = array('hub.mode' => $mode,
 327                           'hub.callback' => $callback,
 328                           'hub.verify' => 'sync',
 329                           'hub.verify_token' => $this->verify_token,
 330                           'hub.secret' => $this->secret,
 331                           'hub.topic' => $this->uri);
 332             $client = new HTTPClient();
 333             if ($this->huburi) {
 334                 $hub = $this->huburi;
 335             } else {
 336                 if (common_config('feedsub', 'fallback_hub')) {
 337                     $hub = common_config('feedsub', 'fallback_hub');
 338                     if (common_config('feedsub', 'hub_user')) {
 339                         $u = common_config('feedsub', 'hub_user');
 340                         $p = common_config('feedsub', 'hub_pass');
 341                         $client->setAuth($u, $p);
 342                     }
 343                 } else {
 344                     throw new FeedSubException('WTF?');
 345                 }
 346             }
 347             $response = $client->post($hub, $headers, $post);
 348             $status = $response->getStatus();
 349             if ($status == 202) {
 350                 common_log(LOG_INFO, __METHOD__ . ': sub req ok, awaiting verification callback');
 351                 return true;
 352             } else if ($status == 204) {
 353                 common_log(LOG_INFO, __METHOD__ . ': sub req ok and verified');
 354                 return true;
 355             } else if ($status >= 200 && $status < 300) {
 356                 common_log(LOG_ERR, __METHOD__ . ": sub req returned unexpected HTTP $status: " . $response->getBody());
 357                 return false;
 358             } else {
 359                 common_log(LOG_ERR, __METHOD__ . ": sub req failed with HTTP $status: " . $response->getBody());
 360                 return false;
 361             }
 362         } catch (Exception $e) {
 363             // wtf!
 364             common_log(LOG_ERR, __METHOD__ . ": error \"{$e->getMessage()}\" hitting hub $this->huburi subscribing to $this->uri");
 365
 366             $orig = clone($this);
 367             $this->verify_token = '';
 368             $this->sub_state = 'inactive';
 369             $this->update($orig);
 370             unset($orig);
 371
 372             return false;
 373         }
 374     }
 375
 376     /**
 377      * Save PuSH subscription confirmation.
 378      * Sets approximate lease start and end times and finalizes state.
 379      *
 380      * @param int $lease_seconds provided hub.lease_seconds parameter, if given
 381      */
 382     public function confirmSubscribe($lease_seconds=0)
 383     {
 384         $original = clone($this);
 385
 386         $this->sub_state = 'active';
 387         $this->sub_start = common_sql_date(time());
 388         if ($lease_seconds > 0) {
 389             $this->sub_end = common_sql_date(time() + $lease_seconds);
 390         } else {
 391             $this->sub_end = null;
 392         }
 393         $this->modified = common_sql_now();
 394
 395         return $this->update($original);
 396     }
 397
 398     /**
 399      * Save PuSH unsubscription confirmation.
 400      * Wipes active PuSH sub info and resets state.
 401      */
 402     public function confirmUnsubscribe()
 403     {
 404         $original = clone($this);
 405
 406         // @fixme these should all be null, but DB_DataObject doesn't save null values...?????
 407         $this->verify_token = '';
 408         $this->secret = '';
 409         $this->sub_state = '';
 410         $this->sub_start = '';
 411         $this->sub_end = '';
 412         $this->modified = common_sql_now();
 413
 414         return $this->update($original);
 415     }
 416
 417     /**
 418      * Accept updates from a PuSH feed. If validated, this object and the
 419      * feed (as a DOMDocument) will be passed to the StartFeedSubHandleFeed
 420      * and EndFeedSubHandleFeed events for processing.
 421      *
 422      * Not guaranteed to be running in an immediate POST context; may be run
 423      * from a queue handler.
 424      *
 425      * Side effects: the feedsub record's lastupdate field will be updated
 426      * to the current time (not published time) if we got a legit update.
 427      *
 428      * @param string $post source of Atom or RSS feed
 429      * @param string $hmac X-Hub-Signature header, if present
 430      */
 431     public function receive($post, $hmac)
 432     {
 433         common_log(LOG_INFO, __METHOD__ . ": packet for \"$this->uri\"! $hmac $post");
 434
 435         if ($this->sub_state != 'active') {
 436             common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH for inactive feed $this->uri (in state '$this->sub_state')");
 437             return;
 438         }
 439
 440         if ($post === '') {
 441             common_log(LOG_ERR, __METHOD__ . ": ignoring empty post");
 442             return;
 443         }
 444
 445         if (!$this->validatePushSig($post, $hmac)) {
 446             // Per spec we silently drop input with a bad sig,
 447             // while reporting receipt to the server.
 448             return;
 449         }
 450
 451         $feed = new DOMDocument();
 452         if (!$feed->loadXML($post)) {
 453             // @fixme might help to include the err message
 454             common_log(LOG_ERR, __METHOD__ . ": ignoring invalid XML");
 455             return;
 456         }
 457
 458         $orig = clone($this);
 459         $this->last_update = common_sql_now();
 460         $this->update($orig);
 461
 462         Event::handle('StartFeedSubReceive', array($this, $feed));
 463         Event::handle('EndFeedSubReceive', array($this, $feed));
 464     }
 465
 466     /**
 467      * Validate the given Atom chunk and HMAC signature against our
 468      * shared secret that was set up at subscription time.
 469      *
 470      * If we don't have a shared secret, there should be no signature.
 471      * If we we do, our the calculated HMAC should match theirs.
 472      *
 473      * @param string $post raw XML source as POSTed to us
 474      * @param string $hmac X-Hub-Signature HTTP header value, or empty
 475      * @return boolean true for a match
 476      */
 477     protected function validatePushSig($post, $hmac)
 478     {
 479         if ($this->secret) {
 480             if (preg_match('/^sha1=([0-9a-fA-F]{40})$/', $hmac, $matches)) {
 481                 $their_hmac = strtolower($matches[1]);
 482                 $our_hmac = hash_hmac('sha1', $post, $this->secret);
 483                 if ($their_hmac === $our_hmac) {
 484                     return true;
 485                 }
 486                 common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bad SHA-1 HMAC: got $their_hmac, expected $our_hmac");
 487             } else {
 488                 common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with bogus HMAC '$hmac'");
 489             }
 490         } else {
 491             if (empty($hmac)) {
 492                 return true;
 493             } else {
 494                 common_log(LOG_ERR, __METHOD__ . ": ignoring PuSH with unexpected HMAC '$hmac'");
 495             }
 496         }
 497         return false;
 498     }
 499 }