Repository: lf-edge/ekuiper
Pull request #2960 (2024-06-28 04:02:58)

feat: support httppull source with auth

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
Attention: Patch coverage is `50.58824%` with `42 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@2407902`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.

| [Files](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) | Patch % | Lines |
|---|---|---|
| [internal/io/http/client.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?src=pr&el=tree&filepath=internal%2Fio%2Fhttp%2Fclient.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vaHR0cC9jbGllbnQuZ28=) | 50.62% | [24 Missing and 16 partials :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |
| [internal/io/http/httppull\_source.go](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?src=pr&el=tree&filepath=internal%2Fio%2Fhttp%2Fhttppull_source.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#diff-aW50ZXJuYWwvaW8vaHR0cC9odHRwcHVsbF9zb3VyY2UuZ28=) | 50.00% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) |

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2960   +/-   ##
============================================
  Coverage             ?   69.69%           
============================================
  Files                ?      312           
  Lines                ?    29938           
  Branches             ?        0           
============================================
  Hits                 ?    20863           
  Misses               ?     7243           
  Partials             ?     1832           
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2960?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/io/http/client.go b/internal/io/http/client.go
index 9a1136d8..5288c8af 100644
--- a/internal/io/http/client.go
+++ b/internal/io/http/client.go
@@ -32,6 +32,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	"github.com/lf-edge/ekuiper/v2/pkg/cert"
 	"github.com/lf-edge/ekuiper/v2/pkg/message"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 // ClientConf is the configuration for http client
@@ -41,6 +42,25 @@ type ClientConf struct {
 	client       *http.Client
 	compressor   message.Compressor   // compressor used to payload compression when specifies compressAlgorithm
 	decompressor message.Decompressor // decompressor used to payload decompression when specifies compressAlgorithm
+
+	// auth related
+	accessConf        *AccessTokenConf
+	refreshConf       *RefreshTokenConf
+	tokenLastUpdateAt time.Time
+	tokens            map[string]interface{}
+}
+
+type AccessTokenConf struct {
+	Url            string `json:"url"`
+	Body           string `json:"body"`
+	Expire         string `json:"expire"`
+	ExpireInSecond int
+}
+
+type RefreshTokenConf struct {
+	Url     string            `json:"url"`
+	Headers map[string]string `json:"headers"`
+	Body    string            `json:"body"`
 }
 
 type RawConf struct {
@@ -50,6 +70,8 @@ type RawConf struct {
 	BodyType string            `json:"bodyType"`
 	Headers  map[string]string `json:"headers"`
 	Timeout  int               `json:"timeout"`
+
+	OAuth map[string]map[string]interface{} `json:"oauth"`
 	// Could be code or body
 	ResponseType string `json:"responseType"`
 	Compression  string `json:"compression"` // Compression specifies the algorithms used to payload compression
@@ -124,6 +146,39 @@ func (cc *ClientConf) InitConf(device string, props map[string]interface{}) erro
 	if err != nil {
 		return err
 	}
+	// validate oAuth. In order to adapt to manager, the validation is closed to allow empty value
+	if c.OAuth != nil {
+		// validate access token
+		if ap, ok := c.OAuth["access"]; ok {
+			accessConf := &AccessTokenConf{}
+			if err := cast.MapToStruct(ap, accessConf); err != nil {
+				return fmt.Errorf("fail to parse the access properties of oAuth: %v", err)
+			}
+			if accessConf.Url == "" {
+				conf.Log.Warnf("access token url is not set, so ignored the oauth setting")
+				c.OAuth = nil
+			} else {
+				// expire time will update every time when access token is refreshed if expired is set
+				cc.accessConf = accessConf
+			}
+		} else {
+			return fmt.Errorf("if setting oAuth, `access` property is required")
+		}
+		// validate refresh token, it is optional
+		if rp, ok := c.OAuth["refresh"]; ok {
+			refreshConf := &RefreshTokenConf{}
+			if err := cast.MapToStruct(rp, refreshConf); err != nil {
+				return fmt.Errorf("fail to parse the refresh token properties: %v", err)
+			}
+			if refreshConf.Url == "" {
+				conf.Log.Warnf("refresh token url is not set, so ignored the refresh setting")
+				delete(c.OAuth, "refresh")
+			} else {
+				cc.refreshConf = refreshConf
+			}
+		}
+	}
+
 	tlscfg, err := cert.GenTLSConfig(props, "http")
 	if err != nil {
 		return err
@@ -146,9 +201,92 @@ func (cc *ClientConf) InitConf(device string, props map[string]interface{}) erro
 			return fmt.Errorf("init payload decompressor failed, %w", err)
 		}
 	}
+	if cc.accessConf != nil {
+		conf.Log.Infof("Try to get access token from %s", cc.accessConf.Url)
+		ctx := mockContext.NewMockContext("none", "httppull_init")
+		cc.tokens = make(map[string]interface{})
+		err := cc.auth(ctx)
+		if err != nil {
+			return fmt.Errorf("fail to authorize by oAuth: %v", err)
+		}
+	}
+	return nil
+}
+
+// initialize the oAuth access token
+func (cc *ClientConf) auth(ctx api.StreamContext) error {
+	resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.accessConf.Url, nil, true, cc.accessConf.Body)
+	if err != nil {
+		return err
+	}
+	tokens, _, err := cc.parseResponse(ctx, resp)
+	if err != nil {
+		return err
+	}
+	cc.tokens = tokens[0]
+	ctx.GetLogger().Infof("Got access token %v", cc.tokens)
+	expireIn, err := ctx.ParseTemplate(cc.accessConf.Expire, cc.tokens)
+	if err != nil {
+		return fmt.Errorf("fail to parse the expire time for access token: %v", err)
+	}
+	cc.accessConf.ExpireInSecond, err = cast.ToInt(expireIn, cast.CONVERT_ALL)
+	if err != nil {
+		return fmt.Errorf("fail to covert the expire time %s for access token: %v", expireIn, err)
+	}
+	if cc.refreshConf != nil {
+		err := cc.refresh(ctx)
+		if err != nil {
+			return err
+		}
+	} else {
+		cc.tokenLastUpdateAt = time.Now()
+	}
 	return nil
 }
 
+func parseHeaders(ctx api.StreamContext, oHeaders map[string]string, data map[string]interface{}) (map[string]string, error) {
+	headers := make(map[string]string, len(oHeaders))
+	var err error
+	for k, v := range oHeaders {
+		headers[k], err = ctx.ParseTemplate(v, data)
+		if err != nil {
+			return nil, fmt.Errorf("fail to parse the header for refresh token request %s: %v", k, err)
+		}
+	}
+	return headers, nil
+}
+
+func (cc *ClientConf) refresh(ctx api.StreamContext) error {
+	if cc.refreshConf != nil {
+		headers := make(map[string]string, len(cc.refreshConf.Headers))
+		var err error
+		for k, v := range cc.refreshConf.Headers {
+			headers[k], err = ctx.ParseTemplate(v, cc.tokens)
+			if err != nil {
+				return fmt.Errorf("fail to parse the header for refresh token request %s: %v", k, err)
+			}
+		}
+
+		resp, err := httpx.Send(conf.Log, cc.client, "json", http.MethodPost, cc.refreshConf.Url, headers, true, cc.refreshConf.Body)
+		if err != nil {
+			return fmt.Errorf("fail to get refresh token: %v", err)
+		}
+		nt, _, err := cc.parseResponse(ctx, resp)
+		if err != nil {
+			return fmt.Errorf("Cannot parse refresh token response to json: %v", err)
+		}
+		for k, v := range nt[0] {
+			if v != nil {
+				cc.tokens[k] = v
+			}
+		}
+		cc.tokenLastUpdateAt = time.Now()
+		return nil
+	} else {
+		return fmt.Errorf("no oAuth config")
+	}
+}
+
 const (
 	BODY_ERR = "response body error"
 	CODE_ERR = "response code error"
@@ -224,6 +362,10 @@ func (cc *ClientConf) parseResponse(ctx api.StreamContext, resp *http.Response)
 	}
 }
 
