code:
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2009-2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('GNUSOCIAL')) { exit(1); }

/**
 * @package OStatusPlugin
 * @maintainer Brion Vibber <brion@status.net>
 */

/*
WebSub (previously PubSubHubbub/PuSH) subscription flow:

    $profile->subscribe()
        sends a sub request to the hub...

    main/push/callback
        hub sends confirmation back to us via GET
        We verify the request, then echo back the challenge.
        On our end, we save the time we subscribed and the lease expiration

    main/push/callback
        hub sends us updates via POST

*/

/**
 * FeedSub handles low-level WebSub (PubSubHubbub/PuSH) subscriptions.
 * Higher-level behavior building OStatus stuff on top is handled
 * under Ostatus_profile.
 */
class FeedSub extends Managed_DataObject
{
    public $__table = 'feedsub';

    public $id;
    public $uri;    // varchar(191)   not 255 because utf8mb4 takes more space

    // WebSub subscription data
    public $huburi;
    public $secret;
    public $sub_state; // subscribe, active, unsubscribe, inactive, nohub
    public $sub_start;
    public $sub_end;
    public $last_update;

    public $created;
    public $modified;

    public static function schemaDef()
    {
        return array(
            'fields' => array(
                'id' => array('type' => 'serial', 'not null' => true, 'description' => 'FeedSub local unique id'),
                'uri' => array('type' => 'varchar', 'not null' => true, 'length' => 191, 'description' => 'FeedSub uri'),
                'huburi' => array('type' => 'text', 'description' => 'FeedSub hub-uri'),
                'secret' => array('type' => 'text', 'description' => 'FeedSub stored secret'),
                'sub_state' => array('type' => 'enum("subscribe","active","unsubscribe","inactive","nohub")', 'not null' => true, 'description' => 'subscription state'),
                'sub_start' => array('type' => 'datetime', 'description' => 'subscription start'),
                'sub_end' => array('type' => 'datetime', 'description' => 'subscription end'),
                'last_update' => array('type' => 'datetime', 'description' => 'when this record was last updated'),
                'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
                'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
            ),
            'primary key' => array('id'),
            'unique keys' => array(
                'feedsub_uri_key' => array('uri'),
            ),
        );
    }

    /**
     * Get the feed uri (http/https)
     */
    public function getUri()
    {
        if (empty($this->uri)) {
            throw new NoUriException($this);
        }
        return $this->uri;
    }

    function getLeaseRemaining()
    {
        if (empty($this->sub_end)) {
            return null;
        }
        return strtotime($this->sub_end) - time();
    }

    /**
     * Do we have a hub? Then we are a WebSub feed.
     * WebSub standard: https://www.w3.org/TR/websub/
     * old: https://en.wikipedia.org/wiki/PubSubHubbub
     *
     * If huburi is empty, then doublecheck that we are not using
     * a fallback hub. If there is a fallback hub, it is only if the
     * sub_state is "nohub" that we assume it's not a WebSub feed.
     */
    public function isWebSub()
    {
        if (empty($this->huburi)
                && (!common_config('feedsub', 'fallback_hub')
                    || $this->sub_state === 'nohub')) {
                // Here we have no huburi set. Also, either there is no
                // fallback hub configured or sub_state is "nohub".
            return false;
        }
        return true;
    }

    /**
     * Fetch the StatusNet-side profile for this feed
     * @return Profile
     */
    public function localProfile()
    {
        if ($this->profile_id) {
            return Profile::getKV('id', $this->profile_id);
        }
        return null;
    }

    /**
     * Fetch the StatusNet-side profile for this feed
     * @return Profile
     */
    public function localGroup()
    {
        if ($this->group_id) {
            return User_group::getKV('id', $this->group_id);
        }
        return null;
    }

    /**
     * @param string $feeduri
     * @return FeedSub
     * @throws FeedSubException if feed is invalid or lacks WebSub setup
     */
    public static function ensureFeed($feeduri)
    {
        $feedsub = self::getKV('uri', $feeduri);
        if ($feedsub instanceof FeedSub) {
            if (!empty($feedsub->huburi)) {
                // If there is already a huburi we don't
                // rediscover it on ensureFeed, call
                // ensureHub to do that (compare ->modified
                // to see if it might be time to do it).
                return $feedsub;
            }
            if ($feedsub->sub_state !== 'inactive') {
                throw new ServerException('Can only ensure WebSub hub for inactive (unsubscribed) feeds.');
            }
            // If huburi is empty we continue with ensureHub
        } else {
            // If we don't have that local feed URI
            // stored then we create a new DB object.
            $feedsub = new FeedSub();
            $feedsub->uri = $feeduri;
            $feedsub->sub_state = 'inactive';
        }

        try {
            // discover the hub uri
            $feedsub->ensureHub();

        } catch (FeedSubNoHubException $e) {
            // Only throw this exception if we can't handle huburi-less feeds
            // (i.e. we have a fallback hub or we can do feed polling (nohub)
            if (!common_config('feedsub', 'fallback_hub') && !common_config('feedsub', 'nohub')) {
                throw $e;
            }
        }

        if (empty($feedsub->id)) {
            // if $feedsub doesn't have an id we'll insert it into the db here
            $feedsub->created = common_sql_now();
            $feedsub->modified = common_sql_now();
            $result = $feedsub->insert();
            if ($result === false) {
                throw new FeedDBException($feedsub);
            }
        }

        return $feedsub;
    }

