joelthompson/README.md

## README.md

      
    Raw
  

              README.md
            
          
    vault_aws_auth

These are python 2 and 3 snippets showing how to generate headers to authenticate with HashiCorp's Vault using the
AWS authentication method. There's also a Ruby implementation which uses version 3 of the AWS SDK for Ruby.
The python scripts look for credentials in the
default boto3 locations;
if you need to supply custom credentials (such as from an AssumeRole call), you would use the
botocore.session.set_credentials
method before calling create_client.
The ruby script looks for credentials from the default SDK locations.
Credits

Thanks to @copumpkin for much of the original python 2 implementation
(provided privately) on which this was based.
Thanks to @stark525 for starting the python 3 port, on which the python 3 implementation is based.

  
## vault_aws_auth.py
#!/usr/bin/env python

import botocore.session
from botocore.awsrequest import create_request_object
import json
import base64

def headers_to_go_style(headers):
    retval = {}
    for k, v in headers.iteritems():
        retval[k] = [v]
    return retval

def generate_vault_request(role_name=""):
    session = botocore.session.get_session()
    # if you have credentials from non-default sources, call
    # session.set_credentials here, before calling session.create_client
    client = session.create_client('sts')
    endpoint = client._endpoint
    operation_model = client._service_model.operation_model('GetCallerIdentity')
    request_dict = client._convert_to_request_dict({}, operation_model)

    awsIamServerId = 'vault.example.com'
    request_dict['headers']['X-Vault-AWS-IAM-Server-ID'] = awsIamServerId

    request = endpoint.create_request(request_dict, operation_model)

    # It's now signed...
    return {
        'iam_http_request_method': request.method,
        'iam_request_url':         base64.b64encode(request.url),
        'iam_request_body':        base64.b64encode(request.body),
        'iam_request_headers':     base64.b64encode(json.dumps(headers_to_go_style(dict(request.headers)))), # It's a CaseInsensitiveDict, which is not JSON-serializable
        'role':                    role_name,
    }

if __name__ == "__main__":
    print json.dumps(generate_vault_request('TestRole'))

## vault_aws_auth.rb
#!/usr/bin/env ruby

require 'base64'
require 'json'

# Tested against v3
require 'aws-sdk'

class VaultHandler < Seahorse::Client::NetHttp::Handler
  def call(context)
    req = context.http_request
    auth_data = {
      "iam_http_request_method" => req.http_method,
      "iam_request_url" => Base64.strict_encode64(req.endpoint.request_uri),
      "iam_request_body" => Base64.strict_encode64(req.body.read),
      "iam_request_headers" => Base64.strict_encode64(JSON.generate(headers(req)))
    }
    resp = Seahorse::Client::Response.new(context: context, data: auth_data)
    resp
  end
end

class VaultPlugin < Seahorse::Client::Plugin
  handler(VaultHandler, step: :send)
end

# We create our own custom sub-class of Aws::STS::Client so that other calls to STS
# will work as normal (i.e., we don't want to intercept ALL STS requests, JUST this
# request to generate Vault login data)
class VaultStsClient < Aws::STS::Client
  set_api(Aws::STS::ClientApi::API)
end

VaultStsClient.add_plugin(VaultPlugin)
sts = VaultStsClient.new()
req = sts.build_request('get_caller_identity')
req.context.http_request.headers['X-Vault-Aws-Iam-Server-Id'] = 'vault.example.com'
auth_data = req.send_request().data
auth_data.each do |key, val|
  puts "#{key}=#{val}"
end

## vault_aws_auth_py3.py
#!/usr/bin/env python3

import boto3
import json
import base64

def headers_to_go_style(headers):
    retval = {}
    for k, v in headers.items():
        if isinstance(v, bytes):
            retval[k] = [str(v, 'ascii')]
        else:
            retval[k] = [v]
    return retval

def generate_vault_request(role_name=""):
    session = boto3.session.Session()
    # if you have credentials from non-default sources, call
    # session.set_credentials here, before calling session.create_client
    client = session.client('sts')
    endpoint = client._endpoint
    operation_model = client._service_model.operation_model('GetCallerIdentity')
    request_dict = client._convert_to_request_dict({}, operation_model)

    awsIamServerId = 'vault.example.com'
    request_dict['headers']['X-Vault-AWS-IAM-Server-ID'] = awsIamServerId

    request = endpoint.create_request(request_dict, operation_model)
    # It's now signed...
    return {
        'iam_http_request_method': request.method,
        'iam_request_url': str(base64.b64encode(request.url.encode('ascii')), 'ascii'),
        'iam_request_body': str(base64.b64encode(request.body.encode('ascii')), 'ascii'),
        'iam_request_headers': str(base64.b64encode(bytes(json.dumps(headers_to_go_style(dict(request.headers))), 'ascii')), 'ascii'), # It's a CaseInsensitiveDict, which is not JSON-serializable
        'role': role_name,
    }

