Skip to content

Commit bd8ce9e

Browse files
SgtCoDFishSgtCoDFish
authored
Merge pull request #664 from jetstack/add_cyberark
Add CyberArk client and mock server
2 parents de7010b + a811241 commit bd8ce9e

File tree

7 files changed

+474
-15
lines changed

7 files changed

+474
-15
lines changed

‎pkg/client/client_cyberark.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package client
2+
3+
import (
4+
"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
5+
)
6+
7+
type CyberArkClient = dataupload.CyberArkClient
8+
9+
var NewCyberArkClient = dataupload.NewCyberArkClient

‎pkg/internal/cyberark/dataupload/dataupload.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
package dataupload
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"crypto/sha256"
7+
"crypto/x509"
8+
"encoding/hex"
9+
"encoding/json"
10+
"fmt"
11+
"io"
12+
"net/http"
13+
"net/url"
14+
15+
"k8s.io/client-go/transport"
16+
17+
"github.com/jetstack/preflight/api"
18+
"github.com/jetstack/preflight/pkg/version"
19+
)
20+
21+
const (
22+
// maxRetrievePresignedUploadURLBodySize is the maximum allowed size for a response body from the
23+
// Retrieve Presigned Upload URL service.
24+
maxRetrievePresignedUploadURLBodySize = 10 * 1024
25+
)
26+
27+
type CyberArkClient struct {
28+
baseURL string
29+
client *http.Client
30+
31+
authenticateRequest func(req *http.Request) error
32+
}
33+
34+
type Options struct {
35+
ClusterName string
36+
ClusterDescription string
37+
}
38+
39+
func NewCyberArkClient(trustedCAs *x509.CertPool, baseURL string, authenticateRequest func(req *http.Request) error) (*CyberArkClient, error) {
40+
cyberClient := &http.Client{}
41+
tr := http.DefaultTransport.(*http.Transport).Clone()
42+
if trustedCAs != nil {
43+
tr.TLSClientConfig.RootCAs = trustedCAs
44+
}
45+
cyberClient.Transport = transport.DebugWrappers(tr)
46+
47+
return &CyberArkClient{
48+
baseURL: baseURL,
49+
client: cyberClient,
50+
authenticateRequest: authenticateRequest,
51+
}, nil
52+
}
53+
54+
func (c *CyberArkClient) PostDataReadingsWithOptions(ctx context.Context, payload api.DataReadingsPost, opts Options) error {
55+
if opts.ClusterName == "" {
56+
return fmt.Errorf("programmer mistake: the cluster name (aka `cluster_id` in the config file) cannot be left empty")
57+
}
58+
59+
encodedBody := &bytes.Buffer{}
60+
checksum := sha256.New()
61+
if err := json.NewEncoder(io.MultiWriter(encodedBody, checksum)).Encode(payload); err != nil {
62+
return err
63+
}
64+
65+
presignedUploadURL, err := c.retrievePresignedUploadURL(ctx, hex.EncodeToString(checksum.Sum(nil)), opts)
66+
if err != nil {
67+
return err
68+
}
69+
70+
req, err := http.NewRequestWithContext(ctx, http.MethodPost, presignedUploadURL, encodedBody)
71+
if err != nil {
72+
return err
73+
}
74+
75+
req.Header.Set("Content-Type", "application/json")
76+
version.SetUserAgent(req)
77+
78+
res, err := c.client.Do(req)
79+
if err != nil {
80+
return err
81+
}
82+
defer res.Body.Close()
83+
84+
if code := res.StatusCode; code < 200 || code >= 300 {
85+
body, _ := io.ReadAll(io.LimitReader(res.Body, 500))
86+
if len(body) == 0 {
87+
body = []byte(`<empty body>`)
88+
}
89+
return fmt.Errorf("received response with status code %d: %s", code, bytes.TrimSpace(body))
90+
}
91+
92+
return nil
93+
}
94+
95+
func (c *CyberArkClient) retrievePresignedUploadURL(ctx context.Context, checksum string, opts Options) (string, error) {
96+
uploadURL, err := url.JoinPath(c.baseURL, "/api/data/kubernetes/upload")
97+
if err != nil {
98+
return "", err
99+
}
100+
101+
request := struct {
102+
ClusterID string `json:"cluster_id"`
103+
ClusterDescription string `json:"cluster_description"`
104+
Checksum string `json:"checksum_sha256"`
105+
}{
106+
ClusterID: opts.ClusterName,
107+
ClusterDescription: opts.ClusterDescription,
108+
Checksum: checksum,
109+
}
110+
111+
encodedBody := &bytes.Buffer{}
112+
if err := json.NewEncoder(encodedBody).Encode(request); err != nil {
113+
return "", err
114+
}
115+
116+
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, encodedBody)
117+
if err != nil {
118+
return "", err
119+
}
120+
121+
req.Header.Set("Content-Type", "application/json")
122+
if err := c.authenticateRequest(req); err != nil {
123+
return "", fmt.Errorf("failed to authenticate request")
124+
}
125+
version.SetUserAgent(req)
126+
127+
res, err := c.client.Do(req)
128+
if err != nil {
129+
return "", err
130+
}
131+
defer res.Body.Close()
132+
133+
if code := res.StatusCode; code < 200 || code >= 300 {
134+
body, _ := io.ReadAll(io.LimitReader(res.Body, 500))
135+
if len(body) == 0 {
136+
body = []byte(`<empty body>`)
137+
}
138+
return "", fmt.Errorf("received response with status code %d: %s", code, bytes.TrimSpace(body))
139+
}
140+
141+
response := struct {
142+
URL string `json:"url"`
143+
}{}
144+
145+
if err := json.NewDecoder(io.LimitReader(res.Body, maxRetrievePresignedUploadURLBodySize)).Decode(&response); err != nil {
146+
if err == io.ErrUnexpectedEOF {
147+
return "", fmt.Errorf("rejecting JSON response from server as it was too large or was truncated")
148+
}
149+
150+
return "", fmt.Errorf("failed to parse JSON from otherwise successful request to start data upload: %s", err)
151+
}
152+
153+
return response.URL, nil
154+
}