    /**
     * ensureHub will only do $this->update if !empty($this->id)
     * because otherwise the object has not been created yet.
     *
     * @param   bool    $rediscovered   Whether the hub info is rediscovered (to avoid endless loop nesting)
     *
     * @return  null    if actively avoiding the database
     *          int     number of rows updated in the database (0 means untouched)
     *
     * @throws  ServerException if something went wrong when updating the database
     *          FeedSubNoHubException   if no hub URL was discovered
     */
    public function ensureHub($rediscovered=false)
    {
        common_debug('Now inside ensureHub again, $rediscovered=='._ve($rediscovered));
        if ($this->sub_state !== 'inactive') {
            common_log(LOG_INFO, sprintf(__METHOD__ . ': Running hub discovery a possibly active feed in %s state for URI %s', _ve($this->sub_state), _ve($this->uri)));
        }

        $discover = new FeedDiscovery();
        $discover->discoverFromFeedURL($this->uri);

        $huburi = $discover->getHubLink();
        if (empty($huburi)) {
            // Will be caught and treated with if statements in regards to
            // fallback hub and feed polling (nohub) configuration.
            throw new FeedSubNoHubException();
        }

        // if we've already got a DB object stored, we want to UPDATE, not INSERT
        $orig = !empty($this->id) ? clone($this) : null;

        $old_huburi = $this->huburi;    // most likely null if we're INSERTing
        $this->huburi = $huburi;

        if (!empty($this->id)) {
            common_debug(sprintf(__METHOD__ . ': Feed uri==%s huburi before=%s after=%s (identical==%s)', _ve($this->uri), _ve($old_huburi), _ve($this->huburi), _ve($old_huburi===$this->huburi)));
            $result = $this->update($orig);
            if ($result === false) {
                // TODO: Get a DB exception class going...
                common_debug('Database update failed for FeedSub id=='._ve($this->id).' with new huburi: '._ve($this->huburi));
                throw new ServerException('Database update failed for FeedSub.');
            }
            if (!$rediscovered) {
                $this->renew();
            }
            return $result;
        }

        return null;    // we haven't done anything with the database
    }

    /**
     * Send a subscription request to the hub for this feed.
     * The hub will later send us a confirmation POST to /main/push/callback.
     *
     * @return void
     * @throws ServerException if feed state is not valid
     */
    public function subscribe($rediscovered=false)
    {
        if ($this->sub_state && $this->sub_state != 'inactive') {
            common_log(LOG_WARNING, sprintf('Attempting to (re)start WebSub subscription to %s in unexpected state %s', $this->getUri(), $this->sub_state));
        }

        if (!Event::handle('FeedSubscribe', array($this))) {
            // A plugin handled it
            return;
        }

        if (empty($this->huburi)) {
            if (common_config('feedsub', 'fallback_hub')) {
                // No native hub on this feed?
                // Use our fallback hub, which handles polling on our behalf.
            } else if (common_config('feedsub', 'nohub')) {
                // For this to actually work, we'll need some polling mechanism.
                // The FeedPoller plugin should take care of it.
                return;
            } else {
                // TRANS: Server exception.
                throw new ServerException(_m('Attempting to start WebSub subscription for feed with no hub.'));
            }
        }

        $this->doSubscribe('subscribe', $rediscovered);
    }

    /**
     * Send a WebSub unsubscription request to the hub for this feed.
     * The hub will later send us a confirmation POST to /main/push/callback.
     * Warning: this will cancel the subscription even if someone else in
     * the system is using it. Most callers will want garbageCollect() instead,
     * which confirms there's no uses left.
     *
     * @throws ServerException if feed state is not valid
     */
    public function unsubscribe() {
        if ($this->sub_state != 'active') {
            common_log(LOG_WARNING, sprintf('Attempting to (re)end WebSub subscription to %s in unexpected state %s', $this->getUri(), $this->sub_state));
        }

        if (!Event::handle('FeedUnsubscribe', array($this))) {
            // A plugin handled it
            return;
        }

        if (empty($this->huburi) && !common_config('feedsub', 'fallback_hub')) {
            /**
             * If the huburi is empty and we don't have a fallback hub,
             * there is nowhere we can send an unsubscribe to.
             *
             * A plugin should handle the FeedSub above and set the proper state
             * if there is no hub. (instead of 'nohub' it should be 'inactive' if
             * the instance has enabled feed polling for feeds that don't publish
             * WebSub/PuSH hubs. FeedPoller is a plugin which enables polling.
             *
             * Secondly, if we don't have the setting "nohub" enabled (i.e.)
             * we're ready to poll ourselves, there is something odd with the
             * database, such as a polling plugin that has been disabled.
             */

            if (!common_config('feedsub', 'nohub')) {
                // TRANS: Server exception.
                throw new ServerException(_m('Attempting to end WebSub subscription for feed with no hub.'));
            }

            return;
        }

        $this->doSubscribe('unsubscribe');
    }

    /**
     * Check if there are any active local uses of this feed, and if not then
     * make sure it's inactive, unsubscribing if necessary.
     *
     * @return boolean true if the subscription is now inactive, false if still active.
     * @throws NoProfileException in FeedSubSubscriberCount for missing Profile entries
     * @throws Exception if something goes wrong in unsubscribe() method
     */
    public function garbageCollect()
    {
        if ($this->sub_state == '' || $this->sub_state == 'inactive') {
            // No active WebSub subscription, we can just leave it be.
            return true;
        }

        // WebSub subscription is either active or in an indeterminate state.
        // Check if we're out of subscribers, and if so send an unsubscribe.
        $count = 0;
        Event::handle('FeedSubSubscriberCount', array($this, &$count));

        if ($count > 0) {
            common_log(LOG_INFO, __METHOD__ . ': ok, ' . $count . ' user(s) left for ' . $this->getUri());
            return false;
        }

        common_log(LOG_INFO, __METHOD__ . ': unsubscribing, no users left for ' . $this->getUri());
        // Unsubscribe throws various Exceptions on failure
        $this->unsubscribe();

        return true;
    }

    static public function renewalCheck()
    {
        $fs = new FeedSub();
        // the "" empty string check is because we historically haven't saved unsubscribed feeds as NULL
        $fs->whereAdd('sub_end IS NOT NULL AND sub_end!="" AND sub_end < NOW() + INTERVAL 1 day');
        if (!$fs->find()) { // find can be both false and 0, depending on why nothing was found
            throw new NoResultException($fs);
        }
        return $fs;
    }