if __name__ == "__main__":
    print(json.dumps(generate_vault_request('TestRole')))
	#!/usr/bin/env python

	import botocore.session
	from botocore.awsrequest import create_request_object
	import json
	import base64

	def headers_to_go_style(headers):
	retval = {}
	for k, v in headers.iteritems():
	retval[k] = [v]
	return retval

	def generate_vault_request(role_name=""):
	session = botocore.session.get_session()
	# if you have credentials from non-default sources, call
	# session.set_credentials here, before calling session.create_client
	client = session.create_client('sts')
	endpoint = client._endpoint
	operation_model = client._service_model.operation_model('GetCallerIdentity')
	request_dict = client._convert_to_request_dict({}, operation_model)

	awsIamServerId = 'vault.example.com'
	request_dict['headers']['X-Vault-AWS-IAM-Server-ID'] = awsIamServerId

	request = endpoint.create_request(request_dict, operation_model)

	# It's now signed...
	return {
	'iam_http_request_method': request.method,
	'iam_request_url': base64.b64encode(request.url),
	'iam_request_body': base64.b64encode(request.body),
	'iam_request_headers': base64.b64encode(json.dumps(headers_to_go_style(dict(request.headers)))), # It's a CaseInsensitiveDict, which is not JSON-serializable
	'role': role_name,
	}

	if __name__ == "__main__":
	print json.dumps(generate_vault_request('TestRole'))
	#!/usr/bin/env ruby

	require 'base64'
	require 'json'

	# Tested against v3
	require 'aws-sdk'

	class VaultHandler < Seahorse::Client::NetHttp::Handler
	def call(context)
	req = context.http_request
	auth_data = {
	"iam_http_request_method" => req.http_method,
	"iam_request_url" => Base64.strict_encode64(req.endpoint.request_uri),
	"iam_request_body" => Base64.strict_encode64(req.body.read),
	"iam_request_headers" => Base64.strict_encode64(JSON.generate(headers(req)))
	}
	resp = Seahorse::Client::Response.new(context: context, data: auth_data)
	resp
	end
	end

	class VaultPlugin < Seahorse::Client::Plugin
	handler(VaultHandler, step: :send)
	end

	# We create our own custom sub-class of Aws::STS::Client so that other calls to STS
	# will work as normal (i.e., we don't want to intercept ALL STS requests, JUST this
	# request to generate Vault login data)
	class VaultStsClient < Aws::STS::Client
	set_api(Aws::STS::ClientApi::API)
	end

	VaultStsClient.add_plugin(VaultPlugin)
	sts = VaultStsClient.new()
	req = sts.build_request('get_caller_identity')
	req.context.http_request.headers['X-Vault-Aws-Iam-Server-Id'] = 'vault.example.com'
	auth_data = req.send_request().data
	auth_data.each do \|key, val\|
	puts "#{key}=#{val}"
	end
	#!/usr/bin/env python3

	import boto3
	import json
	import base64

	def headers_to_go_style(headers):
	retval = {}
	for k, v in headers.items():
	if isinstance(v, bytes):
	retval[k] = [str(v, 'ascii')]
	else:
	retval[k] = [v]
	return retval

	def generate_vault_request(role_name=""):
	session = boto3.session.Session()
	# if you have credentials from non-default sources, call
	# session.set_credentials here, before calling session.create_client
	client = session.client('sts')
	endpoint = client._endpoint
	operation_model = client._service_model.operation_model('GetCallerIdentity')
	request_dict = client._convert_to_request_dict({}, operation_model)

	awsIamServerId = 'vault.example.com'
	request_dict['headers']['X-Vault-AWS-IAM-Server-ID'] = awsIamServerId

	request = endpoint.create_request(request_dict, operation_model)
	# It's now signed...
	return {
	'iam_http_request_method': request.method,
	'iam_request_url': str(base64.b64encode(request.url.encode('ascii')), 'ascii'),
	'iam_request_body': str(base64.b64encode(request.body.encode('ascii')), 'ascii'),
	'iam_request_headers': str(base64.b64encode(bytes(json.dumps(headers_to_go_style(dict(request.headers))), 'ascii')), 'ascii'), # It's a CaseInsensitiveDict, which is not JSON-serializable
	'role': role_name,
	}

	if __name__ == "__main__":
	print(json.dumps(generate_vault_request('TestRole')))