Vault Auth

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')))

The hvac docs have some good examples; I'd suggest checking them out: https://hvac.readthedocs.io/en/stable/usage/auth_methods/aws.html#iam-authentication

"request_dict = client._convert_to_request_dict({}, operation_model)"
The above line throws an exception: ._convert_to_request_dict() missing 1 required positional argument: 'endpoint_url'.
Please help.

@joelthompson
I am trying to use the headers generated by the python3 script in an AWS lambda function but I always get a 400 response. Not sure whats wrong here. Would appreciate any help if someone has achieved this already.

request_body = json.dumps(generate_vault_request())
awsIamServerId = "<vault-addr>/v1/auth/aws/login"
resp = requests.post(url=awsIamServerId, data=request_body)

@srpraharaj did you figure this issue out? Having the same problem. When I add a '' for endpoint it does pass to create the request but then I get this error:

213 | File "/codebuild/output/src965627234/src/provision/secrets/login.py", line 42, in login
214 | sts_request = endpoint.create_request(request_dict, operation_model)
215 | File "/root/.local/lib/python3.10/site-packages/botocore/endpoint.py", line 134, in create_request
216 | self._event_emitter.emit(
217 | File "/root/.local/lib/python3.10/site-packages/botocore/hooks.py", line 412, in emit
218 | return self._emitter.emit(aliased_event_name, **kwargs)
219 | File "/root/.local/lib/python3.10/site-packages/botocore/hooks.py", line 256, in emit
220 | return self._emit(event_name, kwargs)
221 | File "/root/.local/lib/python3.10/site-packages/botocore/hooks.py", line 239, in _emit
222 | response = handler(**kwargs)
223 | File "/root/.local/lib/python3.10/site-packages/botocore/signers.py", line 105, in handler
224 | return self.sign(operation_name, request)
225 | File "/root/.local/lib/python3.10/site-packages/botocore/signers.py", line 189, in sign
226 | auth.add_auth(request)
227 | File "/root/.local/lib/python3.10/site-packages/botocore/auth.py", line 424, in add_auth
228 | canonical_request = self.canonical_request(request)
229 | File "/root/.local/lib/python3.10/site-packages/botocore/auth.py", line 365, in canonical_request
230 | cr.append(self.canonical_headers(headers_to_sign) + '\n')
231 | File "/root/.local/lib/python3.10/site-packages/botocore/auth.py", line 300, in canonical_headers
232 | value = ','.join(
233 | File "/root/.local/lib/python3.10/site-packages/botocore/auth.py", line 301, in <genexpr>
234 | self._header_value(v) for v in headers_to_sign.get_all(key)
235 | File "/root/.local/lib/python3.10/site-packages/botocore/auth.py", line 312, in _header_value
236 | return ' '.join(value.split())
237 | AttributeError: 'NoneType' object has no attribute 'split'

For future reference, the python3 version needs an update to add the endpoint_url.
For STS this would be 'https://sts.amazonaws.com'

request_dict = client._convert_to_request_dict({}, operation_model, endpoint_url)