conjur_iam_client.py

"""
This module provides functionality for obtaining an IAM signed request used for Conjur
authentication or for instantiating a Conjur SDK Python client using IAM authentication.
"""

import datetime
import hashlib
import hmac
import json
import os
import sys
import urllib.parse
from datetime import timedelta
import re

# pylint: disable=import-error
import requests  # pip install requests
from conjur_api import Client
from conjur_api.models import (ConjurConnectionInfo, CredentialsData,
                               SslVerificationMode)
from conjur_api.providers import (AuthnAuthenticationStrategy,
                                  SimpleCredentialsProvider)

# pylint: disable=duplicate-code
# ************* REQUEST VALUES *************
AWS_METADATA_URL = "http://169.254.169.254/latest/meta-data/iam/security-credentials/"
AWS_AVAILABILITY_ZONE = "http://169.254.169.254/latest/meta-data/placement/availability-zone"
METHOD = 'GET'
SERVICE = 'sts'
HOST = 'sts.amazonaws.com'
ENDPOINT = 'https://sts.amazonaws.com'
REQUEST_PARAMETERS = 'Action=GetCallerIdentity&Version=2011-06-15'
# pylint: enable=duplicate-code

class ConjurIAMAuthnException(Exception):
    """
    Exception raised when Conjur IAM authentication fails with a 401 Unauthorized error.
    """
    def __init__(self):
        Exception.__init__(
            self,
            "Conjur IAM authentication failed with 401 - Unauthorized. "
            "Check conjur logs for more information"
        )

class IAMRoleNotAvailableException(Exception):
    """
    Raised when the IAM Role is not available or incorrectly configured.
    """
    def __init__(self):
        Exception.__init__(
            self,
            "Most likely the ec2 instance is configured with no or an incorrect iam role"
        )

class InvalidAwsAccountIdException(Exception):
    """
    Raised when the AWS Account ID is invalid
    """
    def __init__(self):
        Exception.__init__(
            self,
            "The AWS Account ID specified in the CONJUR_AUTHN_LOGIN is invalid and "
            "must be a 12 digit number"
        )

def valid_aws_account_number(host_id):
    """
    Checks if the given host_id contains a valid 12-digit AWS Account ID.
    """
    parts = host_id.split("/")
    account_id = parts[len(parts)-2]
    if len(account_id) == 12:
        return True
    return False

# Key derivation functions. See:
# http://docs.aws.amazon.com/general/latest/gr/signature-v4-examples.html#signature-v4-examples-python
def sign(key, msg):
    """
    Signs the given message using the provided key and HMAC with SHA-256.
    """
    return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()


def get_signature_key(key, date_stamp, region_name, service_name):
    """
    Generates the AWS V4 signing key using the provided key,
    date, region, and service name.
    """
    k_date = sign(('AWS4' + key).encode('utf-8'), date_stamp)
    k_region = sign(k_date, region_name)
    k_service = sign(k_region, service_name)
    k_signing = sign(k_service, 'aws4_request')
    return k_signing

def get_aws_region():
    """
    Get the AWS Region.
    """
    # return requests.get(AWS_AVAILABILITY_ZONE).text[:-1]
    return "us-east-1"

def get_iam_role_name():
    """
    Retrieves the IAM Role Name associated with the current environment.
    """
    token = get_metadata_token()
    headers = {}
    if token:
        headers = {'X-aws-ec2-metadata-token': token}
    res = requests.get(AWS_METADATA_URL, headers=headers, timeout=10)
    return res.text

def get_metadata_token():
    """Request a session token for IMDSv2"""
    url = 'http://169.254.169.254/latest/api/token'
    headers = {'X-aws-ec2-metadata-token-ttl-seconds': '21600'}  # TTL for the token
    try:
        response = requests.put(url, headers=headers, timeout=2)
        if response.status_code == 200:
            return response.text

        return None
    except requests.exceptions.RequestException:
        return None

def get_iam_role_metadata(role_name, token=None):
    """
    Retrieves metadata for the IAM role associated with the current environment.
    """
    headers = {}
    if token:
        headers = {'X-aws-ec2-metadata-token': token}

    try:
        res = requests.get(AWS_METADATA_URL + role_name, headers=headers, timeout=10)
        if res.status_code == 404:
            raise IAMRoleNotAvailableException()

        if res.status_code != 200:
            raise RuntimeError(f"Error retrieving IAM role metadata: {res.status_code}")

        json_dict = json.loads(res.text)

        access_key_id = json_dict["AccessKeyId"]
        secret_access_key = json_dict["SecretAccessKey"]
        token = json_dict["Token"]

        return access_key_id, secret_access_key, token
    except requests.exceptions.RequestException as err:
        raise RuntimeError(f"Failed to get IAM role metadata: {str(err)}") from err

