func NewAwsConfig(
	accessKey string,
	secretKey string,
	regionName string,
	endpoint string,
	disableSSL bool,
) *aws.Config {
	var creds *credentials.Credentials

	if accessKey == "" && secretKey == "" {
		creds = credentials.AnonymousCredentials
	} else {
		creds = credentials.NewStaticCredentials(accessKey, secretKey, "")
	}

	if len(regionName) == 0 {
		regionName = "us-east-1"
	}

	awsConfig := &aws.Config{
		Region:           aws.String(regionName),
		Credentials:      creds,
		S3ForcePathStyle: aws.Bool(true),
		MaxRetries:       aws.Int(maxRetries),
		DisableSSL:       aws.Bool(disableSSL),
	}

	if len(endpoint) != 0 {
		endpoint := fmt.Sprintf("%s", endpoint)
		awsConfig.Endpoint = &endpoint
	}

	return awsConfig
}

// NewDownloader inits and returns a Downloader pointer
func NewDownloader(agrs *cfg.InArgs, conf *cfg.Cfg) (*Downloader, error) {
	if agrs == nil || conf == nil {
		return nil, ErrInvalidArgs
	}
	creds := credentials.NewStaticCredentials(conf.AWSAccessKeyID, conf.AWSSecretKey, "")
	if _, err := creds.Get(); err != nil {
		return nil, err
	}

	awsConf := &aws.Config{Credentials: creds, Region: aws.String(conf.Region)}
	sess := session.New(awsConf)
	client := s3.New(sess, awsConf)

	manager := NewS3DownloadManager(sess)

	d := &Downloader{
		downloadManager: manager,
		args:            agrs,
		pageLister:      client,
		regexp:          regexp.MustCompile(agrs.Regexp),
		workers:         make(chan int, 50),
		fileCreator:     &fsAdapter{},
	}

	d.pageIterator = d.pickPageIterator()
	return d, nil
}

func (p *EBSPlugin) prepare() error {
	if p.AccessKeyID != "" && p.SecretAccessKey != "" {
		p.Credentials = credentials.NewStaticCredentials(p.AccessKeyID, p.SecretAccessKey, "")
	}

	p.EC2 = ec2.New(session.New(&aws.Config{Credentials: p.Credentials, Region: &p.Region}))
	resp, err := p.EC2.DescribeVolumes(&ec2.DescribeVolumesInput{
		Filters: []*ec2.Filter{
			{
				Name: aws.String("attachment.instance-id"),
				Values: []*string{
					&p.InstanceID,
				},
			},
		},
	})
	if err != nil {
		return err
	}
	if resp.NextToken != nil {
		return errors.New("DescribeVolumes response has NextToken")
	}

	p.Volumes = resp.Volumes
	if len(p.Volumes) == 0 {
		return errors.New("DescribeVolumes response has no volumes")
	}

	return nil
}

// Return a new CLIConfig with stored profile settings
func NewFromProfile(name string) (*CLIConfig, error) {
	c := &CLIConfig{}
	c.AssumeRoleInput = new(sts.AssumeRoleInput)
	err := c.Prepare(name)
	if err != nil {
		return nil, err
	}
	sessName, err := c.getSessionName(c.profileCfg.Key("role_session_name").Value())
	if err != nil {
		return nil, err
	}
	c.AssumeRoleInput.RoleSessionName = aws.String(sessName)
	arn := c.profileCfg.Key("role_arn").Value()
	if arn != "" {
		c.AssumeRoleInput.RoleArn = aws.String(arn)
	}
	id := c.profileCfg.Key("external_id").Value()
	if id != "" {
		c.AssumeRoleInput.ExternalId = aws.String(id)
	}
	c.SourceCredentials = credentials.NewStaticCredentials(
		c.profileCred.Key("aws_access_key_id").Value(),
		c.profileCred.Key("aws_secret_access_key").Value(),
		c.profileCred.Key("aws_session_token").Value(),
	)
	return c, nil
}

