No new info detected.

#6920

Committed 22 Nov 2025 06:13PM UTC coverage: 78.218% (+0.007%) from 78.211%

Build # #6920

Build Type

push

coveralls-python

Committed by

teodorescuserban

Commit Message

HDX-10015 Add production config and development tools

Production configuration:
- Add boto3 to requirements.txt for SES API support
- Configure plugin in prod.ini.tpl and common-config-ini.txt
- Add HDX_SMTP_DOMAIN to hdx_theme config declaration
- Update setup_py_helper.sh for plugin packaging

Development tools:
- run-plugin-tests.sh: Quick test runner with docker compose
  - Auto-initializes test environment if needed
  - Reuses running stack for fast iteration
  - Shows coverage for specified plugin only
- TESTING.md: Comprehensive testing documentation

Run Details

13915 of 17790 relevant lines covered (78.22%)

0.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

/ckanext-hdx_smtp_assumerole/ckanext/hdx_smtp_assumerole/helpers/smtp_assume_role.py

# encoding: utf-8

import logging
import boto3
from typing import Dict, Any, Optional
from botocore.exceptions import ClientError, BotoCoreError
from botocore.credentials import InstanceMetadataProvider, InstanceMetadataFetcher

log = logging.getLogger(__name__)


class SMTPAssumeRoleException(Exception):
    """Exception raised when SMTP AssumeRole operations fail."""
    pass


def create_sts_client_with_instance_profile() -> Any:
    """
    Create STS client using ONLY EC2 instance profile credentials.
    Ignores AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY from environment.

    This is necessary because:
    - Developers need static credentials for S3 in local environments
    - STS AssumeRole calls must use instance profile, not static credentials
    - boto3 prioritizes env vars over instance profile by default

    This function explicitly fetches credentials from EC2 instance metadata,
    bypassing environment variables entirely.

    :return: STS client using instance profile credentials only
    :rtype: boto3.client
    :raises SMTPAssumeRoleException: If instance profile credentials cannot be loaded
    """
    try:
        log.debug('Creating STS client using instance profile credentials (ignoring env vars)')

        # Fetch credentials directly from EC2 instance metadata endpoint
        # This bypasses environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
        fetcher = InstanceMetadataFetcher()
        provider = InstanceMetadataProvider(iam_role_fetcher=fetcher)
        credentials = provider.load()

        if credentials is None:
            raise SMTPAssumeRoleException(
                'Failed to load credentials from EC2 instance profile. '
                'Ensure you are running on EC2 with an instance profile attached. '
                'Instance metadata endpoint may be unreachable.'
            )

        log.debug('Successfully loaded credentials from EC2 instance profile')

        # Create STS client with explicit instance profile credentials
        # This ensures env vars (AWS_ACCESS_KEY_ID, etc.) are completely ignored
        sts_client = boto3.client(
            'sts',
            aws_access_key_id=credentials.access_key,
            aws_secret_access_key=credentials.secret_key,
            aws_session_token=credentials.token
        )

        log.debug('Successfully created STS client with instance profile credentials')
        return sts_client

    except SMTPAssumeRoleException:
        # Re-raise our custom exceptions
        raise
    except Exception as e:
        msg = f'Failed to create STS client with instance profile: {str(e)}'
        log.error(msg)
        raise SMTPAssumeRoleException(msg)


def get_account_id_from_sts() -> str:
    """
    Get AWS account ID from STS get-caller-identity.
    Uses ONLY EC2 instance profile credentials (ignores env vars).

    :return: AWS account ID
    :rtype: str
    """
    try:
        # Create STS client using instance profile only (ignores AWS_ACCESS_KEY_ID from env)
        sts_client = create_sts_client_with_instance_profile()

        response = sts_client.get_caller_identity()
        account_id = response['Account']
        log.debug(f'Retrieved AWS account ID from instance profile: {account_id}')
        return account_id
    except SMTPAssumeRoleException:
        # Re-raise our custom exceptions
        raise
    except (ClientError, BotoCoreError) as e:
        msg = f'Failed to get AWS account ID from STS: {str(e)}'
        log.error(msg)
        raise SMTPAssumeRoleException(msg)


