Merge branch 'chore/improve-http-redirects' into release/0.8.0

core/adapters/adapter.go

@@ -25,6 +25,10 @@ var (
 	TaskTypeEthTx = models.MustNewTaskType("ethtx")
 	// TaskTypeEthTxABIEncode is the identifier for the EthTxABIEncode adapter.
 	TaskTypeEthTxABIEncode = models.MustNewTaskType("ethtxabiencode")
+	// TaskTypeHTTPGetWithUnrestrictedNetworkAccess is the identifier for the HTTPGet adapter, with local/private IP access enabled.
+	TaskTypeHTTPGetWithUnrestrictedNetworkAccess = models.MustNewTaskType("httpgetwithunrestrictednetworkaccess")
+	// TaskTypeHTTPPostWithUnrestrictedNetworkAccess is the identifier for the HTTPPost adapter, with local/private IP access enabled.
+	TaskTypeHTTPPostWithUnrestrictedNetworkAccess = models.MustNewTaskType("httppostwithunrestrictednetworkaccess")
 	// TaskTypeHTTPGet is the identifier for the HTTPGet adapter.
 	TaskTypeHTTPGet = models.MustNewTaskType("httpget")
 	// TaskTypeHTTPPost is the identifier for the HTTPPost adapter.
@@ -102,6 +106,12 @@ func For(task models.TaskSpec, config orm.ConfigReader, orm *orm.ORM) (*Pipeline
 	case TaskTypeEthTxABIEncode:
 		ba = &EthTxABIEncode{}
 		err = unmarshalParams(task.Params, ba)
+	case TaskTypeHTTPGetWithUnrestrictedNetworkAccess:
+		ba = &HTTPGet{AllowUnrestrictedNetworkAccess: true}
+		err = unmarshalParams(task.Params, ba)
+	case TaskTypeHTTPPostWithUnrestrictedNetworkAccess:
+		ba = &HTTPPost{AllowUnrestrictedNetworkAccess: true}
+		err = unmarshalParams(task.Params, ba)
 	case TaskTypeHTTPGet:
 		ba = &HTTPGet{}
 		err = unmarshalParams(task.Params, ba)

core/adapters/doc.go

@@ -20,11 +20,28 @@
 // The HTTPGet adapter is used to grab the JSON data from the given URL.
 //  { "type": "HTTPGet", "params": {"get": "https://some-api-example.net/api" }}
 //
+// NOTE: For security, since the URL is untrusted, HTTPGet imposes some
+// restrictions on which IPs may be fetched. Local network and multicast IPs
+// are disallowed by default and attempting to connect will result in an error.
+//
+//
 // HTTPPost
 //
 // Sends a POST request to the specified URL and will return the response.
 //  { "type": "HTTPPost", "params": {"post": "https://weiwatchers.com/api" }}
 //
+// NOTE: For security, since the URL is untrusted, HTTPPost imposes some
+// restrictions on which IPs may be fetched. Local network and multicast IPs
+// are disallowed by default and attempting to connect will result in an error.
+//
+// HTTPGetWithUnrestrictedNetworkAccess
+//
+// Identical to HTTPGet except there are no IP restrictions. Use with caution.
+//
+// HTTPPostWithUnrestrictedNetworkAccess
+//
+// Identical to HTTPPost except there are no IP restrictions. Use with caution.
+//
 // JSONParse
 //
 // The JSONParse adapter will obtain the value(s) for the given field(s).

core/adapters/http.go

@@ -23,22 +23,24 @@ import (
 
 // HTTPGet requires a URL which is used for a GET request when the adapter is called.
 type HTTPGet struct {
-	URL          models.WebURL   `json:"url"`
-	GET          models.WebURL   `json:"get"`
-	Headers      http.Header     `json:"headers"`
-	QueryParams  QueryParameters `json:"queryParams"`
-	ExtendedPath ExtendedPath    `json:"extPath"`
+	URL                            models.WebURL   `json:"url"`
+	GET                            models.WebURL   `json:"get"`
+	Headers                        http.Header     `json:"headers"`
+	QueryParams                    QueryParameters `json:"queryParams"`
+	ExtendedPath                   ExtendedPath    `json:"extPath"`
+	AllowUnrestrictedNetworkAccess bool            `json:"-"`
 }
 
 // HTTPRequestConfig holds the configurable settings for an http request
 type HTTPRequestConfig struct {
-	timeout     time.Duration
-	maxAttempts uint
-	sizeLimit   int64
+	timeout                        time.Duration
+	maxAttempts                    uint
+	sizeLimit                      int64
+	allowUnrestrictedNetworkAccess bool
 }
 
 // TaskType returns the type of Adapter.
-func (h *HTTPGet) TaskType() models.TaskType {
+func (hga *HTTPGet) TaskType() models.TaskType {
 	return TaskTypeHTTPGet
 }
 
@@ -50,6 +52,7 @@ func (hga *HTTPGet) Perform(input models.RunInput, store *store.Store) models.Ru
 		return models.NewRunOutputError(err)
 	}
 	httpConfig := defaultHTTPConfig(store)
+	httpConfig.allowUnrestrictedNetworkAccess = hga.AllowUnrestrictedNetworkAccess
 	return sendRequest(input, request, httpConfig)
 }
 
@@ -75,16 +78,17 @@ func (hga *HTTPGet) GetRequest() (*http.Request, error) {
 
 // HTTPPost requires a URL which is used for a POST request when the adapter is called.
 type HTTPPost struct {
-	URL          models.WebURL   `json:"url"`
-	POST         models.WebURL   `json:"post"`
-	Headers      http.Header     `json:"headers"`
-	QueryParams  QueryParameters `json:"queryParams"`
-	Body         *string         `json:"body,omitempty"`
-	ExtendedPath ExtendedPath    `json:"extPath"`
+	URL                            models.WebURL   `json:"url"`
+	POST                           models.WebURL   `json:"post"`
+	Headers                        http.Header     `json:"headers"`
+	QueryParams                    QueryParameters `json:"queryParams"`
+	Body                           *string         `json:"body,omitempty"`
+	ExtendedPath                   ExtendedPath    `json:"extPath"`
+	AllowUnrestrictedNetworkAccess bool            `json:"-"`
 }
 
 // TaskType returns the type of Adapter.
-func (h *HTTPPost) TaskType() models.TaskType {
+func (hpa *HTTPPost) TaskType() models.TaskType {
 	return TaskTypeHTTPPost
 }
 
@@ -96,6 +100,7 @@ func (hpa *HTTPPost) Perform(input models.RunInput, store *store.Store) models.R
 		return models.NewRunOutputError(err)
 	}
 	httpConfig := defaultHTTPConfig(store)
+	httpConfig.allowUnrestrictedNetworkAccess = hpa.AllowUnrestrictedNetworkAccess
 	return sendRequest(input, request, httpConfig)
 }
 
@@ -159,6 +164,9 @@ func sendRequest(input models.RunInput, request *http.Request, config HTTPReques
 	tr := &http.Transport{
 		DisableCompression: true,
 	}
+	if !config.allowUnrestrictedNetworkAccess {
+		tr.DialContext = restrictedDialContext
+	}
 	client := &http.Client{Transport: tr}
 
 	response, err := withRetry(client, request, config)
@@ -345,5 +353,6 @@ func defaultHTTPConfig(store *store.Store) HTTPRequestConfig {
 		store.Config.DefaultHTTPTimeout(),
 		store.Config.DefaultMaxHTTPAttempts(),
 		store.Config.DefaultHTTPLimit(),
+		false,
 	}
 }

core/adapters/http_allowed_ips.go

@@ -0,0 +1,75 @@
+package adapters
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+)
+
+var privateIPBlocks []*net.IPNet
+
+func init() {
+	for _, cidr := range []string{
+		"127.0.0.0/8",    // IPv4 loopback
+		"10.0.0.0/8",     // RFC1918
+		"172.16.0.0/12",  // RFC1918
+		"192.168.0.0/16", // RFC1918
+		"169.254.0.0/16", // RFC3927 link-local
+		"::1/128",        // IPv6 loopback
+		"fe80::/10",      // IPv6 link-local
+		"fc00::/7",       // IPv6 unique local addr
+	} {
+		_, block, err := net.ParseCIDR(cidr)
+		if err != nil {
+			panic(fmt.Errorf("parse error on %q: %v", cidr, err))
+		}
+		privateIPBlocks = append(privateIPBlocks, block)
+	}
+}
+
+func isRestrictedIP(ip net.IP) bool {
+	if !ip.IsGlobalUnicast() ||
+		ip.IsLoopback() ||
+		ip.IsLinkLocalUnicast() ||
+		ip.IsLinkLocalMulticast() ||
+		ip.IsInterfaceLocalMulticast() ||
+		ip.IsUnspecified() ||
+		ip.Equal(net.IPv4bcast) ||
+		ip.Equal(net.IPv4allsys) ||
+		ip.Equal(net.IPv4allrouter) ||
+		ip.Equal(net.IPv4zero) ||
+		ip.IsMulticast() {
+		return true
+	}
+
+	for _, block := range privateIPBlocks {
+		if block.Contains(ip) {
+			return true
+		}
+	}
+	return false
+}
+
+// restrictedDialContext wraps the Dialer such that after successful connection,
+// we check the IP.
+// If the resolved IP is restricted, close the connection and return an error.
+func restrictedDialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	con, err := (&net.Dialer{
+		// Defaults from GoLang standard http package
+		// https://golang.org/pkg/net/http/#RoundTripper
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+		DualStack: true,
+	}).DialContext(ctx, network, address)
+	if err == nil {
+		// If a connection could be established, ensure its not local or private
+		a, _ := con.RemoteAddr().(*net.TCPAddr)
+
+		if isRestrictedIP(a.IP) {
+			defer con.Close()
+			return nil, fmt.Errorf("disallowed IP %s. Connections to local/private and multicast networks are disabled by default for security reasons. If you really want to allow this, consider using the httpgetwithunrestrictednetworkaccess or httppostwithunrestrictednetworkaccess adapter instead", a.IP.String())
+		}
+	}
+	return con, err
+}

