From 0f40c0b3757659d9ba7967a57d6dd1650bc5bccc Mon Sep 17 00:00:00 2001 From: Louis Seubert Date: Sun, 13 Sep 2026 10:53:42 +0200 Subject: [PATCH] feat: artifacts client --- .forgejo/workflows/default.yml | 4 +- artifact/artifacts.go | 160 +++++++++++++++++++++++++ artifact/artifacts_test.go | 182 ++++++++++++++++++++++++++++ artifact/client.go | 33 +++++ artifact/e2e_test.go | 114 ++++++++++++++++++ artifact/jwt.go | 84 +++++++++++++ artifact/jwt_test.go | 96 +++++++++++++++ artifact/storage.go | 212 +++++++++++++++++++++++++++++++++ artifact/storage_test.go | 173 +++++++++++++++++++++++++++ client.go | 37 +++++- client_test.go | 71 +++++++++++ 11 files changed, 1160 insertions(+), 6 deletions(-) create mode 100644 artifact/artifacts.go create mode 100644 artifact/artifacts_test.go create mode 100644 artifact/client.go create mode 100644 artifact/e2e_test.go create mode 100644 artifact/jwt.go create mode 100644 artifact/jwt_test.go create mode 100644 artifact/storage.go create mode 100644 artifact/storage_test.go create mode 100644 client_test.go diff --git a/.forgejo/workflows/default.yml b/.forgejo/workflows/default.yml index a6f7bb2..4615495 100644 --- a/.forgejo/workflows/default.yml +++ b/.forgejo/workflows/default.yml @@ -12,6 +12,6 @@ jobs: steps: - name: checkout uses: https://code.geekeey.de/actions/checkout@1 - - name: go run + - name: go test run: | - go test + go test ./... diff --git a/artifact/artifacts.go b/artifact/artifacts.go new file mode 100644 index 0000000..963440e --- /dev/null +++ b/artifact/artifacts.go @@ -0,0 +1,160 @@ +package artifact + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "code.geekeey.de/actions/sdk" +) + +type CreateArtifactRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + Name string `json:"name"` + Version int32 `json:"version"` + + ExpiresAt *time.Time `json:"expiresAt,omitempty"` +} + +type CreateArtifactResponse struct { + Ok bool `json:"ok"` + SignedUploadUrl string `json:"signedUploadUrl"` +} + +type FinalizeArtifactRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + Name string `json:"name"` + Size int64 `json:"size"` + Hash string `json:"hash,omitempty"` +} + +type FinalizeArtifactResponse struct { + Ok bool `json:"ok"` + ArtifactId string `json:"artifactId"` +} + +type ListArtifactsRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + NameFilter string `json:"nameFilter"` + IdFilter string `json:"idFilter"` +} + +type Artifact struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + Id string `json:"databaseId"` + Name string `json:"name"` + Size string `json:"size"` + CreatedAt *time.Time `json:"createdAt"` + Digest string `json:"digest"` +} + +type ListArtifactsResponse struct { + Artifacts []Artifact `json:"artifacts"` +} + +type GetSignedArtifactURLRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + Name string `json:"name"` +} + +type GetSignedArtifactURLResponse struct { + SignedUrl string `json:"signedUrl"` +} + +type DeleteArtifactRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + Name string `json:"name"` +} + +type DeleteArtifactResponse struct { + Ok bool `json:"ok"` + ArtifactId string `json:"artifactId"` +} + +func (c *Client) CreateArtifact(ctx context.Context, req CreateArtifactRequest) (*CreateArtifactResponse, error) { + endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact" + var result CreateArtifactResponse + if err := c.twirp(ctx, endpoint, req, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *Client) FinalizeArtifact(ctx context.Context, req FinalizeArtifactRequest) (*FinalizeArtifactResponse, error) { + endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/FinalizeArtifact" + var result FinalizeArtifactResponse + if err := c.twirp(ctx, endpoint, req, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *Client) ListArtifacts(ctx context.Context, req ListArtifactsRequest) (*ListArtifactsResponse, error) { + endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/ListArtifacts" + var result ListArtifactsResponse + if err := c.twirp(ctx, endpoint, req, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *Client) GetSignedArtifactURL(ctx context.Context, req GetSignedArtifactURLRequest) (*GetSignedArtifactURLResponse, error) { + endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/GetSignedArtifactURL" + var result GetSignedArtifactURLResponse + if err := c.twirp(ctx, endpoint, req, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *Client) DeleteArtifact(ctx context.Context, req DeleteArtifactRequest) (*DeleteArtifactResponse, error) { + endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/DeleteArtifact" + var result DeleteArtifactResponse + if err := c.twirp(ctx, endpoint, req, &result); err != nil { + return nil, err + } + return &result, nil +} + +func (c *Client) twirp(ctx context.Context, url string, in any, out any) error { + var b bytes.Buffer + + if err := json.NewEncoder(&b).Encode(in); err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &b) + if err != nil { + return err + } + req.Header.Set("content-type", "application/json") + req.Header.Set("accept", "application/json") + + res, err := c.Do(req) + if err != nil { + return err + } + defer res.Body.Close() + if !sdk.HasContentType(res, "application/json") { + return fmt.Errorf("content-type: %q should be %q", + res.Header.Get("content-type"), "application/json") + } + dec := json.NewDecoder(io.LimitReader(res.Body, 32*1024)) + + if res.StatusCode >= 200 && res.StatusCode < 300 { + return dec.Decode(out) + } else { + body, _ := io.ReadAll(res.Body) + return fmt.Errorf("twirp call %s failed: %d: %s", url, res.StatusCode, string(body)) + } +} diff --git a/artifact/artifacts_test.go b/artifact/artifacts_test.go new file mode 100644 index 0000000..e29b260 --- /dev/null +++ b/artifact/artifacts_test.go @@ -0,0 +1,182 @@ +package artifact + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +type capturedRequest struct { + path string + auth string + body []byte +} + +func newTestServer(t *testing.T, status int, contentType, response string) (*httptest.Server, *capturedRequest) { + t.Helper() + + got := &capturedRequest{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.path = r.URL.Path + got.auth = r.Header.Get("Authorization") + got.body, _ = io.ReadAll(r.Body) + w.Header().Set("content-type", contentType) + w.WriteHeader(status) + _, _ = io.WriteString(w, response) + })) + t.Cleanup(server.Close) + + return server, got +} + +func TestClient_CreateArtifact(t *testing.T) { + t.Parallel() + + server, got := newTestServer(t, http.StatusOK, "application/json", + `{"ok":true,"signedUploadUrl":"https://example.test/upload"}`) + + c := NewClient(server.URL, "secret") + res, err := c.CreateArtifact(context.Background(), CreateArtifactRequest{RunID: "run", JobRunID: "job", Name: "foo", Version: 4}) + if err != nil { + t.Fatal(err) + } + + if !res.Ok || res.SignedUploadUrl != "https://example.test/upload" { + t.Errorf("unexpected response: %+v", res) + } + if got, want := got.path, "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } + if got, want := got.auth, "Bearer secret"; got != want { + t.Errorf("expected Authorization %q to be %q", got, want) + } + var req CreateArtifactRequest + if err := json.Unmarshal(got.body, &req); err != nil { + t.Fatal(err) + } + if req != (CreateArtifactRequest{RunID: "run", JobRunID: "job", Name: "foo", Version: 4}) { + t.Errorf("unexpected request: %+v", req) + } +} + +func TestClient_FinalizeArtifact(t *testing.T) { + t.Parallel() + + server, got := newTestServer(t, http.StatusOK, "application/json", + `{"ok":true,"artifactId":"42"}`) + + c := NewClient(server.URL, "secret") + res, err := c.FinalizeArtifact(context.Background(), FinalizeArtifactRequest{RunID: "run", JobRunID: "job", Name: "foo", Size: 10, Hash: "abc"}) + if err != nil { + t.Fatal(err) + } + + if !res.Ok || res.ArtifactId != "42" { + t.Errorf("unexpected response: %+v", res) + } + if got, want := got.path, "/twirp/github.actions.results.api.v1.ArtifactService/FinalizeArtifact"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } + var req FinalizeArtifactRequest + if err := json.Unmarshal(got.body, &req); err != nil { + t.Fatal(err) + } + if req.Size != 10 || req.Hash != "abc" { + t.Errorf("unexpected request: %+v", req) + } +} + +func TestClient_ListArtifacts(t *testing.T) { + t.Parallel() + + server, got := newTestServer(t, http.StatusOK, "application/json", + `{"artifacts":[{"databaseId":"7","name":"foo"}]}`) + + c := NewClient(server.URL, "secret") + res, err := c.ListArtifacts(context.Background(), ListArtifactsRequest{NameFilter: "foo"}) + if err != nil { + t.Fatal(err) + } + + if len(res.Artifacts) != 1 || res.Artifacts[0].Id != "7" || res.Artifacts[0].Name != "foo" { + t.Errorf("unexpected response: %+v", res) + } + if got, want := got.path, "/twirp/github.actions.results.api.v1.ArtifactService/ListArtifacts"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } +} + +func TestClient_GetSignedArtifactURL(t *testing.T) { + t.Parallel() + + server, got := newTestServer(t, http.StatusOK, "application/json", + `{"signedUrl":"https://example.test/download"}`) + + c := NewClient(server.URL, "secret") + res, err := c.GetSignedArtifactURL(context.Background(), GetSignedArtifactURLRequest{RunID: "run", JobRunID: "job", Name: "foo"}) + if err != nil { + t.Fatal(err) + } + + if res.SignedUrl != "https://example.test/download" { + t.Errorf("unexpected response: %+v", res) + } + if got, want := got.path, "/twirp/github.actions.results.api.v1.ArtifactService/GetSignedArtifactURL"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } +} + +func TestClient_DeleteArtifact(t *testing.T) { + t.Parallel() + + server, got := newTestServer(t, http.StatusOK, "application/json", + `{"ok":true,"artifactId":"7"}`) + + c := NewClient(server.URL, "secret") + res, err := c.DeleteArtifact(context.Background(), DeleteArtifactRequest{RunID: "run", JobRunID: "job", Name: "foo"}) + if err != nil { + t.Fatal(err) + } + + if !res.Ok || res.ArtifactId != "7" { + t.Errorf("unexpected response: %+v", res) + } + if got, want := got.path, "/twirp/github.actions.results.api.v1.ArtifactService/DeleteArtifact"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } +} + +func TestClient_twirp(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + status int + contentType string + body string + wantErr string + }{ + {"bad content type", http.StatusOK, "text/plain", "nope", "content-type"}, + {"server error", http.StatusInternalServerError, "application/json", `{"message":"boom"}`, "twirp call"}, + {"bad json", http.StatusOK, "application/json", "{", "unexpected EOF"}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + server, _ := newTestServer(t, tc.status, tc.contentType, tc.body) + + c := NewClient(server.URL, "secret") + _, err := c.CreateArtifact(context.Background(), CreateArtifactRequest{Name: "foo"}) + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("expected error containing %q, got %v", tc.wantErr, err) + } + }) + } +} diff --git a/artifact/client.go b/artifact/client.go new file mode 100644 index 0000000..3c6c17f --- /dev/null +++ b/artifact/client.go @@ -0,0 +1,33 @@ +// Package artifact provides a client for the GitHub Actions results service. +// The results service exposes the ArtifactService twirp API and the Azure +// block-blob storage used to upload and download workflow artifacts. +package artifact + +import ( + "net/http" + + "code.geekeey.de/actions/sdk" +) + +// Client talks to the results service. It authorizes every request with the +// action runtime token and resolves relative URLs against the results URL. +type Client struct { + http *sdk.Client +} + +// NewClient returns a client that authenticates against the results service at +// base using the given runtime token. +func NewClient(base, runtimeToken string) *Client { + return &Client{http: sdk.NewClient(base, runtimeToken)} +} + +// NewClientFromEnv returns a client configured from the ACTIONS_RESULTS_URL and +// ACTIONS_RUNTIME_TOKEN environment variables. +func NewClientFromEnv(env func(string) string) *Client { + return NewClient(env("ACTIONS_RESULTS_URL"), env("ACTIONS_RUNTIME_TOKEN")) +} + +// Do implements the request executor used by the twirp methods. +func (c *Client) Do(req *http.Request) (*http.Response, error) { + return c.http.Do(req) +} diff --git a/artifact/e2e_test.go b/artifact/e2e_test.go new file mode 100644 index 0000000..ec25ac6 --- /dev/null +++ b/artifact/e2e_test.go @@ -0,0 +1,114 @@ +package artifact + +import ( + "bytes" + "context" + "fmt" + "os" + "testing" + "time" +) + +// TestE2E_ArtifactRoundTrip exercises the real results service. It only runs +// inside a CI pipeline (GITHUB_ACTIONS=true) and is skipped locally so that +// `go test ./...` stays hermetic. Its purpose is to detect drift between the +// results service the runner talks to and the API this client expects. +func TestE2E_ArtifactRoundTrip(t *testing.T) { + if os.Getenv("GITHUB_ACTIONS") != "true" { + t.Skip("e2e test only runs in CI") + } + + resultsURL := os.Getenv("ACTIONS_RESULTS_URL") + runtimeToken := os.Getenv("ACTIONS_RUNTIME_TOKEN") + if resultsURL == "" || runtimeToken == "" { + t.Skip("ACTIONS_RESULTS_URL and ACTIONS_RUNTIME_TOKEN are required") + } + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + + run, job, err := JobInfo() + if err != nil { + t.Fatalf("JobInfo: %v", err) + } + + client := NewClient(resultsURL, runtimeToken) + name := fmt.Sprintf("e2e-%d-%d", os.Getpid(), time.Now().UnixNano()) + + expire := time.Now().Add(1 * time.Hour) + create, err := client.CreateArtifact(ctx, CreateArtifactRequest{ + RunID: run, + JobRunID: job, + Name: name, + Version: 4, + ExpiresAt: &expire, + }) + if err != nil { + t.Fatalf("CreateArtifact: %v", err) + } + if !create.Ok { + t.Fatalf("CreateArtifact not ok: %+v", create) + } + + t.Cleanup(func() { + if _, err := client.DeleteArtifact(context.Background(), DeleteArtifactRequest{RunID: run, JobRunID: job, Name: name}); err != nil { + t.Logf("cleanup DeleteArtifact: %v", err) + } + }) + + content := []byte("artifact e2e " + name) + pushed, err := PushBlob(ctx, nil, bytes.NewReader(content), create.SignedUploadUrl, 0, 0) + if err != nil { + t.Fatalf("PushBlob: %v", err) + } + + finalize, err := client.FinalizeArtifact(ctx, FinalizeArtifactRequest{ + RunID: run, + JobRunID: job, + Name: name, + Size: pushed.Size, + Hash: pushed.SHA256Sum, + }) + if err != nil { + t.Fatalf("FinalizeArtifact: %v", err) + } + if !finalize.Ok { + t.Fatalf("FinalizeArtifact not ok: %+v", finalize) + } + + signed := getSignedArtifactURL(t, ctx, client, run, job, name) + + var buf bytes.Buffer + pulled, err := PullBlob(ctx, nil, &buf, signed.SignedUrl) + if err != nil { + t.Fatalf("PullBlob: %v", err) + } + if !bytes.Equal(buf.Bytes(), content) { + t.Errorf("round trip mismatch: got %q want %q", buf.String(), content) + } + if pulled.SHA256Sum != pushed.SHA256Sum { + t.Errorf("hash mismatch: got %q want %q", pulled.SHA256Sum, pushed.SHA256Sum) + } +} + +// getSignedArtifactURL retries briefly because a freshly finalized artifact may +// not be signed yet. +func getSignedArtifactURL(t *testing.T, ctx context.Context, client *Client, run, job, name string) *GetSignedArtifactURLResponse { + t.Helper() + + var lastErr error + for attempt := 0; attempt < 5; attempt++ { + signed, err := client.GetSignedArtifactURL(ctx, GetSignedArtifactURLRequest{RunID: run, JobRunID: job, Name: name}) + if err == nil && signed.SignedUrl != "" { + return signed + } + lastErr = err + select { + case <-ctx.Done(): + t.Fatalf("GetSignedArtifactURL: %v", ctx.Err()) + case <-time.After(time.Duration(attempt+1) * time.Second): + } + } + t.Fatalf("GetSignedArtifactURL: %v", lastErr) + return nil +} diff --git a/artifact/jwt.go b/artifact/jwt.go new file mode 100644 index 0000000..35176f4 --- /dev/null +++ b/artifact/jwt.go @@ -0,0 +1,84 @@ +package artifact + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "os" + "strings" +) + +// JobInfo returns the backend workflow run and job identifiers encoded in the +// ACTIONS_RUNTIME_TOKEN environment variable. The token is only available from +// the context of an action. +func JobInfo() (run, job string, err error) { + return jobInfo(os.Getenv) +} + +func jobInfo(env func(string) string) (string, string, error) { + var empty string + rt := env("ACTIONS_RUNTIME_TOKEN") + if rt == "" { + return empty, empty, fmt.Errorf("env ACTIONS_RUNTIME_TOKEN is only available from the context of an action") + } + jt, err := newJwtDecode(rt) + if err != nil { + return empty, empty, err + } + scp, ok := jt.Payload["scp"].(string) + if !ok { + return empty, empty, fmt.Errorf("no scp in ACTIONS_RUNTIME_TOKEN") + } + for _, scopes := range strings.Split(scp, " ") { + parts := strings.Split(scopes, ":") + if parts[0] != "Actions.Results" { + continue + } + if len(parts) != 3 { + return empty, empty, fmt.Errorf("invalid scp in ACTIONS_RUNTIME_TOKEN") + } + return parts[1], parts[2], nil + } + + return empty, empty, fmt.Errorf("no backend ids found") +} + +type jwt struct { + Header map[string]any + Payload map[string]any + Signature string +} + +func newJwtDecode(token string) (*jwt, error) { + parts := strings.Split(token, ".") + if len(parts) != 3 { + return nil, fmt.Errorf("invalid JWT: expected 3 segments") + } + + decode := func(seg string) ([]byte, error) { + if l := len(seg) % 4; l != 0 { + seg += strings.Repeat("=", 4-l) + } + return base64.URLEncoding.DecodeString(seg) + } + + headerBytes, err := decode(parts[0]) + if err != nil { + return nil, fmt.Errorf("invalid JWT header: %v", err) + } + var header map[string]any + if err := json.Unmarshal(headerBytes, &header); err != nil { + return nil, fmt.Errorf("invalid JWT header json: %v", err) + } + + payloadBytes, err := decode(parts[1]) + if err != nil { + return nil, fmt.Errorf("invalid JWT payload: %v", err) + } + var payload map[string]any + if err := json.Unmarshal(payloadBytes, &payload); err != nil { + return nil, fmt.Errorf("invalid JWT payload json: %v", err) + } + + return &jwt{Header: header, Payload: payload, Signature: parts[2]}, nil +} diff --git a/artifact/jwt_test.go b/artifact/jwt_test.go new file mode 100644 index 0000000..d91470a --- /dev/null +++ b/artifact/jwt_test.go @@ -0,0 +1,96 @@ +package artifact + +import ( + "encoding/base64" + "encoding/json" + "testing" +) + +func makeToken(t *testing.T, payload map[string]any) string { + t.Helper() + + header := base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"none"}`)) + body, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + return header + "." + base64.RawURLEncoding.EncodeToString(body) + ".signature" +} + +func TestJobInfo(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + token string + run string + job string + wantErr bool + }{ + { + name: "single scope", + token: makeToken(t, map[string]any{"scp": "Actions.Results:123:456"}), + run: "123", + job: "456", + }, + { + name: "scope among others", + token: makeToken(t, map[string]any{"scp": "Actions.Cache:x:y Actions.Results:123:456"}), + run: "123", + job: "456", + }, + { + name: "missing token", + token: "", + wantErr: true, + }, + { + name: "missing scp", + token: makeToken(t, map[string]any{"foo": "bar"}), + wantErr: true, + }, + { + name: "no results scope", + token: makeToken(t, map[string]any{"scp": "Actions.Cache:x:y"}), + wantErr: true, + }, + { + name: "malformed scope", + token: makeToken(t, map[string]any{"scp": "Actions.Results:123"}), + wantErr: true, + }, + { + name: "not a jwt", + token: "not-a-jwt", + wantErr: true, + }, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + env := func(key string) string { + if key == "ACTIONS_RUNTIME_TOKEN" { + return tc.token + } + return "" + } + + run, job, err := jobInfo(env) + if tc.wantErr { + if err == nil { + t.Fatalf("expected error, got run=%q job=%q", run, job) + } + return + } + if err != nil { + t.Fatal(err) + } + if run != tc.run || job != tc.job { + t.Errorf("expected run=%q job=%q, got run=%q job=%q", tc.run, tc.job, run, job) + } + }) + } +} diff --git a/artifact/storage.go b/artifact/storage.go new file mode 100644 index 0000000..534ff17 --- /dev/null +++ b/artifact/storage.go @@ -0,0 +1,212 @@ +package artifact + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/xml" + "fmt" + "io" + "net/http" + "sort" + "sync" +) + +type PullResult struct { + SHA256Sum string + Size int64 +} + +type PushResult struct { + SHA256Sum string + Size int64 +} + +func httpClientOrDefault(c *http.Client) *http.Client { + if c == nil { + return http.DefaultClient + } + return c +} + +// PushBlob uploads the data read from r to the Azure block blob at blobURL as a +// series of blocks which are committed in order. The blobURL is expected to be +// a pre-signed URL. It returns the SHA-256 sum and size of the uploaded data. +func PushBlob(ctx context.Context, client *http.Client, r io.Reader, blobURL string, chunkSize int64, parallelism int) (*PushResult, error) { + client = httpClientOrDefault(client) + if chunkSize <= 0 { + chunkSize = 4 * 1024 * 1024 // 4MB default + } + if parallelism <= 0 { + parallelism = 4 + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + type chunk struct { + index int + data []byte + } + + wrkinp := make(chan chunk) + + var firstErr error + var errOnce sync.Once + fail := func(err error) { + errOnce.Do(func() { + firstErr = err + cancel() + }) + } + + var blockMu sync.Mutex + blocks := map[int]string{} + + var wg sync.WaitGroup + for i := 0; i < parallelism; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for c := range wrkinp { + id := base64.StdEncoding.EncodeToString(fmt.Appendf(nil, "%06d", c.index)) + address := fmt.Sprintf("%s&comp=block&blockid=%s", blobURL, id) + req, err := http.NewRequestWithContext(ctx, http.MethodPut, address, bytes.NewReader(c.data)) + if err != nil { + fail(err) + return + } + req.ContentLength = int64(len(c.data)) + resp, err := client.Do(req) + if err != nil { + fail(err) + return + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + fail(fmt.Errorf("chunk %d upload failed: %s body: %s", c.index, resp.Status, string(body))) + return + } + + blockMu.Lock() + blocks[c.index] = id + blockMu.Unlock() + } + }() + } + + index := 0 + hash := sha256.New() + var size int64 + for { + buf := make([]byte, chunkSize) + n, err := io.ReadFull(r, buf) + if err == io.EOF || err == io.ErrUnexpectedEOF { + if n > 0 { + size += int64(n) + hash.Write(buf[:n]) + select { + case wrkinp <- chunk{index: index, data: buf[:n]}: + index++ + case <-ctx.Done(): + } + } + break + } + if err != nil { + fail(err) + break + } + size += int64(n) + hash.Write(buf) + select { + case wrkinp <- chunk{index: index, data: buf}: + index++ + case <-ctx.Done(): + close(wrkinp) + wg.Wait() + return nil, firstErr + } + } + close(wrkinp) + wg.Wait() + + if firstErr != nil { + return nil, firstErr + } + if len(blocks) == 0 { + return nil, fmt.Errorf("no blocks uploaded") + } + + type Latest struct { + Value string `xml:",chardata"` + } + type BlockList struct { + Latest []Latest `xml:"Latest"` + } + + // The order of the block list determines the order of the resulting blob, so + // the blocks must be committed by ascending index. + indexes := make([]int, 0, len(blocks)) + for i := range blocks { + indexes = append(indexes, i) + } + sort.Ints(indexes) + + blockList := &BlockList{} + for _, i := range indexes { + blockList.Latest = append(blockList.Latest, Latest{Value: blocks[i]}) + } + blockListData, err := xml.Marshal(blockList) + if err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s&comp=blocklist", blobURL), bytes.NewReader(blockListData)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/xml") + req.ContentLength = int64(len(blockListData)) + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("blocklist commit failed: %s body: %s", resp.Status, string(body)) + } + return &PushResult{SHA256Sum: hex.EncodeToString(hash.Sum(nil)), Size: size}, nil +} + +// PullBlob downloads the blob at blobURL, writing its contents to w. The +// blobURL is expected to be a pre-signed URL. It returns the SHA-256 sum and +// size of the downloaded data. +func PullBlob(ctx context.Context, client *http.Client, w io.Writer, blobURL string) (*PullResult, error) { + client = httpClientOrDefault(client) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, blobURL, nil) + if err != nil { + return nil, err + } + resp, err := client.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 32*1024)) + return nil, fmt.Errorf("pull blob failed: %s body: %s", resp.Status, string(body)) + } + + hash := sha256.New() + size, err := io.Copy(io.MultiWriter(w, hash), resp.Body) + if err != nil { + return nil, err + } + return &PullResult{SHA256Sum: hex.EncodeToString(hash.Sum(nil)), Size: size}, nil +} diff --git a/artifact/storage_test.go b/artifact/storage_test.go new file mode 100644 index 0000000..072c685 --- /dev/null +++ b/artifact/storage_test.go @@ -0,0 +1,173 @@ +package artifact + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/xml" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +func sha256Hex(data string) string { + sum := sha256.Sum256([]byte(data)) + return hex.EncodeToString(sum[:]) +} + +func TestPushBlob(t *testing.T) { + t.Parallel() + + const data = "helloworld" // 10 bytes, chunked into 4+4+2 + + var mu sync.Mutex + received := map[string][]byte{} + var blockList []string + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Query().Get("comp") { + case "block": + id := r.URL.Query().Get("blockid") + // Delay the first block so it completes after later ones; the + // committed block list must still be in ascending index order. + raw, _ := base64.StdEncoding.DecodeString(id) + if string(raw) == "000000" { + time.Sleep(20 * time.Millisecond) + } + body, _ := io.ReadAll(r.Body) + mu.Lock() + received[id] = body + mu.Unlock() + w.WriteHeader(http.StatusCreated) + case "blocklist": + body, _ := io.ReadAll(r.Body) + var bl struct { + Latest []struct { + Value string `xml:",chardata"` + } `xml:"Latest"` + } + if err := xml.Unmarshal(body, &bl); err != nil { + t.Errorf("invalid block list xml: %v", err) + } + for _, l := range bl.Latest { + blockList = append(blockList, l.Value) + } + w.WriteHeader(http.StatusCreated) + default: + http.Error(w, "unexpected request", http.StatusBadRequest) + } + })) + defer server.Close() + + res, err := PushBlob(context.Background(), nil, strings.NewReader(data), server.URL+"?sv=test", 4, 3) + if err != nil { + t.Fatal(err) + } + if res.Size != int64(len(data)) { + t.Errorf("expected size %d, got %d", len(data), res.Size) + } + if res.SHA256Sum != sha256Hex(data) { + t.Errorf("expected sha %q, got %q", sha256Hex(data), res.SHA256Sum) + } + + if len(received) != 3 { + t.Fatalf("expected 3 blocks, got %d", len(received)) + } + + wantChunks := []string{"hell", "owor", "ld"} + for i, want := range wantChunks { + id := base64.StdEncoding.EncodeToString([]byte(chunkID(i))) + if got := string(received[id]); got != want { + t.Errorf("block %d: expected %q, got %q", i, want, got) + } + } + + if len(blockList) != 3 { + t.Fatalf("expected 3 block list entries, got %d", len(blockList)) + } + for i, id := range blockList { + raw, err := base64.StdEncoding.DecodeString(id) + if err != nil { + t.Fatalf("invalid block id %q: %v", id, err) + } + if got := string(raw); got != chunkID(i) { + t.Errorf("block list position %d: expected %q, got %q", i, chunkID(i), got) + } + } +} + +func TestPushBlob_Empty(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusCreated) + })) + defer server.Close() + + if _, err := PushBlob(context.Background(), nil, strings.NewReader(""), server.URL+"?sv=test", 4, 3); err == nil { + t.Fatal("expected error for empty input") + } +} + +func TestPushBlob_Error(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + })) + defer server.Close() + + _, err := PushBlob(context.Background(), nil, strings.NewReader("helloworld"), server.URL+"?sv=test", 4, 3) + if err == nil || !strings.Contains(err.Error(), "upload failed") { + t.Fatalf("expected upload error, got %v", err) + } +} + +func TestPullBlob(t *testing.T) { + t.Parallel() + + const data = "helloworld" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, data) + })) + defer server.Close() + + var buf bytes.Buffer + res, err := PullBlob(context.Background(), nil, &buf, server.URL) + if err != nil { + t.Fatal(err) + } + if got := buf.String(); got != data { + t.Errorf("expected %q, got %q", data, got) + } + if res.Size != int64(len(data)) { + t.Errorf("expected size %d, got %d", len(data), res.Size) + } + if res.SHA256Sum != sha256Hex(data) { + t.Errorf("expected sha %q, got %q", sha256Hex(data), res.SHA256Sum) + } +} + +func TestPullBlob_Error(t *testing.T) { + t.Parallel() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "nope", http.StatusNotFound) + })) + defer server.Close() + + if _, err := PullBlob(context.Background(), nil, io.Discard, server.URL); err == nil { + t.Fatal("expected error for non-2xx response") + } +} + +func chunkID(i int) string { + return fmt.Sprintf("%06d", i) +} diff --git a/client.go b/client.go index 9898adb..f49a072 100644 --- a/client.go +++ b/client.go @@ -2,16 +2,25 @@ package sdk import ( "fmt" + "mime" "net/http" "net/url" + "strings" ) +// NewClient returns an HTTP client that resolves relative request URLs against +// base and sets the Authorization header to "Bearer token" on every request. +func NewClient(base, token string) *Client { + return &Client{ + Client: &http.Client{}, + base: base, + token: fmt.Sprintf("Bearer %s", token), + } +} + func (a *Action) Client() *Client { - c := &Client{Client: &http.Client{}} context := a.Context() - c.base = context.APIURL - c.token = fmt.Sprintf("Bearer %s", context.Token) - return c + return NewClient(context.APIURL, context.Token) } type Client struct { @@ -31,3 +40,23 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) { } return c.Client.Do(req) } + +// HasContentType reports whether the response content-type matches the given +// media type. An empty content-type is treated as application/octet-stream. +func HasContentType(r *http.Response, mimetype string) bool { + contentType := r.Header.Get("content-type") + if contentType == "" { + return mimetype == "application/octet-stream" + } + + for _, v := range strings.Split(contentType, ",") { + t, _, err := mime.ParseMediaType(v) + if err != nil { + break + } + if t == mimetype { + return true + } + } + return false +} diff --git a/client_test.go b/client_test.go new file mode 100644 index 0000000..ee13fd2 --- /dev/null +++ b/client_test.go @@ -0,0 +1,71 @@ +package sdk + +import ( + "net/http" + "net/http/httptest" + "testing" +) + +func TestClient_Do(t *testing.T) { + t.Parallel() + + var gotPath, gotAuth string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + c := NewClient(server.URL, "secret") + + req, err := http.NewRequest(http.MethodGet, "/some/path", nil) + if err != nil { + t.Fatal(err) + } + res, err := c.Do(req) + if err != nil { + t.Fatal(err) + } + defer res.Body.Close() + + if got, want := gotAuth, "Bearer secret"; got != want { + t.Errorf("expected Authorization %q to be %q", got, want) + } + if got, want := gotPath, "/some/path"; got != want { + t.Errorf("expected path %q to be %q", got, want) + } +} + +func TestHasContentType(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + contentType string + mimetype string + want bool + }{ + {"exact", "application/json", "application/json", true}, + {"with charset", "application/json; charset=utf-8", "application/json", true}, + {"multiple", "text/plain, application/json", "application/json", true}, + {"wrong type", "text/plain", "application/json", false}, + {"empty matches octet-stream", "", "application/octet-stream", true}, + {"empty does not match json", "", "application/json", false}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + res := &http.Response{Header: http.Header{}} + if tc.contentType != "" { + res.Header.Set("content-type", tc.contentType) + } + if got := HasContentType(res, tc.mimetype); got != tc.want { + t.Errorf("expected HasContentType(%q, %q) to be %v", tc.contentType, tc.mimetype, tc.want) + } + }) + } +}