Allow versionId to be null for Copy,Get,Head API calls (#6942)

Fixes #6935

cmd/api-errors.go

@@ -87,6 +87,7 @@ const (
 	ErrNoSuchBucketPolicy
 	ErrNoSuchKey
 	ErrNoSuchUpload
+	ErrNoSuchVersion
 	ErrNotImplemented
 	ErrPreconditionFailed
 	ErrRequestTimeTooSkewed
@@ -447,6 +448,11 @@ var errorCodeResponse = map[APIErrorCode]APIError{
 		Description:    "The specified multipart upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.",
 		HTTPStatusCode: http.StatusNotFound,
 	},
+	ErrNoSuchVersion: {
+		Code:           "NoSuchVersion",
+		Description:    "Indicates that the version ID specified in the request does not match an existing version.",
+		HTTPStatusCode: http.StatusNotFound,
+	},
 	ErrNotImplemented: {
 		Code:           "NotImplemented",
 		Description:    "A header you provided implies functionality that is not implemented",

cmd/common-main.go

@@ -30,7 +30,6 @@ import (
 	dns2 "github.com/miekg/dns"
 	"github.com/minio/cli"
 	"github.com/minio/minio-go/pkg/set"
-	"github.com/minio/minio/cmd/crypto"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/cmd/logger/target/console"
 	"github.com/minio/minio/cmd/logger/target/http"
@@ -322,20 +321,6 @@ func handleCommonEnvVars() {
 		globalWORMEnabled = bool(wormFlag)
 	}
 
-	kmsConf, err := crypto.NewVaultConfig()
-	if err != nil {
-		logger.Fatal(err, "Unable to initialize hashicorp vault")
-	}
-	if kmsConf.Vault.Endpoint != "" {
-		kms, err := crypto.NewVault(kmsConf)
-		if err != nil {
-			logger.Fatal(err, "Unable to initialize KMS")
-		}
-		globalKMS = kms
-		globalKMSKeyID = kmsConf.Vault.Key.Name
-		globalKMSConfig = kmsConf
-	}
-
 	if compress := os.Getenv("MINIO_COMPRESS"); compress != "" {
 		globalIsCompressionEnabled = strings.EqualFold(compress, "true")
 	}

cmd/config-current.go

@@ -271,8 +271,8 @@ func (s *serverConfig) loadFromEnvs() {
 		s.SetCacheConfig(globalCacheDrives, globalCacheExcludes, globalCacheExpiry, globalCacheMaxUse)
 	}
 
-	if globalKMS != nil {
-		s.KMS = globalKMSConfig
+	if err := Environment.LookupKMSConfig(s.KMS); err != nil {
+		logger.FatalIf(err, "Unable to setup the KMS")
 	}
 
 	if globalIsEnvCompression {
@@ -534,12 +534,8 @@ func (s *serverConfig) loadToCachedConfigs() {
 		globalCacheExpiry = cacheConf.Expiry
 		globalCacheMaxUse = cacheConf.MaxUse
 	}
-	if globalKMS == nil {
-		globalKMSConfig = s.KMS
-		if kms, err := crypto.NewVault(globalKMSConfig); err == nil {
-			globalKMS = kms
-			globalKMSKeyID = globalKMSConfig.Vault.Key.Name
-		}
+	if err := Environment.LookupKMSConfig(s.KMS); err != nil {
+		logger.FatalIf(err, "Unable to setup the KMS")
 	}
 
 	if !globalIsCompressionEnabled {

cmd/crypto/vault.go

@@ -19,200 +19,133 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"os"
-	"strconv"
 	"strings"
 	"time"
 
 	vault "github.com/hashicorp/vault/api"
 )
 
-const (
-	// vaultEndpointEnv Vault endpoint environment variable
-	vaultEndpointEnv = "MINIO_SSE_VAULT_ENDPOINT"
-	// vaultAuthTypeEnv type of vault auth to be used
-	vaultAuthTypeEnv = "MINIO_SSE_VAULT_AUTH_TYPE"
-	// vaultAppRoleIDEnv  Vault AppRole ID environment variable
-	vaultAppRoleIDEnv = "MINIO_SSE_VAULT_APPROLE_ID"
-	// vaultAppSecretIDEnv Vault AppRole Secret environment variable
-	vaultAppSecretIDEnv = "MINIO_SSE_VAULT_APPROLE_SECRET"
-	// vaultKeyVersionEnv Vault Key Version environment variable
-	vaultKeyVersionEnv = "MINIO_SSE_VAULT_KEY_VERSION"
-	// vaultKeyNameEnv Vault Encryption Key Name environment variable
-	vaultKeyNameEnv = "MINIO_SSE_VAULT_KEY_NAME"
-
-	// vaultCAPath is the path to a directory of PEM-encoded CA
-	// cert files to verify the Vault server SSL certificate.
-	vaultCAPath = "MINIO_SSE_VAULT_CAPATH"
-)
-
 var (
 	//ErrKMSAuthLogin is raised when there is a failure authenticating to KMS
 	ErrKMSAuthLogin = errors.New("Vault service did not return auth info")
 )
 
+// VaultKey represents vault encryption key-ring.
+type VaultKey struct {
+	Name    string `json:"name"`    // The name of the encryption key-ring
+	Version int    `json:"version"` // The key version
+}
+
+// VaultAuth represents vault authentication type.
+// Currently the only supported authentication type is AppRole.
+type VaultAuth struct {
+	Type    string       `json:"type"`    // The authentication type
+	AppRole VaultAppRole `json:"approle"` // The AppRole authentication credentials
+}
+
+// VaultAppRole represents vault AppRole authentication credentials
+type VaultAppRole struct {
+	ID     string `json:"id"`     // The AppRole access ID
+	Secret string `json:"secret"` // The AppRole secret
+}
+
+// VaultConfig represents vault configuration.
+type VaultConfig struct {
+	Endpoint  string    `json:"endpoint"` // The vault API endpoint as URL
+	CAPath    string    `json:"-"`        // The path to PEM-encoded certificate files used for mTLS. Currently not used in config file.
+	Auth      VaultAuth `json:"auth"`     // The vault authentication configuration
+	Key       VaultKey  `json:"key-id"`   // The named key used for key-generation / decryption.
+	Namespace string    `json:"-"`        // The vault namespace of enterprise vault instances
+}
+
+// vaultService represents a connection to a vault KMS.
 type vaultService struct {
 	config        *VaultConfig
 	client        *vault.Client
 	leaseDuration time.Duration
 }
 
-// return transit secret engine's path for generate data key operation
-func (v *vaultService) genDataKeyEndpoint(key string) string {
-	return "/transit/datakey/plaintext/" + key
+var _ KMS = (*vaultService)(nil) // compiler check that *vaultService implements KMS
+
+// empty/default vault configuration used to check whether a particular is empty.
+var emptyVaultConfig = VaultConfig{}
+
+// IsEmpty returns true if the vault config struct is an
+// empty configuration.
+func (v *VaultConfig) IsEmpty() bool { return *v == emptyVaultConfig }
+
+// Verify returns a nil error if the vault configuration
+// is valid. A valid configuration is either empty or
+// contains valid non-default values.
+func (v *VaultConfig) Verify() (err error) {
+	if v.IsEmpty() {
+		return // an empty configuration is valid
+	}
+	switch {
+	case v.Endpoint == "":
+		err = errors.New("crypto: missing hashicorp vault endpoint")
+	case strings.ToLower(v.Auth.Type) != "approle":
+		err = fmt.Errorf("crypto: invalid hashicorp vault authentication type: %s is not supported", v.Auth.Type)
+	case v.Auth.AppRole.ID == "":
+		err = errors.New("crypto: missing hashicorp vault AppRole ID")
+	case v.Auth.AppRole.Secret == "":
+		err = errors.New("crypto: missing hashicorp vault AppSecret ID")
+	case v.Key.Name == "":
+		err = errors.New("crypto: missing hashicorp vault key name")
+	case v.Key.Version < 0:
+		err = errors.New("crypto: invalid hashicorp vault key version: The key version must not be negative")
+	}
+	return
 }
 
-// return transit secret engine's path for decrypt operation
-func (v *vaultService) decryptEndpoint(key string) string {
-	return "/transit/decrypt/" + key
-}
+// NewVault initializes Hashicorp Vault KMS by authenticating
+// to Vault with the credentials in config and gets a client
+// token for future api calls.
+func NewVault(config VaultConfig) (KMS, error) {
+	if config.IsEmpty() {
+		return nil, errors.New("crypto: the hashicorp vault configuration must not be empty")
+	}
+	if err := config.Verify(); err != nil {
+		return nil, err
+	}
 
-// VaultKey represents vault encryption key-id name & version
-type VaultKey struct {
-	Name    string `json:"name"`
-	Version int    `json:"version"`
-}
+	vaultCfg := vault.Config{Address: config.Endpoint}
+	if err := vaultCfg.ConfigureTLS(&vault.TLSConfig{CAPath: config.CAPath}); err != nil {
+		return nil, err
+	}
+	client, err := vault.NewClient(&vaultCfg)
+	if err != nil {
+		return nil, err
+	}
+	if config.Namespace != "" {
+		client.SetNamespace(config.Namespace)
+	}
 
-// VaultAuth represents vault auth type to use. For now, AppRole is the only supported
-// auth type.
-type VaultAuth struct {
-	Type    string       `json:"type"`
-	AppRole VaultAppRole `json:"approle"`
-}
-
-// VaultAppRole represents vault approle credentials
-type VaultAppRole struct {
-	ID     string `json:"id"`
-	Secret string `json:"secret"`
-}
-
-// VaultConfig holds config required to start vault service
-type VaultConfig struct {
-	Endpoint string    `json:"endpoint"`
-	Auth     VaultAuth `json:"auth"`
-	Key      VaultKey  `json:"key-id"`
-}
-
-// validate whether all required env variables needed to start vault service have
-// been set
-func validateVaultConfig(c *VaultConfig) error {
-	if c.Endpoint == "" {
-		return fmt.Errorf("Missing hashicorp vault endpoint - %s is empty", vaultEndpointEnv)
+	payload := map[string]interface{}{
+		"role_id":   config.Auth.AppRole.ID,
+		"secret_id": config.Auth.AppRole.Secret,
 	}
-	if strings.ToLower(c.Auth.Type) != "approle" {
-		return fmt.Errorf("Unsupported hashicorp vault auth type - %s", vaultAuthTypeEnv)
-	}
-	if c.Auth.AppRole.ID == "" {
-		return fmt.Errorf("Missing hashicorp vault AppRole ID - %s is empty", vaultAppRoleIDEnv)
-	}
-	if c.Auth.AppRole.Secret == "" {
-		return fmt.Errorf("Missing hashicorp vault AppSecret ID - %s is empty", vaultAppSecretIDEnv)
-	}
-	if c.Key.Name == "" {
-		return fmt.Errorf("Invalid value set in environment variable %s", vaultKeyNameEnv)
-	}
-	if c.Key.Version < 0 {
-		return fmt.Errorf("Invalid value set in environment variable %s", vaultKeyVersionEnv)
-	}
-	return nil
-}
-
-// authenticate to vault with app role id and app role secret, and get a client access token, lease duration
-func getVaultAccessToken(client *vault.Client, appRoleID, appSecret string) (token string, duration int, err error) {
-	data := map[string]interface{}{
-		"role_id":   appRoleID,
-		"secret_id": appSecret,
-	}
-	resp, e := client.Logical().Write("auth/approle/login", data)
-	if e != nil {
-		return token, duration, e
+	resp, err := client.Logical().Write("auth/approle/login", payload)
+	if err != nil {
+		return nil, err
 	}
 	if resp.Auth == nil {
-		return token, duration, ErrKMSAuthLogin
+		return nil, ErrKMSAuthLogin
 	}
-	return resp.Auth.ClientToken, resp.Auth.LeaseDuration, nil
+
+	client.SetToken(resp.Auth.ClientToken)
+	v := &vaultService{client: client, config: &config, leaseDuration: time.Duration(resp.Auth.LeaseDuration)}
+	v.renewToken()
+	return v, nil
 }
 
-// NewVaultConfig sets KMSConfig from environment
-// variables and performs validations.
-func NewVaultConfig() (KMSConfig, error) {
-	kc := KMSConfig{}
-	endpoint := os.Getenv(vaultEndpointEnv)
-	roleID := os.Getenv(vaultAppRoleIDEnv)
-	roleSecret := os.Getenv(vaultAppSecretIDEnv)
-	keyName := os.Getenv(vaultKeyNameEnv)
-	keyVersion := 0
-	authType := "approle"
-	if versionStr := os.Getenv(vaultKeyVersionEnv); versionStr != "" {
-		version, err := strconv.Atoi(versionStr)
-		if err != nil {
-			return kc, fmt.Errorf("Unable to parse %s value (`%s`)", vaultKeyVersionEnv, versionStr)
-		}
-		keyVersion = version
-	}
-	// return if none of the vault env variables are configured
-	if (endpoint == "") && (roleID == "") && (roleSecret == "") && (keyName == "") && (keyVersion == 0) {
-		return kc, nil
-	}
-	c := VaultConfig{
-		Endpoint: endpoint,
-		Auth: VaultAuth{
-			Type: authType,
-			AppRole: VaultAppRole{
-				ID:     roleID,
-				Secret: roleSecret,
-			},
-		},
-		Key: VaultKey{
-			Version: keyVersion,
-			Name:    keyName,
-		},
-	}
-	if err := validateVaultConfig(&c); err != nil {
-		return kc, err
-	}
-	kc.Vault = c
-	return kc, nil
-}
-
-// NewVault initializes Hashicorp Vault KMS by
-// authenticating to Vault with the credentials in KMSConfig,
-// and gets a client token for future api calls.
-func NewVault(kmsConf KMSConfig) (KMS, error) {
-	config := kmsConf.Vault
-	vconfig := &vault.Config{
-		Address: config.Endpoint,
-	}
-	if err := vconfig.ConfigureTLS(&vault.TLSConfig{
-		CAPath: os.Getenv(vaultCAPath),
-	}); err != nil {
-		return nil, err
-	}
-	c, err := vault.NewClient(vconfig)
-	if err != nil {
-		return nil, err
-	}
-
-	if ns, ok := os.LookupEnv("MINIO_SSE_VAULT_NAMESPACE"); ok {
-		c.SetNamespace(ns)
-	}
-
-	accessToken, leaseDuration, err := getVaultAccessToken(c, config.Auth.AppRole.ID, config.Auth.AppRole.Secret)
-	if err != nil {
-		return nil, err
-	}
-	// authenticate and get the access token
-	c.SetToken(accessToken)
-	v := vaultService{client: c, config: &config, leaseDuration: time.Duration(leaseDuration)}
-	v.renewToken(c)
-	return &v, nil
-}
-
-func (v *vaultService) renewToken(c *vault.Client) {
+// renewToken starts a new go-routine which renews
+// the vault authentication token periodically.
+func (v *vaultService) renewToken() {
 	retryDelay := 1 * time.Minute
 	go func() {
 		for {
-			s, err := c.Auth().Token().RenewSelf(int(v.leaseDuration))
+			s, err := v.client.Auth().Token().RenewSelf(int(v.leaseDuration))
 			if err != nil {
 				time.Sleep(retryDelay)
 				continue
@@ -223,48 +156,54 @@ func (v *vaultService) renewToken(c *vault.Client) {
 	}()
 }
 
-// Generates a random plain text key, sealed plain text key from
-// Vault. It returns the plaintext key and sealed plaintext key on success
+// GenerateKey returns a new plaintext key, generated by the KMS,
+// and a sealed version of this plaintext key encrypted using the
+// named key referenced by keyID. It also binds the generated key
+// cryptographically to the provided context.
 func (v *vaultService) GenerateKey(keyID string, ctx Context) (key [32]byte, sealedKey []byte, err error) {
-	contextStream := new(bytes.Buffer)
-	ctx.WriteTo(contextStream)
+	var contextStream bytes.Buffer
+	ctx.WriteTo(&contextStream)
 
 	payload := map[string]interface{}{
 		"context": base64.StdEncoding.EncodeToString(contextStream.Bytes()),
 	}
-	s, err1 := v.client.Logical().Write(v.genDataKeyEndpoint(keyID), payload)
-
-	if err1 != nil {
-		return key, sealedKey, err1
+	s, err := v.client.Logical().Write(fmt.Sprintf("/transit/datakey/plaintext/%s", keyID), payload)
+	if err != nil {
+		return key, sealedKey, err
 	}
 	sealKey := s.Data["ciphertext"].(string)
 	plainKey, err := base64.StdEncoding.DecodeString(s.Data["plaintext"].(string))
 	if err != nil {
-		return key, sealedKey, err1
+		return key, sealedKey, err
 	}
 	copy(key[:], []byte(plainKey))
 	return key, []byte(sealKey), nil
 }
 
-// unsealKMSKey unseals the sealedKey using the Vault master key
-// referenced by the keyID. The plain text key is returned on success.
+// UnsealKey returns the decrypted sealedKey as plaintext key.
+// Therefore it sends the sealedKey to the KMS which decrypts
+// it using the named key referenced by keyID and responses with
+// the plaintext key.
+//
+// The context must be same context as the one provided while
+// generating the plaintext key / sealedKey.
 func (v *vaultService) UnsealKey(keyID string, sealedKey []byte, ctx Context) (key [32]byte, err error) {
-	contextStream := new(bytes.Buffer)
-	ctx.WriteTo(contextStream)
+	var contextStream bytes.Buffer
+	ctx.WriteTo(&contextStream)
+
 	payload := map[string]interface{}{
 		"ciphertext": string(sealedKey),
 		"context":    base64.StdEncoding.EncodeToString(contextStream.Bytes()),
 	}
-	s, err1 := v.client.Logical().Write(v.decryptEndpoint(keyID), payload)
-	if err1 != nil {
-		return key, err1
+	s, err := v.client.Logical().Write(fmt.Sprintf("/transit/decrypt/%s", keyID), payload)
+	if err != nil {
+		return key, err
 	}
 	base64Key := s.Data["plaintext"].(string)
-	plainKey, err1 := base64.StdEncoding.DecodeString(base64Key)
-	if err1 != nil {
-		return key, err1
+	plainKey, err := base64.StdEncoding.DecodeString(base64Key)
+	if err != nil {
+		return key, err
 	}
 	copy(key[:], []byte(plainKey))
-
 	return key, nil
 }

cmd/environment.go

@@ -0,0 +1,166 @@
+// Minio Cloud Storage, (C) 2016, 2017, 2018 Minio, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/minio/minio/cmd/crypto"
+)
+
+const (
+	// EnvKMSMasterKey is the environment variable used to specify
+	// a KMS master key used to protect SSE-S3 per-object keys.
+	// Valid values must be of the from: "KEY_ID:32_BYTE_HEX_VALUE".
+	EnvKMSMasterKey = "MINIO_SSE_MASTER_KEY"
+)
+
+const (
+	// EnvVaultEndpoint is the environment variable used to specify
+	// the vault HTTPS endpoint.
+	EnvVaultEndpoint = "MINIO_SSE_VAULT_ENDPOINT"
+
+	// EnvVaultAuthType is the environment variable used to specify
+	// the authentication type for vault.
+	EnvVaultAuthType = "MINIO_SSE_VAULT_AUTH_TYPE"
+
+	// EnvVaultAppRoleID is the environment variable used to specify
+	// the vault AppRole ID.
+	EnvVaultAppRoleID = "MINIO_SSE_VAULT_APPROLE_ID"
+
+	// EnvVaultAppSecretID is the environment variable used to specify
+	// the vault AppRole secret corresponding to the AppRole ID.
+	EnvVaultAppSecretID = "MINIO_SSE_VAULT_APPROLE_SECRET"
+
+	// EnvVaultKeyVersion is the environment variable used to specify
+	// the vault key version.
+	EnvVaultKeyVersion = "MINIO_SSE_VAULT_KEY_VERSION"
+
+	// EnvVaultKeyName is the environment variable used to specify
+	// the vault named key-ring. In the S3 context it's referred as
+	// customer master key ID (CMK-ID).
+	EnvVaultKeyName = "MINIO_SSE_VAULT_KEY_NAME"
+
+	// EnvVaultCAPath is the environment variable used to specify the
+	// path to a directory of PEM-encoded CA cert files. These CA cert
+	// files are used to authenticate Minio to Vault over mTLS.
+	EnvVaultCAPath = "MINIO_SSE_VAULT_CAPATH"
+
+	// EnvVaultNamespace is the environment variable used to specify
+	// vault namespace. The vault namespace is used if the enterprise
+	// version of Hashicorp Vault is used.
+	EnvVaultNamespace = "MINIO_SSE_VAULT_NAMESPACE"
+)
+
+// Environment provides functions for accessing environment
+// variables.
+var Environment = environment{}
+
+type environment struct{}
+
+// Get retrieves the value of the environment variable named
+// by the key. If the variable is present in the environment the
+// value (which may be empty) is returned. Otherwise it returns
+// the specified default value.
+func (environment) Get(key, defaultValue string) string {
+	if v, ok := os.LookupEnv(key); ok {
+		return v
+	}
+	return defaultValue
+}
+
+// Lookup retrieves the value of the environment variable named
+// by the key. If the variable is present in the environment the
+// value (which may be empty) is returned and the boolean is true.
+// Otherwise the returned value will be empty and the boolean will
+// be false.
+func (environment) Lookup(key string) (string, bool) { return os.LookupEnv(key) }
+
+// LookupKMSConfig extracts the KMS configuration provided by environment
+// variables and merge them with the provided KMS configuration. The
+// merging follows the following rules:
+//
+// 1. A valid value provided as environment variable is higher prioritized
+// than the provided configuration and overwrites the value from the
+// configuration file.
+//
+// 2. A value specified as environment variable never changes the configuration
+// file. So it is never made a persistent setting.
+//
+// It sets the global KMS configuration according to the merged configuration
+// on success.
+func (env environment) LookupKMSConfig(config crypto.KMSConfig) (err error) {
+	// Lookup Hashicorp-Vault configuration & overwrite config entry if ENV var is present
+	config.Vault.Endpoint = env.Get(EnvVaultEndpoint, config.Vault.Endpoint)
+	config.Vault.CAPath = env.Get(EnvVaultCAPath, config.Vault.CAPath)
+	config.Vault.Auth.Type = env.Get(EnvVaultAuthType, config.Vault.Auth.Type)
+	config.Vault.Auth.AppRole.ID = env.Get(EnvVaultAppRoleID, config.Vault.Auth.AppRole.ID)
+	config.Vault.Auth.AppRole.Secret = env.Get(EnvVaultAppSecretID, config.Vault.Auth.AppRole.Secret)
+	config.Vault.Key.Name = env.Get(EnvVaultKeyName, config.Vault.Key.Name)
+	config.Vault.Namespace = env.Get(EnvVaultNamespace, config.Vault.Namespace)
+	keyVersion := env.Get(EnvVaultKeyVersion, strconv.Itoa(config.Vault.Key.Version))
+	config.Vault.Key.Version, err = strconv.Atoi(keyVersion)
+	if err != nil {
+		return fmt.Errorf("Invalid ENV variable: Unable to parse %s value (`%s`)", EnvVaultKeyVersion, keyVersion)
+	}
+	if err = config.Vault.Verify(); err != nil {
+		return err
+	}
+
+	// Lookup KMS master keys - only available through ENV.
+	if masterKey, ok := env.Lookup(EnvKMSMasterKey); ok {
+		if !config.Vault.IsEmpty() { // Vault and KMS master key provided
+			return errors.New("Ambiguous KMS configuration: vault configuration and a master key are provided at the same time")
+		}
+		globalKMSKeyID, globalKMS, err = parseKMSMasterKey(masterKey)
+		if err != nil {
+			return err
+		}
+	}
+	if !config.Vault.IsEmpty() {
+		globalKMS, err = crypto.NewVault(config.Vault)
+		if err != nil {
+			return err
+		}
+		globalKMSKeyID = config.Vault.Key.Name
+	}
+	return nil
+}
+
+// parseKMSMasterKey parses the value of the environment variable
+// `EnvKMSMasterKey` and returns a key-ID and a master-key KMS on success.
+func parseKMSMasterKey(envArg string) (string, crypto.KMS, error) {
+	values := strings.SplitN(envArg, ":", 2)
+	if len(values) != 2 {
+		return "", nil, fmt.Errorf("Invalid KMS master key: %s does not contain a ':'", envArg)
+	}
+	var (
+		keyID  = values[0]
+		hexKey = values[1]
+	)
+	if len(hexKey) != 64 { // 2 hex bytes = 1 byte
+		return "", nil, fmt.Errorf("Invalid KMS master key: %s not a 32 bytes long HEX value", hexKey)
+	}
+	var masterKey [32]byte
+	if _, err := hex.Decode(masterKey[:], []byte(hexKey)); err != nil {
+		return "", nil, fmt.Errorf("Invalid KMS master key: %s not a 32 bytes long HEX value", hexKey)
+	}
+	return keyID, crypto.NewKMS(masterKey), nil
+}

cmd/gateway/nas/gateway-nas.go

@@ -117,7 +117,7 @@ func (g *NAS) Production() bool {
 }
 
 // IsListenBucketSupported returns whether listen bucket notification is applicable for this gateway.
-func (g *NAS) IsListenBucketSupported() bool {
+func (n *nasObjects) IsListenBucketSupported() bool {
 	return false
 }
 

cmd/globals.go

@@ -230,8 +230,6 @@ var (
 	globalKMSKeyID string
 	// Allocated KMS
 	globalKMS crypto.KMS
-	// KMS config
-	globalKMSConfig crypto.KMSConfig
 
 	// Is compression include extensions/content-types set.
 	globalIsEnvCompression bool

cmd/http/listener.go

@@ -178,6 +178,9 @@ type httpListener struct {
 // isRoutineNetErr returns true if error is due to a network timeout,
 // connect reset or io.EOF and false otherwise
 func isRoutineNetErr(err error) bool {
+	if err == nil {
+		return false
+	}
 	if nErr, ok := err.(*net.OpError); ok {
 		// Check if the error is a tcp connection reset
 		if syscallErr, ok := nErr.Err.(*os.SyscallError); ok {
@@ -188,7 +191,8 @@ func isRoutineNetErr(err error) bool {
 		// Check if the error is a timeout
 		return nErr.Timeout()
 	}
-	return err == io.EOF
+	// check for io.EOF and also some times io.EOF is wrapped is another error type.
+	return err == io.EOF || err.Error() == "EOF"
 }
 
 // start - starts separate goroutine for each TCP listener.  A valid insecure/TLS HTTP new connection is passed to httpListener.acceptCh.

cmd/http/listener_test.go

@@ -20,6 +20,7 @@ import (
 	"bufio"
 	"bytes"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -801,6 +802,10 @@ func TestIgnoreErr(t *testing.T) {
 			err:  &net.OpError{Err: &myTimeoutErr{timeout: true}},
 			want: true,
 		},
+		{
+			err:  errors.New("EOF"),
+			want: true,
+		},
 		{
 			err:  &net.OpError{Err: &myTimeoutErr{timeout: false}},
 			want: false,
@@ -809,6 +814,10 @@ func TestIgnoreErr(t *testing.T) {
 			err:  io.ErrUnexpectedEOF,
 			want: false,
 		},
+		{
+			err:  nil,
+			want: false,
+		},
 	}
 
 	for i, tc := range testCases {

cmd/http/server.go

@@ -36,7 +36,7 @@ const (
 	DefaultShutdownTimeout = 5 * time.Second
 
 	// DefaultTCPKeepAliveTimeout - default TCP keep alive timeout for accepted connection.
-	DefaultTCPKeepAliveTimeout = 10 * time.Second
+	DefaultTCPKeepAliveTimeout = 30 * time.Second
 
 	// DefaultReadTimeout - default timout to read data from accepted connection.
 	DefaultReadTimeout = 5 * time.Minute

cmd/notification.go

@@ -537,6 +537,9 @@ func (args eventArgs) ToEvent() event.Event {
 	if args.EventName != event.ObjectRemovedDelete {
 		newEvent.S3.Object.ETag = args.Object.ETag
 		newEvent.S3.Object.Size = args.Object.Size
+		if args.Object.IsCompressed() {
+			newEvent.S3.Object.Size = args.Object.GetActualSize()
+		}
 		newEvent.S3.Object.ContentType = args.Object.ContentType
 		newEvent.S3.Object.UserMetadata = args.Object.UserDefined
 	}

cmd/object-handlers.go

@@ -218,11 +218,13 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r
 			writeErrorResponse(w, ErrInvalidFileHeaderInfo, r.URL, guessIsBrowserReq(r))
 			return
 		}
-		if selectReq.OutputSerialization.CSV.QuoteFields != s3select.CSVQuoteFieldsAlways &&
-			selectReq.OutputSerialization.CSV.QuoteFields != s3select.CSVQuoteFieldsAsNeeded &&
-			selectReq.OutputSerialization.CSV.QuoteFields != "" {
-			writeErrorResponse(w, ErrInvalidQuoteFields, r.URL, guessIsBrowserReq(r))
-			return
+		if selectReq.OutputSerialization.CSV != nil {
+			if selectReq.OutputSerialization.CSV.QuoteFields != s3select.CSVQuoteFieldsAlways &&
+				selectReq.OutputSerialization.CSV.QuoteFields != s3select.CSVQuoteFieldsAsNeeded &&
+				selectReq.OutputSerialization.CSV.QuoteFields != "" {
+				writeErrorResponse(w, ErrInvalidQuoteFields, r.URL, guessIsBrowserReq(r))
+				return
+			}
 		}
 		if len(selectReq.InputSerialization.CSV.RecordDelimiter) > 2 {
 			writeErrorResponse(w, ErrInvalidRequestParameter, r.URL, guessIsBrowserReq(r))
@@ -321,6 +323,11 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
 	bucket := vars["bucket"]
 	object := vars["object"]
 
+	if vid := r.URL.Query().Get("versionId"); vid != "" && vid != "null" {
+		writeErrorResponse(w, ErrNoSuchVersion, r.URL, guessIsBrowserReq(r))
+		return
+	}
+
 	var (
 		opts ObjectOptions
 		err  error
@@ -481,7 +488,7 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
 		return
 	}
 	if crypto.S3.IsRequested(r.Header) || crypto.S3KMS.IsRequested(r.Header) { // If SSE-S3 or SSE-KMS present -> AWS fails with undefined error
-		writeErrorResponse(w, ErrBadRequest, r.URL, guessIsBrowserReq(r))
+		writeErrorResponseHeadersOnly(w, ErrBadRequest)
 		return
 	}
 
@@ -489,6 +496,11 @@ func (api objectAPIHandlers) HeadObjectHandler(w http.ResponseWriter, r *http.Re
 	bucket := vars["bucket"]
 	object := vars["object"]
 
+	if vid := r.URL.Query().Get("versionId"); vid != "" && vid != "null" {
+		writeErrorResponseHeadersOnly(w, ErrNoSuchVersion)
+		return
+	}
+
 	getObjectInfo := objectAPI.GetObjectInfo
 	if api.CacheAPI() != nil {
 		getObjectInfo = api.CacheAPI().GetObjectInfo
@@ -695,6 +707,22 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		cpSrcPath = r.Header.Get("X-Amz-Copy-Source")
 	}
 
+	// Check https://docs.aws.amazon.com/AmazonS3/latest/dev/ObjectVersioning.html
+	// Regardless of whether you have enabled versioning, each object in your bucket
+	// has a version ID. If you have not enabled versioning, Amazon S3 sets the value
+	// of the version ID to null. If you have enabled versioning, Amazon S3 assigns a
+	// unique version ID value for the object.
+	if u, err := url.Parse(cpSrcPath); err == nil {
+		// Check if versionId query param was added, if yes then check if
+		// its non "null" value, we should error out since we do not support
+		// any versions other than "null".
+		if vid := u.Query().Get("versionId"); vid != "" && vid != "null" {
+			writeErrorResponse(w, ErrNoSuchVersion, r.URL, guessIsBrowserReq(r))
+			return
+		}
+		cpSrcPath = u.Path
+	}
+
 	srcBucket, srcObject := path2BucketAndObject(cpSrcPath)
 	// If source object is empty or bucket is empty, reply back invalid copy source.
 	if srcObject == "" || srcBucket == "" {
@@ -786,13 +814,15 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		length = actualSize
 	}
 
+	var compressMetadata map[string]string
 	// No need to compress for remote etcd calls
 	// Pass the decompressed stream to such calls.
 	isCompressed := objectAPI.IsCompressionSupported() && isCompressible(r.Header, srcObject) && !isRemoteCallRequired(ctx, srcBucket, dstBucket, objectAPI)
 	if isCompressed {
-		// Storing the compression metadata.
-		srcInfo.UserDefined[ReservedMetadataPrefix+"compression"] = compressionAlgorithmV1
-		srcInfo.UserDefined[ReservedMetadataPrefix+"actual-size"] = strconv.FormatInt(actualSize, 10)
+		compressMetadata = make(map[string]string, 2)
+		// Preserving the compression metadata.
+		compressMetadata[ReservedMetadataPrefix+"compression"] = compressionAlgorithmV1
+		compressMetadata[ReservedMetadataPrefix+"actual-size"] = strconv.FormatInt(actualSize, 10)
 		// Remove all source encrypted related metadata to
 		// avoid copying them in target object.
 		crypto.RemoveInternalEntries(srcInfo.UserDefined)
@@ -935,6 +965,11 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		return
 	}
 
+	// Store the preserved compression metadata.
+	for k, v := range compressMetadata {
+		srcInfo.UserDefined[k] = v
+	}
+
 	// We need to preserve the encryption headers set in EncryptRequest,
 	// so we do not want to override them, copy them instead.
 	for k, v := range encMetadata {

cmd/posix.go

@@ -71,6 +71,7 @@ type posix struct {
 	connected bool
 
 	diskMount bool // indicates if the path is an actual mount.
+	driveSync bool // indicates if the backend is synchronous.
 
 	// Disk usage metrics
 	stopUsageCh chan struct{}
@@ -191,6 +192,15 @@ func newPosix(path string) (*posix, error) {
 		diskMount:   mountinfo.IsLikelyMountPoint(path),
 	}
 
+	var pf BoolFlag
+	if driveSync := os.Getenv("MINIO_DRIVE_SYNC"); driveSync != "" {
+		pf, err = ParseBoolFlag(driveSync)
+		if err != nil {
+			return nil, err
+		}
+		p.driveSync = bool(pf)
+	}
+
 	if !p.diskMount {
 		go p.diskUsage(globalServiceDoneCh)
 	}
@@ -1013,8 +1023,14 @@ func (s *posix) AppendFile(volume, path string, buf []byte) (err error) {
 		return errFaultyDisk
 	}
 
-	// Create file if not found
-	w, err := s.openFile(volume, path, os.O_CREATE|os.O_APPEND|os.O_WRONLY)
+	var w *os.File
+	// Create file if not found, additionally also enables synchronous
+	// operation if asked by the user.
+	if s.driveSync {
+		w, err = s.openFile(volume, path, os.O_CREATE|os.O_SYNC|os.O_APPEND|os.O_WRONLY)
+	} else {
+		w, err = s.openFile(volume, path, os.O_CREATE|os.O_APPEND|os.O_WRONLY)
+	}
 	if err != nil {
 		return err
 	}

cmd/web-handlers.go

@@ -258,7 +258,8 @@ func (web *webAPIHandlers) ListBuckets(r *http.Request, args *WebGenericArgs, re
 		listBuckets = web.CacheAPI().ListBuckets
 	}
 
-	if _, _, authErr := webRequestAuthenticate(r); authErr != nil {
+	claims, owner, authErr := webRequestAuthenticate(r)
+	if authErr != nil {
 		return toJSONError(authErr)
 	}
 
@@ -270,10 +271,19 @@ func (web *webAPIHandlers) ListBuckets(r *http.Request, args *WebGenericArgs, re
 		}
 		for _, dnsRecord := range dnsBuckets {
 			bucketName := strings.Trim(dnsRecord.Key, "/")
-			reply.Buckets = append(reply.Buckets, WebBucketInfo{
-				Name:         bucketName,
-				CreationDate: dnsRecord.CreationDate,
-			})
+			if globalIAMSys.IsAllowed(iampolicy.Args{
+				AccountName:     claims.Subject,
+				Action:          iampolicy.Action(policy.GetObjectAction),
+				BucketName:      bucketName,
+				ConditionValues: getConditionValues(r, ""),
+				IsOwner:         owner,
+				ObjectName:      "",
+			}) {
+				reply.Buckets = append(reply.Buckets, WebBucketInfo{
+					Name:         bucketName,
+					CreationDate: dnsRecord.CreationDate,
+				})
+			}
 		}
 	} else {
 		buckets, err := listBuckets(context.Background())
@@ -281,10 +291,19 @@ func (web *webAPIHandlers) ListBuckets(r *http.Request, args *WebGenericArgs, re
 			return toJSONError(err)
 		}
 		for _, bucket := range buckets {
-			reply.Buckets = append(reply.Buckets, WebBucketInfo{
-				Name:         bucket.Name,
-				CreationDate: bucket.Created,
-			})
+			if globalIAMSys.IsAllowed(iampolicy.Args{
+				AccountName:     claims.Subject,
+				Action:          iampolicy.Action(policy.GetObjectAction),
+				BucketName:      bucket.Name,
+				ConditionValues: getConditionValues(r, ""),
+				IsOwner:         owner,
+				ObjectName:      "",
+			}) {
+				reply.Buckets = append(reply.Buckets, WebBucketInfo{
+					Name:         bucket.Name,
+					CreationDate: bucket.Created,
+				})
+			}
 		}
 	}
 

docs/config/README.md

@@ -132,9 +132,9 @@ By default, parity for objects with standard storage class is set to `N/2`, and
 
 ## Environment only settings
 
-#### Browser
+### Browser
 
-Enable or disable access to web UI. By default it is set to `on`. You may override this field with ``MINIO_BROWSER`` environment variable.
+Enable or disable access to web UI. By default it is set to `on`. You may override this field with `MINIO_BROWSER` environment variable.
 
 Example:
 
@@ -145,7 +145,7 @@ minio server /data
 
 ### Domain
 
-By default, Minio supports path-style requests that are of the format http://mydomain.com/bucket/object. MINIO_DOMAIN environment variable is used to enable virtual-host-style requests. If the request `Host` header matches with `(.+).mydomain.com` then the matched pattern `$1` is used as bucket and the path is used as object. More information on path-style and virtual-host-style [here](http://docs.aws.amazon.com/AmazonS3/latest/dev/RESTAPI.html)
+By default, Minio supports path-style requests that are of the format http://mydomain.com/bucket/object. `MINIO_DOMAIN` environment variable is used to enable virtual-host-style requests. If the request `Host` header matches with `(.+).mydomain.com` then the matched pattern `$1` is used as bucket and the path is used as object. More information on path-style and virtual-host-style [here](http://docs.aws.amazon.com/AmazonS3/latest/dev/RESTAPI.html)
 Example:
 
 ```sh
@@ -153,6 +153,17 @@ export MINIO_DOMAIN=mydomain.com
 minio server /data
 ```
 
+### Drive Sync
+
+By default, Minio writes to disk in synchronous mode for all metadata operations. Set `MINIO_DRIVE_SYNC` environment variable to enable synchronous mode for all data operations as well.
+
+Example:
+
+```sh
+export MINIO_DRIVE_SYNC=on
+minio server /data
+```
+
 ## Explore Further
 
 * [Minio Quickstart Guide](https://docs.minio.io/docs/minio-quickstart-guide)

docs/gateway/gcs.md

@@ -1,20 +1,26 @@
 # Minio GCS Gateway [![Slack](https://slack.minio.io/slack?type=svg)](https://slack.minio.io)
-Minio GCS Gateway adds Amazon S3 compatibility to Google Cloud Storage.
 
-## Run Minio Gateway for GCS
-### Create service account key for GCS and get the credentials file
-1. Go to the [API Console Credentials page](https://console.developers.google.com/project/_/apis/credentials).
-2. Select your project or create a new project. Note down your project ID.
-3. On the Credentials page, select the __Create credentials__ drop-down, then select __Service account key__.
-4. From the __Service account__ drop-down, select __New service account__
-5. Fill up __Service account name__ and __Service account ID__
-6. For the __Role__, click the select dropdown to choose __Storage__ -> __Storage Admin__ _(Full control of GCS resources)_
-7. Click the __Create__ button. This will download a credentials file to your desktop. Let's call this credentials.json
+Minio GCS Gateway allows you to access Google Cloud Storage (GCS) with Amazon S3-compatible APIs
 
-Note: Alternate ways to setup *Application Default Credentials* is explained [here](https://developers.google.com/identity/protocols/application-default-credentials)
+- [Run Minio Gateway for GCS](#run-minio-gateway-for-gcs)
+- [Test Using Minio Browser](#test-using-minio-browser)
+- [Test Using Minio Client](#test-using-minio-client)
 
-### Using Docker
-```
+## <a name="run-minio-gateway-for-gcs"></a>1. Run Minio Gateway for GCS
+
+### 1.1 Create a Service Account key for GCS and get the Credentials File
+1. Navigate to the [API Console Credentials page](https://console.developers.google.com/project/_/apis/credentials).
+2. Select a project or create a new project. Note the project ID.
+3. Select the **Create credentials** dropdown on the **Credentials** page, and click **Service account key**.
+4. Select **New service account** from the **Service account** dropdown.
+5. Populate the **Service account name** and **Service account ID**.
+6. Click the dropdown for the **Role** and choose **Storage** > **Storage Admin** *(Full control of GCS resources)*.
+7. Click the **Create** button to download a credentials file and rename it to `credentials.json`.
+
+**Note:** For alternate ways to set up *Application Default Credentials*, see [Setting Up Authentication for Server to Server Production Applications](https://developers.google.com/identity/protocols/application-default-credentials).
+
+### 1.1 Run Minio GCS Gateway Using Docker
+```sh
 docker run -p 9000:9000 --name gcs-s3 \
  -v /path/to/credentials.json:/credentials.json \
  -e "GOOGLE_APPLICATION_CREDENTIALS=/credentials.json" \
@@ -23,47 +29,60 @@ docker run -p 9000:9000 --name gcs-s3 \
  minio/minio gateway gcs yourprojectid
 ```
 
-### Using Binary
-```
+### 1.2 Run Minio GCS Gateway Using the Minio Binary
+
+```sh
 export GOOGLE_APPLICATION_CREDENTIALS=/path/to/credentials.json
 export MINIO_ACCESS_KEY=minioaccesskey
 export MINIO_SECRET_KEY=miniosecretkey
 minio gateway gcs yourprojectid
 ```
 
-## Test using Minio Browser
-Minio Gateway comes with an embedded web based object browser. Point your web browser to http://127.0.0.1:9000 to ensure that your server has started successfully.
+## <a name="test-using-minio-browser"></a>2. Test Using Minio Browser
+
+Minio Gateway comes with an embedded web-based object browser that outputs content to http://127.0.0.1:9000. To test that Minio Gateway is running, open a web browser, navigate to http://127.0.0.1:9000, and ensure that the object browser is displayed.
 
 ![Screenshot](https://github.com/minio/minio/blob/master/docs/screenshots/minio-browser-gateway.png?raw=true)
 
-## Test using Minio Client `mc`
-`mc` provides a modern alternative to UNIX commands such as ls, cat, cp, mirror, diff etc. It supports filesystems and Amazon S3 compatible cloud storage services.
+## <a name="test-using-minio-client"></a>3. Test Using Minio Client
 
-### Configure `mc`
-```
+Minio Client is a command-line tool called `mc` that provides UNIX-like commands for interacting with the server (e.g. ls, cat, cp, mirror, diff, find, etc.).  `mc` supports file systems and Amazon S3-compatible cloud storage services (AWS Signature v2 and v4).
+
+### 3.1 Configure the Gateway using Minio Client
+
+Use the following command to configure the gateway:
+
+```sh
 mc config host add mygcs http://gateway-ip:9000 minioaccesskey miniosecretkey
 ```
 
-### List containers on GCS
-```
+### 3.2 List Containers on GCS
+
+Use the following command to list the containers on GCS:
+
+```sh
 mc ls mygcs
+```
+
+A response similar to this one should be displayed:
+
+```
 [2017-02-22 01:50:43 PST]     0B ferenginar/
 [2017-02-26 21:43:51 PST]     0B my-container/
 [2017-02-26 22:10:11 PST]     0B test-container1/
 ```
 
-### Known limitations
-Gateway inherits the following GCS limitations:
+### 3.3 Known limitations
+Minio Gateway has the following limitations when used with GCS:
 
-- Only read-only or write-only bucket policy supported at bucket level, all other variations will return API Notimplemented error.
-- _List Multipart Uploads_ and _List Object parts_ always returns empty list. i.e Client will need to remember all the parts that it has uploaded and use it for _Complete Multipart Upload_
+* It only supports read-only and write-only bucket policies at the bucket level; all other variations will return `API Not implemented`.
+* The `List Multipart Uploads` and `List Object parts` commands always return empty lists. Therefore, the client must store all of the parts that it has uploaded and use that information when invoking the `_Complete Multipart Upload` command.
 
 Other limitations:
 
-- Bucket notification APIs are not supported.
+* Bucket notification APIs are not supported.
 
-## Explore Further
+## <a name="explore-further"></a>4. Explore Further
 - [`mc` command-line interface](https://docs.minio.io/docs/minio-client-quickstart-guide)
 - [`aws` command-line interface](https://docs.minio.io/docs/aws-cli-with-minio)
 - [`minio-go` Go SDK](https://docs.minio.io/docs/golang-client-quickstart-guide)
-

docs/kms/README.md

@@ -1,26 +1,49 @@
 # KMS Quickstart Guide [![Slack](https://slack.minio.io/slack?type=svg)](https://slack.minio.io)
 
-KMS feature allows you to use Vault to generate and manage encryption keys for use by the minio server to encrypt objects. This document explains how to configure Minio with Vault as KMS.
+Minio uses a key-management-system (KMS) to support SSE-S3. If a client requests SSE-S3 or auto-encryption
+is enabled the Minio server encrypts each object with an unique object key which is protected by a master key
+managed by the KMS. Usually many/all object keys are protected by a single master key.
+
+Minio supports two different KMS concepts:
+ - External KMS:
+   Minio can be configured to use an external KMS i.e. [Hashicorp Vault](https://www.vaultproject.io/).
+   An external KMS decouples Minio as storage system from key-management. An external KMS can
+   be managed by a dedicated security team and allows to grant/deny access to (certain) objects
+   by en/disabling the corresponding master keys on demand.
+   However, an external KMS causes configuration and management overhead. 
+ - Direct KMS master keys:
+   Minio can also be configured to directly use a master key specified by the ENV. variable `MINIO_SSE_MASTER_KEY`.
+   Direct master keys are useful if the storage backend is not on the same machine as the Minio server - e.g.
+   if network drives or Minio gateway is used - and an external KMS would cause too much management overhead.  
+   Note: If the Minio server machine is ever compromised, then the master key must also be 
+   treated as compromised.
 
 ## Get started
 
 ### 1. Prerequisites
 Install Minio - [Minio Quickstart Guide](https://docs.minio.io/docs/minio-quickstart-guide).
 
-### 2. Configure Vault
-Vault as Key Management System requires following to be configured in Vault
+### 2. Setup a KMS
 
-- [transit](https://www.vaultproject.io/docs/secrets/transit/index.html) backend configured with a named encryption key-ring
-- [AppRole](https://www.vaultproject.io/docs/auth/approle.html) based authentication with read/update policy for transit backend. In particular, read and update policy are required for the [Generate Data Key](https://www.vaultproject.io/api/secret/transit/index.html#generate-data-key) endpoint and [Decrypt Data](https://www.vaultproject.io/api/secret/transit/index.html#decrypt-data) endpoint.
+Either use Hashicorp Vault as external KMS or specify a master key directly depending on your use case.
+
+#### 2.1 Setup Hashicorp Vault
 
 Here is a sample quick start for configuring vault with a transit backend and Approle with correct policy 
-#### 2.1 Start Vault server in dev mode
+
+Minio requires the following Vault setup:
+- The [transit backend](https://www.vaultproject.io/api/secret/transit/index.html) configured with a named encryption key-ring
+- [AppRole](https://www.vaultproject.io/docs/auth/approle.html) based authentication with read/update policy for transit backend. In particular, read and update policy are required for the [Generate Data Key](https://www.vaultproject.io/api/secret/transit/index.html#generate-data-key) endpoint and [Decrypt Data](https://www.vaultproject.io/api/secret/transit/index.html#decrypt-data) endpoint.
+
+**2.1.1 Start Vault server in dev mode**
+
 In dev mode, Vault server runs in-memory and starts unsealed. Note that running Vault in dev mode is insecure and any data stored in the Vault is lost upon restart.
 ```
 vault server -dev
 ```
 
-#### 2.2 Set up vault transit backend and create an app role
+**2.1.2 Set up vault transit backend and create an app role**
+
 ```
 cat > vaultpolicy.hcl <<EOF
 path "transit/datakey/plaintext/my-minio-key" { 
@@ -50,11 +73,9 @@ vault write -f auth/approle/role/my-role/secret-id
 
 The AppRole ID, AppRole Secret Id, Vault endpoint and Vault key name can now be used to start minio server with Vault as KMS.
 
-Note: If [Vault Namespaces](https://learn.hashicorp.com/vault/operations/namespaces) are in use, VAULT_NAMESPACE variable needs to be set before setting approle and transit secrets engine.
+**2.1.3 Vault Environment variables**
 
-### 3. Environment variables
-
-You'll need the Vault endpoint, AppRole ID, AppRole SecretID and encryption key-ring name defined in step 2.2
+You'll need the Vault endpoint, AppRole ID, AppRole SecretID and encryption key-ring name defined in step 2.1.2
 
 ```sh
 export MINIO_SSE_VAULT_APPROLE_ID=9b56cc08-8258-45d5-24a3-679876769126
@@ -64,16 +85,24 @@ export MINIO_SSE_VAULT_KEY_NAME=my-minio-key
 minio server ~/export
 ```
 
-Optionally set `MINIO_SSE_VAULT_CAPATH` as the path to a directory of PEM-encoded CA cert files to verify the Vault server SSL certificate.
-```
-export MINIO_SSE_VAULT_CAPATH=/home/user/custom-pems
-```
-
 Optionally set `MINIO_SSE_VAULT_NAMESPACE` if AppRole and Transit Secrets engine have been scoped to Vault Namespace
+
 ```
 export MINIO_SSE_VAULT_NAMESPACE=ns1
 ```
-### 4. Test your setup
+
+Note: If [Vault Namespaces](https://learn.hashicorp.com/vault/operations/namespaces) are in use, MINIO_SSE_VAULT_NAMESPACE variable needs to be set before setting approle and transit secrets engine.
+
+#### 2.2 Specify a master key
+
+A KMS master key consists of a master-key ID (CMK) and the 256 bit master key encoded as HEX value separated by a `:`.
+A KMS master key can be specified directly using:
+
+```sh
+export MINIO_SSE_MASTER_KEY=my-minio-key:6368616e676520746869732070617373776f726420746f206120736563726574
+```
+
+### 3. Test your setup
 
 To test this setup, start minio server with environment variables set in Step 3, and server is ready to handle SSE-S3 requests.
 

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '2'
 # 9001 through 9004.
 services:
  minio1:
-  image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+  image: minio/minio:RELEASE.2018-12-06T01-27-43Z
   volumes:
    - data1:/data
   ports:
@@ -15,7 +15,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio2:
-  image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+  image: minio/minio:RELEASE.2018-12-06T01-27-43Z
   volumes:
    - data2:/data
   ports:
@@ -25,7 +25,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio3:
-  image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+  image: minio/minio:RELEASE.2018-12-06T01-27-43Z
   volumes:
    - data3:/data
   ports:
@@ -35,7 +35,7 @@ services:
    MINIO_SECRET_KEY: minio123
   command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
  minio4:
-  image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+  image: minio/minio:RELEASE.2018-12-06T01-27-43Z
   volumes:
    - data4:/data
   ports:

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.1'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -24,7 +24,7 @@ services:
       - access_key
 
   minio2:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -46,7 +46,7 @@ services:
       - access_key
 
   minio3:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -68,7 +68,7 @@ services:
       - access_key
 
   minio4:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -24,7 +24,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio2:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -46,7 +46,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio3:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -68,7 +68,7 @@ services:
     command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export
 
   minio4:
-    image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+    image: minio/minio:RELEASE.2018-12-06T01-27-43Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/kubernetes/minio-distributed-daemonset.yaml

@@ -30,7 +30,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+        image: minio/minio:RELEASE.2018-12-06T01-27-43Z
         # Unfortunately you must manually define each server. Perhaps autodiscovery via DNS can be implemented in the future.
         args:
         - server

docs/orchestration/kubernetes/minio-distributed-statefulset.yaml

@@ -21,7 +21,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+        image: minio/minio:RELEASE.2018-12-06T01-27-43Z
         args:
         - server
         - http://minio-0.minio.default.svc.cluster.local/data

docs/orchestration/kubernetes/minio-gcs-gateway-deployment.yaml

@@ -21,7 +21,7 @@ spec:
       containers:
       - name: minio
         # Pulls the default Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+        image: minio/minio:RELEASE.2018-12-06T01-27-43Z
         args:
         - gateway
         - gcs

docs/orchestration/kubernetes/minio-standalone-deployment.yaml

@@ -29,7 +29,7 @@ spec:
         - name: data 
           mountPath: "/data"
         # Pulls the lastest Minio image from Docker Hub
-        image: minio/minio:RELEASE.2018-11-30T03-56-59Z
+        image: minio/minio:RELEASE.2018-12-06T01-27-43Z
         args:
         - server
         - /data

pkg/dns/etcd_dns.go

@@ -85,16 +85,28 @@ func (c *coreDNS) list(key string) ([]SrvRecord, error) {
 		}
 		srvRecord.Key = strings.TrimPrefix(string(n.Key), key)
 		srvRecord.Key = strings.TrimSuffix(srvRecord.Key, srvRecord.Host)
-		srvRecords = append(srvRecords, srvRecord)
+		// SRV records are stored in the following form
+		// /skydns/net/miniocloud/bucket1, so this function serves multiple
+		// purposes basically when we do a Get(bucketName) this function
+		// should return a single DNS record for any input 'bucketName'.
+		//
+		// In all other situations when we want to list all DNS records,
+		// which is handled in the else clause.
+		if key != msg.Path(fmt.Sprintf(".%s.", c.domainName), defaultPrefixPath) {
+			if srvRecord.Key == "/" {
+				srvRecords = append(srvRecords, srvRecord)
+			}
+		} else {
+			srvRecords = append(srvRecords, srvRecord)
+		}
 
 	}
-	if srvRecords != nil {
-		sort.Slice(srvRecords, func(i int, j int) bool {
-			return srvRecords[i].Key < srvRecords[j].Key
-		})
-	} else {
+	if len(srvRecords) == 0 {
 		return nil, ErrNoEntriesFound
 	}
+	sort.Slice(srvRecords, func(i int, j int) bool {
+		return srvRecords[i].Key < srvRecords[j].Key
+	})
 	return srvRecords, nil
 }
 
@@ -123,9 +135,18 @@ func (c *coreDNS) Put(bucket string) error {
 // Removes DNS entries added in Put().
 func (c *coreDNS) Delete(bucket string) error {
 	key := msg.Path(fmt.Sprintf("%s.%s.", bucket, c.domainName), defaultPrefixPath)
-	ctx, cancel := context.WithTimeout(context.Background(), defaultContextTimeout)
-	_, err := c.etcdClient.Delete(ctx, key)
-	defer cancel()
+	srvRecords, err := c.list(key)
+	if err != nil {
+		return err
+	}
+	for _, record := range srvRecords {
+		dctx, dcancel := context.WithTimeout(context.Background(), defaultContextTimeout)
+		if _, err = c.etcdClient.Delete(dctx, key+"/"+record.Host); err != nil {
+			dcancel()
+			return err
+		}
+		dcancel()
+	}
 	return err
 }
 

pkg/s3select/format/csv/csv.go

@@ -57,9 +57,12 @@ type Options struct {
 	// SQL expression meant to be evaluated.
 	Expression string
 
-	// What the outputted CSV will be delimited by .
+	// Output CSV will be delimited by.
 	OutputFieldDelimiter string
 
+	// Output CSV record will be delimited by.
+	OutputRecordDelimiter string
+
 	// Size of incoming object
 	StreamSize int64
 
@@ -68,6 +71,9 @@ type Options struct {
 
 	// Progress enabled, enable/disable progress messages.
 	Progress bool
+
+	// Output format type, supported values are CSV and JSON
+	OutputType format.Type
 }
 
 // cinput represents a record producing input from a formatted object.
@@ -147,6 +153,9 @@ func (reader *cinput) readHeader() error {
 		reader.firstRow = nil
 	} else {
 		reader.firstRow, readErr = reader.reader.Read()
+		if readErr != nil {
+			return format.ErrCSVParsingError
+		}
 		reader.header = make([]string, len(reader.firstRow))
 		for i := range reader.firstRow {
 			reader.header[i] = "_" + strconv.Itoa(i)
@@ -173,8 +182,13 @@ func (reader *cinput) Read() ([]byte, error) {
 	if dec != nil {
 		var data []byte
 		var err error
-		for i, value := range dec {
-			data, err = sjson.SetBytes(data, reader.header[i], value)
+		// Navigate column values in reverse order to preserve
+		// the input order for AWS S3 compatibility, because
+		// sjson adds json key/value pairs in first in last out
+		// fashion. This should be fixed in sjson ideally. Following
+		// work around is needed to circumvent this issue for now.
+		for i := len(dec) - 1; i >= 0; i-- {
+			data, err = sjson.SetBytes(data, reader.header[i], dec[i])
 			if err != nil {
 				return nil, err
 			}
@@ -184,11 +198,16 @@ func (reader *cinput) Read() ([]byte, error) {
 	return nil, nil
 }
 
-// OutputFieldDelimiter - returns the delimiter specified in input request
+// OutputFieldDelimiter - returns the requested output field delimiter.
 func (reader *cinput) OutputFieldDelimiter() string {
 	return reader.options.OutputFieldDelimiter
 }
 
+// OutputRecordDelimiter - returns the requested output record delimiter.
+func (reader *cinput) OutputRecordDelimiter() string {
+	return reader.options.OutputFieldDelimiter
+}
+
 // HasHeader - returns true or false depending upon the header.
 func (reader *cinput) HasHeader() bool {
 	return reader.options.HasHeader
@@ -285,11 +304,16 @@ func (reader *cinput) CreateProgressXML() (string, error) {
 	return xml.Header + string(out), nil
 }
 
-// Type - return the data format type {
+// Type - return the data format type
 func (reader *cinput) Type() format.Type {
 	return format.CSV
 }
 
+// OutputType - return the data format type
+func (reader *cinput) OutputType() format.Type {
+	return reader.options.OutputType
+}
+
 // ColNameErrs is a function which makes sure that the headers are requested are
 // present in the file otherwise it throws an error.
 func (reader *cinput) ColNameErrs(columnNames []string) error {

pkg/s3select/format/json/json.go

@@ -22,6 +22,7 @@ import (
 	"io"
 
 	"github.com/minio/minio/pkg/s3select/format"
+	"github.com/tidwall/gjson"
 )
 
 // Options options are passed to the underlying encoding/json reader.
@@ -40,24 +41,32 @@ type Options struct {
 	// SQL expression meant to be evaluated.
 	Expression string
 
-	// What the outputted  will be delimited by .
+	// Input record delimiter.
 	RecordDelimiter string
 
+	// Output CSV will be delimited by.
+	OutputFieldDelimiter string
+
+	// Output record delimiter.
+	OutputRecordDelimiter string
+
 	// Size of incoming object
 	StreamSize int64
 
-	// True if Type is DOCUMENTS
-	Type bool
+	// True if DocumentType is DOCUMENTS
+	DocumentType bool
 
 	// Progress enabled, enable/disable progress messages.
 	Progress bool
+
+	// Output format type, supported values are CSV and JSON
+	OutputType format.Type
 }
 
 // jinput represents a record producing input from a  formatted file or pipe.
 type jinput struct {
 	options         *Options
 	reader          *bufio.Reader
-	firstRow        []string
 	header          []string
 	minOutputLength int
 	stats           struct {
@@ -79,7 +88,6 @@ func New(opts *Options) (format.Select, error) {
 	reader.stats.BytesScanned = opts.StreamSize
 	reader.stats.BytesProcessed = 0
 	reader.stats.BytesReturned = 0
-
 	return reader, nil
 }
 
@@ -95,7 +103,7 @@ func (reader *jinput) UpdateBytesProcessed(size int64) {
 
 // Read the file and returns
 func (reader *jinput) Read() ([]byte, error) {
-	data, err := reader.reader.ReadBytes('\n')
+	data, _, err := reader.reader.ReadLine()
 	if err != nil {
 		if err == io.EOF || err == io.ErrClosedPipe {
 			err = nil
@@ -103,17 +111,32 @@ func (reader *jinput) Read() ([]byte, error) {
 			err = format.ErrJSONParsingError
 		}
 	}
+	if err == nil {
+		var header []string
+		gjson.ParseBytes(data).ForEach(func(key, value gjson.Result) bool {
+			header = append(header, key.String())
+			return true
+		})
+		reader.header = header
+	}
 	return data, err
 }
 
-// OutputFieldDelimiter - returns the delimiter specified in input request
+// OutputFieldDelimiter - returns the delimiter specified in input request,
+// for JSON output this value is empty, but does have a value when
+// output type is CSV.
 func (reader *jinput) OutputFieldDelimiter() string {
-	return ","
+	return reader.options.OutputFieldDelimiter
+}
+
+// OutputRecordDelimiter - returns the delimiter specified in input request, after each JSON record.
+func (reader *jinput) OutputRecordDelimiter() string {
+	return reader.options.OutputRecordDelimiter
 }
 
 // HasHeader - returns true or false depending upon the header.
 func (reader *jinput) HasHeader() bool {
-	return false
+	return true
 }
 
 // Expression - return the Select Expression for
@@ -128,7 +151,7 @@ func (reader *jinput) UpdateBytesReturned(size int64) {
 
 // Header returns a nil in case of
 func (reader *jinput) Header() []string {
-	return nil
+	return reader.header
 }
 
 // CreateStatXML is the function which does the marshaling from the stat
@@ -171,6 +194,11 @@ func (reader *jinput) Type() format.Type {
 	return format.JSON
 }
 
+// OutputType - return the data format type {
+func (reader *jinput) OutputType() format.Type {
+	return reader.options.OutputType
+}
+
 // ColNameErrs - this is a dummy function for JSON input type.
 func (reader *jinput) ColNameErrs(columnNames []string) error {
 	return nil

pkg/s3select/format/select.go

@@ -22,10 +22,12 @@ import "encoding/xml"
 // https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectSELECTContent.html
 type Select interface {
 	Type() Type
+	OutputType() Type
 	Read() ([]byte, error)
 	Header() []string
 	HasHeader() bool
 	OutputFieldDelimiter() string
+	OutputRecordDelimiter() string
 	UpdateBytesProcessed(int64)
 	Expression() string
 	UpdateBytesReturned(int64)

pkg/s3select/input.go

@@ -65,40 +65,60 @@ func New(reader io.Reader, size int64, req ObjectSelectRequest) (s3s format.Sele
 
 	//  Initializating options for CSV
 	if req.InputSerialization.CSV != nil {
-		if req.OutputSerialization.CSV.FieldDelimiter == "" {
-			req.OutputSerialization.CSV.FieldDelimiter = ","
-		}
 		if req.InputSerialization.CSV.FileHeaderInfo == "" {
 			req.InputSerialization.CSV.FileHeaderInfo = CSVFileHeaderInfoNone
 		}
 		if req.InputSerialization.CSV.RecordDelimiter == "" {
 			req.InputSerialization.CSV.RecordDelimiter = "\n"
 		}
-		s3s, err = csv.New(&csv.Options{
-			HasHeader:            req.InputSerialization.CSV.FileHeaderInfo == CSVFileHeaderInfoUse,
-			RecordDelimiter:      req.InputSerialization.CSV.RecordDelimiter,
-			FieldDelimiter:       req.InputSerialization.CSV.FieldDelimiter,
-			Comments:             req.InputSerialization.CSV.Comments,
-			Name:                 "S3Object", // Default table name for all objects
-			ReadFrom:             reader,
-			Compressed:           string(req.InputSerialization.CompressionType),
-			Expression:           cleanExpr(req.Expression),
-			OutputFieldDelimiter: req.OutputSerialization.CSV.FieldDelimiter,
-			StreamSize:           size,
-			HeaderOpt:            req.InputSerialization.CSV.FileHeaderInfo == CSVFileHeaderInfoUse,
-			Progress:             req.RequestProgress.Enabled,
-		})
+		options := &csv.Options{
+			Name:            "S3Object", // Default table name for all objects
+			HasHeader:       req.InputSerialization.CSV.FileHeaderInfo == CSVFileHeaderInfoUse,
+			RecordDelimiter: req.InputSerialization.CSV.RecordDelimiter,
+			FieldDelimiter:  req.InputSerialization.CSV.FieldDelimiter,
+			Comments:        req.InputSerialization.CSV.Comments,
+			ReadFrom:        reader,
+			Compressed:      string(req.InputSerialization.CompressionType),
+			Expression:      cleanExpr(req.Expression),
+			StreamSize:      size,
+			HeaderOpt:       req.InputSerialization.CSV.FileHeaderInfo == CSVFileHeaderInfoUse,
+			Progress:        req.RequestProgress.Enabled,
+		}
+		if req.OutputSerialization.CSV != nil {
+			if req.OutputSerialization.CSV.FieldDelimiter == "" {
+				req.OutputSerialization.CSV.FieldDelimiter = ","
+			}
+			options.OutputFieldDelimiter = req.OutputSerialization.CSV.FieldDelimiter
+			options.OutputRecordDelimiter = req.OutputSerialization.CSV.RecordDelimiter
+			options.OutputType = format.CSV
+		}
+		if req.OutputSerialization.JSON != nil {
+			options.OutputRecordDelimiter = req.OutputSerialization.JSON.RecordDelimiter
+			options.OutputType = format.JSON
+		}
+		// Initialize CSV input type
+		s3s, err = csv.New(options)
 	} else if req.InputSerialization.JSON != nil {
-		//  Initializating options for JSON
-		s3s, err = json.New(&json.Options{
-			Name:       "S3Object", // Default table name for all objects
-			ReadFrom:   reader,
-			Compressed: string(req.InputSerialization.CompressionType),
-			Expression: cleanExpr(req.Expression),
-			StreamSize: size,
-			Type:       req.InputSerialization.JSON.Type == JSONTypeDocument,
-			Progress:   req.RequestProgress.Enabled,
-		})
+		options := &json.Options{
+			Name:         "S3Object", // Default table name for all objects
+			ReadFrom:     reader,
+			Compressed:   string(req.InputSerialization.CompressionType),
+			Expression:   cleanExpr(req.Expression),
+			StreamSize:   size,
+			DocumentType: req.InputSerialization.JSON.Type == JSONTypeDocument,
+			Progress:     req.RequestProgress.Enabled,
+		}
+		if req.OutputSerialization.JSON != nil {
+			options.OutputRecordDelimiter = req.OutputSerialization.JSON.RecordDelimiter
+			options.OutputType = format.JSON
+		}
+		if req.OutputSerialization.CSV != nil {
+			options.OutputFieldDelimiter = req.OutputSerialization.CSV.FieldDelimiter
+			options.OutputRecordDelimiter = req.OutputSerialization.CSV.RecordDelimiter
+			options.OutputType = format.CSV
+		}
+		// Initialize JSON input type
+		s3s, err = json.New(options)
 	}
 	return s3s, err
 }

pkg/s3select/select.go

@@ -201,14 +201,19 @@ func processSelectReq(reqColNames []string, alias string, wc sqlparser.Expr, lre
 		lrecords = math.MaxInt64
 	}
 
-	columnsKv, err := columnsIndex(reqColNames, f)
-	if err != nil {
-		rowCh <- Row{
-			err: err,
+	var results []string
+	var columnsKv []columnKv
+	if f.Type() == format.CSV {
+		var err error
+		columnsKv, err = columnsIndex(reqColNames, f)
+		if err != nil {
+			rowCh <- Row{
+				err: err,
+			}
+			return
 		}
-		return
+		results = make([]string, len(columnsKv))
 	}
-	var results = make([]string, len(columnsKv))
 
 	for {
 		record, err := f.Read()
@@ -228,6 +233,19 @@ func processSelectReq(reqColNames []string, alias string, wc sqlparser.Expr, lre
 			return
 		}
 
+		// For JSON multi-line input type columns needs
+		// to be handled for each record.
+		if f.Type() == format.JSON {
+			columnsKv, err = columnsIndex(reqColNames, f)
+			if err != nil {
+				rowCh <- Row{
+					err: err,
+				}
+				return
+			}
+			results = make([]string, len(columnsKv))
+		}
+
 		f.UpdateBytesProcessed(int64(len(record)))
 
 		// Return in case the number of record reaches the LIMIT
@@ -250,17 +268,17 @@ func processSelectReq(reqColNames []string, alias string, wc sqlparser.Expr, lre
 		if condition {
 			// if its an asterix we just print everything in the row
 			if reqColNames[0] == "*" && fnNames[0] == "" {
-				switch f.Type() {
+				switch f.OutputType() {
 				case format.CSV:
 					for i, kv := range columnsKv {
 						results[i] = gjson.GetBytes(record, kv.Key).String()
 					}
 					rowCh <- Row{
-						record: strings.Join(results, f.OutputFieldDelimiter()) + "\n",
+						record: strings.Join(results, f.OutputFieldDelimiter()) + f.OutputRecordDelimiter(),
 					}
 				case format.JSON:
 					rowCh <- Row{
-						record: string(record) + "\n",
+						record: string(record) + f.OutputRecordDelimiter(),
 					}
 				}
 			} else if alias != "" {