func listS3(wg *sync.WaitGroup, index int, results chan<- []string) {

	var creds *credentials.Credentials
	if *accessFlag != "" && *secretFlag != "" {
		creds = credentials.NewStaticCredentials(*accessFlag, *secretFlag, "")
	} else {
		creds = credentials.AnonymousCredentials
	}
	sess := session.New(aws.NewConfig().WithCredentials(creds).WithRegion(*regionFlag).WithEndpoint(*endpointFlag).WithS3ForcePathStyle(true))

	prefix := fmt.Sprintf("%s%x", *prefixFlag, index)

	svc := s3.New(sess)
	inputparams := &s3.ListObjectsInput{
		Bucket: aws.String(*bucketFlag),
		Prefix: aws.String(prefix),
	}

	result := make([]string, 0, 1000)

	svc.ListObjectsPages(inputparams, func(page *s3.ListObjectsOutput, lastPage bool) bool {

		for _, value := range page.Contents {
			result = append(result, *value.Key)
		}

		if lastPage {
			results <- result
			wg.Done()
			return false
		} else {
			return true
		}
	})
}

func TestSignWithRequestBody_Overwrite(t *testing.T) {
	creds := credentials.NewStaticCredentials("AKID", "SECRET", "SESSION")
	signer := NewSigner(creds)

	var expectBody []byte

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		b, err := ioutil.ReadAll(r.Body)
		r.Body.Close()
		assert.NoError(t, err)
		assert.Equal(t, len(expectBody), len(b))
		w.WriteHeader(http.StatusOK)
	}))

	req, err := http.NewRequest("GET", server.URL, strings.NewReader("invalid body"))

	_, err = signer.Sign(req, nil, "service", "region", time.Now())
	req.ContentLength = 0

	assert.NoError(t, err)

	resp, err := http.DefaultClient.Do(req)
	assert.NoError(t, err)
	assert.Equal(t, http.StatusOK, resp.StatusCode)
}

// MustClient will use the cache cluster ID to describe
// the cache cluster and instantiate a memcache.Client
// with the cache nodes returned from AWS.
func (e *ElastiCache) MustClient() *memcache.Client {
	var creds *credentials.Credentials
	if e.AccessKey != "" {
		creds = credentials.NewStaticCredentials(e.AccessKey, e.SecretKey, "")
	} else {
		creds = credentials.NewEnvCredentials()
	}

	ecclient := elasticache.New(session.New(&aws.Config{
		Credentials: creds,
		Region:      &e.Region,
	}))

	resp, err := ecclient.DescribeCacheClusters(&elasticache.DescribeCacheClustersInput{
		CacheClusterId:    &e.ClusterID,
		ShowCacheNodeInfo: aws.Bool(true),
	})
	if err != nil {
		log.Fatalf("unable to describe cache cluster: %s", err)
	}

	var nodes []string
	for _, cluster := range resp.CacheClusters {
		for _, cnode := range cluster.CacheNodes {
			addr := fmt.Sprintf("%s:%d", *cnode.Endpoint.Address, *cnode.Endpoint.Port)
			nodes = append(nodes, addr)
		}
	}

	return memcache.New(nodes...)
}

// GetSessions returns the current session from the SessionProvider.
func (dsp *ConfigurableSessionProvider) GetSession() *session.Session {
	dsp.sessionMutex.Lock()
	defer dsp.sessionMutex.Unlock()

	if dsp._session != nil {
		return dsp._session
	}

	cfgs := []*aws.Config{
		aws.NewConfig().WithRegion(dsp.ConfigRegion),
	}

	if dsp.ConfigAccessKey != "" {
		cfgs = append(cfgs, aws.NewConfig().WithCredentials(credentials.NewStaticCredentials(
			dsp.ConfigAccessKey,
			dsp.ConfigSecretKey,
			dsp.ConfigSessionToken,
		)))
	} else if !dsp.ConfigDisableENVCredentials {
		// NOTE: We may want to init all credential providers in one config, as they might overwrite each other
		cfgs = append(cfgs, aws.NewConfig().WithCredentials(credentials.NewCredentials(&credentials.EnvProvider{})))
	} else {
		panic("no valid configuration parameters for aws credentials found.")
	}

	dsp._session = session.New(cfgs...)
	return dsp._session
}