def create_canonical_request(amzdate, token, signed_headers, payload_hash):
    """
    Creates the canonical request string for signing the AWS request.
    """
    # ************* TASK 1: CREATE A CANONICAL REQUEST *************
    # http://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html

    # Step 1 is to define the verb (GET, POST, etc.)--already done.

    # Step 2: Create canonical URI--the part of the URI from domain to query
    # string (use '/' if no path)
    canonical_uri = '/'

    # Step 3: Create the canonical query string. In this example (a GET request),
    # request parameters are in the query string. Query string values must
    # be URL-encoded (space=%20). The parameters must be sorted by name.
    # For this example, the query string is pre-formatted in the request_parameters variable.
    canonical_querystring = REQUEST_PARAMETERS

    # Step 4: Create payload hash (hash of the request body content). For GET
    # requests, the payload is an empty string ("").
    # payload_hash = hashlib.sha256(('').encode('utf-8')).hexdigest()

    # Step 5: Create the canonical headers and signed headers. Header names
    # must be trimmed and lowercase, and sorted in code point order from
    # low to high. Note that there is a trailing \n.
    canonical_headers = 'host:' + HOST + '\n' + 'x-amz-content-sha256:' + payload_hash + '\n' + \
        'x-amz-date:' + amzdate + '\n' + 'x-amz-security-token:' + token + '\n'

    # Step 6: Create the list of signed headers. This lists the headers
    # in the canonical_headers list, delimited with ";" and in alpha order.
    # Note: The request can include any headers; canonical_headers and
    # signed_headers lists those that you want to be included in the
    # hash of the request. "Host" and "x-amz-date" are always required.
    # signed_headers = 'host;x-amz-content-sha256;x-amz-date;x-amz-security-token'

    # Step 7: Combine elements to create canonical request
    canonical_request = METHOD + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + \
        canonical_headers + '\n' + signed_headers + '\n' + payload_hash

    return canonical_request

# pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
def create_conjur_iam_api_key(iam_role_name=None, access_key=None, secret_key=None, token=None):
    """
    Creates an IAM API key for Conjur authentication using the provided IAM role and credentials.
    """
    if iam_role_name is None:
        iam_role_name = get_iam_role_name()

    metadata_token = get_metadata_token()

    if access_key is None and secret_key is None and token is None:
        access_key, secret_key, token = get_iam_role_metadata(iam_role_name, metadata_token)

    region = get_aws_region()

    if access_key is None or secret_key is None:
        print('No access key is available.')
        sys.exit()

    # Create a date for headers and the credential string
    date = datetime.datetime.now(datetime.timezone.utc)
    amzdate = date.strftime('%Y%m%dT%H%M%SZ')
    datestamp = date.strftime('%Y%m%d') # Date w/o time, used in credential scope

    # ************* TASK 1: CREATE A CANONICAL REQUEST *************
    signed_headers = 'host;x-amz-content-sha256;x-amz-date;x-amz-security-token'
    payload_hash = hashlib.sha256(('').encode('utf-8')).hexdigest()
    canonical_request = create_canonical_request(amzdate, token, signed_headers, payload_hash)

    # ************* TASK 2: CREATE THE STRING TO SIGN*************
    # Match the algorithm to the hashing algorithm you use, either SHA-1 or
    # SHA-256 (recommended)
    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = datestamp + '/' + region + '/' + SERVICE + '/' + 'aws4_request'
    string_to_sign = algorithm + '\n' + amzdate + '\n' + credential_scope + \
        '\n' + hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()

    # ************* TASK 3: CALCULATE THE SIGNATURE *************
    # Create the signing key using the function defined above.
    signing_key = get_signature_key(secret_key, datestamp, region, SERVICE)

    # Sign the string_to_sign using the signing_key
    signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()

    # ************* TASK 4: ADD SIGNING INFORMATION TO THE REQUEST *************
    # The signing information can be either in a query string value or in
    # a header named Authorization. This code shows how to use a header.
    # Create authorization header and add to request headers
    authorization_header = (
        algorithm + ' ' +
        'Credential=' + access_key + '/' + credential_scope + ', ' +
        'SignedHeaders=' + signed_headers + ', ' +
        'Signature=' + signature
    )

    # The request can include any headers, but MUST include "host", "x-amz-date",
    # and (for this scenario) "Authorization". "host" and "x-amz-date" must
    # be included in the canonical_headers and signed_headers, as noted
    # earlier. Order here is not significant.
    # Python note: The 'host' header is added automatically by the Python 'requests' library.

    headers = {
        'host': HOST,
        'x-amz-date': amzdate,
        'x-amz-security-token': token,
        'x-amz-content-sha256': payload_hash,
        'authorization': authorization_header
    }

    # ************* SEND THE REQUEST *************
    return f'{headers}'.replace("'", '"')

def get_conjur_iam_session_token(
    appliance_url, account, service_id, host_id, cert_file,
    iam_role_name=None, access_key=None, secret_key=None, token=None, ssl_verify=True
):
    """
    Retrieves the Conjur IAM session token for the provided service and IAM role credentials.
    """

    if not valid_aws_account_number(host_id):
        raise InvalidAwsAccountIdException()

    appliance_url = appliance_url.rstrip("/")
    url = (
        f"{appliance_url}/authn-iam/{service_id}/{account}/"
        f"{urllib.parse.quote(host_id, safe='')}/authenticate"
    )

    iam_api_key = create_conjur_iam_api_key(iam_role_name, access_key, secret_key, token)

    # If cert file is not provided then assume conjur is using valid certificate
    if cert_file is None:
        cert_file = True

    # If ssl_verify is explicitly false then ignore ssl certificate even if cert_file is set
    if not ssl_verify:
        cert_file = False

    res = requests.post(url=url,data=iam_api_key,verify=cert_file,timeout=10)

    if res.status_code == 401:
        raise ConjurIAMAuthnException()
    return res.text

