Source code for runway.cfngin.hooks.ssm.parameter

"""AWS SSM Parameter Store hooks."""
# pylint: disable=no-self-argument
from __future__ import annotations

import json
import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast

from pydantic import Extra, validator
from typing_extensions import Literal, TypedDict

from ....compat import cached_property
from ....utils import BaseModel, JsonEncoder
from ..protocols import CfnginHookProtocol
from ..utils import TagDataModel

if TYPE_CHECKING:
    from mypy_boto3_ssm.client import SSMClient
    from mypy_boto3_ssm.literals import ParameterTierType
    from mypy_boto3_ssm.type_defs import ParameterTypeDef, TagTypeDef

    from ...._logging import RunwayLogger
    from ....context import CfnginContext
else:
    ParameterTierType = Literal["Advanced", "Intelligent-Tiering", "Standard"]

LOGGER = cast("RunwayLogger", logging.getLogger(__name__))

# PutParameterResultTypeDef but without metadata
_PutParameterResultTypeDef = TypedDict(
    "_PutParameterResultTypeDef", {"Tier": ParameterTierType, "Version": int}
)


[docs]class ArgsDataModel(BaseModel): """Parameter hook args. Attributes: allowed_pattern: A regular expression used to validate the parameter value. data_type: The data type for a String parameter. Supported data types include plain text and Amazon Machine Image IDs. description: Information about the parameter. force: Skip checking the current value of the parameter, just put it. Can be used alongside ``overwrite`` to always update a parameter. key_id: The KMS Key ID that you want to use to encrypt a parameter. Either the default AWS Key Management Service (AWS KMS) key automatically assigned to your AWS account or a custom key. Required for parameters that use the ``SecureString`` data type. name: The fully qualified name of the parameter that you want to add to the system. overwrite: Allow overwriting an existing parameter. policies: One or more policies to apply to a parameter. This field takes a JSON array. tags: Optional metadata that you assign to a resource. tier: The parameter tier to assign to a parameter. type: The type of parameter. value: The parameter value that you want to add to the system. Standard parameters have a value limit of 4 KB. Advanced parameters have a value limit of 8 KB. """ allowed_pattern: Optional[str] = None data_type: Optional[Literal["aws:ec2:image", "text"]] = None description: Optional[str] = None force: bool = False key_id: Optional[str] = None name: str overwrite: bool = True policies: Optional[str] = None tags: Optional[List[TagDataModel]] = None tier: ParameterTierType = "Standard" type: Literal["String", "StringList", "SecureString"] value: Optional[str] = None
[docs] class Config: """Model configuration.""" allow_population_by_field_name = True extra = Extra.ignore fields = { "allowed_pattern": {"alias": "AllowedPattern"}, "data_type": {"alias": "DataType"}, "description": {"alias": "Description"}, "key_id": {"alias": "KeyId"}, "name": {"alias": "Name"}, "overwrite": {"alias": "Overwrite"}, "policies": {"alias": "Policies"}, "tags": {"alias": "Tags"}, "tier": {"alias": "Tier"}, "type": {"alias": "Type"}, "value": {"alias": "Value"}, }
@validator("policies", allow_reuse=True, pre=True) def _convert_policies(cls, v: Union[List[Dict[str, Any]], str, Any]) -> str: """Convert policies to acceptable value.""" if isinstance(v, str): return v if isinstance(v, list): return json.dumps(v, cls=JsonEncoder) raise TypeError( f"unexpected type {type(v)}; permitted: Optional[Union[List[Dict[str, Any]], str]]" ) @validator("tags", allow_reuse=True, pre=True) def _convert_tags( cls, v: Union[Dict[str, str], List[Dict[str, str]], Any] ) -> List[Dict[str, str]]: """Convert tags to acceptable value.""" if isinstance(v, list): return v if isinstance(v, dict): return [{"Key": k, "Value": v} for k, v in v.items()] raise TypeError( f"unexpected type {type(v)}; permitted: " "Optional[Union[Dict[str, str], List[Dict[str, str]]]" )
class _Parameter(CfnginHookProtocol): """AWS SSM Parameter Store Parameter.""" args: ArgsDataModel def __init__( # pylint: disable=super-init-not-called self, context: CfnginContext, *, name: str, type: Literal[ # pylint: disable=redefined-builtin "String", "StringList", "SecureString" ], **kwargs: Any, ) -> None: """Instantiate class. Args: context: CFNgin context object. name: The fully qualified name of the parameter that you want to add to the system. type: The type of parameter. """ self.args = ArgsDataModel.parse_obj({"name": name, "type": type, **kwargs}) self.ctx = context @cached_property def client(self) -> SSMClient: """AWS SSM client.""" return self.ctx.get_session().client("ssm") def delete(self) -> bool: """Delete parameter.""" try: self.client.delete_parameter(Name=self.args.name) LOGGER.info("deleted SSM Parameter %s", self.args.name) except self.client.exceptions.ParameterNotFound: LOGGER.info("delete parameter skipped; %s not found", self.args.name) return True def get(self) -> ParameterTypeDef: """Get parameter.""" if self.args.force: # bypass getting current value return {} try: return self.client.get_parameter( Name=self.args.name, WithDecryption=True ).get("Parameter", {}) except self.client.exceptions.ParameterNotFound: LOGGER.verbose("parameter %s does not exist", self.args.name) return {} def get_current_tags(self) -> List[TagTypeDef]: """Get Tags currently applied to Parameter.""" try: return self.client.list_tags_for_resource( ResourceId=self.args.name, ResourceType="Parameter" ).get("TagList", []) except ( self.client.exceptions.InvalidResourceId, self.client.exceptions.ParameterNotFound, ): return [] def post_deploy(self) -> _PutParameterResultTypeDef: """Run during the *post_deploy* stage.""" result = self.put() self.update_tags() return result def post_destroy(self) -> bool: """Run during the *post_destroy* stage.""" return self.delete() def pre_deploy(self) -> _PutParameterResultTypeDef: """Run during the *pre_deploy* stage.""" result = self.put() self.update_tags() return result def pre_destroy(self) -> bool: """Run during the *pre_destroy* stage.""" return self.delete() def put(self) -> _PutParameterResultTypeDef: """Put parameter.""" if not self.args.value: LOGGER.info( "skipped putting SSM Parameter; value provided for %s is falsy", self.args.name, ) return {"Tier": self.args.tier, "Version": 0} current_param = self.get() if current_param.get("Value") != self.args.value: try: result = self.client.put_parameter( **self.args.dict( by_alias=True, exclude_none=True, exclude={"force", "tags"} ) ) except self.client.exceptions.ParameterAlreadyExists: LOGGER.warning( "parameter %s already exists; to overwrite it's value, " 'set the overwrite field to "true"', self.args.name, ) return { "Tier": current_param.get("Tier", self.args.tier), "Version": current_param.get("Version", 0), } else: result: _PutParameterResultTypeDef = { "Tier": current_param.get("Tier", self.args.tier), "Version": current_param.get("Version", 0), } LOGGER.info("put SSM Parameter %s", self.args.name) return result def update_tags(self) -> None: """Update tags.""" current_tags = self.get_current_tags() if self.args.tags and current_tags: diff_tag_keys = list( {i["Key"] for i in current_tags} ^ {i.key for i in self.args.tags} ) elif self.args.tags: diff_tag_keys = [] else: diff_tag_keys = [i["Key"] for i in current_tags] try: if diff_tag_keys: diff_tag_keys.sort() self.client.remove_tags_from_resource( ResourceId=self.args.name, ResourceType="Parameter", TagKeys=diff_tag_keys, ) LOGGER.debug( "removed tags for parameter %s: %s", self.args.name, diff_tag_keys ) if self.args.tags: tags_to_add = [ cast("TagTypeDef", tag.dict(by_alias=True)) for tag in self.args.tags ] self.client.add_tags_to_resource( ResourceId=self.args.name, ResourceType="Parameter", Tags=tags_to_add, ) LOGGER.debug( "added tags to parameter %s: %s", self.args.name, [tag["Key"] for tag in tags_to_add], ) except self.client.exceptions.InvalidResourceId: LOGGER.info( "skipped updating tags; parameter %s does not exist", self.args.name ) else: LOGGER.info("updated tags for parameter %s", self.args.name)
[docs]class SecureString(_Parameter): """AWS SSM Parameter Store SecureString Parameter."""
[docs] def __init__( self, context: CfnginContext, *, name: str, **kwargs: Any, ) -> None: """Instantiate class. Args: context: CFNgin context object. name: The fully qualified name of the parameter that you want to add to the system. """ for k in ["Type", "type"]: # ensure neither of these are set kwargs.pop(k, None) super().__init__(context, name=name, type="SecureString", **kwargs)