From 046e7995ab98c344e4bdd0fbadcd94f9201c2d4d Mon Sep 17 00:00:00 2001 From: Louis Seubert Date: Sun, 5 Oct 2025 20:39:02 +0200 Subject: [PATCH] wip --- artifacts.go | 78 ++++++++++++++++++ save/github.go => client.go | 52 +++++++++++- load/main.go | 28 +++++++ save/main.go | 65 ++++++++++++++- storage.go | 154 ++++++++++++++++++++++++++++++++++++ 5 files changed, 370 insertions(+), 7 deletions(-) create mode 100644 artifacts.go rename save/github.go => client.go (71%) create mode 100644 storage.go diff --git a/artifacts.go b/artifacts.go new file mode 100644 index 0000000..3babd4d --- /dev/null +++ b/artifacts.go @@ -0,0 +1,78 @@ +package artifacts + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "os" + "strings" +) + +// run, job, err +func JobInfo() (string, string, error) { + rt := os.Getenv("ACTIONS_RUNTIME_TOKEN") + var empty string + 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]interface{} + Payload map[string]interface{} + 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]interface{} + 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]interface{} + 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/save/github.go b/client.go similarity index 71% rename from save/github.go rename to client.go index 81dc5fd..c897ebe 100644 --- a/save/github.go +++ b/client.go @@ -1,4 +1,4 @@ -package main +package artifacts import ( "bytes" @@ -40,7 +40,7 @@ func (c *ActionClient) Do(req *http.Request) (*http.Response, error) { } func HasContentType(r *http.Response, mimetype string) bool { - contentType := r.Header.Get("Content-type") + contentType := r.Header.Get("content-type") if contentType == "" { return mimetype == "application/octet-stream" } @@ -65,6 +65,10 @@ type GitHub struct { client ApiClient } +func NewGitHub(client ApiClient) *GitHub { + return &GitHub{client: client} +} + type CreateArtifactRequest struct { RunID string `json:"workflowRunBackendId"` JobRunID string `json:"workflowJobRunBackendId"` @@ -83,7 +87,7 @@ type FinalizeArtifactRequest struct { RunID string `json:"workflowRunBackendId"` JobRunID string `json:"workflowJobRunBackendId"` Name string `json:"name"` - Size int `json:"size"` + Size int64 `json:"size"` Hash string `json:"hash,omitempty"` } @@ -92,6 +96,46 @@ type FinalizeArtifactResponse struct { ArtifactId string `json:"artifactId"` } +type ListArtifactsRequest struct { + RunID string `json:"workflowRunBackendId"` + JobRunID string `json:"workflowJobRunBackendId"` + NameFilter string `json:nameFilter` + IdFilter string `json:idFilter` +} + +type ListArtifactsResponse struct { + Artifacts []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"` + } `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 (gh *GitHub) CreateArtifact(ctx context.Context, req CreateArtifactRequest) (*CreateArtifactResponse, error) { endpoint := "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact" var result CreateArtifactResponse @@ -121,7 +165,7 @@ func (gh *GitHub) twirp(ctx context.Context, url string, in any, out any) error if err != nil { return err } - req.Header.Set("Content-Type", "application/json") + req.Header.Set("content-type", "application/json") req.Header.Set("Accept", "application/json") res, err := gh.client.Do(req) diff --git a/load/main.go b/load/main.go index da29a2c..b80b539 100644 --- a/load/main.go +++ b/load/main.go @@ -1,4 +1,32 @@ package main +import ( + "context" + "os" + "os/signal" + + "code.geekeey.de/actions/sdk" +) + func main() { + action := &SaveArtifactAction{Action: sdk.New()} + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) + defer cancel() + if err := action.Run(ctx); err != nil { + action.Errorf("%s", err) + os.Exit(1) + } +} + +type SaveArtifactAction struct { + *sdk.Action + name string +} + +func (action *SaveArtifactAction) setup() error { + return nil +} + +func (action *SaveArtifactAction) Run(ctx context.Context) error { + return nil } diff --git a/save/main.go b/save/main.go index 7a98ac2..7693ffc 100644 --- a/save/main.go +++ b/save/main.go @@ -1,11 +1,18 @@ package main import ( + "archive/tar" "context" + "fmt" + "io" "os" "os/signal" + "runtime" + "time" + "code.geekeey.de/actions/artifacts" "code.geekeey.de/actions/sdk" + "code.geekeey.de/actions/sdk/iox" ) func main() { @@ -20,9 +27,7 @@ func main() { type SaveArtifactAction struct { *sdk.Action - repository string - path string - reference string + name string } func (action *SaveArtifactAction) setup() error { @@ -33,5 +38,59 @@ func (action *SaveArtifactAction) Run(ctx context.Context) error { if err := action.setup(); err != nil { return err } + + run, job, err := artifacts.JobInfo() + if err != nil { + return fmt.Errorf("unable to get job info: %v", err) + } + + github := artifacts.NewGitHub(artifacts.NewActionClient()) + + expire := time.Now().Add(20 * time.Hour) + create, err := github.CreateArtifact(ctx, artifacts.CreateArtifactRequest{ + RunID: run, + JobRunID: job, + Name: action.name, + Version: 4, + ExpiresAt: &expire, + }) + if err != nil { + return fmt.Errorf("cannot create artifact %w", err) + } + if !create.Ok { + return fmt.Errorf("cannot get pre-signed URL") + } + + rd, err := action.CreateArtifact(ctx) + + res, err := artifacts.PushBlob_(ctx, rd, create.SignedUploadUrl, 1024*1024*1, runtime.NumCPU()) + + finish, err := github.FinalizeArtifact(ctx, artifacts.FinalizeArtifactRequest{ + RunID: run, + JobRunID: job, + Name: action.name, + Size: res.Size, + Hash: res.SHA256Sum, + }) + if err != nil { + return fmt.Errorf("cannot finish artifact %w", err) + } + if !create.Ok { + return fmt.Errorf("cannot finish artifact upload") + } + + action.Noticef("created artifact: %s", finish.ArtifactId) + return nil } + +func (action *SaveArtifactAction) CreateArtifact(ctx context.Context) (io.Reader, error) { + rd, wr, err := os.Pipe() + go func() { + defer wr.Close() + + t := tar.NewWriter(wr) + t.AddFS(iox.NewGlobFS(os.DirFS(""), "")) + }() + return rd, err +} diff --git a/storage.go b/storage.go new file mode 100644 index 0000000..f6a83ba --- /dev/null +++ b/storage.go @@ -0,0 +1,154 @@ +package artifacts + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/xml" + "fmt" + "io" + "net/http" + "sync" +) + +type PullResult struct { + SHA256Sum string + Size int64 +} +type PushResult struct { + SHA256Sum string + Size int64 +} + +func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxParallel int) (int, string, error) { + var empty string + if chunkSize <= 0 { + chunkSize = 4 * 1024 * 1024 // 4MB default + } + if maxParallel <= 0 { + maxParallel = 4 + } + + type chunk struct { + index int + data []byte + } + wrkinp := make(chan chunk) + wrkerr := make(chan error, maxParallel) + blockIDs := make([]string, 0) + var blockMu sync.Mutex + var wg sync.WaitGroup + + // Worker pool for parallel uploading + for i := 0; i < maxParallel; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for c := range wrkinp { + id := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%06d", c.index))) + address := fmt.Sprintf("%s&comp=block&blockid=%s", url, id) + req, err := http.NewRequestWithContext(ctx, "PUT", address, bytes.NewReader(c.data)) + if err != nil { + wrkerr <- err + return + } + req.ContentLength = int64(len(c.data)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + wrkerr <- err + return + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + wrkerr <- fmt.Errorf("chunk %d upload failed: %s body: %s", c.index, resp.Status, string(body)) + return + } + + blockMu.Lock() + defer blockMu.Unlock() + blockIDs = append(blockIDs, id) + } + }() + } + + // Chunk the data and send to workers + index := 0 + hash := sha256.New() + for { + buf := make([]byte, chunkSize) + n, err := io.ReadFull(data, buf) + if err == io.EOF || err == io.ErrUnexpectedEOF { + if n > 0 { + hash.Write(buf[:n]) + wrkinp <- chunk{index: index, data: buf[:n]} + index++ + } + break + } + if err != nil { + close(wrkinp) + wg.Wait() + return empty, err + } + hash.Write(buf) + wrkinp <- chunk{index: index, data: buf} + index++ + } + close(wrkinp) + wg.Wait() + + select { + case err := <-wrkerr: + return empty, err + default: + } + + type BlockList struct { + Latest []string `xml:"Latest"` + } + + type Latest struct { + Value string `xml:",chardata"` + } + + // Sort block IDs to ensure correct order + blockMu.Lock() + defer blockMu.Unlock() + if len(blockIDs) == 0 { + return empty, fmt.Errorf("no blocks uploaded") + } + blockList := &BlockList{} + for i := 0; i < len(blockIDs); i++ { + blockList.Latest = append(blockList.Blocks, xmlBlock{XMLName: xml.Name{Local: "Latest"}, ID: blockIDs[i]}) + } + blockListData, err := xml.Marshal(blockList) + if err != nil { + return empty, err + } + + req, err := http.NewRequestWithContext(ctx, "PUT", fmt.Sprintf("%s&comp=blocklist", url), bytes.NewReader(blockListData)) + if err != nil { + return empty, err + } + req.Header.Set("Content-Type", "application/xml") + req.ContentLength = int64(len(blockListData)) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return empty, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return empty, fmt.Errorf("blocklist commit failed: %s body: %s", resp.Status, string(body)) + } + return hex.EncodeToString(hash.Sum(nil)), nil +} + +func PushBlob_(ctx context.Context, r io.Reader, blobURL string, chunkSize int64, parallelism int) (*PushResult, error) { +} + +func PullBlob_(ctx context.Context, w io.Writer, blobURL string, chunkSize int64, parallelism int) (*PullResult, error) { +}