    public function renew($rediscovered=false)
    {
        common_debug('FeedSub is being renewed for uri=='._ve($this->uri).' on huburi=='._ve($this->huburi));
        $this->subscribe($rediscovered);
    }

    /**
     * Setting to subscribe means it is _waiting_ to become active. This
     * cannot be done in a transaction because there is a chance that the
     * remote script we're calling (as in the case of PuSHpress) performs
     * the lookup _while_ we're POSTing data, which means the transaction
     * never completes (PushcallbackAction gets an 'inactive' state).
     *
     * @return boolean true when everything is ok (throws Exception on fail)
     * @throws Exception on failure, can be HTTPClient's or our own.
     */
    protected function doSubscribe($mode, $rediscovered=false)
    {
        $msg = null;    // carries descriptive error message to enduser (no remote data strings!)

        $orig = clone($this);
        if ($mode == 'subscribe') {
            $this->secret = common_random_hexstr(32);
        }
        $this->sub_state = $mode;
        $this->update($orig);
        unset($orig);

        try {
            $callback = common_local_url('pushcallback', array('feed' => $this->id));
            $headers = array('Content-Type: application/x-www-form-urlencoded');
            $post = array('hub.mode' => $mode,
                          'hub.callback' => $callback,
                          'hub.verify' => 'async',  // TODO: deprecated, remove when noone uses PuSH <0.4 (only 'async' method used there)
                          'hub.verify_token' => 'Deprecated-since-PuSH-0.4', // TODO: rm!

                          'hub.lease_seconds' => 2592000,   // 3600*24*30, request approximately month long lease (may be changed by hub)
                          'hub.secret' => $this->secret,
                          'hub.topic' => $this->getUri());
            $client = new HTTPClient();
            if ($this->huburi) {
                $hub = $this->huburi;
            } else {
                if (common_config('feedsub', 'fallback_hub')) {
                    $hub = common_config('feedsub', 'fallback_hub');
                    if (common_config('feedsub', 'hub_user')) {
                        $u = common_config('feedsub', 'hub_user');
                        $p = common_config('feedsub', 'hub_pass');
                        $client->setAuth($u, $p);
                    }
                } else {
                    throw new FeedSubException('Server could not find a usable WebSub hub.');
                }
            }
            $response = $client->post($hub, $headers, $post);
            $status = $response->getStatus();
            // WebSub specificed response status code
            if ($status == 202  || $status == 204) {
                common_log(LOG_INFO, __METHOD__ . ': sub req ok, awaiting verification callback');
                return;
            } else if ($status >= 200 && $status < 300) {
                common_log(LOG_ERR, __METHOD__ . ": sub req returned unexpected HTTP $status: " . $response->getBody());
                $msg = sprintf(_m("Unexpected HTTP status: %d"), $status);
            } else if ($status == 422 && !$rediscovered) {
                // Error code regarding something wrong in the data (it seems
                // that we're talking to a WebSub hub at least, so let's check
                // our own data to be sure we're not mistaken somehow, which
                // means rediscovering hub data (the boolean parameter means
                // we avoid running this part over and over and over and over):

                common_debug('Running ensureHub again due to 422 status, $rediscovered=='._ve($rediscovered));
                $discoveryResult = $this->ensureHub(true);
                common_debug('ensureHub is now done and its result was: '._ve($discoveryResult));
            } else {
                common_log(LOG_ERR, __METHOD__ . ": sub req failed with HTTP $status: " . $response->getBody());
            }
        } catch (Exception $e) {
            common_log(LOG_ERR, __METHOD__ . ": error \"{$e->getMessage()}\" hitting hub {$this->huburi} subscribing to {$this->getUri()}");

            // Reset the subscription state.
            $orig = clone($this);
            $this->sub_state = 'inactive';
            $this->update($orig);

            // Throw the Exception again.
            throw $e;
        }
        throw new ServerException("{$mode} request failed" . (!is_null($msg) ? " ($msg)" : '.'));
    }

    /**
     * Save WebSub subscription confirmation.
     * Sets approximate lease start and end times and finalizes state.
     *
     * @param int $lease_seconds provided hub.lease_seconds parameter, if given
     */
    public function confirmSubscribe($lease_seconds)
    {
        $original = clone($this);

        $this->sub_state = 'active';
        $this->sub_start = common_sql_date(time());
        if ($lease_seconds > 0) {
            $this->sub_end = common_sql_date(time() + $lease_seconds);
        } else {
            $this->sub_end = null;  // Backwards compatibility to StatusNet (PuSH <0.4 supported permanent subs)
        }
        $this->modified = common_sql_now();

        common_debug(__METHOD__ . ': Updating sub state and metadata for '.$this->getUri());
        return $this->update($original);
    }

    /**
     * Save WebSub unsubscription confirmation.
     * Wipes active WebSub sub info and resets state.
     */
    public function confirmUnsubscribe()
    {
        $original = clone($this);

        // @fixme these should all be null, but DB_DataObject doesn't save null values...?????
        $this->secret = '';
        $this->sub_state = '';
        $this->sub_start = '';
        $this->sub_end = '';
        $this->modified = common_sql_now();

        return $this->update($original);
    }