+func (cc *ClientConf) parseHeaders(ctx api.StreamContext) (map[string]string, error) {
+	return parseHeaders(ctx, cc.config.Headers, cc.tokens)
+}
+
 func decode(data []byte) ([]map[string]interface{}, error) {
 	var r1 interface{}
 	err := json.Unmarshal(data, &r1)
diff --git a/internal/io/http/httppull_source.go b/internal/io/http/httppull_source.go
index 4224f1db..93b6b755 100644
--- a/internal/io/http/httppull_source.go
+++ b/internal/io/http/httppull_source.go
@@ -59,7 +59,11 @@ func (hps *HttpPullSource) Provision(ctx api.StreamContext, configs map[string]a
 }
 
 func (hps *HttpPullSource) doPull(ctx api.StreamContext) ([]map[string]any, error) {
-	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, hps.config.Headers, true, []byte(hps.config.Body))
+	headers, err := hps.parseHeaders(ctx)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := httpx.Send(ctx.GetLogger(), hps.client, hps.config.BodyType, hps.config.Method, hps.config.Url, headers, true, []byte(hps.config.Body))
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/internal/io/http/client_test.go b/internal/io/http/client_test.go
index 7589024d..1e36d4c6 100644
--- a/internal/io/http/client_test.go
+++ b/internal/io/http/client_test.go
@@ -16,9 +16,12 @@ package http
 
 import (
 	"encoding/json"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 )
 
 func TestInitConf(t *testing.T) {
@@ -134,3 +137,31 @@ func TestDecode(t *testing.T) {
 		require.Equal(t, tc.got, g)
 	}
 }
+
+func TestClientAuth(t *testing.T) {
+	server := createServer()
+	defer func() {
+		server.Close()
+	}()
+
+	c := &ClientConf{}
+	ctx := mockContext.NewMockContext("1", "2")
+	require.NoError(t, c.InitConf("", map[string]interface{}{
+		"oauth": map[string]interface{}{
+			"access": map[string]interface{}{
+				"url":    fmt.Sprintf("%s/auth", server.URL),
+				"expire": "3600",
+				"body":   `{"a":1}`,
+			},
+			"refresh": map[string]interface{}{
+				"url": fmt.Sprintf("%s/refresh", server.URL),
+				"headers": map[string]interface{}{
+					"a": "{{.message}}",
+				},
+				"body": `{"a":1}`,
+			},
+		},
+	}))
+	require.NoError(t, c.auth(ctx))
+	require.NoError(t, c.refresh(ctx))
+}
diff --git a/internal/io/http/httppull_source_test.go b/internal/io/http/httppull_source_test.go
index 12020d23..e76cedaa 100644
--- a/internal/io/http/httppull_source_test.go
+++ b/internal/io/http/httppull_source_test.go
@@ -61,11 +61,33 @@ func handleErr(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(resp)
 }
 
+func handleAuth(w http.ResponseWriter, r *http.Request) {
+	resp := Response{
+		Message: "auth",
+		Code:    200,
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(resp)
+}
+
+func handleRefresh(w http.ResponseWriter, r *http.Request) {
+	resp := Response{
+		Message: "refresh",
+		Code:    200,
+	}
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(http.StatusCreated)
+	json.NewEncoder(w).Encode(resp)
+}
+
 func createServer() *httptest.Server {
 	router := http.NewServeMux()
 	router.HandleFunc("/get", handleGet)
 	router.HandleFunc("/post", handlePost)
 	router.HandleFunc("/err", handleErr)
+	router.HandleFunc("/auth", handleAuth)
+	router.HandleFunc("/refresh", handleRefresh)
 	server := httptest.NewServer(router)
 	return server
 }