def build_role_arn(role_name_or_arn: str, region: Optional[str] = None) -> str:
    """
    Build full role ARN from role name or return ARN if already provided.
    If only role name is provided, account ID is deduced from STS.

    :param role_name_or_arn: Role name (e.g., 'my-role') or full ARN
    :type role_name_or_arn: str
    :param region: AWS region (optional, for logging)
    :type region: str
    :return: Full role ARN
    :rtype: str
    """
    if not role_name_or_arn:
        raise SMTPAssumeRoleException('Role name or ARN is required')

    # Check if already an ARN
    if role_name_or_arn.startswith('arn:aws:iam::'):
        log.debug(f'Using provided role ARN: {role_name_or_arn}')
        return role_name_or_arn

    # Build ARN from role name
    account_id = get_account_id_from_sts()
    role_arn = f'arn:aws:iam::{account_id}:role/{role_name_or_arn}'
    log.debug(f'Built role ARN: {role_arn}')
    return role_arn


def assume_role_for_smtp(role_name_or_arn: str, region: str, session_name: str = 'ckan-ses-session', duration_seconds: int = 3600) -> Dict[str, Any]:
    """
    Assume AWS IAM role and return temporary credentials for SES API.
    Uses ONLY EC2 instance profile credentials (ignores env vars).

    This function:
    1. Builds full role ARN (deduces account ID if needed)
    2. Assumes the role using STS with instance profile only
    3. Returns temporary credentials (access_key, secret_key, session_token)

    Note: This function explicitly uses instance profile credentials and ignores
    AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY from environment. This allows
    using static credentials for other services (like S3) without interfering
    with STS AssumeRole calls.

    The returned credentials are used with SES API (not SMTP), which supports
    session tokens from temporary credentials.

    :param role_name_or_arn: Role name or full ARN to assume
    :type role_name_or_arn: str
    :param region: AWS region for SES API
    :type region: str
    :param session_name: Session name for AssumeRole (default: 'ckan-ses-session')
    :type session_name: str
    :param duration_seconds: Duration of temporary credentials (default: 3600)
    :type duration_seconds: int
    :return: Dictionary with access_key, secret_key, session_token, expiration
    :rtype: dict
    """
    try:
        # Build role ARN
        role_arn = build_role_arn(role_name_or_arn, region)

        # Create STS client using instance profile only (ignores AWS_ACCESS_KEY_ID from env)
        sts_client = create_sts_client_with_instance_profile()
        log.debug(f'Assuming role: {role_arn} with session: {session_name}')

        response = sts_client.assume_role(
            RoleArn=role_arn,
            RoleSessionName=session_name,
            DurationSeconds=duration_seconds
        )

        credentials = response['Credentials']
        access_key = credentials['AccessKeyId']
        secret_key = credentials['SecretAccessKey']
        session_token = credentials['SessionToken']
        expiration = credentials['Expiration']

        log.debug(f'Successfully assumed role for SES API. Credentials expire at: {expiration}')

        return {
            'access_key': access_key,
            'secret_key': secret_key,
            'session_token': session_token,
            'expiration': expiration,
        }

    except SMTPAssumeRoleException:
        # Re-raise our custom exceptions
        raise
    except (ClientError, BotoCoreError) as e:
        msg = f'Failed to assume role for SMTP: {str(e)}'
        log.error(msg)
        raise SMTPAssumeRoleException(msg)
    except Exception as e:
        msg = f'Unexpected error during SMTP AssumeRole: {str(e)}'
        log.error(msg)
        raise SMTPAssumeRoleException(msg)