    /**
     * Accept updates from a WebSub feed. If validated, this object and the
     * feed (as a DOMDocument) will be passed to the StartFeedSubHandleFeed
     * and EndFeedSubHandleFeed events for processing.
     *
     * Not guaranteed to be running in an immediate POST context; may be run
     * from a queue handler.
     *
     * Side effects: the feedsub record's lastupdate field will be updated
     * to the current time (not published time) if we got a legit update.
     *
     * @param string $post source of Atom or RSS feed
     * @param string $hmac X-Hub-Signature header, if present
     */
    public function receive($post, $hmac)
    {
        common_log(LOG_INFO, sprintf(__METHOD__.': packet for %s with HMAC %s', _ve($this->getUri()), _ve($hmac)));

        if (!in_array($this->sub_state, array('active', 'nohub'))) {
            common_log(LOG_ERR, sprintf(__METHOD__.': ignoring WebSub for inactive feed %s (in state %s)', _ve($this->getUri()), _ve($this->sub_state)));
            return;
        }

        if ($post === '') {
            common_log(LOG_ERR, __METHOD__ . ": ignoring empty post");
            return;
        }

        try {
            if (!$this->validatePushSig($post, $hmac)) {
                // Per spec we silently drop input with a bad sig,
                // while reporting receipt to the server.
                return;
            }

            $this->receiveFeed($post);

        } catch (FeedSubBadPushSignatureException $e) {
            // We got a signature, so something could be wrong. Let's check to see if
            // maybe upstream has switched to another hub. Let's fetch feed and then
            // compare rel="hub" with $this->huburi, which is done in $this->ensureHub()

            $this->ensureHub(true);
        }
    }

    /**
     * All our feed URIs should be URLs.
     */
    public function importFeed()
    {
        $feed_url = $this->getUri();

        // Fetch the URL
        try {
            common_log(LOG_INFO, sprintf('Importing feed backlog from %s', $feed_url));
            $feed_xml = HTTPClient::quickGet($feed_url, 'application/atom+xml');
        } catch (Exception $e) {
            throw new FeedSubException("Could not fetch feed from URL '%s': %s (%d).\n", $feed_url, $e->getMessage(), $e->getCode());
        }

        return $this->receiveFeed($feed_xml);
    }

    protected function receiveFeed($feed_xml)
    {
        // We're passed the XML for the Atom feed as $feed_xml,
        // so read it into a DOMDocument and process.
        $feed = new DOMDocument();
        if (!$feed->loadXML($feed_xml)) {
            // @fixme might help to include the err message
            common_log(LOG_ERR, __METHOD__ . ": ignoring invalid XML");
            return;
        }

        $orig = clone($this);
        $this->last_update = common_sql_now();
        $this->update($orig);

        Event::handle('StartFeedSubReceive', array($this, $feed));
        Event::handle('EndFeedSubReceive', array($this, $feed));
    }

    /**
     * Validate the given Atom chunk and HMAC signature against our
     * shared secret that was set up at subscription time.
     *
     * If we don't have a shared secret, there should be no signature.
     * If we do, our calculated HMAC should match theirs.
     *
     * @param string $post raw XML source as POSTed to us
     * @param string $hmac X-Hub-Signature HTTP header value, or empty
     * @return boolean true for a match
     */
    protected function validatePushSig($post, $hmac)
    {
        if ($this->secret) {
            // {3,16} because shortest hash algorithm name is 3 characters (md2,md4,md5) and longest
            // is currently 11 characters, but we'll leave some margin in the end...
            if (preg_match('/^([0-9a-zA-Z\-\,]{3,16})=([0-9a-fA-F]+)$/', $hmac, $matches)) {
                $hash_algo  = strtolower($matches[1]);
                $their_hmac = strtolower($matches[2]);
                common_debug(sprintf(__METHOD__ . ': WebSub push from feed %s uses HMAC algorithm %s with value: %s', _ve($this->getUri()), _ve($hash_algo), _ve($their_hmac)));

                if (!in_array($hash_algo, hash_algos())) {
                    // We can't handle this at all, PHP doesn't recognize the algorithm name ('md5', 'sha1', 'sha256' etc: https://secure.php.net/manual/en/function.hash-algos.php)
                    common_log(LOG_ERR, sprintf(__METHOD__.': HMAC algorithm %s unsupported, not found in PHP hash_algos()', _ve($hash_algo)));
                    return false;
                } elseif (!is_null(common_config('security', 'hash_algos')) && !in_array($hash_algo, common_config('security', 'hash_algos'))) {
                    // We _won't_ handle this because there is a list of accepted hash algorithms and this one is not in it.
                    common_log(LOG_ERR, sprintf(__METHOD__.': Whitelist for HMAC algorithms exist, but %s is not included.', _ve($hash_algo)));
                    return false;
                }

                $our_hmac = hash_hmac($hash_algo, $post, $this->secret);
                if ($their_hmac !== $our_hmac) {
                    common_log(LOG_ERR, sprintf(__METHOD__.': ignoring WebSub push with bad HMAC hash: got %s, expected %s for feed %s from hub %s', _ve($their_hmac), _ve($our_hmac), _ve($this->getUri()), _ve($this->huburi)));
                    throw new FeedSubBadPushSignatureException('Incoming WebSub push signature did not match expected HMAC hash.');
                }
                return true;

            } else {
                common_log(LOG_ERR, sprintf(__METHOD__.': ignoring WebSub push with bogus HMAC==', _ve($hmac)));
            }
        } else {
            if (empty($hmac)) {
                return true;
            } else {
                common_log(LOG_ERR, sprintf(__METHOD__.': ignoring WebSub push with unexpected HMAC==%s', _ve($hmac)));
            }
        }
        return false;
    }

    public function delete($useWhere=false)
    {
        try {
            $oprofile = Ostatus_profile::getKV('feeduri', $this->getUri());
            if ($oprofile instanceof Ostatus_profile) {
                // Check if there's a profile. If not, handle the NoProfileException below
                $profile = $oprofile->localProfile();
            }
        } catch (NoProfileException $e) {
            // If the Ostatus_profile has no local Profile bound to it, let's clean it out at the same time
            $oprofile->delete();
        } catch (NoUriException $e) {
            // FeedSub->getUri() can throw a NoUriException, let's just go ahead and delete it
        }
        return parent::delete($useWhere);
    }
}


code:
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('GNUSOCIAL')) { exit(1); }

/**
 * WebSub (previously PuSH) feed subscription record
 * @package Hub
 * @author Brion Vibber <brion@status.net>
 */
class HubSub extends Managed_DataObject
{
    public $__table = 'hubsub';