func NewS3Client(accessKey string, secretKey string, regionName string, endpoint string) (S3Client, error) {
	var creds *credentials.Credentials

	if accessKey == "" && secretKey == "" {
		creds = credentials.AnonymousCredentials
	} else {
		creds = credentials.NewStaticCredentials(accessKey, secretKey, "")
	}

	if len(regionName) == 0 {
		regionName = "us-east-1"
	}

	awsConfig := &aws.Config{
		Region:      &regionName,
		Credentials: creds,
	}

	if len(endpoint) != 0 {
		endpoint := fmt.Sprintf("https://%s", endpoint)
		awsConfig.Endpoint = &endpoint
	}

	client := s3.New(awsConfig)

	return &s3client{
		client: client,
	}, nil
}

func (s *GoofysTest) SetUpSuite(t *C) {
	//addr := "play.minio.io:9000"
	const LOCAL_TEST = true

	if LOCAL_TEST {
		addr := "127.0.0.1:8080"

		err := s.waitFor(t, addr)
		t.Assert(err, IsNil)

		s.awsConfig = &aws.Config{
			//Credentials: credentials.AnonymousCredentials,
			Credentials:      credentials.NewStaticCredentials("foo", "bar", ""),
			Region:           aws.String("us-west-2"),
			Endpoint:         aws.String(addr),
			DisableSSL:       aws.Bool(true),
			S3ForcePathStyle: aws.Bool(true),
			MaxRetries:       aws.Int(0),
			//Logger: t,
			//LogLevel: aws.LogLevel(aws.LogDebug),
			//LogLevel: aws.LogLevel(aws.LogDebug | aws.LogDebugWithHTTPBody),
		}
	} else {
		s.awsConfig = &aws.Config{
			Region:     aws.String("us-west-2"),
			DisableSSL: aws.Bool(true),
		}
	}
	s.s3 = s3.New(s.awsConfig)

	_, err := s.s3.ListBuckets(nil)
	t.Assert(err, IsNil)
}

// NewSNSPublisher will initiate the SNS client.
// If no credentials are passed in with the config,
// the publisher is instantiated with the AWS_ACCESS_KEY
// and the AWS_SECRET_KEY environment variables.
func NewSNSPublisher(cfg *config.SNS) (*SNSPublisher, error) {
	p := &SNSPublisher{}

	if cfg.Topic == "" {
		return p, errors.New("SNS topic name is required")
	}
	p.topic = cfg.Topic

	if cfg.Region == "" {
		return p, errors.New("SNS region is required")
	}

	var creds *credentials.Credentials
	if cfg.AccessKey != "" {
		creds = credentials.NewStaticCredentials(cfg.AccessKey, cfg.SecretKey, "")
	} else {
		creds = credentials.NewEnvCredentials()
	}

	p.sns = sns.New(session.New(&aws.Config{
		Credentials: creds,
		Region:      &cfg.Region,
	}))
	return p, nil
}

func New(conf *AwsConfig) (*AwsImages, error) {
	checkCfg := "Please check your configuration"

	if len(conf.Regions) == 0 {
		return nil, errors.New("AWS Regions are not set. " + checkCfg)
	}

	if conf.AccessKey == "" {
		return nil, errors.New("AWS Access Key is not set. " + checkCfg)
	}

	if conf.SecretKey == "" {
		return nil, errors.New("AWS Secret Key is not set. " + checkCfg)
	}

	// increase the timeout
	timeout := time.Second * 30
	client := &http.Client{
		Transport: &http.Transport{TLSHandshakeTimeout: timeout},
		Timeout:   timeout,
	}

	creds := credentials.NewStaticCredentials(conf.AccessKey, conf.SecretKey, "")
	awsCfg := &awsclient.Config{
		Credentials: creds,
		HTTPClient:  client,
		Logger:      awsclient.NewDefaultLogger(),
	}

	m := newMultiRegion(awsCfg, filterRegions(conf.Regions, conf.RegionsExclude))
	return &AwsImages{
		services: m,
		images:   make(map[string][]*ec2.Image),
	}, nil
}

