Merge pull request #33 from grycap/dev-slangarita

oidc-token and fix bugs

cmd/cluster_add.go

@@ -37,11 +37,17 @@ func clusterAddFunc(cmd *cobra.Command, args []string) error {
 	var err error
 
 	oidcAccountName, _ := cmd.Flags().GetString("oidc-account-name")
+	oidcToken, _ := cmd.Flags().GetString("oidc-token")
 	if oidcAccountName != "" {
 		if len(args) != 2 {
 			cmd.SilenceUsage = false
 			return errors.New("if the \"--oidc-account-name\" flag is set only 2 arguments are allowed")
 		}
+	} else if oidcToken != "" {
+		if len(args) != 2 {
+			cmd.SilenceUsage = false
+			return errors.New("if the \"--oidc-token\" flag is set only 2 arguments are allowed")
+		}
 	} else {
 		if len(args) == 2 {
 			cmd.SilenceUsage = false
@@ -76,7 +82,7 @@ func clusterAddFunc(cmd *cobra.Command, args []string) error {
 
 	disableSSL, _ := cmd.Flags().GetBool("disable-ssl")
 
-	err = conf.AddCluster(configPath, identifier, endpoint, username, pass, oidcAccountName, !disableSSL)
+	err = conf.AddCluster(configPath, identifier, endpoint, username, pass, oidcAccountName, oidcToken, !disableSSL)
 	if err != nil {
 		return err
 	}
@@ -88,7 +94,7 @@ func clusterAddFunc(cmd *cobra.Command, args []string) error {
 
 func makeClusterAddCmd() *cobra.Command {
 	clusterAddCmd := &cobra.Command{
-		Use:     "add IDENTIFIER ENDPOINT {USERNAME {PASSWORD | --password-stdin} | --oidc-account-name ACCOUNT}",
+		Use:     "add IDENTIFIER ENDPOINT {USERNAME {PASSWORD | --password-stdin} | --oidc-account-name ACCOUNT | --oidc-token TOKEN}",
 		Short:   "Add a new existing cluster to oscar-cli",
 		Args:    cobra.RangeArgs(2, 4),
 		Aliases: []string{"a"},
@@ -98,6 +104,7 @@ func makeClusterAddCmd() *cobra.Command {
 	clusterAddCmd.Flags().Bool("disable-ssl", false, "disable verification of ssl certificates for the added cluster")
 	clusterAddCmd.Flags().Bool("password-stdin", false, "take the password from stdin")
 	clusterAddCmd.Flags().StringP("oidc-account-name", "o", "", "OIDC account name to authenticate using oidc-agent. Note that oidc-agent must be started and properly configured\n(See: https://indigo-dc.gitbook.io/oidc-agent/)")
+	clusterAddCmd.Flags().StringP("oidc-token", "t", "", "OIDC token to authenticate using oidc-token. Note that oidc-token must be started and properly configured\n(See: https://mytoken.data.kit.edu/)")
 
 	return clusterAddCmd
 }

cmd/service_job.go

@@ -17,10 +17,15 @@ limitations under the License.
 package cmd
 
 import (
+	"bytes"
+	"encoding/base64"
 	"errors"
+	"fmt"
+	"io"
+	"os"
 
 	"github.com/grycap/oscar-cli/pkg/config"
-	"github.com/grycap/oscar-cli/pkg/storage"
+	"github.com/grycap/oscar-cli/pkg/service"
 	"github.com/spf13/cobra"
 )
 
@@ -37,30 +42,79 @@ func serviceJobFunc(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
+	// Check if the cluster has auth
+	endpoint, _ := cmd.Flags().GetString("endpoint")
+	token, _ := cmd.Flags().GetString("token")
+
+	if endpoint != "" && token == "" {
+		// Error missing token
+		return errors.New("you must specify a service token with the flag \"--token\"")
+	}
+	if token != "" && endpoint == "" {
+		// Error missing endpoint
+		return errors.New("you must specify a the cluster endpoint with the flag \"--endpoint\"")
+	}
 	// Parse input (only --input or --text-input are allowed) (AND one of them is required)
-	inputFilePath, _ := cmd.Flags().GetString("input")
-	if inputFilePath == "" {
-		return errors.New("you must specify \"--input\" or \"--text-input\" flag")
+	inputFile, _ := cmd.Flags().GetString("file-input")
+	textInput, _ := cmd.Flags().GetString("text-input")
+
+	if inputFile == "" && textInput == "" {
+		return errors.New("you must specify \"--file-input\" or \"--text-input\" flag")
+	}
+	if inputFile != "" && textInput != "" {
+		return errors.New("you only can specify one of \"--file-input\" or \"--text-input\" flags")
 	}
+
+	var inputReader io.Reader = bytes.NewBufferString(textInput)
+
+	if inputFile != "" {
+		// Open the file
+		file, err := os.Open(inputFile)
+		defer file.Close()
+		if err != nil {
+			return fmt.Errorf("unable to read the file \"%s\"", inputFile)
+		}
+		// Set the file as the inputReader
+		inputReader = file
+	}
+
+	// Make pipe to encode file stream
+	reader, writer := io.Pipe()
+	encoder := base64.NewEncoder(base64.StdEncoding, writer)
+
+	// Copy the file to the encoder in a goroutine to avoid blocking the execution
+	go func() {
+		_, err := io.Copy(encoder, inputReader)
+		encoder.Close()
+		if err != nil {
+			writer.CloseWithError(err)
+		}
+		writer.Close()
+	}()
 	// Make the request
-	err = storage.PutFile(conf.Oscar[cluster], args[0], DEFAULT_PROVIDER, inputFilePath, "")
+	resBody, err := service.JobService(conf.Oscar[cluster], args[0], token, endpoint, reader)
 	if err != nil {
 		return err
 	}
+	defer resBody.Close()
 
 	return nil
 }
 
 func makeServiceJobCmd() *cobra.Command {
 	serviceRunCmd := &cobra.Command{
-		Use:     "job SERVICE_NAME --input",
+		Use:     "job SERVICE_NAME {--file-input | --text-input}",
 		Short:   "Invoke a service asynchronously (only compatible with MinIO providers)",
 		Args:    cobra.ExactArgs(1),
 		Aliases: []string{"job", "j"},
-		RunE:    serviceRunFunc,
+		RunE:    serviceJobFunc,
 	}
 
+	serviceRunCmd.Flags().StringP("cluster", "c", "", "set the cluster")
+	serviceRunCmd.Flags().StringP("endpoint", "e", "", "endpoint of a non registered cluster")
+	serviceRunCmd.Flags().StringP("token", "t", "", "token of the service")
 	serviceRunCmd.Flags().StringP("file-input", "f", "", "input file for the request")
+	serviceRunCmd.Flags().StringP("text-input", "i", "", "text input string for the request")
 
 	return serviceRunCmd
 }

cmd/service_run.go

@@ -55,14 +55,14 @@ func serviceRunFunc(cmd *cobra.Command, args []string) error {
 		return errors.New("you must specify a the cluster endpoint with the flag \"--endpoint\"")
 	}
 	// Parse input (only --input or --text-input are allowed) (AND one of them is required)
-	inputFile, _ := cmd.Flags().GetString("input")
+	inputFile, _ := cmd.Flags().GetString("file-input")
 	textInput, _ := cmd.Flags().GetString("text-input")
 	outputFile, _ := cmd.Flags().GetString("output")
 	if inputFile == "" && textInput == "" {
-		return errors.New("you must specify \"--input\" or \"--text-input\" flag")
+		return errors.New("you must specify \"--file-input\" or \"--text-input\" flag")
 	}
 	if inputFile != "" && textInput != "" {
-		return errors.New("you only can specify one of \"--input\" or \"--text-input\" flags")
+		return errors.New("you only can specify one of \"--file-input\" or \"--text-input\" flags")
 	}
 
 	var inputReader io.Reader = bytes.NewBufferString(textInput)
@@ -160,7 +160,7 @@ func serviceRunFunc(cmd *cobra.Command, args []string) error {
 
 func makeServiceRunCmd() *cobra.Command {
 	serviceRunCmd := &cobra.Command{
-		Use:     "run SERVICE_NAME {--input | --text-input}",
+		Use:     "run SERVICE_NAME {--file-input | --text-input}",
 		Short:   "Invoke a service synchronously (a Serverless backend in the cluster is required)",
 		Args:    cobra.ExactArgs(1),
 		Aliases: []string{"invoke", "r"},

pkg/cluster/cluster.go

@@ -51,6 +51,7 @@ type Cluster struct {
 	AuthUser        string `json:"auth_user,omitempty"`
 	AuthPassword    string `json:"auth_password,omitempty"`
 	OIDCAccountName string `json:"oidc_account_name,omitempty"`
+	OIDCToken       string `json:"oidc_token,omitempty"`
 	SSLVerify       bool   `json:"ssl_verify"`
 	Memory          string `json:"memory"`
 	LogLevel        string `json:"log_level"`
@@ -108,6 +109,11 @@ func (cluster *Cluster) GetClient(args ...int) *http.Client {
 			token:     token,
 			transport: transport,
 		}
+	} else if cluster.OIDCToken != "" {
+		transport = &tokenRoundTripper{
+			token:     cluster.OIDCToken,
+			transport: transport,
+		}
 	} else {
 		// Use basic auth
 		transport = &basicAuthRoundTripper{

pkg/config/config.go

@@ -119,13 +119,14 @@ func (config *Config) writeConfig(configPath string) (err error) {
 }
 
 // AddCluster adds a new cluster to the config
-func (config *Config) AddCluster(configPath string, id string, endpoint string, authUser string, authPassword string, oidcAccountName string, sslVerify bool) error {
+func (config *Config) AddCluster(configPath string, id string, endpoint string, authUser string, authPassword string, oidcAccountName string, oidcToken string, sslVerify bool) error {
 	// Add (or overwrite) the new cluster
 	config.Oscar[id] = &cluster.Cluster{
 		Endpoint:        endpoint,
 		AuthUser:        authUser,
 		AuthPassword:    authPassword,
 		OIDCAccountName: oidcAccountName,
+		OIDCToken:       oidcToken,
 		SSLVerify:       sslVerify,
 		Memory:          defaultMemory,
 		LogLevel:        defaultLogLevel,

pkg/service/service.go

@@ -37,6 +37,7 @@ import (
 
 const servicesPath = "/system/services"
 const runPath = "/run"
+const jobPath = "/job"
 
 // FDL represents a Functions Definition Language file
 type FDL struct {
@@ -282,3 +283,66 @@ func RunService(c *cluster.Cluster, name string, token string, endpoint string,
 
 	return res.Body, nil
 }
+
+// JobService invokes a service asynchronously
+func JobService(c *cluster.Cluster, name string, token string, endpoint string, input io.Reader) (responseBody io.ReadCloser, err error) {
+
+	var jobServiceURL *url.URL
+	if token != "" {
+		jobServiceURL, err = url.Parse(endpoint)
+	} else {
+		jobServiceURL, err = url.Parse(c.Endpoint)
+	}
+
+	if err != nil {
+		return nil, cluster.ErrParsingEndpoint
+	}
+	jobServiceURL.Path = path.Join(jobServiceURL.Path, jobPath, name)
+	// Make the request
+	req, err := http.NewRequest(http.MethodPost, jobServiceURL.String(), input)
+	if err != nil {
+		return nil, cluster.ErrMakingRequest
+	}
+
+	var res *http.Response
+	if token != "" {
+		bearer := "Bearer " + strings.TrimSpace(token)
+		req.Header.Add("Authorization", bearer)
+
+		client := &http.Client{}
+		res, err = client.Do(req)
+	} else {
+
+		// Get the service
+		svc, err := GetService(c, name)
+		if err != nil {
+			return nil, err
+		}
+		// Add service's token if defined (OSCAR >= v2.2.0)
+		if svc.Token != "" {
+			bearer := "Bearer " + strings.TrimSpace(svc.Token)
+			req.Header.Add("Authorization", bearer)
+		}
+		// Update cluster client timeout
+		client := c.GetClient()
+		client.Timeout = time.Second * 300
+
+		// Update client transport to remove basic auth
+		client.Transport = &http.Transport{
+			// Enable/disable ssl verification
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: !c.SSLVerify},
+		}
+
+		res, err = client.Do(req)
+	}
+
+	if err != nil {
+		return nil, cluster.ErrSendingRequest
+	}
+
+	if err := cluster.CheckStatusCode(res); err != nil {
+		return nil, err
+	}
+
+	return res.Body, nil
+}