File Coverage

blib/lib/Mojolicious/Plugin/PubSubHubbub.pm

Criterion	Covered	Total	%
statement	256	308	83.1
branch	106	172	61.6
condition	45	93	48.3
subroutine	28	30	93.3
pod	3	5	60.0
total	438	608	72.0

package Mojolicious::Plugin::PubSubHubbub;
use Mojo::Base 'Mojolicious::Plugin';
use Mojo::UserAgent;
use Mojo::DOM;
use Mojo::ByteStream 'b';
use Mojo::Util qw/secure_compare hmac_sha1_sum/;

our $VERSION = '0.20';

# Todo:
# - Be compliant with https://www.w3.org/TR/websub/
# - Prevent log injection
# - Make everything async (top priority)
# - Maybe allow something like ->feed_to_json (look at superfeedr)
# - Test ->discover

# Default lease seconds before automatic subscription refreshing
has lease_seconds => ( 9 * 24 * 60 * 60 );
has hub => 'http://pubsubhubbub.appspot.com/';

my $FEED_TYPE_RE   = qr{^(?i:application/(atom|r(?:ss|df))\+xml)};
my $FEED_ENDING_RE = qr{(?i:\.(r(?:ss|df)|atom))$};

# User Agent Name
my $UA_NAME = __PACKAGE__ . ' v' . $VERSION;

# Prototypes
sub _add_topics;

# Register plugin
sub register {
  my ($plugin, $mojo, $param) = @_;

  $param ||= {};

  # Load parameter from Config file
  if (my $config_param = $mojo->config('PubSubHubbub')) {
    $param = { %$param, %$config_param };
  };

  my $helpers = $mojo->renderer->helpers;

  # Load 'callback' plugin
  unless (exists $helpers->{'callback'}) {
    $mojo->plugin('Util::Callback');
  };

  # Set callbacks on registration
  $mojo->callback([qw/pubsub_accept pubsub_verify/] => $param);

  # Load 'endpoint' plugin
  unless (exists $helpers->{'endpoint'}) {
    $mojo->plugin('Util::Endpoint');
  };

  # Load 'randomstring' plugin
  $mojo->plugin('Util::RandomString' => {
    pubsub_challenge => {
      length => 12,
      alphabet => [ 'A' .. 'Z', 'a' .. 'z', 0 .. 9 ]
    }
  });

  # Set hub attribute
  if ($param->{hub}) {
    $plugin->hub($param->{hub});
  };

  # Establish an endpoint
  $mojo->endpoint('pubsub-hub' => $plugin->hub);

  # Set lease_seconds attribute
  if ($param->{lease_seconds}) {
    $plugin->lease_seconds($param->{lease_seconds});
  };

  # Add 'pubsub' shortcut
  $mojo->routes->add_shortcut(
    pubsub => sub {
      my ($route, $param) = @_;

      # Set param default to 'cb'
      $param ||= 'cb';

      # 'hub' is currently not supported
      return unless $param eq 'cb';

      # Set PubSubHubbub endpoints
      $route->endpoint('pubsub-callback');

      # Add 'callback' route
      $route->to(
        cb => sub {
          my $c = shift;

          # Hook on verification
          return $plugin->verify($c) if $c->param('hub.mode');

          # Hook on callback
          return $plugin->callback($c);
        });
    });

  # Return plugin object
  $mojo->helper(
    'pubsub._plugin' => sub {
      $plugin;
    });

  $mojo->helper(
    'pubsub.publish' => sub {
      $plugin->publish( @_ );
    });

  # Add 'subscribe' and 'unsubscribe' helper
  foreach my $action (qw(subscribe unsubscribe)) {
    $mojo->helper(
      "pubsub.${action}" => sub {
        $plugin->_change_subscription(shift, mode => $action, @_);
      });
  };

  $mojo->helper(
    'pubsub.discover' => sub {
      $plugin->discover( @_ )
    }
  );
};


# Ping a hub for topics
sub publish {
  my $plugin = shift;
  my $c      = shift;

  # Nothing to publish or no hub defined
  return unless @_ || !$plugin->hub;

  # Set all urls
  my @urls = map($c->endpoint($_), @_);

  # Create post message
  my %post = (
    'hub.mode' => 'publish',
    'hub.url'  => \@urls
  );

  # Get user agent
  my $ua = Mojo::UserAgent->new(
    max_redirects => 3,
    name => $UA_NAME
  );

  my $msg = 'Cannot ping hub';
  $msg .= ' - maybe no SSL support' if index($plugin->hub, 'https') == 0;

  # Blocking
  # Post to hub
  my $tx = $ua->post( $plugin->hub => form => \%post );

  my $res = $tx->result;

  # No response
  unless ($res) {
    $c->app->log->warn($msg);
    return;
  };

  # is 2xx, incl. 204 aka successful
  return 1 if $res->is_success;

  # Not successful
  return;
};