func New(config Config) (*Client, error) {
	credentials := credentials.NewStaticCredentials(config.AccessKey, config.SecretKey, "")
	sdkConfig := &aws.Config{
		Credentials: credentials,
		Region:      aws.String(config.Region),
	}

	session := session.New(sdkConfig)

	if config.CloudFormationWaitTimeout == 0 {
		return nil, fmt.Errorf("AWS config CloudFormationWaitTimeout must be a positive timeout")
	}

	ec2EndpointConfig, err := config.getEndpoint("ec2")
	if err != nil {
		return nil, err
	}
	cloudformationEndpointConfig, err := config.getEndpoint("cloudformation")
	if err != nil {
		return nil, err
	}
	iamEndpointConfig, err := config.getEndpoint("iam")
	if err != nil {
		return nil, err
	}

	return &Client{
		EC2:            ec2.New(session, ec2EndpointConfig),
		CloudFormation: cloudformation.New(session, cloudformationEndpointConfig),
		IAM:            iam.New(session, iamEndpointConfig),
		Clock:          clockImpl{},
		CloudFormationWaitTimeout: config.CloudFormationWaitTimeout,
	}, nil
}

func main() {
	if accessKey == "" {
		die("AWS_ACCESS_KEY is not set")
	}
	if secretKey == "" {
		die("AWS_SECRET_KEY is not set")
	}
	if hostedZone == "" {
		die("ROUTE53_HOSTED_ZONE is not set")
	}

	opts := &dnsclient.Options{
		Creds:       credentials.NewStaticCredentials(accessKey, secretKey, ""),
		HostedZone:  hostedZone,
		Log:         logging.NewCustom("dnsclient", os.Getenv("ROUTE53_DEBUG") == "1"),
		SyncTimeout: 5 * time.Minute,
	}

	if d, err := time.ParseDuration(os.Getenv("ROUTE53_TIMEOUT")); err == nil {
		opts.SyncTimeout = d
	}

	opts.Log.Debug("Options: %# v", opts)

	var err error
	client, err = dnsclient.NewRoute53Client(opts)
	if err != nil {
		die(err)
	}

	if err := Resources.Main(os.Args[1:]); err != nil {
		die(err)
	}
}

func (sb signerBuilder) BuildSigner() signer {
	endpoint := "https://" + sb.ServiceName + "." + sb.Region + ".amazonaws.com"
	var req *http.Request
	if sb.Method == "POST" {
		body := []byte(sb.Query.Encode())
		reader := bytes.NewReader(body)
		req, _ = http.NewRequest(sb.Method, endpoint, reader)
		req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
		req.Header.Add("Content-Length", string(len(body)))
	} else {
		req, _ = http.NewRequest(sb.Method, endpoint, nil)
		req.URL.RawQuery = sb.Query.Encode()
	}

	sig := signer{
		Request: req,
		Time:    sb.SignTime,
		Credentials: credentials.NewStaticCredentials(
			"AKID",
			"SECRET",
			sb.SessionToken),
	}

	if os.Getenv("DEBUG") != "" {
		sig.Debug = aws.LogDebug
		sig.Logger = aws.NewDefaultLogger()
	}

	return sig
}

func (h *awsAuthHeaderer) SignRequest(r *http.Request) *http.Request {
	region := h.region

	var body io.ReadSeeker
	if r.Body != nil {
		body = r.Body.(io.ReadSeeker)
	}

	if len(region) == 0 {
		region = guessAWSRegion(r.URL.Host)
	}
	v4.Sign(&request.Request{
		ClientInfo: metadata.ClientInfo{
			SigningRegion: region,
			SigningName:   awsS3Service,
		},
		Config: aws.Config{
			Credentials: credentials.NewStaticCredentials(h.accessKeyID, h.secretAccessKey, ""),
		},
		HTTPRequest: r,
		Body:        body,
		Time:        time.Now(),
	})

	return r
}

