Send request with SigV4 in python using boto3

sigv4_using_http_client.py

from boto3.session import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
from http.client import HTTPConnection, HTTPSConnection
import json
import os
from urllib.parse import urlparse

def sigv4_request(
  url,
  method='GET',
  body=None,
  params=None,
  headers=None,
  service='execute-api',
  region=os.environ['AWS_REGION'],
  credentials=Session().get_credentials().get_frozen_credentials()
):
  """Sends an HTTP request signed with SigV4

  Args:
    url: The request URL (e.g. 'https://www.example.com').
    method: The request method (e.g. 'GET', 'POST', 'PUT', 'DELETE'). Defaults to 'GET'.
    body: The request body (e.g. json.dumps({ 'foo': 'bar' })). Defaults to None.
    params: The request query params (e.g. { 'foo': 'bar' }). Defaults to None.
    headers: The request headers (e.g. { 'content-type': 'application/json' }). Defaults to None.
    service: The AWS service name. Defaults to 'execute-api'.
    region: The AWS region id. Defaults to the env var 'AWS_REGION'.
    credentials: The AWS credentials. Defaults to the current boto3 session's credentials.
  Returns:
     The HTTP response
  """

  # sign request
  req = AWSRequest(
    method=method,
    url=url,
    data=body,
    params=params,
    headers=headers
  )
  SigV4Auth(credentials, service, region).add_auth(req)
  req = req.prepare()
    
  # parse URL
  u = urlparse(req.url)
  path_and_query = u.path if u.query is None else u.path + '?' + u.query

  # send request
  cnn = HTTPSConnection(u.hostname) if u.scheme == 'https' else HTTPConnection(u.hostname)
  cnn.request(
    req.method,
    path_and_query,
    headers=req.headers,
    body=req.body
  )
  return cnn.getresponse()

def __init__():
  # send request
  response = sigv4_request(
    'https://www.example.com',
    method='POST',
    headers={
      'content-type': 'application/json',
    },
    body=json.dumps({
      'foo': 'bar'
    }),
    region='us-west-2',
    credentials=Credentials(
      'ACCESS_KEY_ID',
      'SECRET_ACCESS_KEY',
      'SESSION_TOKEN'
    )
  )
  
  # do something with response
  print(response.status)
  print(json.loads(response.read().decode()))

sigv4_using_requests.py

from boto3.session import Session
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
import json
import os
from requests import request

def sigv4_request(
  url,
  method='GET',
  body=None,
  params=None,
  headers=None,
  service='execute-api',
  region=os.environ['AWS_REGION'],
  credentials=Session().get_credentials().get_frozen_credentials()
):
  """Sends an HTTP request signed with SigV4

  Args:
    url: The request URL (e.g. 'https://www.example.com').
    method: The request method (e.g. 'GET', 'POST', 'PUT', 'DELETE'). Defaults to 'GET'.
    body: The request body (e.g. json.dumps({ 'foo': 'bar' })). Defaults to None.
    params: The request query params (e.g. { 'foo': 'bar' }). Defaults to None.
    headers: The request headers (e.g. { 'content-type': 'application/json' }). Defaults to None.
    service: The AWS service name. Defaults to 'execute-api'.
    region: The AWS region id. Defaults to the env var 'AWS_REGION'.
    credentials: The AWS credentials. Defaults to the current boto3 session's credentials.
  Returns:
     The HTTP response
  """

  # sign request
  req = AWSRequest(
    method=method,
    url=url,
    data=body,
    params=params,
    headers=headers
  )
  SigV4Auth(credentials, service, region).add_auth(req)
  req = req.prepare()

  # send request
  return request(
    method=req.method,
    url=req.url,
    headers=req.headers,
    data=req.body
  )

def __init__():
  # send request
  response = sigv4_request(
    'https://www.example.com',
    method='POST',
    headers={
      'content-type': 'application/json',
    },
    body=json.dumps({
      'foo': 'bar'
    }),
    # In Lambda functions, you can omit 'region' and 'credentials' to use the Lambda's region and credentials
  )
  
  # do something with response
  print(response.status_code)
  print(response.json())

lifesaver, thanks for sharing

thank you so much!!!
you saved my soul!

hi can i use that to send an inference request to aws sagemaker?

@geraldstanje Hello, I haven't tested this code with sagemaker. In theory, as long as the endpoint is protected with SigV4 it should work. You'll have to test it yourself though.

dont you need a aws_key_id and aws_session for that?

@geraldstanje Yes, see here.

i will test it.