# Verify a changed subscription or automatically refresh
sub verify {
  my $plugin = shift;
  my $c = shift;

  # Good request
  if ($c->param('hub.topic') &&
      $c->param('hub.challenge') &&
      $c->param('hub.mode') =~ /^(?:un)?subscribe$/) {

    my $challenge = $c->param('hub.challenge');

    my %param;
    foreach (qw/mode
                topic
                verify
                lease_seconds
                verify_token/) {
      $param{$_} = $c->param("hub.$_") if $c->param("hub.$_");
    };

    # Get verification callback
    my $ok = $c->callback(
      pubsub_verify => \%param
    );

    # Render challenge
    return $c->render(
      'status' => 200,
      'format' => 'txt',
      'data'   => $challenge
    ) if $ok;
  };

  # Not found
  return $c->reply->not_found;
};


# Discover links from header
# This is extremely simplified from https://tools.ietf.org/html/rfc5988
sub _discover_header_links {
  my $header = shift;

  my $header_hash = $header->to_hash(1);

  my @links = (@{$header_hash->{Link} // []}, @{$header_hash->{link} // []});
  my %links;

  # Iterate through all header links
  foreach (@links) {

    # Make multiline headers one line
    $_ = join(' ', @$_) if ref $_;

    # Check for link with correct relation
    if ($_ =~ /^\<([^>]+?)\>(.*?rel\s*=\s*"(self|hub|alternate)".*?)$/mi) {

      # Create new link hash
      my %link = ( href => $1, rel  => $3 );

      # There may be more than one reference
      my $check = $2;

      # Set type
      if ($check =~ /type\s*=\s*"([^"]+?)"/omi) {
        my $type = $1;
        next if $type && $type !~ $FEED_TYPE_RE;
        $link{type} = $type;
        $link{short_type} = $1;
      };

      # Set title
      if ($check =~ /title\s*=\s*"([^"]+?)"/omi) {
        $link{title} = $1;
      };

      # Check file ending for short type
      unless ($link{short_type}) {
        $link{short_type} = $1 if $link{href} =~ $FEED_ENDING_RE;
      };

      # Push found link
      my $rel = $link{rel};
      $links{$rel} //= [];
      push(@{$links{$rel}}, \%link);
    };
  };

  # Return array
  return \%links;
};


# Discover links from dom tree
sub _discover_dom_links {
  my $dom = shift;

  my %links;

  # Find alternate representations
  $dom->find('link[rel="alternate"], link[rel="self"], link[rel="hub"]')->each(
    sub {
      my ($href, $rel, $type, $title) = @{$_->attr}{qw/href rel type title/};

      # Is no supported type
      return if $type && $type !~ $FEED_TYPE_RE;

      # Set short type
      my $short_type = $1 if $1;

      return unless $href && $rel;

      # Create new link hash
      my %link = ( href => $href, rel  => $rel );

      # Short type yet not known
      unless ($short_type) {

        # Set short type by file ending
        $link{short_type} = $1 if $href =~ m/\.(r(?:ss|df)|atom)$/i;
      }

      # Set short type
      else {
        $link{short_type} = $short_type;
      };

      # Set title and type
      $link{title} = $title if $title;
      $link{type}  = $type if $type;

      # Push found link
      $links{$rel} //= [];
      push(@{$links{$rel}}, \%link);
    }
  );

  # Return array
  return \%links;
};