‎pkg/internal/cyberark/dataupload/dataupload_test.go‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
package dataupload_test
2+
3+
import (
4+
"context"
5+
"crypto/x509"
6+
"encoding/pem"
7+
"fmt"
8+
"net/http"
9+
"testing"
10+
"time"
11+
12+
"github.com/stretchr/testify/require"
13+
14+
"github.com/jetstack/preflight/api"
15+
"github.com/jetstack/preflight/pkg/internal/cyberark/dataupload"
16+
)
17+
18+
func TestCyberArkClient_PostDataReadingsWithOptions(t *testing.T) {
19+
fakeTime := time.Unix(123, 0)
20+
defaultPayload := api.DataReadingsPost{
21+
AgentMetadata: &api.AgentMetadata{
22+
Version: "test-version",
23+
ClusterID: "test",
24+
},
25+
DataGatherTime: fakeTime,
26+
DataReadings: []*api.DataReading{
27+
{
28+
ClusterID: "success-cluster-id",
29+
DataGatherer: "test-gatherer",
30+
Timestamp: api.Time{Time: fakeTime},
31+
Data: map[string]interface{}{"test": "data"},
32+
SchemaVersion: "v1",
33+
},
34+
},
35+
}
36+
defaultOpts := dataupload.Options{
37+
ClusterName: "success-cluster-id",
38+
ClusterDescription: "success-cluster-description",
39+
}
40+
41+
setToken := func(token string) func(*http.Request) error {
42+
return func(req *http.Request) error {
43+
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token))
44+
return nil
45+
}
46+
}
47+
48+
tests := []struct {
49+
name string
50+
payload api.DataReadingsPost
51+
authenticate func(req *http.Request) error
52+
opts dataupload.Options
53+
requireFn func(t *testing.T, err error)
54+
}{
55+
{
56+
name: "successful upload",
57+
payload: defaultPayload,
58+
opts: defaultOpts,
59+
authenticate: setToken("success-token"),
60+
requireFn: func(t *testing.T, err error) {
61+
require.NoError(t, err)
62+
},
63+
},
64+
{
65+
name: "error when cluster name is empty",
66+
payload: defaultPayload,
67+
opts: dataupload.Options{ClusterName: ""},
68+
authenticate: setToken("success-token"),
69+
requireFn: func(t *testing.T, err error) {
70+
require.ErrorContains(t, err, "programmer mistake: the cluster name")
71+
},
72+
},
73+
{
74+
name: "error when bearer token is incorrect",
75+
payload: defaultPayload,
76+
opts: defaultOpts,
77+
authenticate: setToken("fail-token"),
78+
requireFn: func(t *testing.T, err error) {
79+
require.ErrorContains(t, err, "received response with status code 500: should authenticate using the correct bearer token")
80+
},
81+
},
82+
{
83+
name: "invalid JSON from server (RetrievePresignedUploadURL step)",
84+
payload: defaultPayload,
85+
opts: dataupload.Options{ClusterName: "invalid-json-retrieve-presigned", ClusterDescription: defaultOpts.ClusterDescription},
86+
authenticate: setToken("success-token"),
87+
requireFn: func(t *testing.T, err error) {
88+
require.ErrorContains(t, err, "rejecting JSON response from server as it was too large or was truncated")
89+
},
90+
},
91+
{
92+
name: "500 from server (PostData step)",
93+
payload: defaultPayload,
94+
opts: dataupload.Options{ClusterName: "invalid-response-post-data", ClusterDescription: defaultOpts.ClusterDescription},
95+
authenticate: setToken("success-token"),
96+
requireFn: func(t *testing.T, err error) {
97+
require.ErrorContains(t, err, "received response with status code 500: mock error")
98+
},
99+
},
100+
}
101+
102+
for _, tc := range tests {
103+
t.Run(tc.name, func(t *testing.T) {
104+
server := dataupload.MockDataUploadServer()
105+
defer server.Close()
106+
107+
certPool := x509.NewCertPool()
108+
require.True(t, certPool.AppendCertsFromPEM(pem.EncodeToMemory(&pem.Block{
109+
Type: "CERTIFICATE",
110+
Bytes: server.Server.TLS.Certificates[0].Certificate[0],
111+
})))
112+
113+
cyberArkClient, err := dataupload.NewCyberArkClient(certPool, server.Server.URL, tc.authenticate)
114+
require.NoError(t, err)
115+
116+
err = cyberArkClient.PostDataReadingsWithOptions(context.TODO(), tc.payload, tc.opts)
117+
tc.requireFn(t, err)
118+
})
119+
}
120+
}

0 commit comments

Comments
 (0)