    public $hashkey; // sha1(topic . '|' . $callback); (topic, callback) key is too long for myisam in utf8
    public $topic;      // varchar(191)   not 255 because utf8mb4 takes more space
    public $callback;   // varchar(191)   not 255 because utf8mb4 takes more space
    public $secret;
    public $sub_start;
    public $sub_end;
    public $created;
    public $modified;

    static function hashkey($topic, $callback)
    {
        return sha1($topic . '|' . $callback);
    }

    public static function getByHashkey($topic, $callback)
    {
        return self::getKV('hashkey', self::hashkey($topic, $callback));
    }

    public static function schemaDef()
    {
        return array(
            'fields' => array(
                'hashkey' => array('type' => 'char', 'not null' => true, 'length' => 40, 'description' => 'HubSub hashkey'),
                'topic' => array('type' => 'varchar', 'not null' => true, 'length' => 191, 'description' => 'HubSub topic'),
                'callback' => array('type' => 'varchar', 'not null' => true, 'length' => 191, 'description' => 'HubSub callback'),
                'secret' => array('type' => 'text', 'description' => 'HubSub stored secret'),
                'sub_start' => array('type' => 'datetime', 'description' => 'subscription start'),
                'sub_end' => array('type' => 'datetime', 'description' => 'subscription end'),
                'errors' => array('type' => 'integer', 'not null' => true, 'default' => 0, 'description' => 'Queue handling error count, is reset on success.'),
                'error_start' => array('type' => 'datetime', 'default' => null, 'description' => 'time of first error since latest success, should be null if no errors have been counted'),
                'last_error' => array('type' => 'datetime', 'default' => null, 'description' => 'time of last failure, if ever'),
                'last_error_msg' => array('type' => 'text', 'default' => null, 'description' => 'Last error _message_'),
                'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
                'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
            ),
            'primary key' => array('hashkey'),
            'indexes' => array(
                'hubsub_callback_idx' => array('callback'),
                'hubsub_topic_idx' => array('topic'),
            ),
        );
    }

    function getErrors()
    {
        return intval($this->errors);
    }

    // $msg is only set if $error_count is 0
    function setErrors($error_count, $msg=null)
    {
        assert(is_int($error_count));
        if (!is_int($error_count) || $error_count < 0) {
            common_log(LOG_ERR, 'HubSub->setErrors was given a bad value: '._ve($error_count));
            throw new ServerException('HubSub error count must be an integer higher or equal to 0.');
        }

        $orig = clone($this);
        $now = common_sql_now();

        if ($error_count === 1) {
            // Record when the errors started
            $this->error_start = $now;
        }
        if ($error_count > 0) {
            // Record this error's occurrence in time
            $this->last_error = $now;
            $this->last_error_msg = $msg;
        } else {
            $this->error_start = null;
            $this->last_error = null;
            $this->last_error_msg = null;
        }

        $this->errors = $error_count;
        $this->update($orig);
    }

    function resetErrors()
    {
        return $this->setErrors(0);
    }

    function incrementErrors($msg=null)
    {
        return $this->setErrors($this->getErrors()+1, $msg);
    }

    /**
     * Validates a requested lease length, sets length plus
     * subscription start & end dates.
     *
     * Does not save to database -- use before insert() or update().
     *
     * @param int $length in seconds
     */
    function setLease($length)
    {
        common_debug('WebSub hub got requested lease_seconds=='._ve($length));
        assert(is_int($length));

        $min = 86400;   // 3600*24 (one day)
        $max = 86400 * 30;

        if ($length == 0) {
            // We want to garbage collect dead subscriptions!
            $length = $max;
        } elseif( $length < $min) {
            $length = $min;
        } else if ($length > $max) {
            $length = $max;
        }

        common_debug('WebSub hub after sanitation: lease_seconds=='._ve($length));
        $this->sub_start = common_sql_now();
        $this->sub_end = common_sql_date(time() + $length);
    }

    function getLeaseTime()
    {
        if (empty($this->sub_start) || empty($this->sub_end)) {
            return null;
        }
        $length = strtotime($this->sub_end) - strtotime($this->sub_start);
        assert($length > 0);
        return $length;
    }

    function getLeaseRemaining()
    {
        if (empty($this->sub_end)) {
            return null;
        }
        return strtotime($this->sub_end) - time();
    }

    /**
     * Schedule a future verification ping to the subscriber.
     * If queues are disabled, will be immediate.
     *
     * @param string $mode 'subscribe' or 'unsubscribe'
     * @param string $token hub.verify_token value, if provided by client
     */
    function scheduleVerify($mode, $token=null, $retries=null)
    {
        if ($retries === null) {
            $retries = intval(common_config('ostatus', 'hub_retries'));
        }
        $data = array('sub' => clone($this),
                      'mode' => $mode,
                      'token' => $token,    // let's put it in there if remote uses PuSH <0.4
                      'retries' => $retries);
        $qm = QueueManager::get();
        $qm->enqueue($data, 'hubconf');
    }

    public function getTopic()
    {
        return $this->topic;
    }