core/adapters/http_allowed_ips_test.go

@@ -0,0 +1,44 @@
+package adapters
+
+import (
+	"github.com/stretchr/testify/assert"
+	"net"
+	"testing"
+)
+
+func TestHttpAllowedIPS_isRestrictedIP(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		ip           net.IP
+		isRestricted bool
+	}{
+		{net.ParseIP("1.1.1.1"), false},
+		{net.ParseIP("216.239.32.10"), false},
+		{net.ParseIP("2001:4860:4860::8888"), false},
+		{net.ParseIP("127.0.0.1"), true},
+		{net.ParseIP("255.255.255.255"), true},
+		{net.ParseIP("224.0.0.1"), true},
+		{net.ParseIP("224.0.0.2"), true},
+		{net.ParseIP("224.1.1.1"), true},
+		{net.ParseIP("0.0.0.0"), true},
+		{net.ParseIP("192.168.0.1"), true},
+		{net.ParseIP("192.168.1.255"), true},
+		{net.ParseIP("255.255.255.255"), true},
+		{net.ParseIP("10.0.0.1"), true},
+		{net.ParseIP("::1"), true},
+		{net.ParseIP("fd57:03f9:9ef5:8a81::1"), true},
+		{net.ParseIP("FD00::1"), true},
+		{net.ParseIP("FF02::1"), true},
+		{net.ParseIP("FE80:0000:0000:0000:abcd:abcd:abcd:abcd"), true},
+		{net.IP{0xff, 0x01, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01}, true},
+		{net.IP{0xff, 0x02, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01}, true},
+		{net.IP{0xff, 0x02, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x02}, true},
+	}
+
+	for _, test := range tests {
+		t.Run(test.ip.String(), func(t *testing.T) {
+			assert.Equal(t, test.isRestricted, isRestrictedIP(test.ip))
+		})
+	}
+}

