nsqd: support POST auth

apps/nsqd/options.go

@@ -134,6 +134,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 
  authHTTPAddresses := app.StringArray{}
  flagSet.Var(&authHTTPAddresses, "auth-http-address", "<addr>:<port> or a full url to query auth server (may be given multiple times)")
+ flagSet.String("auth-http-request-method", opts.AuthHTTPRequestMethod, "HTTP method to use for auth server requests")
  flagSet.String("broadcast-address", opts.BroadcastAddress, "address that will be registered with lookupd (defaults to the OS hostname)")
  flagSet.Int("broadcast-tcp-port", opts.BroadcastTCPPort, "TCP port that will be registered with lookupd (defaults to the TCP port that this nsqd is listening on)")
  flagSet.Int("broadcast-http-port", opts.BroadcastHTTPPort, "HTTP port that will be registered with lookupd (defaults to the HTTP port that this nsqd is listening on)")

internal/auth/authorizations.go

@@ -76,13 +76,13 @@ func (a *State) IsExpired() bool {
 }
 
 func QueryAnyAuthd(authd []string, remoteIP string, tlsEnabled bool, commonName string, authSecret string,
- clientTLSConfig *tls.Config, connectTimeout time.Duration, requestTimeout time.Duration) (*State, error) {
+ clientTLSConfig *tls.Config, connectTimeout time.Duration, requestTimeout time.Duration, httpRequestMethod string) (*State, error) {
  var retErr error
  start := rand.Int()
  n := len(authd)
  for i := 0; i < n; i++ {
  a := authd[(i+start)%n]
- authState, err := QueryAuthd(a, remoteIP, tlsEnabled, commonName, authSecret, clientTLSConfig, connectTimeout, requestTimeout)
+ authState, err := QueryAuthd(a, remoteIP, tlsEnabled, commonName, authSecret, clientTLSConfig, connectTimeout, requestTimeout, httpRequestMethod)
  if err != nil {
  es := fmt.Sprintf("failed to auth against %s - %s", a, err)
  if retErr != nil {
@@ -97,7 +97,8 @@ func QueryAnyAuthd(authd []string, remoteIP string, tlsEnabled bool, commonName
 }
 
 func QueryAuthd(authd string, remoteIP string, tlsEnabled bool, commonName string, authSecret string,
- clientTLSConfig *tls.Config, connectTimeout time.Duration, requestTimeout time.Duration) (*State, error) {
+ clientTLSConfig *tls.Config, connectTimeout time.Duration, requestTimeout time.Duration, httpRequestMethod string) (*State, error) {
+ var authState State
  v := url.Values{}
  v.Set("remote_ip", remoteIP)
  if tlsEnabled {
@@ -110,15 +111,21 @@ func QueryAuthd(authd string, remoteIP string, tlsEnabled bool, commonName strin
 
  var endpoint string
  if strings.Contains(authd, "://") {
- endpoint = fmt.Sprintf("%s?%s", authd, v.Encode())
+ endpoint = authd
  } else {
- endpoint = fmt.Sprintf("http://%s/auth?%s", authd, v.Encode())
+ endpoint = fmt.Sprintf("http://%s/auth", authd)
  }
 
- var authState State
  client := http_api.NewClient(clientTLSConfig, connectTimeout, requestTimeout)
- if err := client.GETV1(endpoint, &authState); err != nil {
- return nil, err
+ if httpRequestMethod == "post" {
+ if err := client.POSTV1(endpoint, v, &authState); err != nil {
+ return nil, err
+ }
+ } else {
+ endpoint = fmt.Sprintf("%s?%s", endpoint, v.Encode())
+ if err := client.GETV1(endpoint, &authState); err != nil {
+ return nil, err
+ }
  }
 
  // validation on response

internal/clusterinfo/data.go

@@ -878,7 +878,7 @@ func (c *ClusterInfo) nsqlookupdPOST(addrs []string, uri string, qs string) erro
  for _, addr := range addrs {
  endpoint := fmt.Sprintf("http://%s/%s?%s", addr, uri, qs)
  c.logf("CI: querying nsqlookupd %s", endpoint)
- err := c.client.POSTV1(endpoint)
+ err := c.client.POSTV1(endpoint, nil, nil)
  if err != nil {
  errs = append(errs, err)
  }
@@ -894,7 +894,7 @@ func (c *ClusterInfo) producersPOST(pl Producers, uri string, qs string) error {
  for _, p := range pl {
  endpoint := fmt.Sprintf("http://%s/%s?%s", p.HTTPAddress(), uri, qs)
  c.logf("CI: querying nsqd %s", endpoint)
- err := c.client.POSTV1(endpoint)
+ err := c.client.POSTV1(endpoint, nil, nil)
  if err != nil {
  errs = append(errs, err)
  }

internal/http_api/api_request.go

@@ -1,6 +1,7 @@
 package http_api
 
 import (
+ "bytes"
  "crypto/tls"
  "encoding/json"
  "fmt"
@@ -86,14 +87,26 @@ retry:
 
 // PostV1 is a helper function to perform a V1 HTTP request
 // and parse our NSQ daemon's expected response format, with deadlines.
-func (c *Client) POSTV1(endpoint string) error {
+func (c *Client) POSTV1(endpoint string, data url.Values, v interface{}) error {
 retry:
- req, err := http.NewRequest("POST", endpoint, nil)
+ var reqBody io.Reader
+ if data != nil {
+ js, err := json.Marshal(data)
+ if err != nil {
+ return fmt.Errorf("failed to marshal POST data to endpoint: %v", endpoint)
+ }
+ reqBody = bytes.NewBuffer(js)
+ }
+
+ req, err := http.NewRequest("POST", endpoint, reqBody)
  if err != nil {
  return err
  }
 
  req.Header.Add("Accept", "application/vnd.nsq; version=1.0")
+ if reqBody != nil {
+ req.Header.Add("Content-Type", "application/json")
+ }
 
  resp, err := c.c.Do(req)
  if err != nil {
@@ -116,6 +129,10 @@ retry:
  return fmt.Errorf("got response %s %q", resp.Status, body)
  }
 
+ if v != nil {
+ return json.Unmarshal(body, &v)
+ }
+
  return nil
 }
 

nsqd/client_v2.go

@@ -659,7 +659,9 @@ func (c *clientV2) QueryAuthd() error {
  remoteIP, tlsEnabled, commonName, c.AuthSecret,
  c.nsqd.clientTLSConfig,
  c.nsqd.getOpts().HTTPClientConnectTimeout,
- c.nsqd.getOpts().HTTPClientRequestTimeout)
+ c.nsqd.getOpts().HTTPClientRequestTimeout,
+ c.nsqd.getOpts().AuthHTTPRequestMethod,
+ )
  if err != nil {
  return err
  }

nsqd/nsqd.go

@@ -135,6 +135,10 @@ func New(opts *Options) (*NSQD, error) {
  }
  n.clientTLSConfig = clientTLSConfig
 
+ if opts.AuthHTTPRequestMethod != "post" && opts.AuthHTTPRequestMethod != "get" {
+ return nil, errors.New("--auth-http-request-method must be post or get")
+ }
+
  for _, v := range opts.E2EProcessingLatencyPercentiles {
  if v <= 0 || v > 1 {
  return nil, fmt.Errorf("invalid E2E processing latency percentile: %v", v)

nsqd/nsqd_test.go

@@ -336,11 +336,11 @@ func TestCluster(t *testing.T) {
  test.Nil(t, err)
 
  url := fmt.Sprintf("http://%s/topic/create?topic=%s", nsqd.RealHTTPAddr(), topicName)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url, nil, nil)
  test.Nil(t, err)
 
  url = fmt.Sprintf("http://%s/channel/create?topic=%s&channel=ch", nsqd.RealHTTPAddr(), topicName)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url, nil, nil)
  test.Nil(t, err)
 
  // allow some time for nsqd to push info to nsqlookupd
@@ -394,7 +394,7 @@ func TestCluster(t *testing.T) {
  test.Equal(t, "ch", lr.Channels[0])
 
  url = fmt.Sprintf("http://%s/topic/delete?topic=%s", nsqd.RealHTTPAddr(), topicName)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(url, nil, nil)
  test.Nil(t, err)
 
  // allow some time for nsqd to push info to nsqlookupd

nsqd/options.go

@@ -27,6 +27,7 @@ type Options struct {
  BroadcastHTTPPort int `flag:"broadcast-http-port"`
  NSQLookupdTCPAddresses []string `flag:"lookupd-tcp-address" cfg:"nsqlookupd_tcp_addresses"`
  AuthHTTPAddresses []string `flag:"auth-http-address" cfg:"auth_http_addresses"`
+ AuthHTTPRequestMethod string `flag:"auth-http-request-method" cfg:"auth_http_request_method"`
  HTTPClientConnectTimeout time.Duration `flag:"http-client-connect-timeout" cfg:"http_client_connect_timeout"`
  HTTPClientRequestTimeout time.Duration `flag:"http-client-request-timeout" cfg:"http_client_request_timeout"`
 
@@ -110,6 +111,7 @@ func NewOptions() *Options {
 
  NSQLookupdTCPAddresses: make([]string, 0),
  AuthHTTPAddresses: make([]string, 0),
+ AuthHTTPRequestMethod: "get",
 
  HTTPClientConnectTimeout: 2 * time.Second,
  HTTPClientRequestTimeout: 5 * time.Second,

nsqd/protocol_v2_test.go

@@ -18,6 +18,7 @@ import (
  "os"
  "runtime"
  "strconv"
+ "strings"
  "sync"
  "sync/atomic"
  "testing"
@@ -1476,24 +1477,30 @@ func TestClientAuth(t *testing.T) {
  authSuccess := ""
  tlsEnabled := false
  commonName := ""
- runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName)
+ httpAuthRequestMethod := "get"
+ runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName, httpAuthRequestMethod)
 
  // now one that will succeed
  authResponse = `{"ttl":10, "authorizations":
  [{"topic":"test", "channels":[".*"], "permissions":["subscribe","publish"]}]
  }`
  authError = ""
  authSuccess = `{"identity":"","identity_url":"","permission_count":1}`
- runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName)
+ runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName, httpAuthRequestMethod)
 
  // one with TLS enabled
  tlsEnabled = true
  commonName = "test.local"
- runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName)
+ runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName, httpAuthRequestMethod)
+
+ // test POST based authentication
+ httpAuthRequestMethod = "post"
+ runAuthTest(t, authResponse, authSecret, authError, authSuccess, tlsEnabled, commonName, httpAuthRequestMethod)
+
 }
 
 func runAuthTest(t *testing.T, authResponse string, authSecret string, authError string,
- authSuccess string, tlsEnabled bool, commonName string) {
+ authSuccess string, tlsEnabled bool, commonName string, httpAuthRequestMethod string) {
  var err error
  var expectedRemoteIP string
  expectedTLS := "false"
@@ -1503,11 +1510,23 @@ func runAuthTest(t *testing.T, authResponse string, authSecret string, authError
 
  authd := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
  t.Logf("in test auth handler %s", r.RequestURI)
- r.ParseForm()
- test.Equal(t, expectedRemoteIP, r.Form.Get("remote_ip"))
- test.Equal(t, expectedTLS, r.Form.Get("tls"))
- test.Equal(t, commonName, r.Form.Get("common_name"))
- test.Equal(t, authSecret, r.Form.Get("secret"))
+ test.Equal(t, httpAuthRequestMethod, strings.ToLower(r.Method))
+
+ var values url.Values
+
+ if r.Method == "POST" {
+ err = json.NewDecoder(r.Body).Decode(&values)
+ if err != nil {
+ t.Error(err)
+ }
+ } else {
+ r.ParseForm()
+ values = r.Form
+ }
+ test.Equal(t, expectedRemoteIP, values.Get("remote_ip"))
+ test.Equal(t, expectedTLS, values.Get("tls"))
+ test.Equal(t, commonName, values.Get("common_name"))
+ test.Equal(t, authSecret, values.Get("secret"))
  fmt.Fprint(w, authResponse)
  }))
  defer authd.Close()
@@ -1519,6 +1538,7 @@ func runAuthTest(t *testing.T, authResponse string, authSecret string, authError
  opts.Logger = test.NewTestLogger(t)
  opts.LogLevel = LOG_DEBUG
  opts.AuthHTTPAddresses = []string{addr.Host}
+ opts.AuthHTTPRequestMethod = httpAuthRequestMethod
  if tlsEnabled {
  opts.TLSCert = "./test/certs/server.pem"
  opts.TLSKey = "./test/certs/server.key"

nsqlookupd/nsqlookupd_test.go

@@ -220,7 +220,7 @@ func TestTombstoneRecover(t *testing.T) {
 
  endpoint := fmt.Sprintf("http://%s/topic/tombstone?topic=%s&node=%s:%d",
  httpAddr, topicName, HostAddr, HTTPPort)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint, nil, nil)
  test.Nil(t, err)
 
  pr := ProducersDoc{}
@@ -263,7 +263,7 @@ func TestTombstoneUnregister(t *testing.T) {
 
  endpoint := fmt.Sprintf("http://%s/topic/tombstone?topic=%s&node=%s:%d",
  httpAddr, topicName, HostAddr, HTTPPort)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint, nil, nil)
  test.Nil(t, err)
 
  pr := ProducersDoc{}
@@ -348,7 +348,7 @@ func TestTombstonedNodes(t *testing.T) {
 
  endpoint := fmt.Sprintf("http://%s/topic/tombstone?topic=%s&node=%s:%d",
  httpAddr, topicName, HostAddr, HTTPPort)
- err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint)
+ err = http_api.NewClient(nil, ConnectTimeout, RequestTimeout).POSTV1(endpoint, nil, nil)
  test.Nil(t, err)
 
  producers, _ = ci.GetLookupdProducers(lookupdHTTPAddrs)