    /**
     * Send a verification ping to subscriber, and if confirmed apply the changes.
     * This may create, update, or delete the database record.
     *
     * @param string $mode 'subscribe' or 'unsubscribe'
     * @param string $token hub.verify_token value, if provided by client
     * @throws ClientException on failure
     */
    function verify($mode, $token=null)
    {
        assert($mode == 'subscribe' || $mode == 'unsubscribe');

        $challenge = common_random_hexstr(32);
        $params = array('hub.mode' => $mode,
                        'hub.topic' => $this->getTopic(),
                        'hub.challenge' => $challenge);
        if ($mode == 'subscribe') {
            $params['hub.lease_seconds'] = $this->getLeaseTime();
        }
        if ($token !== null) {  // TODO: deprecated in PuSH 0.4
            $params['hub.verify_token'] = $token;   // let's put it in there if remote uses PuSH <0.4
        }

        // Any existing query string parameters must be preserved
        $url = $this->callback;
        if (strpos($url, '?') !== false) {
            $url .= '&';
        } else {
            $url .= '?';
        }
        $url .= http_build_query($params, '', '&');

        $request = new HTTPClient();
        $response = $request->get($url);
        $status = $response->getStatus();

        if ($status >= 200 && $status < 300) {
            common_log(LOG_INFO, "Verified {$mode} of {$this->callback}:{$this->getTopic()}");
        } else {
            // TRANS: Client exception. %s is a HTTP status code.
            throw new ClientException(sprintf(_m('Hub subscriber verification returned HTTP %s.'),$status));
        }

        $old = HubSub::getByHashkey($this->getTopic(), $this->callback);
        if ($mode == 'subscribe') {
            if ($old instanceof HubSub) {
                $this->update($old);
            } else {
                $ok = $this->insert();
            }
        } else if ($mode == 'unsubscribe') {
            if ($old instanceof HubSub) {
                $old->delete();
            } else {
                // That's ok, we're already unsubscribed.
            }
        }
    }

    // set the hashkey automagically on insert
    protected function onInsert()
    {
        $this->setHashkey();
        $this->created = common_sql_now();
        $this->modified = common_sql_now();
    }

    // update the hashkey automagically if needed
    protected function onUpdateKeys(Managed_DataObject $orig)
    {
        if ($this->topic !== $orig->topic || $this->callback !== $orig->callback) {
            $this->setHashkey();
        }
    }

    protected function setHashkey()
    {
        $this->hashkey = self::hashkey($this->topic, $this->callback);
    }

    /**
     * Schedule delivery of a 'fat ping' to the subscriber's callback
     * endpoint. If queues are disabled, this will run immediately.
     *
     * @param string $atom well-formed Atom feed
     * @param int $retries optional count of retries if POST fails; defaults to hub_retries from config or 0 if unset
     */
    function distribute($atom, $retries=null)
    {
        if ($retries === null) {
            $retries = intval(common_config('ostatus', 'hub_retries'));
        }

        $data = array('topic' => $this->getTopic(),
                      'callback' => $this->callback,
                      'atom' => $atom,
                      'retries' => $retries);
        common_log(LOG_INFO, sprintf('Queuing WebSub: %s to %s', _ve($data['topic']), _ve($data['callback'])));
        $qm = QueueManager::get();
        $qm->enqueue($data, 'hubout');
    }

    /**
     * Queue up a large batch of pushes to multiple subscribers
     * for this same topic update.
     *
     * If queues are disabled, this will run immediately.
     *
     * @param string $atom well-formed Atom feed
     * @param array $pushCallbacks list of callback URLs
     */
    function bulkDistribute($atom, array $pushCallbacks)
    {
        if (empty($pushCallbacks)) {
            common_log(LOG_ERR, 'Callback list empty for bulkDistribute.');
            return false;
        }
        $data = array('atom' => $atom,
                      'topic' => $this->getTopic(),
                      'pushCallbacks' => $pushCallbacks);
        common_log(LOG_INFO, "Queuing WebSub batch: {$this->getTopic()} to ".count($pushCallbacks)." sites");
        $qm = QueueManager::get();
        $qm->enqueue($data, 'hubprep');
        return true;
    }

    /**
     * @return  boolean     true/false for HTTP response
     * @throws  Exception   for lower-than-HTTP errors (such as NS lookup failure, connection refused...)
     */
    public static function pushAtom($topic, $callback, $atom, $secret=null, $hashalg='sha1')
    {
        $headers = array('Content-Type: application/atom+xml');
        if ($secret) {
            $hmac = hash_hmac($hashalg, $atom, $secret);
            $headers[] = "X-Hub-Signature: {$hashalg}={$hmac}";
        } else {
            $hmac = '(none)';
        }
        common_log(LOG_INFO, sprintf('About to WebSub-push feed to %s for %s, HMAC %s', _ve($callback), _ve($topic), _ve($hmac)));

        $request = new HTTPClient();
        $request->setConfig(array('follow_redirects' => false));
        $request->setBody($atom);

        // This will throw exception on non-HTTP failures
        try {
            $response = $request->post($callback, $headers);
        } catch (Exception $e) {
            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())));
            throw $e;
        }

        return $response->isOk();
    }

    /**
     * Send a 'fat ping' to the subscriber's callback endpoint
     * containing the given Atom feed chunk.
     *
     * Determination of which feed items to send should be done at
     * a higher level; don't just shove in a complete feed!
     *
     * FIXME: Add 'failed' incremental count.
     *
     * @param string $atom well-formed Atom feed
     * @return  boolean     Whether the PuSH was accepted or not.
     * @throws Exception (HTTP or general)
     */
    function push($atom)
    {
        try {
            $success = self::pushAtom($this->getTopic(), $this->callback, $atom, $this->secret);
            if ($success) {
                return true;
            } elseif ('https' === parse_url($this->callback, PHP_URL_SCHEME)) {
                // Already HTTPS, no need to check remote http/https migration issues
                return false;
            }
            // if pushAtom returned false and we didn't try an HTTPS endpoint,
            // let's try HTTPS too (assuming only http:// and https:// are used ;))

        } catch (Exception $e) {
            if ('https' === parse_url($this->callback, PHP_URL_SCHEME)) {
                // Already HTTPS, no need to check remote http/https migration issues
                throw $e;
            }
        }

        // We failed the WebSub push, but it might be that the remote site has changed their configuration to HTTPS
        common_debug('WebSub HTTPSFIX: push failed, so we need to see if it can be the remote http->https migration issue.');

        // XXX: DO NOT trust a Location header here, _especially_ from 'http' protocols,
        // but not 'https' either at least if we don't do proper CA verification. Trust that
        // the most common change here is simply switching 'http' to 'https' and we will
        // solve 99% of all of these issues for now. There should be a proper mechanism
        // if we want to change the callback URLs, preferrably just manual resubscriptions
        // from the remote side, combined with implemented WebSub subscription timeouts.

        // Test if the feed callback for this node has already been migrated to HTTPS in our database
        // (otherwise we'd get collisions when inserting it further down)
        $httpscallback = preg_replace('/^http/', 'https', $this->callback, 1);
        $alreadyreplaced = self::getByHashKey($this->getTopic(), $httpscallback);
        if ($alreadyreplaced instanceof HubSub) {
            // Let's remove the old HTTP callback object.
            $this->delete();

            // XXX: I think this means we might lose a message or two when
            //      remote side migrates to HTTPS because we only try _once_
            //      for _one_ WebSub push. The rest of the posts already
            //      stored in our queue (if any) will not find a HubSub
            //      object. This could maybe be fixed by handling migration
            //      in HubOutQueueHandler while handling the item there.
            common_debug('WebSub HTTPSFIX: Pushing Atom to HTTPS callback instead of HTTP, because of switch to HTTPS since enrolled in queue.');
            return self::pushAtom($this->getTopic(), $httpscallback, $atom, $this->secret);
        }

        common_debug('WebSub HTTPSFIX: callback to '._ve($this->callback).' for '._ve($this->getTopic()).' trying HTTPS callback: '._ve($httpscallback));
        $success = self::pushAtom($this->getTopic(), $httpscallback, $atom, $this->secret);
        if ($success) {
            // Yay, we made a successful push to https://, let's remember this in the future!
            $orig = clone($this);
            $this->callback = $httpscallback;
            // NOTE: hashkey will be set in $this->onUpdateKeys($orig) through updateWithKeys
            $this->updateWithKeys($orig);
            return true;
        }

        // If there have been any exceptions thrown before, they're handled
        // higher up. This function's return value is just whether the WebSub
        // push was accepted or not.
        return $success;
    }
}