func TestIgnorePreResignRequestWithValidCreds(t *testing.T) {
	svc := awstesting.NewClient(&aws.Config{
		Credentials: credentials.NewStaticCredentials("AKID", "SECRET", "SESSION"),
		Region:      aws.String("us-west-2"),
	})
	r := svc.NewRequest(
		&request.Operation{
			Name:       "BatchGetItem",
			HTTPMethod: "POST",
			HTTPPath:   "/",
		},
		nil,
		nil,
	)
	r.ExpireTime = time.Minute * 10

	SignSDKRequest(r)
	sig := r.HTTPRequest.URL.Query().Get("X-Amz-Signature")

	signSDKRequestWithCurrTime(r, func() time.Time {
		// Simulate one second has passed so that signature's date changes
		// when it is resigned.
		return time.Now().Add(1 * time.Second)
	})
	assert.NotEqual(t, sig, r.HTTPRequest.URL.Query().Get("X-Amz-Signature"))
}

// New initializes a new S3 client connection based on config.
func New() *S3Client {
	var (
		cfg *aws.Config
	)

	if config.S3.Endpoint != "" {
		cfg = &aws.Config{
			Endpoint:         aws.String(config.S3.Endpoint),
			DisableSSL:       aws.Bool(strings.HasPrefix(config.S3.Endpoint, "http://")),
			Region:           aws.String(config.S3.Region),
			S3ForcePathStyle: aws.Bool(config.S3.PathStyle),
		}
	} else {
		cfg = &aws.Config{
			Region:           aws.String(config.S3.Region),
			S3ForcePathStyle: aws.Bool(config.S3.PathStyle),
		}
	}

	if config.S3.Access != "" && config.S3.Secret != "" {
		cfg.Credentials = credentials.NewStaticCredentials(
			config.S3.Access,
			config.S3.Secret,
			"",
		)
	}

	return &S3Client{
		client: s3.New(
			session.New(),
			cfg,
		),
	}
}

func Credentials(req *Request) *credentials.Credentials {
	if req != nil {
		if access, ok := req.ResourceProperties["AccessId"].(string); ok && access != "" {
			if secret, ok := req.ResourceProperties["SecretAccessKey"].(string); ok && secret != "" {
				return credentials.NewStaticCredentials(access, secret, "")
			}
		}
	}

	if os.Getenv("AWS_ACCESS") != "" {
		return credentials.NewStaticCredentials(os.Getenv("AWS_ACCESS"), os.Getenv("AWS_SECRET"), "")
	}

	// return credentials.NewCredentials(&credentials.EC2RoleProvider{})
	return credentials.NewEnvCredentials()
}

func initAWS() {
	defer Track("initAWS", Now(), debugOut)

	AWSSession = session.New()

	// Region
	if GlobalConfig.Get("awsRegion") != "" {
		// CLI trumps
		AWSSession.Config.Region = aws.String(GlobalConfig.Get("awsRegion"))
	} else if os.Getenv("AWS_REGION") == "" {
		// Grab it from this EC2 instace
		region, err := ec2metadata.New(session.New()).Region()
		if err != nil {
			fmt.Printf("Cannot set AWS region: '%v'\n", err)
			os.Exit(1)
		}
		AWSSession.Config.Region = aws.String(region)
	}

	// Creds
	if GlobalConfig.Get("awsAccessKey") != "" && GlobalConfig.Get("awsSecretKey") != "" {
		// CLI trumps
		creds := credentials.NewStaticCredentials(
			GlobalConfig.Get("awsAccessKey"),
			GlobalConfig.Get("awsSecretKey"),
			"")
		AWSSession.Config.Credentials = creds
	}

}