func()

in internal/pkg/cli/run_local.go [737:875]
118 lines of code
21 McCabe index (conditional complexity)

func (o *runLocalOpts) taskRoleCredentials(ctx context.Context) (map[string]string, error) {
	// assumeRoleMethod tries to directly call sts:AssumeRole for TaskRole using default session
	// calls sts:AssumeRole through aws-sdk-go here https://github.com/aws/aws-sdk-go/blob/ac58203a9054cc9d901429bdd94edfc0a7a1de46/aws/credentials/stscreds/assume_role_provider.go#L352
	assumeRoleMethod := func() (map[string]string, error) {
		taskDef, err := o.ecsClient.TaskDefinition(o.appName, o.envName, o.wkldName)
		if err != nil {
			return nil, err
		}

		taskRoleSess, err := o.sessProvider.FromRole(aws.StringValue(taskDef.TaskRoleArn), o.targetEnv.Region)
		if err != nil {
			return nil, err
		}

		return sessionEnvVars(ctx, taskRoleSess)
	}

	// ecsExecMethod tries to use ECS Exec to retrive credentials from running container
	ecsExecMethod := func() (map[string]string, error) {
		svcDesc, err := o.ecsClient.DescribeService(o.appName, o.envName, o.wkldName)
		if err != nil {
			return nil, fmt.Errorf("describe ECS service for %s in environment %s: %w", o.wkldName, o.envName, err)
		}

		stdoutReader, err := o.captureStdout()
		if err != nil {
			return nil, err
		}
		defer o.releaseStdout()

		// try exec on each container within the service
		var wg sync.WaitGroup
		containerErr := make(chan error)
		for _, task := range svcDesc.Tasks {
			taskID, err := awsecs.TaskID(aws.StringValue(task.TaskArn))
			if err != nil {
				return nil, err
			}

			for _, container := range task.Containers {
				wg.Add(1)
				containerName := aws.StringValue(container.Name)
				go func() {
					defer wg.Done()
					err := o.ecsExecutor.ExecuteCommand(awsecs.ExecuteCommandInput{
						Cluster:   svcDesc.ClusterName,
						Command:   fmt.Sprintf("/bin/sh -c %q\n", curlContainerCredentialsCmd),
						Task:      taskID,
						Container: containerName,
					})
					if err != nil {
						containerErr <- fmt.Errorf("container %s in task %s: %w", containerName, taskID, err)
					}
				}()
			}
		}

		// wait for containers to finish and reset stdout
		containersFinished := make(chan struct{})
		go func() {
			wg.Wait()
			o.releaseStdout()
			close(containersFinished)
		}()

		type containerCredentialsOutput struct {
			AccessKeyId     string
			SecretAccessKey string
			Token           string
		}

		// parse stdout to try and find credentials
		credsResult := make(chan map[string]string)
		parseErr := make(chan error)
		go func() {
			select {
			case <-containersFinished:
				buf, err := io.ReadAll(stdoutReader)
				if err != nil {
					parseErr <- err
					return
				}
				lines := bytes.Split(buf, []byte("\n"))
				var creds containerCredentialsOutput
				for _, line := range lines {
					err := json.Unmarshal(line, &creds)
					if err != nil {
						continue
					}
					credsResult <- map[string]string{
						"AWS_ACCESS_KEY_ID":     creds.AccessKeyId,
						"AWS_SECRET_ACCESS_KEY": creds.SecretAccessKey,
						"AWS_SESSION_TOKEN":     creds.Token,
					}
					return
				}
				parseErr <- errors.New("all containers failed to retrieve credentials")
			case <-ctx.Done():
				return
			}
		}()

		var containerErrs []error
		for {
			select {
			case creds := <-credsResult:
				return creds, nil
			case <-ctx.Done():
				return nil, ctx.Err()
			case err := <-parseErr:
				return nil, errors.Join(append([]error{err}, containerErrs...)...)
			case err := <-containerErr:
				containerErrs = append(containerErrs, err)
			}
		}
	}

	credentialsChain := []func() (map[string]string, error){
		assumeRoleMethod,
		ecsExecMethod,
	}

	credentialsChainWrappedErrs := []string{
		"assume role",
		"ecs exec",
	}

	// return TaskRole credentials from first successful method
	var errs []error
	for errIndex, method := range credentialsChain {
		vars, err := method()
		if err == nil {
			return vars, nil
		}
		errs = append(errs, fmt.Errorf("%s: %w", credentialsChainWrappedErrs[errIndex], err))
	}

	return nil, &errTaskRoleRetrievalFailed{errs}
}