repo_name: andremichelle/gnu

license: mit


code:
<?php
/**
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2010, StatusNet, Inc.
 *
 * A sample module to show best practices for StatusNet plugins
 *
 * PHP version 5
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @package   StatusNet
 * @author    James Walker <james@status.net>
 * @copyright 2010 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html AGPL 3.0
 * @link      http://status.net/
 */

if (!defined('GNUSOCIAL')) { exit(1); }

class Magicsig extends Managed_DataObject
{
    const PUBLICKEYREL = 'magic-public-key';

    const DEFAULT_KEYLEN = 1024;
    const DEFAULT_SIGALG = 'RSA-SHA256';

    public $__table = 'magicsig';

    /**
     * Key to user.id/profile.id for the local user whose key we're storing.
     *
     * @var int
     */
    public $user_id;

    /**
     * Flattened string representation of the key pair; callers should
     * usually use $this->publicKey and $this->privateKey directly,
     * which hold live \phpseclib\Crypt\RSA key objects.
     *
     * @var string
     */
    public $keypair;

    /**
     * Crypto algorithm used for this key; currently only RSA-SHA256 is supported.
     *
     * @var string
     */
    public $alg;

    /**
     * Public RSA key; gets serialized in/out via $this->keypair string.
     *
     * @var \phpseclib\Crypt\RSA
     */
    public $publicKey;

    /**
     * PrivateRSA key; gets serialized in/out via $this->keypair string.
     *
     * @var \phpseclib\Crypt\RSA
     */
    public $privateKey;

    public function __construct($alg=self::DEFAULT_SIGALG)
    {
        $this->alg = $alg;
    }

    /**
     * Fetch a Magicsig object from the cache or database on a field match.
     *
     * @param string $k
     * @param mixed $v
     * @return Magicsig
     */
    static function getKV($k, $v=null)
    {
        $obj =  parent::getKV($k, $v);
        if ($obj instanceof Magicsig) {
            $obj->importKeys(); // Loads \phpseclib\Crypt\RSA objects etc.

            // Throw out a big fat warning for keys of less than 1024 bits. (
            // The only case these show up in would be imported or
            // legacy very-old-StatusNet generated keypairs.
            if (strlen($obj->publicKey->modulus->toBits()) < 1024) {
                common_log(LOG_WARNING, sprintf('Salmon key with <1024 bits (%d) belongs to profile with id==%d',
                                            strlen($obj->publicKey->modulus->toBits()),
                                            $obj->user_id));
            }
        }

        return $obj;
    }

    public static function schemaDef()
    {
        return array(
            'fields' => array(
                'user_id' => array('type' => 'int', 'not null' => true, 'description' => 'user id'),
                'keypair' => array('type' => 'text', 'description' => 'keypair text representation'),
                'alg' => array('type' => 'varchar', 'length' => 64, 'description' => 'algorithm'),
            ),
            'primary key' => array('user_id'),
            'foreign keys' => array(
                'magicsig_user_id_fkey' => array('profile', array('user_id' => 'id')),
            ),
        );
    }

    /**
     * Save this keypair into the database.
     *
     * Overloads default insert behavior to encode the live key objects
     * as a flat string for storage.
     *
     * @return mixed
     */
    function insert()
    {
        $this->keypair = $this->toString(true);

        return parent::insert();
    }

    /**
     * Generate a new keypair for a local user and store in the database.
     *
     * Warning: this can be very slow on systems without the GMP module.
     * Runtimes of 20-30 seconds are not unheard-of.
     *
     * FIXME: More than 1024 bits please. But StatusNet _discards_ non-1024 bits,
     *        so we'll have to wait the last mohican out before switching defaults.
     *
     * @param User $user the local user (since we don't have remote private keys)
     */
    public static function generate(User $user, $bits=self::DEFAULT_KEYLEN, $alg=self::DEFAULT_SIGALG)
    {
        $magicsig = new Magicsig($alg);
        $magicsig->user_id = $user->id;

        $rsa = new \phpseclib\Crypt\RSA();

        $keypair = $rsa->createKey($bits);

        $magicsig->privateKey = new \phpseclib\Crypt\RSA();
        $magicsig->privateKey->load($keypair['privatekey']);

        $magicsig->publicKey = new \phpseclib\Crypt\RSA();
        $magicsig->publicKey->load($keypair['publickey']);

        $magicsig->insert();        // will do $this->keypair = $this->toString(true);
        $magicsig->importKeys();    // seems it's necessary to re-read keys from text keypair

        return $magicsig;
    }

