123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344 |
- <?php
- if (!defined('STATUSNET')) {
- exit(1);
- }
- class HubSub extends Managed_DataObject
- {
- public $__table = 'hubsub';
- public $hashkey;
- public $topic;
- public $callback;
- 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'),
- '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 setLease($length)
- {
- common_debug('PuSH hub got requested lease_seconds=='._ve($length));
- assert(is_int($length));
- $min = 86400;
- $max = 86400 * 30;
- if ($length == 0) {
-
- $length = $max;
- } elseif( $length < $min) {
- $length = $min;
- } else if ($length > $max) {
- $length = $max;
- }
- common_debug('PuSH 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();
- }
-
- 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,
- 'retries' => $retries);
- $qm = QueueManager::get();
- $qm->enqueue($data, 'hubconf');
- }
- public function getTopic()
- {
- return $this->topic;
- }
-
- 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) {
- $params['hub.verify_token'] = $token;
- }
-
- $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 {
-
- 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 {
-
- }
- }
- }
-
- function insert()
- {
- $this->hashkey = self::hashkey($this->getTopic(), $this->callback);
- $this->created = common_sql_now();
- $this->modified = common_sql_now();
- return parent::insert();
- }
-
- function distribute($atom, $retries=null)
- {
- if ($retries === null) {
- $retries = intval(common_config('ostatus', 'hub_retries'));
- }
-
-
-
-
- $sub = HubSub::getByHashkey($this->getTopic(), $this->callback);
- $data = array('sub' => $sub,
- 'atom' => $atom,
- 'retries' => $retries);
- common_log(LOG_INFO, "Queuing PuSH: {$this->getTopic()} to {$this->callback}");
- $qm = QueueManager::get();
- $qm->enqueue($data, 'hubout');
- }
-
- 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 PuSH batch: {$this->getTopic()} to ".count($pushCallbacks)." sites");
- $qm = QueueManager::get();
- $qm->enqueue($data, 'hubprep');
- return true;
- }
-
- function push($atom)
- {
- $headers = array('Content-Type: application/atom+xml');
- if ($this->secret) {
- $hmac = hash_hmac('sha1', $atom, $this->secret);
- $headers[] = "X-Hub-Signature: sha1=$hmac";
- } else {
- $hmac = '(none)';
- }
- common_log(LOG_INFO, "About to push feed to $this->callback for {$this->getTopic()}, HMAC $hmac");
- $request = new HTTPClient();
- $request->setConfig(array('follow_redirects' => false));
- $request->setBody($atom);
- try {
- $response = $request->post($this->callback, $headers);
- if ($response->isOk()) {
- return true;
- }
- } catch (Exception $e) {
- $response = null;
- common_debug('PuSH callback to '._ve($this->callback).' for '._ve($this->getTopic()).' failed with exception: '._ve($e->getMessage()));
- }
-
-
-
-
-
-
-
- if ('http' === parse_url($this->callback, PHP_URL_SCHEME)) {
-
- $httpscallback = preg_replace('/^http/', 'https', $this->callback, 1);
- $alreadyreplaced = self::getByHashKey($this->getTopic(), $httpscallback);
- if ($alreadyreplaced instanceof HubSub) {
- $this->delete();
- throw new AlreadyFulfilledException('The remote side has already established an HTTPS callback, deleting the legacy HTTP entry.');
- }
- common_debug('PuSH callback to '._ve($this->callback).' for '._ve($this->getTopic()).' trying HTTPS callback: '._ve($httpscallback));
- $response = $request->post($httpscallback, $headers);
- if ($response->isOk()) {
- $orig = clone($this);
- $this->callback = $httpscallback;
- $this->hashkey = self::hashkey($this->getTopic(), $this->callback);
- $this->updateWithKeys($orig);
- return true;
- }
- }
-
- if (is_null($response)) {
-
-
- throw new Exception(sprintf(_m('HTTP request failed without response to URL: %s'), _ve(isset($httpscallback) ? $httpscallback : $this->callback)));
- }
-
- throw new Exception(sprintf(_m('Callback returned status: %1$s. Body: %2$s'),
- $response->getStatus(),trim($response->getBody())));
- }
- }
|