def get_version_from_changelog(file_path):
    """
    Extracts the first semantic version number from a changelog file.

    Args:
        file_path (str): Path to the changelog.md file.

    Returns:
        str or None: The first version number found in the format X.Y.Z, or None if not found.
    """
    cached_version = os.environ.get('INTEGRATION_VERSION')
    if cached_version:
        return cached_version.strip()
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            content = file.read()

        # Match version numbers like [1.2.3]
        match = re.search(r'## \[(\d+\.\d+\.\d+)\]', content)
        if match:
            version = match.group(1)
            os.environ['INTEGRATION_VERSION'] = version  # cache it
            return version
    except FileNotFoundError:
        print(f"Warning: {file_path} not found. Using default version.")
    version = "0.0.0"
    os.environ['INTEGRATION_VERSION'] = version
    return version

# If using IAM roles with conjur via the python3 api client use this function.
# The client will not support auto-refreshing of token when using iam authentication
# so it is recommended to call this
# method everytime you make a client request.
# An issue/enhancement has ben created on the conjur-python3-api
# github to address this issue however this is a work around for the time being.
# pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
def create_conjur_iam_client(
    appliance_url, account, service_id, host_id, cert_file,
    iam_role_name=None, access_key=None, secret_key=None,
    token=None, ssl_verify=True
):
    """
    Create Conjur IAM Client
    """
    appliance_url = appliance_url.rstrip("/")
    # create our client with a placeholder api key
    connection_info = ConjurConnectionInfo(
        conjur_url=appliance_url, account=account, cert_file=cert_file
    )
    credentials = CredentialsData(
        username=host_id, api_key="placeholder", machine=appliance_url
    )
    credentials_provider = SimpleCredentialsProvider()
    credentials_provider.save(credentials)
    del credentials
    authn_provider = AuthnAuthenticationStrategy(credentials_provider)
    ssl_verification_mode=SslVerificationMode.CA_BUNDLE

    if cert_file is None:
        ssl_verification_mode=SslVerificationMode.INSECURE

    client = Client(
        connection_info,
        authn_strategy=authn_provider,
        ssl_verification_mode=ssl_verification_mode,
        async_mode=False
    )

    client.set_integration_name("IAM Client")
    client.set_integration_type("cybr-secretsmanager-python-sdk")
    client.set_integration_version(get_version_from_changelog('CHANGELOG.md'))
    client.set_vendor_name("AWS")

    # telemetry changes
    # latest_version = Client.get_latest_version(
    #     os.path.join(os.path.dirname(__file__), 'CHANGELOG.md')
    # )
    #client.set_top_source_name("cour_iam_client/"+latest_version)

    # now obtain the iam session_token
    session_token = get_conjur_iam_session_token(
      appliance_url, account, service_id, host_id, cert_file, iam_role_name,
      access_key, secret_key, token, ssl_verify
    )

    # override the _api_token with the token created in get_conjur_iam_session_token
    # pylint: disable=W0212
    client._api._api_token = session_token
    client._api.api_token_expiration = (
        datetime.datetime.now() + timedelta(minutes=8)
    )

    return client

def create_conjur_iam_client_from_env(
    iam_role_name=None,
    access_key=None,
    secret_key=None,
    token=None,
    ssl_verify=True
):
    """
    Create Conjur IAM Client from environmental variables.
    """
    try:
        appliance_url = os.environ['CONJUR_APPLIANCE_URL']
        account = os.environ['CONJUR_ACCOUNT']
        service_id = os.environ['AUTHN_IAM_SERVICE_ID']
        host_id = os.environ['CONJUR_AUTHN_LOGIN']
        cert_file = None
        if 'CONJUR_CERT_FILE' in os.environ:
            cert_file = os.environ['CONJUR_CERT_FILE']
        return create_conjur_iam_client(
            appliance_url, account, service_id, host_id, cert_file,
            iam_role_name, access_key, secret_key, token, ssl_verify
        )

    except KeyError as err:
        raise KeyError(f"Failed to retrieve environment variable: {err}") from err


# Examples of using methods:
# get_conjur_iam_session_token(os.environ['CONJUR_APPLIANCE_URL'], os.environ['CONJUR_ACCOUNT'],
# os.environ['AUTHN_IAM_SERVICE_ID'],
# os.environ['CONJUR_AUTHN_LOGIN'], os.environ['CONJUR_CERT_FILE'])
# create_conjur_iam_client(os.environ['CONJUR_APPLIANCE_URL'], os.environ['CONJUR_ACCOUNT'],
# os.environ['AUTHN_IAM_SERVICE_ID'],
# os.environ['CONJUR_AUTHN_LOGIN'], os.environ['CONJUR_CERT_FILE'])