    /**
     * Encode the keypair or public key as a string.
     *
     * @param boolean $full_pair set to true to include the private key.
     * @return string
     */
    public function toString($full_pair=false, $base64url=true)
    {
        $base64_func = $base64url ? 'Magicsig::base64_url_encode' : 'base64_encode';
        $mod = call_user_func($base64_func, $this->publicKey->modulus->toBytes());
        $exp = call_user_func($base64_func, $this->publicKey->exponent->toBytes());

        $private_exp = '';
        if ($full_pair && $this->privateKey instanceof \phpseclib\Crypt\RSA && $this->privateKey->exponent->toBytes()) {
            $private_exp = '.' . call_user_func($base64_func, $this->privateKey->exponent->toBytes());
        }

        return 'RSA.' . $mod . '.' . $exp . $private_exp;
    }

    public function toFingerprint()
    {
        // This assumes a specific behaviour from toString, to format as such:
        //    "RSA." + base64(pubkey.modulus_as_bytes) + "." + base64(pubkey.exponent_as_bytes)
        // We don't want the base64 string to be the "url encoding" version because it is not
        // as common in programming libraries. And we want it to be base64 encoded since ASCII
        // representation avoids any problems with NULL etc. in less forgiving languages and also
        // just easier to debug...
        return strtolower(hash('sha256', $this->toString(false, false)));
    }

    public function exportPublicKey($type='PKCS1')
    {
        $this->publicKey->setPublicKey();
        return $this->publicKey->getPublicKey($type);
    }

    /**
     * importKeys will load the object's keypair string, which initiates
     * loadKey() and configures \phpseclib\Crypt\RSA objects.
     *
     * @param string $keypair optional, otherwise the object's "keypair" property will be used
     */
    public function importKeys($keypair=null)
    {
        $this->keypair = $keypair===null ? $this->keypair : preg_replace('/\s+/', '', $keypair);

        // parse components
        if (!preg_match('/RSA\.([^\.]+)\.([^\.]+)(\.([^\.]+))?/', $this->keypair, $matches)) {
            common_debug('Magicsig error: RSA key not found in provided string.');
            throw new ServerException('RSA key not found in keypair string.');
        }

        $mod = $matches[1];
        $exp = $matches[2];
        if (!empty($matches[4])) {
            $private_exp = $matches[4];
        } else {
            $private_exp = false;
        }

        $this->loadKey($mod, $exp, 'public');
        if ($private_exp) {
            $this->loadKey($mod, $private_exp, 'private');
        }
    }

    /**
     * Fill out $this->privateKey or $this->publicKey with a \phpseclib\Crypt\RSA object
     * representing the give key (as mod/exponent pair).
     *
     * @param string $mod base64url-encoded
     * @param string $exp base64url-encoded exponent
     * @param string $type one of 'public' or 'private'
     */
    public function loadKey($mod, $exp, $type = 'public')
    {
        $rsa = new \phpseclib\Crypt\RSA();
        $rsa->setHash($this->getHash());
        $rsa->modulus = new \phpseclib\Math\BigInteger(Magicsig::base64_url_decode($mod), 256);
        $rsa->k = strlen($rsa->modulus->toBytes());
        $rsa->exponent = new \phpseclib\Math\BigInteger(Magicsig::base64_url_decode($exp), 256);

        if ($type == 'private') {
            $this->privateKey = $rsa;
        } else {
            $this->publicKey = $rsa;
        }
    }

    public function loadPublicKeyPKCS1($key)
    {
        $rsa = new \phpseclib\Crypt\RSA();
        if (!$rsa->setPublicKey($key, 'PKCS1')) {
            throw new ServerException('Could not load PKCS1 public key. We probably got this from a remote Diaspora node as the profile public key.');
        }
        $this->publicKey = $rsa;
    }

    /**
     * Returns the name of the crypto algorithm used for this key.
     *
     * @return string
     */
    public function getName()
    {
        return $this->alg;
    }

    /**
     * Returns the name of a hash function to use for signing with this key.
     *
     * @return string
     */
    public function getHash()
    {
        switch ($this->alg) {
        case 'RSA-SHA256':
            return 'sha256';
        }
        throw new ServerException('Unknown or unsupported hash algorithm for Salmon');
    }

    /**
     * Generate base64-encoded signature for the given byte string
     * using our private key.
     *
     * @param string $bytes as raw byte string
     * @return string base64url-encoded signature
     */
    public function sign($bytes)
    {
        $sig = $this->privateKey->sign($bytes, \phpseclib\Crypt\RSA::PADDING_PKCS1);
        if ($sig === false) {
            throw new ServerException('Could not sign data');
        }
        return Magicsig::base64_url_encode($sig);
    }

    /**
     *
     * @param string $signed_bytes as raw byte string
     * @param string $signature as base64url encoded
     * @return boolean
     */
    public function verify($signed_bytes, $signature)
    {
        $signature = self::base64_url_decode($signature);
        return $this->publicKey->verify($signed_bytes, $signature, \phpseclib\Crypt\RSA::PADDING_PKCS1);
    }

    /**
     * URL-encoding-friendly base64 variant encoding.
     *
     * @param string $input
     * @return string
     */
    public static function base64_url_encode($input)
    {
        return strtr(base64_encode($input), '+/', '-_');
    }

    /**
     * URL-encoding-friendly base64 variant decoding.
     *
     * @param string $input
     * @return string
     */
    public static function base64_url_decode($input)
    {
        return base64_decode(strtr($input, '-_', '+/'));
    }
}