# Heuristically sort links to best match the topic
sub _discover_sort_links {
  my $links = shift;

  my ($topic, $hub);

  # Get self link as topic
  if ($links->{self}) {

    # Find best match of all returned links
    foreach my $link (@{$links->{self}}) {
      $topic ||= $link;
      if ($link->{short_type} && !$topic->{short_type}) {
        $topic = $link;
      };
    };
  };

  # Get hub
  if ($links->{hub}) {

    # Find best match of all returned links
    foreach my $link (@{$links->{hub}}) {
      $hub ||= $link;
      if ($link->{short_type} && !$hub->{short_type}) {
        $hub = $link;
      };
    };
  };

  # Already found topic and hub
  return ($topic, $hub) if $topic && $hub;

  # Check alternates
  my $alternate = $links->{alternate};

  # Search in alternate representations for best match
  if ($alternate) {

    # Iterate through all alternate links
    # and check their titles
    foreach my $link (@$alternate) {

      # No title given
      unless ($link->{title}) {
        $link->{pref} = 2;
      }

      # Guess which feed is best based on the title
      elsif ($link->{title} =~ /(?i:feed|stream)/i) {

        # This is more likely a comment feed
        if ($link->{title} =~ /[ck]omment/i) {
          $link->{pref} = 1;
        }

        # This may be the correct feed
        else {
          $link->{pref} = 3;
        };
      }

      # Don't know ...
      else {
        $link->{pref} = 2;
      };
    };

    # Get best topic
    ($topic) = (sort {

      # Sort by title
      if ($a->{pref} < $b->{pref}) {
        return 1;
      }
      elsif ($a->{pref} > $b->{pref}) {
        return -1;
      }
      # Sort by type
      elsif ($a->{short_type} gt $b->{short_type}) {
        return 1;
      }
      elsif ($a->{short_type} lt $b->{short_type}) {
        return -1;
      }
      # Sort by length
      elsif (length($a->{href}) > length($b->{href})) {
        return 1;
      }
      elsif (length($a->{href}) <= length($b->{href})) {
        return -1;
      }
      # Equal
      else {
        return -1;
      };
    } @$alternate);
  };

  # Maybe empty ... maybe not
  return ($topic, $hub);
};


# Discover topic and hub based on a URI
# That's a rather complex heuristic, but should gain good results
sub discover {
  my $plugin = shift;
  my $c = shift;

  # No uri given
  return () unless $_[0];

  # Get uri
  my $base = Mojo::URL->new( shift ) or return ();

  # Set base to uri
  $base->base($c->req->url);

  # Initialize UserAgent
  my $ua = Mojo::UserAgent->new(
    max_redirects => 3,
    name => $UA_NAME
  );

  # Initialize variables
  my ($hub, $topic, $nbase, $ntopic);

  # Retrieve resource
  my $tx = $ua->get($base);

  unless ($tx->error) {

    # Change base after possible redirects
    $base = $tx->req->url;

    # Get response
    my $res = $tx->res;

    # Check sorted header links
    ($topic, $hub) = _discover_sort_links(
      _discover_header_links($res->headers)
    );

    # Fine
    unless ($topic && $hub) {

      my $dom = $res->dom;

      # Check sorted dom links
      ($topic, $hub) = _discover_sort_links(
        _discover_dom_links($dom)
      );
    };

    # Fine
    if ($topic && !$hub) {

      # Initialize new UserAgent
      $ua = Mojo::UserAgent->new(
        max_redirects => 3,
        name => $UA_NAME
      );

      # Set new base base
      $nbase = Mojo::URL->new($topic->{href})->base($base)->to_abs;

      # Retrieve resource
      $tx = $ua->get($nbase);

      # Request was successful
      unless ($tx->error) {

        # Change nbase after possible redirects
        $nbase = $tx->req->url;

        # Get response
        $res = $tx->res;

        # Check sorted header links
        ($ntopic, $hub) = _discover_sort_links(
          _discover_header_links($res->headers)
        );


        unless ($ntopic && $hub) {

          # Check sorted dom links
          ($ntopic, $hub) = _discover_sort_links(
            _discover_dom_links($res->dom)
          );
        };
      }

      # Reset nbase as no connection occurred
      else {
        $nbase = undef;
      };
    };
  };

  # Make relative path for topics and hubs absolute
  $hub = Mojo::URL->new($hub->{href})->base( $nbase || $base )->to_abs if $hub;

  # New topic is set
  if ($ntopic) {
    $topic = Mojo::URL->new($ntopic->{href})->base($nbase)->to_abs;
  }

  # Old topic is set
  elsif ($topic) {
    $topic = Mojo::URL->new($topic->{href})->base($base)->to_abs;
  };

  # Return
  return ($topic, $hub);
};