1	# encoding: utf-8
2
3	import logging	1✔
4	import boto3	1✔
5	from typing import Dict, Any, Optional	1✔
6	from botocore.exceptions import ClientError, BotoCoreError	1✔
7	from botocore.credentials import InstanceMetadataProvider, InstanceMetadataFetcher	1✔
8
9	log = logging.getLogger(__name__)	1✔
10
11
12	class SMTPAssumeRoleException(Exception):	1✔
13	"""Exception raised when SMTP AssumeRole operations fail."""
14	pass	1✔
15
16
17	def create_sts_client_with_instance_profile() -> Any:	1✔
18	"""
19	Create STS client using ONLY EC2 instance profile credentials.
20	Ignores AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY from environment.
21
22	This is necessary because:
23	- Developers need static credentials for S3 in local environments
24	- STS AssumeRole calls must use instance profile, not static credentials
25	- boto3 prioritizes env vars over instance profile by default
26
27	This function explicitly fetches credentials from EC2 instance metadata,
28	bypassing environment variables entirely.
29
30	:return: STS client using instance profile credentials only
31	:rtype: boto3.client
32	:raises SMTPAssumeRoleException: If instance profile credentials cannot be loaded
33	"""
34	try:	1✔
35	log.debug('Creating STS client using instance profile credentials (ignoring env vars)')	1✔
36
37	# Fetch credentials directly from EC2 instance metadata endpoint
38	# This bypasses environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
39	fetcher = InstanceMetadataFetcher()	1✔
40	provider = InstanceMetadataProvider(iam_role_fetcher=fetcher)	1✔
41	credentials = provider.load()	1✔
42
43	if credentials is None:	1✔
44	raise SMTPAssumeRoleException(	×
45	'Failed to load credentials from EC2 instance profile. '
46	'Ensure you are running on EC2 with an instance profile attached. '
47	'Instance metadata endpoint may be unreachable.'
48	)
49
50	log.debug('Successfully loaded credentials from EC2 instance profile')	1✔
51
52	# Create STS client with explicit instance profile credentials
53	# This ensures env vars (AWS_ACCESS_KEY_ID, etc.) are completely ignored
54	sts_client = boto3.client(	1✔
55	'sts',
56	aws_access_key_id=credentials.access_key,
57	aws_secret_access_key=credentials.secret_key,
58	aws_session_token=credentials.token
59	)
60
61	log.debug('Successfully created STS client with instance profile credentials')	1✔
62	return sts_client	1✔
63
64	except SMTPAssumeRoleException:	1✔
65	# Re-raise our custom exceptions
66	raise	×
67	except Exception as e:	1✔
68	msg = f'Failed to create STS client with instance profile: {str(e)}'	1✔
69	log.error(msg)	1✔
70	raise SMTPAssumeRoleException(msg)	1✔
71
72
73	def get_account_id_from_sts() -> str:	1✔
74	"""
75	Get AWS account ID from STS get-caller-identity.
76	Uses ONLY EC2 instance profile credentials (ignores env vars).
77
78	:return: AWS account ID
79	:rtype: str
80	"""
81	try:	1✔
82	# Create STS client using instance profile only (ignores AWS_ACCESS_KEY_ID from env)
83	sts_client = create_sts_client_with_instance_profile()	1✔
84
85	response = sts_client.get_caller_identity()	1✔
86	account_id = response['Account']	1✔
87	log.debug(f'Retrieved AWS account ID from instance profile: {account_id}')	1✔
88	return account_id	1✔
89	except SMTPAssumeRoleException:	1✔
90	# Re-raise our custom exceptions
91	raise	×
92	except (ClientError, BotoCoreError) as e:	1✔
93	msg = f'Failed to get AWS account ID from STS: {str(e)}'	1✔
94	log.error(msg)	1✔
95	raise SMTPAssumeRoleException(msg)	1✔
96
97
98	def build_role_arn(role_name_or_arn: str, region: Optional[str] = None) -> str:	1✔
99	"""
100	Build full role ARN from role name or return ARN if already provided.
101	If only role name is provided, account ID is deduced from STS.
102
103	:param role_name_or_arn: Role name (e.g., 'my-role') or full ARN
104	:type role_name_or_arn: str
105	:param region: AWS region (optional, for logging)
106	:type region: str
107	:return: Full role ARN
108	:rtype: str
109	"""
110	if not role_name_or_arn:	1✔
111	raise SMTPAssumeRoleException('Role name or ARN is required')	1✔
112
113	# Check if already an ARN
114	if role_name_or_arn.startswith('arn:aws:iam::'):	1✔
115	log.debug(f'Using provided role ARN: {role_name_or_arn}')	1✔
116	return role_name_or_arn	1✔
117
118	# Build ARN from role name
119	account_id = get_account_id_from_sts()	1✔
120	role_arn = f'arn:aws:iam::{account_id}:role/{role_name_or_arn}'	1✔
121	log.debug(f'Built role ARN: {role_arn}')	1✔
122	return role_arn	1✔
123
124
125	def assume_role_for_smtp(role_name_or_arn: str, region: str, session_name: str = 'ckan-ses-session', duration_seconds: int = 3600) -> Dict[str, Any]:	1✔
126	"""
127	Assume AWS IAM role and return temporary credentials for SES API.
128	Uses ONLY EC2 instance profile credentials (ignores env vars).
129
130	This function:
131	1. Builds full role ARN (deduces account ID if needed)
132	2. Assumes the role using STS with instance profile only
133	3. Returns temporary credentials (access_key, secret_key, session_token)
134
135	Note: This function explicitly uses instance profile credentials and ignores
136	AWS_ACCESS_KEY_ID/AWS_SECRET_ACCESS_KEY from environment. This allows
137	using static credentials for other services (like S3) without interfering
138	with STS AssumeRole calls.
139
140	The returned credentials are used with SES API (not SMTP), which supports
141	session tokens from temporary credentials.
142
143	:param role_name_or_arn: Role name or full ARN to assume
144	:type role_name_or_arn: str
145	:param region: AWS region for SES API
146	:type region: str
147	:param session_name: Session name for AssumeRole (default: 'ckan-ses-session')
148	:type session_name: str
149	:param duration_seconds: Duration of temporary credentials (default: 3600)
150	:type duration_seconds: int
151	:return: Dictionary with access_key, secret_key, session_token, expiration
152	:rtype: dict
153	"""
154	try:	1✔
155	# Build role ARN
156	role_arn = build_role_arn(role_name_or_arn, region)	1✔
157
158	# Create STS client using instance profile only (ignores AWS_ACCESS_KEY_ID from env)
159	sts_client = create_sts_client_with_instance_profile()	1✔
160	log.debug(f'Assuming role: {role_arn} with session: {session_name}')	1✔
161
162	response = sts_client.assume_role(	1✔
163	RoleArn=role_arn,
164	RoleSessionName=session_name,
165	DurationSeconds=duration_seconds
166	)
167
168	credentials = response['Credentials']	1✔
169	access_key = credentials['AccessKeyId']	1✔
170	secret_key = credentials['SecretAccessKey']	1✔
171	session_token = credentials['SessionToken']	1✔
172	expiration = credentials['Expiration']	1✔
173
174	log.debug(f'Successfully assumed role for SES API. Credentials expire at: {expiration}')	1✔
175
176	return {	1✔
177	'access_key': access_key,
178	'secret_key': secret_key,
179	'session_token': session_token,
180	'expiration': expiration,
181	}
182
183	except SMTPAssumeRoleException:	1✔
184	# Re-raise our custom exceptions
185	raise	×
186	except (ClientError, BotoCoreError) as e:	1✔
187	msg = f'Failed to assume role for SMTP: {str(e)}'	1✔
188	log.error(msg)	1✔
189	raise SMTPAssumeRoleException(msg)	1✔
190	except Exception as e:	×
191	msg = f'Unexpected error during SMTP AssumeRole: {str(e)}'	×
192	log.error(msg)	×
193	raise SMTPAssumeRoleException(msg)	×

OCHA-DAP / hdx-ckan / #6920

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous