runway.cfngin.utils module¶

CFNgin utilities.

class runway.cfngin.utils.Extractor[source]¶

Bases: object

Base class for extractors.

__init__(archive: Path | None = None) → None[source]¶

Instantiate class.

Parameters:: archive (str) – Archive path.

set_archive(dir_name: Path) → None[source]¶

Update archive filename to match directory name & extension.

Parameters:: dir_name – Archive directory name

class runway.cfngin.utils.SOARecord[source]¶

Bases: object

Represents an SOA record.

__init__(record: ResourceRecordSetOutputTypeDef) → None[source]¶: Instantiate class.

class runway.cfngin.utils.SOARecordText[source]¶

Bases: object

Represents the actual body of an SOARecord.

__init__(record_text: str) → None[source]¶: Instantiate class.

__str__() → str[source]¶: Convert an instance of this class to a string.

class runway.cfngin.utils.SourceProcessor[source]¶

Bases: object

Makes remote python package sources available in current environment.

__init__(sources: CfnginPackageSourcesDefinitionModel, cache_dir: Path) → None[source]¶

Process a config’s defined package sources.

Parameters:

sources – Package sources from CFNgin config dictionary.
cache_dir – Path where remote sources will be cached.

create_cache_directories() → None[source]¶: Ensure that SourceProcessor cache directories exist.

static determine_git_ls_remote_ref(config: GitCfnginPackageSourceDefinitionModel) → str[source]¶

Determine the ref to be used with the “git ls-remote” command.

Parameters:: config – Git package source config.
Returns:: A branch reference or “HEAD”.

determine_git_ref(config: GitCfnginPackageSourceDefinitionModel) → str[source]¶

Determine the ref to be used for git checkout.

Parameters:: config – Git package source config.
Returns:: A commit id or tag name.

fetch_git_package(config: GitCfnginPackageSourceDefinitionModel) → None[source]¶

Make a remote git repository available for local use.

Parameters:: config – Package source config.

fetch_local_package(config: LocalCfnginPackageSourceDefinitionModel) → None[source]¶

Make a local path available to current CFNgin config.

Parameters:: config – Package source config.

fetch_s3_package(config: S3CfnginPackageSourceDefinitionModel) → None[source]¶

Make a remote S3 archive available for local use.

Parameters:: config – Package source config.

get_package_sources() → None[source]¶: Make remote python packages available for local use.

static git_ls_remote(uri: str, ref: str) → str[source]¶

Determine the latest commit id for a given ref.

Parameters:

uri – Git URI.
ref – Git ref.

sanitize_git_path(uri: str, ref: str | None = None) → str[source]¶

Take a git URI and ref and converts it to a directory safe path.

Parameters:

uri – Git URI. (e.g. git@github.com:foo/bar.git)
ref – Git ref to be appended to the path.

Returns:

Directory name for the supplied uri

static sanitize_uri_path(uri: str) → str[source]¶

Take a URI and converts it to a directory safe path.

Parameters:: uri – URI to sanitize.
Returns:: Directory name for the supplied uri.

update_paths_and_config(config: GitCfnginPackageSourceDefinitionModel | LocalCfnginPackageSourceDefinitionModel | S3CfnginPackageSourceDefinitionModel, pkg_dir_name: str, pkg_cache_dir: Path | None = None) → None[source]¶

Handle remote source defined sys.paths & configs.

Parameters:

config – Package source config.
pkg_dir_name – Directory name of the CFNgin archive.
pkg_cache_dir – Fully qualified path to CFNgin cache cache directory.

class runway.cfngin.utils.TarExtractor[source]¶

Bases: Extractor

Extracts tar archives.

extract(destination: Path) → None[source]¶: Extract the archive.

class runway.cfngin.utils.TarGzipExtractor[source]¶

Bases: Extractor

Extracts compressed tar archives.

extract(destination: Path) → None[source]¶: Extract the archive.

class runway.cfngin.utils.ZipExtractor[source]¶

Bases: Extractor

Extracts zip archives.

extract(destination: Path) → None[source]¶: Extract the archive.

runway.cfngin.utils.camel_to_snake(name: str) → str[source]¶

Convert CamelCase to snake_case.

Parameters:: name (str) – The name to convert from CamelCase to snake_case.
Returns:: Converted string.
Return type:: str

runway.cfngin.utils.cf_safe_name(name: str) → str[source]¶

Convert a name to a safe string for a CloudFormation resource.

Given a string, returns a name that is safe for use as a CloudFormation Resource. (ie: Only alphanumeric characters)

runway.cfngin.utils.convert_class_name(kls: type) → str[source]¶

Get a string that represents a given class.

Parameters:: kls – The class being analyzed for its name.
Returns:: The name of the given kls.

runway.cfngin.utils.create_route53_zone(client: Route53Client, zone_name: str) → str[source]¶

Create the given zone_name if it doesn’t already exists.

Also sets the SOA negative caching TTL to something short (300 seconds).

Parameters:

client – The connection used to interact with Route53’s API.
zone_name – The name of the DNS hosted zone to create.

Returns:

The zone id returned from AWS for the existing, or newly created zone.

runway.cfngin.utils.ensure_s3_bucket(s3_client: S3Client, bucket_name: str, bucket_region: str | None = None, *, create: bool = True, persist_graph: bool = False) → None[source]¶

Ensure an s3 bucket exists, if it does not then create it.

Parameters:

s3_client – An s3 client used to verify and create the bucket.
bucket_name – The bucket being checked/created.
bucket_region – The region to create the bucket in. If not provided, will be determined by s3_client’s region.
create – Whether to create the bucket if it does not exist.
persist_graph – Check bucket for recommended settings. If creating a new bucket, it will be created with recommended settings.

runway.cfngin.utils.get_client_region(client: Any) → str[source]¶

Get the region from a boto3 client.

Parameters:: client – The client to get the region from.
Returns:: AWS region string.

runway.cfngin.utils.get_hosted_zone_by_name(client: Route53Client, zone_name: str) → str | None[source]¶

Get the zone id of an existing zone by name.

Parameters:

client – The connection used to interact with Route53’s API.
zone_name – The name of the DNS hosted zone to create.

Returns:

The Id of the Hosted Zone.

runway.cfngin.utils.get_or_create_hosted_zone(client: Route53Client, zone_name: str) → str[source]¶

Get the Id of an existing zone, or create it.

Parameters:

client – The connection used to interact with Route53’s API.
zone_name – The name of the DNS hosted zone to create.

Returns:

The Id of the Hosted Zone.

runway.cfngin.utils.get_s3_endpoint(client: Any) → str[source]¶

Get the s3 endpoint for the given boto3 client.

Parameters:: client – The client to get the endpoint from.
Returns:: The AWS endpoint for the client.

runway.cfngin.utils.get_soa_record(client: Route53Client, zone_id: str, zone_name: str) → SOARecord[source]¶

Get the SOA record for zone_name from zone_id.

Parameters:

client – The connection used to interact with Route53’s API.
zone_id – The AWS Route53 zone id of the hosted zone to query.
zone_name – The name of the DNS hosted zone to create.

Returns:

An object representing the parsed SOA record returned from AWS Route53.

runway.cfngin.utils.is_within_directory(directory: Path | str, target: str) → bool[source]¶

Check if file is in directory.

Determines if the provided path is within a specific directory or its subdirectories.

Parameters:

directory – Path of the directory we’re checking.
target – Path of the file we’re checking for containment.

Returns:

True if the target is in the directory or subdirectories, False otherwise.

Return type:

bool

runway.cfngin.utils.parse_cloudformation_template(template: str) → dict[str, Any][source]¶

Parse CFN template string.

Leverages the vendored aws-cli yamlhelper to handle JSON or YAML templates.

Parameters:: template – The template body.

runway.cfngin.utils.parse_zone_id(full_zone_id: str) → str[source]¶: Parse the returned hosted zone id and returns only the ID itself.

runway.cfngin.utils.read_value_from_path(value: str, *, root_path: Path | None = None) → str[source]¶

Enable translators to read values from files.

The value can be referred to with the file:// prefix.

Example

conf_key: ${kms file://kms_value.txt}

runway.cfngin.utils.s3_bucket_location_constraint(region: str | None) → str | None[source]¶

Return the appropriate LocationConstraint info for a new S3 bucket.

When creating a bucket in a region OTHER than us-east-1, you need to specify a LocationConstraint inside the CreateBucketConfiguration argument. This function helps you determine the right value given a given client.

Parameters:: region – The region where the bucket will be created in.
Returns:: The string to use with the given client for creating a bucket.

runway.cfngin.utils.safe_tar_extract(tar: TarFile, path: Path | str = '.', members: list[TarInfo] | None = None, *, numeric_owner: bool = False) → None[source]¶

Safely extract the contents of a tar file to a specified directory.

This code is modified from a PR provided to Runway project to address CVE-2007-4559.

Parameters:

tar – The tar file object that will be extracted.
path – The directory to extract the tar into.
members – List of TarInfo objects to extract.
numeric_owner – Enable usage of owner and group IDs when extracting.

Raises:

Exception – If any tar file tries to go outside the specified area.

runway.cfngin.utils.stack_template_key_name(blueprint: Blueprint) → str[source]¶

Given a blueprint, produce an appropriate key name.

Parameters:: blueprint – The blueprint object to create the key from.
Returns:: Key name resulting from blueprint.

runway.cfngin.utils.uppercase_first_letter(string_: str) → str[source]¶: Return string with first character upper case.

runway.cfngin.utils.yaml_to_ordered_dict(stream: str, loader: type[~yaml.loader.Loader | ~yaml.loader.SafeLoader] = <class 'yaml.loader.SafeLoader'>) → OrderedDict[str, Any][source]¶

yaml.load alternative with preserved dictionary order.

Parameters:

stream – YAML string to load.
loader – PyYAML loader class. Defaults to safe load.