eba4fd1
package main
import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/signal"
"sort"
"strconv"
"strings"
"sync"
"syscall"
"time"
"github.com/nektos/act/pkg/common"
"github.com/nektos/act/pkg/exprparser"
"github.com/nektos/act/pkg/model"
"github.com/nektos/act/pkg/runner"
"github.com/nektos/act/pkg/workflowpattern"
"github.com/robfig/cron/v3"
log "github.com/sirupsen/logrus"
"gopkg.in/yaml.v3"
)
type request struct {
Patterns []string `json:"patterns"`
Paths []string `json:"paths"`
TriggerEvent string `json:"trigger_event"`
Branch string `json:"branch"`
Tag string `json:"tag"`
Action string `json:"action"`
ID json.RawMessage `json:"id"`
Op string `json:"op"`
Source string `json:"source"`
File string `json:"file"`
Job string `json:"job"`
Expression string `json:"expression"`
Value interface{} `json:"value"`
StatusCheck bool `json:"status_check"`
Matrix map[string]interface{} `json:"matrix"`
Needs map[string]exprparser.Needs `json:"needs"`
Jobs map[string]*model.WorkflowCallResult `json:"jobs"`
Github map[string]interface{} `json:"github"`
Event map[string]interface{} `json:"event"`
Inputs map[string]interface{} `json:"inputs"`
Env map[string]string `json:"env"`
Vars map[string]string `json:"vars"`
Secrets map[string]string `json:"secrets"`
Workdir string `json:"workdir"`
Platforms map[string]string `json:"platforms"`
NodeRuntimes map[string]string `json:"node_runtimes"`
ActionCacheDir string `json:"action_cache_dir"`
ActionServer string `json:"action_server"`
ActionToken string `json:"action_token"`
DockerHost string `json:"docker_host"`
RuntimeURL string `json:"runtime_url"`
RuntimeToken string `json:"runtime_token"`
Token string `json:"token"`
TimeoutSeconds uint64 `json:"timeout_seconds"`
}
type output struct {
ID json.RawMessage `json:"id,omitempty"`
Event string `json:"event"`
Data interface{} `json:"data,omitempty"`
Error string `json:"error,omitempty"`
}
type emitter struct {
mu sync.Mutex
encoder *json.Encoder
id json.RawMessage
masks []string
allowSensitiveResult bool
}
func (e *emitter) send(event string, data interface{}, err error) {
e.mu.Lock()
defer e.mu.Unlock()
item := output{ID: e.id, Event: event, Data: data}
if err != nil {
item.Error = err.Error()
}
encoded, marshalErr := json.Marshal(item)
if marshalErr != nil {
return
}
text := string(encoded)
masks := e.masks
if event == "result" && e.allowSensitiveResult {
masks = nil
}
for _, mask := range masks {
if mask != "" {
escaped, _ := json.Marshal(mask)
text = strings.ReplaceAll(text, string(escaped[1:len(escaped)-1]), "***")
}
}
_, _ = fmt.Fprintln(os.Stdout, text)
}
type eventHook struct{ emit *emitter }
func (h eventHook) Levels() []log.Level { return log.AllLevels }
func (h eventHook) Fire(entry *log.Entry) error {
h.emit.send("log", map[string]interface{}{"text": entry.Message + "\n"}, nil)
return nil
}
func readWorkflow(req request) (*model.Workflow, map[string]interface{}, error) {
workflow, err := model.ReadWorkflow(strings.NewReader(req.Source), true)
if err != nil {
return nil, nil, err
}
workflow.File = req.File
var raw map[string]interface{}
if err = yaml.Unmarshal([]byte(req.Source), &raw); err != nil {
return nil, nil, err
}
triggers := map[string]interface{}{}
switch value := raw["on"].(type) {
case string:
triggers[value] = map[string]interface{}{}
case []interface{}:
for _, name := range value {
triggers[fmt.Sprint(name)] = map[string]interface{}{}
}
case map[string]interface{}:
for name, value := range value {
if value == nil {
value = map[string]interface{}{}
}
triggers[name] = value
}
default:
return nil, nil, fmt.Errorf("on must declare a trigger")
}
allowed := map[string]bool{"push": true, "pull_request": true, "workflow_dispatch": true, "schedule": true, "repository_dispatch": true, "workflow_call": true, "workflow_run": true}
for event := range triggers {
if !allowed[event] {
return nil, nil, fmt.Errorf("event %q has no GitData event source", event)
}
}
if len(triggers) == 0 || len(workflow.Jobs) == 0 {
return nil, nil, fmt.Errorf("workflow must declare triggers and jobs")
}
raw["on"] = triggers
marks := map[string]int{}
var visit func(string) error
visit = func(id string) error {
if marks[id] == 1 {
return fmt.Errorf("dependency cycle at job %q", id)
}
if marks[id] == 2 {
return nil
}
job, ok := workflow.Jobs[id]
if !ok {
return fmt.Errorf("unknown job %q", id)
}
marks[id] = 1
if job.Uses == "" && len(job.Steps) == 0 {
return fmt.Errorf("job %q must declare steps", id)
}
for _, dep := range job.Needs() {
if err := visit(dep); err != nil {
return err
}
}
marks[id] = 2
return nil
}
for id := range workflow.Jobs {
if err := visit(id); err != nil {
return nil, nil, err
}
}
return workflow, raw, nil
}
func runContext(req request, wf *model.Workflow, emit *emitter) *runner.RunContext {
if req.Env == nil {
req.Env = map[string]string{}
}
if req.Github == nil {
req.Github = map[string]interface{}{}
}
if req.Workdir == "" {
req.Workdir = "."
}
if req.ActionServer == "" {
req.ActionServer = "https://github.com"
}
for name, need := range req.Needs {
job := wf.Jobs[name]
if job == nil {
job = &model.Job{}
wf.Jobs[name] = job
}
job.Outputs = need.Outputs
job.Result = need.Result
}
event, _ := json.Marshal(req.Event)
cfg := &runner.Config{Workdir: req.Workdir, ActionCacheDir: req.ActionCacheDir, Env: req.Env, Secrets: req.Secrets, Vars: req.Vars, Token: req.Token,
Platforms: req.Platforms, GitHubInstance: "github.com", EventName: fmt.Sprint(req.Github["event_name"]), Actor: fmt.Sprint(req.Github["actor"]),
NoSkipCheckout: true, AutoRemove: true, BindWorkdir: false, LogOutput: true, ContainerDaemonSocket: req.DockerHost,
GitDataGithub: req.Github, GitDataInputs: req.Inputs, GitDataJobs: req.Jobs, GitDataActionServer: req.ActionServer, GitDataActionToken: req.ActionToken,
GitDataRuntimeURL: req.RuntimeURL, GitDataRuntimeToken: req.RuntimeToken, GitDataNodeRuntimes: req.NodeRuntimes}
if req.RuntimeURL != "" {
cfg.ArtifactServerPath = "gitdata"
}
if emit != nil {
cfg.GitDataObserver = func(event runner.GitDataEvent) { emit.send(event.Type, event, nil) }
}
rc := &runner.RunContext{Config: cfg, Run: &model.Run{Workflow: wf, JobID: req.Job}, Matrix: req.Matrix, EventJSON: string(event), StepResults: map[string]*model.StepResult{}}
ctx := context.Background()
rc.ExprEval = rc.NewExpressionEvaluator(ctx)
rc.Name = rc.ExprEval.Interpolate(ctx, rc.Run.String())
rc.JobName = rc.Name
return rc
}
func evaluateValue(ctx context.Context, rc *runner.RunContext, value interface{}) (interface{}, error) {
var node yaml.Node
if err := node.Encode(value); err != nil {
return nil, err
}
if err := rc.NewExpressionEvaluator(ctx).EvaluateYamlNode(ctx, &node); err != nil {
return nil, err
}
var result interface{}
err := node.Decode(&result)
return result, err
}
func stringValue(value interface{}) string {
if value == nil {
return ""
}
return fmt.Sprint(value)
}
func boolValue(value interface{}, fallback bool) bool {
if value == nil {
return fallback
}
if b, ok := value.(bool); ok {
return b
}
b, err := strconv.ParseBool(fmt.Sprint(value))
if err != nil {
return fallback
}
return b
}
func object(value interface{}) map[string]interface{} {
m, _ := value.(map[string]interface{})
if m == nil {
return map[string]interface{}{}
}
return m
}
func expand(req request, wf *model.Workflow, raw map[string]interface{}, rc *runner.RunContext) (interface{}, error) {
ctx := context.Background()
job := wf.GetJob(req.Job)
enabled, err := runner.EvalBool(ctx, rc.ExprEval, job.If.Value, exprparser.DefaultStatusCheckSuccess)
if err != nil {
return nil, err
}
if !enabled {
return map[string]interface{}{"skipped": true, "instances": []interface{}{}}, nil
}
rawJob := object(object(raw["jobs"])[req.Job])
strategy := object(rawJob["strategy"])
if job.Strategy != nil {
if err := rc.ExprEval.EvaluateYamlNode(ctx, &job.Strategy.RawMatrix); err != nil {
return nil, err
}
}
matrices, err := job.GetMatrixes()
if err != nil {
return nil, err
}
if len(matrices) > 256 {
return nil, fmt.Errorf("matrix exceeds 256 instances")
}
instances := []interface{}{}
for index, matrix := range matrices {
rc.Matrix = matrix
rc.ExprEval = rc.NewExpressionEvaluator(ctx)
fields := map[string]interface{}{}
for _, name := range []string{"name", "runs-on", "environment", "timeout-minutes", "continue-on-error", "concurrency", "with", "permissions"} {
value, e := evaluateValue(ctx, rc, rawJob[name])
if e != nil {
return nil, fmt.Errorf("job %s %s: %w", req.Job, name, e)
}
fields[name] = value
}
labels := []string{}
switch label := fields["runs-on"].(type) {
case string:
labels = append(labels, label)
case []interface{}:
for _, l := range label {
labels = append(labels, stringValue(l))
}
case map[string]interface{}:
switch ls := label["labels"].(type) {
case string:
labels = append(labels, ls)
case []interface{}:
for _, l := range ls {
labels = append(labels, stringValue(l))
}
}
}
if job.Uses == "" && len(labels) == 0 {
return nil, fmt.Errorf("job %s runs-on has no labels", req.Job)
}
environment := fields["environment"]
if env, ok := environment.(map[string]interface{}); ok {
environment = env["name"]
}
timeout := uint64(21600)
if fields["timeout-minutes"] != nil {
minutes, e := strconv.ParseFloat(stringValue(fields["timeout-minutes"]), 64)
if e != nil || minutes <= 0 {
return nil, fmt.Errorf("invalid timeout-minutes")
}
timeout = uint64(minutes * 60)
}
failFast, e := evaluateValue(ctx, rc, strategy["fail-fast"])
if e != nil {
return nil, e
}
maxParallel := len(matrices)
if strategy["max-parallel"] != nil {
v, e := evaluateValue(ctx, rc, strategy["max-parallel"])
if e != nil {
return nil, e
}
n, e := strconv.Atoi(stringValue(v))
if e != nil || n <= 0 {
return nil, fmt.Errorf("invalid max-parallel")
}
maxParallel = n
}
name := stringValue(fields["name"])
if name == "" {
name = req.Job
}
instances = append(instances, map[string]interface{}{"index": index, "matrix": matrix, "name": name, "runs_on": labels, "environment": stringValue(environment), "timeout_seconds": timeout, "continue_on_error": boolValue(fields["continue-on-error"], false), "fail_fast": boolValue(failFast, true), "max_parallel": maxParallel, "concurrency": fields["concurrency"], "permissions": fields["permissions"], "with": fields["with"], "uses": job.Uses, "group": object(fields["runs-on"])["group"]})
}
return map[string]interface{}{"skipped": false, "instances": instances}, nil
}
func execute(req request, rc *runner.RunContext, emit *emitter) (interface{}, error) {
if err := os.MkdirAll(req.Workdir, 0750); err != nil {
return nil, err
}
signalCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
jobCtx := signalCtx
cancel := func() {}
if req.TimeoutSeconds > 0 {
jobCtx, cancel = context.WithTimeout(signalCtx, time.Duration(req.TimeoutSeconds)*time.Second)
}
defer cancel()
logger := log.New()
logger.SetOutput(io.Discard)
logger.SetLevel(log.InfoLevel)
logger.AddHook(eventHook{emit: emit})
ctx := common.WithLogger(context.Background(), logger)
ctx = common.WithJobCancelContext(ctx, jobCtx)
ctx = common.WithJobErrorContainer(ctx)
hardCtx, hardCancel := context.WithCancel(ctx)
defer hardCancel()
go func() {
select {
case <-jobCtx.Done():
timer := time.NewTimer(10 * time.Second)
defer timer.Stop()
select {
case <-timer.C:
hardCancel()
case <-hardCtx.Done():
}
case <-hardCtx.Done():
}
}()
ctx = hardCtx
executor, err := rc.Executor()
if err != nil {
return nil, err
}
err = executor(ctx)
result := rc.Run.Job().Result
if result == "" {
result = "success"
}
if errors.Is(jobCtx.Err(), context.DeadlineExceeded) {
result = "timed_out"
} else if jobCtx.Err() != nil {
result = "cancelled"
} else if err != nil || common.JobError(ctx) != nil {
result = "failure"
}
outputs := rc.Run.Job().Outputs
for name, value := range outputs {
for _, secret := range append(emit.masks, rc.Masks...) {
if secret != "" && strings.Contains(value, secret) {
delete(outputs, name)
break
}
}
}
message := ""
if err != nil {
message = err.Error()
}
if jobErr := common.JobError(ctx); jobErr != nil {
message = jobErr.Error()
}
return map[string]interface{}{"result": result, "outputs": outputs, "error": message}, nil
}
func handle(req request, emit *emitter) (result interface{}, err error) {
defer func() {
if recovered := recover(); recovered != nil {
result = nil
err = fmt.Errorf("engine panic: %v", recovered)
}
}()
if req.Op == "version" {
return map[string]interface{}{"protocol_version": 2, "upstream": "v0.2.89", "commit": "4f411281417e88660bea1c1a1749aa71ae0bd60f"}, nil
}
if req.Op == "oidc_sign" || req.Op == "oidc_jwks" {
return oidc(req)
}
if req.Op == "match" {
patterns, e := workflowpattern.CompilePatterns(req.Patterns...)
if e != nil {
return nil, e
}
return !workflowpattern.Skip(patterns, req.Paths, &workflowpattern.EmptyTraceWriter{}), nil
}
if req.Op == "cron" {
schedule, e := cron.ParseStandard(req.Expression)
if e != nil {
return nil, e
}
timestamp, e := strconv.ParseFloat(fmt.Sprint(req.Value), 64)
if e != nil {
return nil, e
}
return schedule.Next(time.Unix(int64(timestamp), 0).UTC()).Unix(), nil
}
wf, raw, err := readWorkflow(req)
if err != nil {
return nil, err
}
if req.Op == "parse" {
return raw, nil
}
if req.Op == "trigger" {
return triggerMatches(req, raw)
}
if req.Job == "" {
ids := wf.GetJobIDs()
sort.Strings(ids)
req.Job = ids[0]
}
if wf.Jobs[req.Job] == nil {
return nil, fmt.Errorf("unknown job %q", req.Job)
}
rc := runContext(req, wf, emit)
switch req.Op {
case "evaluate":
if req.Expression != "" || req.StatusCheck {
check := exprparser.DefaultStatusCheckNone
if req.StatusCheck {
check = exprparser.DefaultStatusCheckSuccess
}
return rc.GitDataEvaluate(context.Background(), req.Expression, check)
}
return evaluateValue(context.Background(), rc, req.Value)
case "expand":
return expand(req, wf, raw, rc)
case "execute":
return execute(req, rc, emit)
default:
return nil, fmt.Errorf("unknown operation %q", req.Op)
}
}
func main() {
log.SetOutput(os.Stderr)
log.SetLevel(log.WarnLevel)
scanner := bufio.NewScanner(os.Stdin)
scanner.Buffer(make([]byte, 65536), 8*1024*1024)
for scanner.Scan() {
var req request
emit := &emitter{encoder: json.NewEncoder(os.Stdout)}
if err := json.Unmarshal(scanner.Bytes(), &req); err != nil {
emit.send("result", nil, err)
continue
}
emit.id = req.ID
emit.allowSensitiveResult = req.Op != "execute"
for _, secret := range req.Secrets {
emit.masks = append(emit.masks, secret)
}
emit.masks = append(emit.masks, req.Token, req.RuntimeToken, req.ActionToken)
result, err := handle(req, emit)
emit.send("result", result, err)
}
if err := scanner.Err(); err != nil {
_, _ = fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}