# subscribe or unsubscribe from a topic
sub _change_subscription {
  my $plugin = shift;
  my $c      = shift;
  my %param  = @_;

  my $log = $c->app->log;

  # Get callback endpoint
  # Works only if endpoints provided
  unless ($param{callback} ||= $c->endpoint('pubsub-callback')) {
    $log->error('You have to specify a callback endpoint') and return;
  };

  # No topic or hub url given
  unless (exists $param{topic} &&
            $param{topic} =~ m{^https?://}i &&
            exists $param{hub}) {
    $log->warn('You have to specify a topic and a hub');
    return;
  };

  my $mode = $param{mode};

  # delete lease seconds if no integer
  if (exists $param{lease_seconds} &&
        ($mode eq 'unsubscribe' || $param{lease_seconds} !~ /^\d+$/)
      ) {
    delete $param{lease_seconds};
  };

  # Set to default
  $param{lease_seconds} ||= $plugin->lease_seconds if $mode eq 'subscribe';

  # Render post string
  my %post = ( callback => $param{callback} );
  foreach ( qw/mode topic verify lease_seconds secret/ ) {
    $post{ $_ } = $param{ $_ } if exists $param{ $_ } && $param{ $_ };
  };

  # Use verify token
  $post{verify_token} =
    exists $param{verify_token} ?
    $param{verify_token} :
    ($param{verify_token} =
	   $c->random_string('pubsub_challenge'));

  $post{verify} = "${_}sync" foreach ('a', '');

  my $mojo = $c->app;

  $mojo->plugins->emit_hook(
    "before_pubsub_$mode" => ($c, \%param, \%post)
  );

  # Prefix all parameters
  %post = map { 'hub.' . $_ => $post{$_} } keys %post;

  # Get user agent
  my $ua = Mojo::UserAgent->new(
    max_redirects => 3,
    name => $UA_NAME
  );

  # Send subscription change to hub
  my $tx = $ua->post($param{hub} => form => \%post);

  my $res = $tx->result;

  # No response
  unless ($res) {
    my $msg = 'Cannot ping hub';
    $msg .= ' - maybe no SSL support' if index($param{hub}, 'https') == 0;
    $log->warn($msg);
    return;
  };

  $mojo->plugins->emit_hook(
    "after_pubsub_$mode" => (
      $c, $param{hub}, \%post, $res->code, $res->body
    ));

  # is 2xx, incl. 204 aka successful and 202 aka accepted
  my $success = $res->is_success ? 1 : 0;

  return ($success, $res->{body}) if wantarray;
  return $success;
};


# Incoming data callback
sub callback {
  my $plugin = shift;
  my $c      = shift;
  my $log    = $c->app->log;

  my $ct = $c->req->headers->header('Content-Type') || 'unknown';
  my $type;

  # Is Atom
  if ($ct =~ m{^application/atom\+xml}) {
    $type = 'atom';
  }

  # Is RSS
  elsif ($ct =~ m{^application/r(?:ss|df)\+xml}) {
    $type = 'rss';
  }

  # Unsupported content type
  else {
    $log->warn("Unsupported media type: $ct") if $c->req->body;
    return _render_fail($c);
  };

  my $dom = Mojo::DOM->new(xml => 1, charset => 'UTF-8');

  # Parse fat ping
  $dom->parse(b($c->req->body)->decode->to_string);

  # Find topics in Payload
  my $topics = _find_topics($type, $dom);

  # No topics to process - but technically fine
  return _render_success($c) unless $topics->[0];

  # Save unfiltered topics for later comparison
  my @old_topics = @$topics;

  # Check for secret and which topics are wanted
  ($topics, my $secret, my $x_hub_on_behalf_of) =
    $c->callback(pubsub_accept => $type, $topics);

  $x_hub_on_behalf_of ||= 1;

  # No topics to process
  # return _render_success( $c => $x_hub_on_behalf_of )
  return _render_success( $c => 1 ) unless scalar @$topics;

  # Todo: Async with on(finish => ..)

  # Secret is needed
  if ($secret) {

    # Unable to verify secret
    unless ( _check_signature( $c, $secret )) {

      $log->debug(
        'Unable to verify secret for ' . join('; ', @$topics)
      );

      # return _render_success( $c => $x_hub_on_behalf_of );
      return _render_success( $c => 1 );
    };
  };

  # Some topics are unwanted
  if (@$topics != @old_topics) {

    # filter dom based on topics
    $topics = _filter_topics($dom, $topics);
  };

  $c->app->plugins->emit_hook(
    on_pubsub_content => $c, $type, $dom
  );

  # Successful
  return _render_success( $c => $x_hub_on_behalf_of );
};


# Find topics of entries
sub _find_topics {
  my $type = shift;
  my $dom  = shift;

  # Get all source links
  my $links = $dom->find('source > link[rel="self"][href]');

  # Save href as topics
  my @topics = @{ $links->map( sub { $_->attr('href') } ) } if $links;

  # Find all entries, regardless if rss or atom
  my $entries = $dom->find('item, feed > entry');

  # Not every entry has a source
  if ($links->size != $entries->size) {

    # One feed or entry
    my $link = $dom->at(
      'feed > link[rel="self"][href],' .
        'channel > link[rel="self"][href]'
      );

    my $self_href;

    # Channel or feed link
    if ($link) {
      $self_href = $link->attr('href');
    }

    # Source of first item in RSS
    elsif (!$self_href && $type eq 'rss') {

      # Possible
      $link = $dom->at('item > source');
      $self_href = $link->attr('url') if $link;
    };

    # Add topic to all entries
    _add_topics($type, $dom, $self_href) if $self_href;

    # Get all source links
    $links = $dom->find('source > link[rel="self"][href]');

    # Save href as topics
    @topics = @{ $links->map( sub { $_->attr('href') } ) } if $links;
  };

  # Unify list
  if (@topics > 1) {
    my %topics = map { $_ => 1 } @topics;
    @topics = sort keys %topics;
  };

  return \@topics;
};


# Add topic to entries
sub _add_topics {
  state $atom_ns = 'http://www.w3.org/2005/Atom';

  my ($type, $dom, $self_href) = @_;

  my $link = qq{};

  # Add source information to each entry
  $dom->find('item, entry')->each(
    sub {
      my $entry = shift;
      my $source;

      # Sources are found
      if (my $sources = $entry->find('source')) {
        foreach my $s (@$sources) {
          $source = $s and last if $s->namespace eq $atom_ns;
        };
      };

      # No source found
      unless ($source) {
        $source = $entry->append_content(qq{})
          ->at(qq{source[xmlns="$atom_ns"]});
      }

      # Link already there
      elsif ($source->at('link[rel="self"][href]')) {
        return $dom;
      };

      # Add link
      $source->append_content( $link );
    });

  return $dom;
};


# filter entries based on their topic
sub _filter_topics {
  my $dom     = shift;

  my %allowed = map { $_ => 1 } @{ shift(@_) };

  my $links = $dom->find(
    'feed > entry > source > link[rel="self"][href],' .
      'item  > source > link[rel="self"][href]'
    );

  my %topics;

  # Delete entries that are not allowed
  $links->each(
    sub {
      my $l = shift;
      my $href = $l->attr('href');

      # entry is not allowed
      unless (exists $allowed{$href}) {
        $l->parent->parent->replace('');
      }

      # Entry is fine and found
      else {
        $topics{$href} = 1;
      };
    });

  return [ sort keys %topics ];
};


# Check signature
sub _check_signature {
  my ($c, $secret) = @_;

  my $req = $c->req;

  # Get signature
  my $signature = $req->headers->header('X-Hub-Signature');

  # Signature expected but not given
  return unless $signature;

  # Delete signature prefix - don't remind, if it's not there.
  $signature =~ s/^sha1=//i;

  # Generate check signature
  my $signature_check = hmac_sha1_sum $req->body, $secret;

  # Return true  if signature check succeeds
  return secure_compare $signature, $signature_check;
};


# Render success
sub _render_success {
  my $c = shift;
  my $x_hub_on_behalf_of = shift;

  # Set X-Hub-On-Behalf-Of header
  if ($x_hub_on_behalf_of &&
        $x_hub_on_behalf_of =~ s/^\s*(\d+)\s*$/$1/) {

    # Set X-Hub-On-Behalf-Of header
    $c->res->headers->header(
      'X-Hub-On-Behalf-Of' => $x_hub_on_behalf_of
    );
  };

  # Render success with no content
  return $c->render(
    status => 204,
    format => 'txt',
    data   => ''
  );
};


# Render fail
sub _render_fail {
  my $c = shift;

  my $fail =<<'FAIL';



    PubSubHubbub Endpoint


    PubSubHubbub Endpoint

      This is an endpoint for the
      PubSubHubbub protocol

    Your request was not correct.


FAIL

  return $c->render(
    data   => $fail,
    status => 400  # bad request
  );
};


1;


__END__