Merge pull request #129 from chihaya/jwt

add support for jwt validation of infohashes

tracker/announce.go

@@ -18,8 +18,14 @@ func (tkr *Tracker) HandleAnnounce(ann *models.Announce, w Writer) (err error) {
 		}
 	}
 
-	torrent, err := tkr.FindTorrent(ann.Infohash)
+	if tkr.Config.JWKSetURI != "" {
+		err := tkr.validateJWT(ann.JWT, ann.Infohash)
+		if err != nil {
+			return err
+		}
+	}
 
+	torrent, err := tkr.FindTorrent(ann.Infohash)
 	if err == models.ErrTorrentDNE && tkr.Config.CreateOnAnnounce {
 		torrent = &models.Torrent{
 			Infohash: ann.Infohash,

tracker/jwt.go

@@ -0,0 +1,147 @@
+// Copyright 2015 The Chihaya Authors. All rights reserved.
+// Use of this source code is governed by the BSD 2-Clause license,
+// which can be found in the LICENSE file.
+
+package tracker
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+
+	oidchttp "github.com/coreos/go-oidc/http"
+	"github.com/coreos/go-oidc/jose"
+	"github.com/golang/glog"
+)
+
+const jwkTTLFallback = 5 * time.Minute
+
+func (tkr *Tracker) updateJWKSetForever() {
+	defer tkr.shutdownWG.Done()
+
+	client := &http.Client{Timeout: 5 * time.Second}
+
+	// Get initial JWK Set.
+	err := tkr.updateJWKSet(client)
+	if err != nil {
+		glog.Warningf("Failed to get initial JWK Set: %s", err)
+	}
+
+	for {
+		select {
+		case <-tkr.shuttingDown:
+			return
+
+		case <-time.After(tkr.Config.JWKSetUpdateInterval.Duration):
+			err = tkr.updateJWKSet(client)
+			if err != nil {
+				glog.Warningf("Failed to update JWK Set: %s", err)
+			}
+		}
+	}
+}
+
+type jwkSet struct {
+	Keys       []jose.JWK `json:"keys"`
+	Issuer     string     `json:"issuer"`
+	validUntil time.Time
+}
+
+func (tkr *Tracker) updateJWKSet(client *http.Client) error {
+	glog.Info("Attemping to update JWK Set")
+	resp, err := client.Get(tkr.Config.JWKSetURI)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	var jwks jwkSet
+	err = json.NewDecoder(resp.Body).Decode(&jwks)
+	if err != nil {
+		return err
+	}
+
+	if len(jwks.Keys) == 0 {
+		return errors.New("Failed to find any keys from JWK Set URI")
+	}
+
+	if jwks.Issuer == "" {
+		return errors.New("Failed to find any issuer from JWK Set URI")
+	}
+
+	ttl, _, _ := oidchttp.Cacheable(resp.Header)
+	if ttl == 0 {
+		ttl = jwkTTLFallback
+	}
+	jwks.validUntil = time.Now().Add(ttl)
+
+	tkr.jwkSet = jwks
+	glog.Info("Successfully updated JWK Set")
+	return nil
+}
+
+func validateJWTSignature(jwt *jose.JWT, jwkSet *jwkSet) (bool, error) {
+	for _, jwk := range jwkSet.Keys {
+		v, err := jose.NewVerifier(jwk)
+		if err != nil {
+			return false, err
+		}
+
+		if err := v.Verify(jwt.Signature, []byte(jwt.Data())); err == nil {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
+func (tkr *Tracker) validateJWT(jwtStr, infohash string) error {
+	jwkSet := tkr.jwkSet
+	if time.Now().After(jwkSet.validUntil) {
+		return fmt.Errorf("Failed verify JWT due to stale JWK Set")
+	}
+
+	jwt, err := jose.ParseJWT(jwtStr)
+	if err != nil {
+		return err
+	}
+
+	validated, err := validateJWTSignature(&jwt, &jwkSet)
+	if err != nil {
+		return err
+	} else if !validated {
+		return errors.New("Failed to verify JWT with all available verifiers")
+	}
+
+	claims, err := jwt.Claims()
+	if err != nil {
+		return err
+	}
+
+	if claimedIssuer, ok, err := claims.StringClaim("iss"); claimedIssuer != jwkSet.Issuer || err != nil || !ok {
+		return errors.New("Failed to validate JWT issuer claim")
+	}
+
+	if claimedAudience, ok, err := claims.StringClaim("aud"); claimedAudience != tkr.Config.JWTAudience || err != nil || !ok {
+		return errors.New("Failed to validate JWT audience claim")
+	}
+
+	claimedInfohash, ok, err := claims.StringClaim("infohash")
+	if err != nil || !ok {
+		return errors.New("Failed to validate JWT infohash claim")
+	}
+
+	unescapedInfohash, err := url.QueryUnescape(claimedInfohash)
+	if err != nil {
+		return errors.New("Failed to unescape JWT infohash claim")
+	}
+
+	if unescapedInfohash != infohash {
+		return errors.New("Failed to match infohash claim with requested infohash")
+	}
+
+	return nil
+}

tracker/models/models.go

@@ -136,6 +136,7 @@ type Announce struct {
 	NumWant    int      `json:"numwant"`
 	PeerID     string   `json:"peer_id"`
 	Uploaded   uint64   `json:"uploaded"`
+	JWT        string   `json:"jwt"`
 
 	Torrent *Torrent `json:"-"`
 	Peer    *Peer    `json:"-"`

tracker/tracker.go

@@ -7,6 +7,7 @@
 package tracker
 
 import (
+	"sync"
 	"time"
 
 	"github.com/golang/glog"
@@ -19,6 +20,12 @@ import (
 // independently of the underlying data transports used.
 type Tracker struct {
 	Config *config.Config
+
+	jwkSet jwkSet
+
+	shuttingDown chan struct{}
+	shutdownWG   sync.WaitGroup
+
 	*Storage
 }
 
@@ -26,16 +33,25 @@ type Tracker struct {
 // Maintenance routines are automatically spawned in the background.
 func New(cfg *config.Config) (*Tracker, error) {
 	tkr := &Tracker{
-		Config:  cfg,
-		Storage: NewStorage(cfg),
+		Config:       cfg,
+		Storage:      NewStorage(cfg),
+		shuttingDown: make(chan struct{}),
 	}
 
+	glog.Info("Starting garbage collection goroutine")
+	tkr.shutdownWG.Add(1)
 	go tkr.purgeInactivePeers(
 		cfg.PurgeInactiveTorrents,
 		time.Duration(float64(cfg.MinAnnounce.Duration)*cfg.ReapRatio),
 		cfg.ReapInterval.Duration,
 	)
 
+	if tkr.Config.JWKSetURI != "" {
+		glog.Info("Starting JWK Set update goroutine")
+		tkr.shutdownWG.Add(1)
+		go tkr.updateJWKSetForever()
+	}
+
 	if cfg.ClientWhitelistEnabled {
 		tkr.LoadApprovedClients(cfg.ClientWhitelist)
 	}
@@ -45,8 +61,8 @@ func New(cfg *config.Config) (*Tracker, error) {
 
 // Close gracefully shutdowns a Tracker by closing any database connections.
 func (tkr *Tracker) Close() error {
-
-	// TODO(jzelinskie): shutdown purgeInactivePeers goroutine.
+	close(tkr.shuttingDown)
+	tkr.shutdownWG.Wait()
 
 	return nil
 }
@@ -73,13 +89,21 @@ type Writer interface {
 // purgeInactivePeers periodically walks the torrent database and removes
 // peers that haven't announced recently.
 func (tkr *Tracker) purgeInactivePeers(purgeEmptyTorrents bool, threshold, interval time.Duration) {
-	for _ = range time.NewTicker(interval).C {
-		before := time.Now().Add(-threshold)
-		glog.V(0).Infof("Purging peers with no announces since %s", before)
+	defer tkr.shutdownWG.Done()
 
-		err := tkr.PurgeInactivePeers(purgeEmptyTorrents, before)
-		if err != nil {
-			glog.Errorf("Error purging torrents: %s", err)
+	for {
+		select {
+		case <-tkr.shuttingDown:
+			return
+
+		case <-time.After(interval):
+			before := time.Now().Add(-threshold)
+			glog.V(0).Infof("Purging peers with no announces since %s", before)
+
+			err := tkr.PurgeInactivePeers(purgeEmptyTorrents, before)
+			if err != nil {
+				glog.Errorf("Error purging torrents: %s", err)
+			}
 		}
 	}
 }