core/adapters/http_test.go

@@ -83,9 +83,10 @@ func TestHTTPGet_Perform(t *testing.T) {
 			defer cleanup()
 
 			hga := adapters.HTTPGet{
-				URL:         cltest.WebURL(t, mock.URL),
-				Headers:     test.headers,
-				QueryParams: test.queryParams,
+				URL:                            cltest.WebURL(t, mock.URL),
+				Headers:                        test.headers,
+				QueryParams:                    test.queryParams,
+				AllowUnrestrictedNetworkAccess: true,
 			}
 			assert.Equal(t, test.queryParams, hga.QueryParams)
 
@@ -111,8 +112,12 @@ func TestHTTP_TooLarge(t *testing.T) {
 		verb    string
 		factory func(models.WebURL) adapters.BaseAdapter
 	}{
-		{"GET", func(url models.WebURL) adapters.BaseAdapter { return &adapters.HTTPGet{URL: url} }},
-		{"POST", func(url models.WebURL) adapters.BaseAdapter { return &adapters.HTTPPost{URL: url} }},
+		{"GET", func(url models.WebURL) adapters.BaseAdapter {
+			return &adapters.HTTPGet{URL: url, AllowUnrestrictedNetworkAccess: true}
+		}},
+		{"POST", func(url models.WebURL) adapters.BaseAdapter {
+			return &adapters.HTTPPost{URL: url, AllowUnrestrictedNetworkAccess: true}
+		}},
 	}
 	for _, test := range tests {
 		t.Run(test.verb, func(t *testing.T) {
@@ -131,6 +136,38 @@ func TestHTTP_TooLarge(t *testing.T) {
 	}
 }
 
+func TestHTTP_PerformWithRestrictedIP(t *testing.T) {
+	cfg := orm.NewConfig()
+	store := &store.Store{Config: cfg}
+
+	tests := []struct {
+		verb    string
+		factory func(models.WebURL) adapters.BaseAdapter
+	}{
+		{"GET", func(url models.WebURL) adapters.BaseAdapter {
+			return &adapters.HTTPGet{URL: url, AllowUnrestrictedNetworkAccess: false}
+		}},
+		{"POST", func(url models.WebURL) adapters.BaseAdapter {
+			return &adapters.HTTPPost{URL: url, AllowUnrestrictedNetworkAccess: false}
+		}},
+	}
+	for _, test := range tests {
+		t.Run(test.verb, func(t *testing.T) {
+			input := cltest.NewRunInputWithResult("inputValue")
+			payload := ""
+			mock, _ := cltest.NewHTTPMockServer(t, http.StatusOK, test.verb, payload)
+			defer mock.Close()
+
+			h := test.factory(cltest.WebURL(t, mock.URL))
+			result := h.Perform(input, store)
+
+			require.Error(t, result.Error())
+			assert.Contains(t, result.Error().Error(), "disallowed IP")
+			assert.Equal(t, "", result.Result().String())
+		})
+	}
+}
+
 func stringRef(str string) *string {
 	return &str
 }
@@ -263,10 +300,11 @@ func TestHttpPost_Perform(t *testing.T) {
 			defer cleanup()
 
 			hpa := adapters.HTTPPost{
-				URL:         cltest.WebURL(t, mock.URL),
-				Headers:     test.headers,
-				QueryParams: test.queryParams,
-				Body:        test.body,
+				URL:                            cltest.WebURL(t, mock.URL),
+				Headers:                        test.headers,
+				QueryParams:                    test.queryParams,
+				Body:                           test.body,
+				AllowUnrestrictedNetworkAccess: true,
 			}
 			assert.Equal(t, test.queryParams, hpa.QueryParams)
 
@@ -653,3 +691,15 @@ func TestHTTP_BuildingURL(t *testing.T) {
 		})
 	}
 }
+
+func TestHTTP_JSONDeserializationDoesNotSetAllowUnrestrictedNetworkAccess(t *testing.T) {
+	hga := adapters.HTTPGet{}
+	err := json.Unmarshal([]byte(`{"allowUnrestrictedNetworkAccess": true}`), &hga)
+	require.NoError(t, err)
+	assert.False(t, hga.AllowUnrestrictedNetworkAccess)
+
+	hpa := adapters.HTTPPost{}
+	err = json.Unmarshal([]byte(`{"allowUnrestrictedNetworkAccess": true}`), &hpa)
+	require.NoError(t, err)
+	assert.False(t, hpa.AllowUnrestrictedNetworkAccess)
+}

core/internal/features_test.go

@@ -572,7 +572,7 @@ func TestIntegration_WeiWatchers(t *testing.T) {
 	require.NoError(t, app.Start())
 
 	j := cltest.NewJobWithLogInitiator()
-	post := cltest.NewTask(t, "httppost", fmt.Sprintf(`{"url":"%v"}`, mockServer.URL))
+	post := cltest.NewTask(t, "httppostwithunrestrictednetworkaccess", fmt.Sprintf(`{"url":"%v"}`, mockServer.URL))
 	tasks := []models.TaskSpec{post}
 	j.Tasks = tasks
 	j = cltest.CreateJobSpecViaWeb(t, app, j)

core/internal/testdata/hello_world_job.json

@@ -1,7 +1,7 @@
 {
   "initiators": [{ "type": "web" }],
   "tasks": [
-    { "type": "HttpGet", "params": {
+    { "type": "HTTPGetWithUnrestrictedNetworkAccess", "params": {
         "get": "https://bitstamp.net/api/ticker/",
         "headers": {
           "Key1": ["value"],

integration-scripts/src/sendEthlogTransaction.ts

@@ -46,7 +46,12 @@ async function sendEthlogTransaction({
         _comment: 'Trigger on logs emitted by ethLog contract',
       },
     ],
-    tasks: [{ type: 'HttpPost', params: { url: echoServerUrl } }],
+    tasks: [
+      {
+        type: 'HttpPostWithUnrestrictedNetworkAccess',
+        params: { url: echoServerUrl },
+      },
+    ],
   }
   const specsUrl = url.resolve(chainlinkUrl, '/v2/specs')
   const Job = await request.post(specsUrl, { json: job }).catch((e: any) => {