🚀 Create, manage and connect to servers in your AWS account with no effort ☁️
import click
from botocore.exceptions import BotoCoreError, ClientError

from holy_cli import __version__
from holy_cli.log import getLogger

from .global_commands import teardown, update
from .server_commands import server


# Global CLI entry point to catch unhandled botocore exceptions
def entry_point():
    try:
        cli()
    except BotoCoreError as err:
        click.echo(f"Error: {err}", err=True)
        return 1
    except ClientError as err:
        getLogger().error(err.response["Error"])
        click.echo(
            ("Error: {message}").format(message=err.response["Error"]["Message"]),
            err=True,
        )
        return 1


@click.group()
@click.version_option(__version__, prog_name="holy-cli")
def cli() -> None:
    pass


cli.add_command(server)
cli.add_command(teardown)
cli.add_command(update)

src/holy_cli/cli/__init__.py
holy-cli/cli
import click

from holy_cli.cloud.aws.actions import AWSActions
from holy_cli.log import setLoggerToStream
from holy_cli.util import version_up_to_date


@click.command()
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def teardown(**kwargs) -> None:
    """Remove all holy infrastructure"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    click.confirm(
        "Are you sure you want to remove all holy infrastructure?", abort=True
    )

    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.teardown()

    click.echo("All holy infrastructure removed")


@click.command()
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def update(**kwargs) -> None:
    """Check for the latest version"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    if version_up_to_date():
        click.echo("Version is up to date")
    else:
        click.echo(
            "A new version is available, please run: pip install --upgrade holy-cli"
        )

src/holy_cli/cli/global_commands.py
holy-cli/cli
import click
from tabulate import tabulate

from holy_cli.cloud.aws import AWS_ARCHITECTURE_VALUES, AWS_OS_USER_MAPPING
from holy_cli.cloud.aws.actions import AWSActions
from holy_cli.cloud.options import CreateServerOptions, ServerDTO
from holy_cli.exceptions import AbortError
from holy_cli.log import setLoggerToStream


@click.group()
def server() -> None:
    """Manage servers"""
    pass


@server.command(name="list")
@click.option(
    "-r",
    "--running",
    help="Only show servers that are running",
    default=False,
    is_flag=True,
    show_default=True,
)
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def list_cmd(**kwargs) -> None:
    """List all servers"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    servers = actions.list_servers(kwargs["running"])

    click.echo(tabulate(servers, headers="keys", tablefmt="simple_grid"))


@server.command()
@click.argument("name")
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def info(**kwargs) -> None:
    """View info about a server"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    info = actions.get_server_info(server)
    table = list(map(list, info.items()))

    click.echo(tabulate(table, tablefmt="simple_grid"))


@server.command(short_help="SSH into a server")
@click.argument("name")
@click.option("-u", "--username", help="Username to SSH in with")
@click.option(
    "-s",
    "--save",
    help="Save details to SSH config file (~/.ssh/config)",
    default=False,
    is_flag=True,
    show_default=True,
)
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def ssh(**kwargs) -> None:
    """
    SSH into a server. Examples:

    # SSH straight in:

    holy server ssh my_server

    # Save SSH config for use later (VS Code, ssh command etc):

    holy server ssh my_server --save

    # SSH in with a particular username:

    holy server ssh my_server --username=root
    """
    if kwargs.get("verbose"):
        setLoggerToStream()

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.ssh_into_server(server, kwargs.get("username"), kwargs["save"])

    if kwargs["save"]:
        click.echo("SSH details written to ~/.ssh/config")


@server.command()
@click.argument("name")
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def start(**kwargs) -> None:
    """Start a server"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.start_server(server)


@server.command()
@click.argument("name")
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def stop(**kwargs) -> None:
    """Stop a server"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.stop_server(server)


@server.command()
@click.argument("name")
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def delete(**kwargs) -> None:
    """Delete a server"""
    if kwargs.get("verbose"):
        setLoggerToStream()

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.delete_server(server)


@server.command(short_help="Manage server ports")
@click.argument("name")
@click.option("--action", help="The action to take (open or close)", required=True)
@click.option("--port", help="The port number", required=True, type=int)
@click.option("--ip", help="IP address to restrict port access to")
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def port(**kwargs) -> None:
    """
    Manage inbound server ports. Examples:

    # Open port 80 to the world:

    holy server port my_server --action=open --port=80

    # Close port 80 to the world:

    holy server port my_server --action=close --port=80

    # Open port 80 to a specific IP address:

    holy server port my_server --action=open --port=80 --ip=1.2.3.4
    """
    if kwargs.get("verbose"):
        setLoggerToStream()

    valid_actions = ("open", "close")

    if kwargs["action"] not in valid_actions:
        raise AbortError(
            "Invalid action value, must be one of: " + ", ".join(valid_actions)
        )

    server = ServerDTO(kwargs["name"])
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    actions.change_port(server, kwargs["port"], kwargs["action"], kwargs.get("ip"))

    click.echo(
        f"Port {kwargs['port']} has been {'opened' if kwargs['action'] == 'open' else 'closed'} to {kwargs.get('ip') or 'the world'}"
    )


@server.command(short_help="Create a new server")
@click.argument("name", required=False)
@click.option(
    "--os",
    help="Operating system: " + ", ".join(AWS_OS_USER_MAPPING.keys()),
    default="amazon-linux",
    show_default=True,
)
@click.option(
    "--architecture",
    help="CPU architecture: " + ", ".join(AWS_ARCHITECTURE_VALUES),
    default="x86_64",
    show_default=True,
)
@click.option(
    "--image-id",
    help="Amazon machine image ID to use (overrides OS and architecture options)",
)
@click.option(
    "--type",
    help="Instance type to use",
    default="t2.micro",
    show_default=True,
    required=True,
)
@click.option(
    "--disk-size",
    help="Disk size in GB",
    type=int,
    default=8,
    show_default=True,
    required=True,
)
@click.option(
    "--ports",
    help="Port numbers to open (comma seperated list)",
    default="22,80,443",
    show_default=True,
)
@click.option(
    "--actions",
    help="Allowed actions the server can make when calling other AWS services (comma seperated list e.g. s3:*,logs:GetLogEvents)",
)
@click.option(
    "--script",
    help="Path to a script file to run on the server upon creation",
    type=click.Path(exists=True),
)
@click.option(
    "--iam-profile",
    help="IAM instance profile name or ARN to use (overrides actions option)",
)
@click.option(
    "--subnet-id",
    help="A specific subnet ID to launch in",
)
@click.option("--region", help="AWS region to use")
@click.option("--profile", help="AWS profile to use")
@click.option("-v", "--verbose", help="Show verbose output", count=True)
def create(**kwargs) -> None:
    """
    Create a new server. Examples:

    # No options, create with a random name using amazon-linux OS on a micro instance (within the free tier):

    holy server create

    # Create with a specific name, using ubuntu on a large compute optimized instance with 30GB disk space:

    holy server create my_server --os=ubuntu:22 --type=c4.large --disk-size=30

    # Create with a specific Amazon machine image:

    holy server create my_server --image-id="ami-00d5053dee71cee04"

    # Create and open ports 22 and 3000 to the world:

    holy server create my_server --ports="22,3000"

    # Create and allow the server to list all S3 buckets:

    holy server create my_server --actions="s3:ListAllMyBuckets"

    # Create and run a script once launched:

    holy server create my_server --script=/path/to/install_software.sh
    """
    if kwargs.get("verbose"):
        setLoggerToStream()

    options = CreateServerOptions.load_from_cli(**kwargs)
    actions = AWSActions.load_from_cli(kwargs.get("region"), kwargs.get("profile"))
    instance = actions.create_server(options)
    ssh_cmd = f"holy server ssh {options.name}"

    if kwargs.get("region"):
        ssh_cmd += f" --region={kwargs['region']}"

    if kwargs.get("profile"):
        ssh_cmd += f" --profile={kwargs['profile']}"

    click.echo(
        f"Your server is ready:\n\nIP: {instance.public_ip_address}\nDNS: {instance.public_dns_name}\n\nTo connect run: {ssh_cmd}"
    )

src/holy_cli/cli/server_commands.py
holy-cli/cli
AWS_TAG_KEY = "holy-cli"
AWS_TAG_VALUE = "1"

AWS_OS_USER_MAPPING = {
    "amazon-linux": "ec2-user",
    "ubuntu:22": "ubuntu",
    "ubuntu:20": "ubuntu",
    "ubuntu:18": "ubuntu",
    "ubuntu:16": "ubuntu",
    "rhel:9": "ec2-user",
    "rhel:8": "ec2-user",
    "rhel:7": "ec2-user",
}

AWS_ARCHITECTURE_VALUES = ("x86_64", "arm64")

src/holy_cli/cloud/aws/__init__.py
holy-cli/cli
from __future__ import annotations

import time
from typing import List, Optional

from botocore.exceptions import ClientError
from mypy_boto3_ec2.service_resource import Instance
from yaspin import yaspin

from holy_cli.config import Config, GlobalConfig
from holy_cli.exceptions import AbortError
from holy_cli.log import getLogger

from ..options import CreateServerOptions, ServerDTO
from . import AWS_OS_USER_MAPPING
from .iam import IAMWrapper
from .image import ImageWrapper
from .instance import InstanceWrapper
from .key_pair import KeyPairWrapper
from .security_group import SecurityGroupWrapper
from .ssh import SSHWrapper
from .vpc import VPCWrapper


class AWSActions:
    def __init__(self, config: Config) -> None:
        self.config = config
        self.log = getLogger()
        self.vpc = VPCWrapper(self.config)
        self.key_pair = KeyPairWrapper(self.config)
        self.image = ImageWrapper(self.config)
        self.security_group = SecurityGroupWrapper(self.config)
        self.iam = IAMWrapper(self.config)
        self.instance = InstanceWrapper(self.config)

    def teardown(self) -> None:
        def teardown_retry(attempts: int) -> None:
            self.log.debug(f"Teardown attempt #{attempts}")

            try:
                self.security_group.teardown()
                self.key_pair.teardown()
                self.iam.teardown()
                self.vpc.teardown()
            except ClientError as err:
                if err.response["Error"]["Code"] == "DependencyViolation":
                    if attempts < 10:
                        time.sleep(5)
                        teardown_retry(attempts + 1)
                        return
                raise

        with yaspin(text="Removing infrastructure", color="yellow") as spinner:
            try:
                self.instance.teardown()
                teardown_retry(0)

                spinner.ok("✅ ")
            except:
                spinner.fail("💥 ")
                raise

    def create_server(self, options: CreateServerOptions) -> Instance:
        self.log.info(f"Creating server {options.name} with ID {options.id}")

        with yaspin(text=f"Creating server {options.name}", color="yellow") as spinner:
            try:
                if self.instance.exists(options.id):
                    raise AbortError(
                        "Server already exists, please choose a different name"
                    )

                # Use provided subnet / VPC
                if options.subnet_id:
                    subnet = self.vpc.get_subnet_by_id(options.subnet_id)

                    if subnet is None:
                        raise AbortError("Subnet not found")

                    vpc = subnet.vpc
                else:
                    # Create or use holy VPC
                    vpc = self.vpc.get_vpc()

                    if vpc is None:
                        vpc = self.vpc.create()
                        spinner.write("> Created VPC")

                    subnet = list(vpc.subnets.all())[0]

                self.log.info(f"VPC ID: {vpc.id}")
                self.log.info(f"Subnet ID: {subnet.id}")

                # Create a new key pair
                key_pair = self.key_pair.create(options.id)
                self.log.info(f"Key pair name: {key_pair.name}")
                spinner.write("> Created key pair")

                # Use the specified AMI image or find one based on the OS and architecture
                if options.image_id:
                    image = self.image.get_image_by_id(options.image_id)

                    if image is None:
                        raise AbortError("Image not found")
                else:
                    image = self.image.find_image_choices(
                        options.os, options.architecture
                    )
                    self.log.info(f"Image ID: {image.id}")
                    spinner.write("> Found AMI image")

                # Create a new security group
                sg = self.security_group.create(
                    vpc.id, options.id, options.name, options.ports
                )
                self.log.info(f"Security group ID: {sg.id}")
                spinner.write("> Created security group")

                # Create a new IAM role and instance profile
                iam_profile_for_actions = None

                if options.actions and not options.iam_profile:
                    iam_profile_for_actions = self.iam.create_instance_profile(
                        options.id, options.actions
                    )
                    self.log.info(f"IAM profile ARN: {iam_profile_for_actions.arn}")
                    spinner.write("> Created IAM role")

                try:
                    instance = self.instance.create(
                        server_id=options.id,
                        server_name=options.name,
                        os=options.os,
                        subnet_id=subnet.id,
                        image_id=image.id,
                        root_device_name=image.root_device_name,
                        instance_type=options.type,
                        key_pair_name=key_pair.name,
                        security_group_id=sg.id,
                        disk_size=options.disk_size,
                        script_file=options.script_file,
                        iam_profile=options.iam_profile,
                    )

                    self.log.info(f"Instance ID: {instance.id}")
                except ClientError:
                    # Remove anything created at this stage so not to cause name conflicts
                    key_pair.delete()
                    self.key_pair.delete_key_file(key_pair.name)
                    sg.delete()

                    if iam_profile_for_actions:
                        self.iam.delete(options.id)

                    raise

                spinner.write("> Created instance")
                spinner.write("> Waiting for instance to start...")
                instance.wait_until_running()

                # Attach the IAM instance profile once running because it takes several seconds for the permissions to propagate
                if iam_profile_for_actions:
                    self.instance.associate_iam_instance_profile(
                        instance.id, iam_profile_for_actions.arn
                    )
                    spinner.write("> Attached IAM role")

                # Reload the instance data so that we can get the public IP and DNS
                instance.reload()

                spinner.ok("✅ ")
            except:
                spinner.fail("💥 ")
                raise

        return instance

    def get_server_info(self, server: ServerDTO) -> dict:
        instance = self.instance.get_by_id(server.id)
        name = self.instance.get_tag_value(instance.tags, "Name")
        os = self.instance.get_tag_value(instance.tags, "holy-cli:os")

        if instance.state["Name"] != "terminated":
            _, key_file_path = self.key_pair.get_name_and_path(server.id)
        else:
            key_file_path = "-"

        if os in AWS_OS_USER_MAPPING:
            username = AWS_OS_USER_MAPPING[os]
        else:
            username = "-"

        volume_size = "-"
        ports = []

        if len(instance.block_device_mappings) > 0:
            volume_id = instance.block_device_mappings[0]["Ebs"]["VolumeId"]
            volume = self.instance.get_volume(volume_id)

            if volume:
                volume_size = f"{volume.size}GB"

        if len(instance.security_groups) > 0:
            sg = self.security_group.get_by_server_id(server.id)

            if sg:
                ports = [perm["FromPort"] for perm in list(sg.ip_permissions)]
                ports.sort()

        return {
            "AWS ID": instance.id,
            "Holy ID": server.id,
            "Name": name,
            "State": instance.state["Name"],
            "Availability Zone": instance.placement["AvailabilityZone"],
            "OS": os or "-",
            "Architecture": instance.architecture,
            "Type": instance.instance_type,
            "Disk Size": volume_size,
            "Open Ports": ", ".join(map(str, ports)),
            "Public IP": instance.public_ip_address or "-",
            "Private IP": instance.private_ip_address or "-",
            "DNS": instance.public_dns_name or "-",
            "SSH Username": username,
            "SSH Key": key_file_path,
        }

    def list_servers(self, only_running: bool) -> List[dict]:
        results = []
        instances = self.instance.get_all()

        for instance in instances:
            if only_running and instance.state["Name"] != "running":
                continue

            name = self.instance.get_tag_value(instance.tags, "Name")
            os = self.instance.get_tag_value(instance.tags, "holy-cli:os")

            results.append(
                {
                    "Name": name,
                    "State": instance.state["Name"],
                    "OS": os or "-",
                    "Type": instance.instance_type,
                    "IP": instance.public_ip_address or "-",
                    "DNS": instance.public_dns_name or "-",
                }
            )

        if len(results) == 0:
            results.append(
                {
                    "Name": "No existing servers",
                }
            )

        return results

    def ssh_into_server(
        self, server: ServerDTO, username: Optional[str], save: bool
    ) -> None:
        instance = self.instance.get_by_id(server.id)
        _, key_file_path = self.key_pair.get_name_and_path(server.id)

        ssh = SSHWrapper(self.config)

        if save:
            ssh.save_to_file(instance, key_file_path, username)
        else:
            ssh.ssh_into_instance(instance, key_file_path, username)

    def start_server(self, server: ServerDTO) -> None:
        with yaspin(text=f"Starting server {server.name}", color="yellow") as spinner:
            try:
                instance = self.instance.get_by_id(server.id)
                instance.start()
                instance.wait_until_running()

                spinner.ok("✅ ")
            except:
                spinner.fail("💥 ")
                raise

    def stop_server(self, server: ServerDTO) -> None:
        with yaspin(text=f"Stopping server {server.name}", color="yellow") as spinner:
            try:
                instance = self.instance.get_by_id(server.id)
                instance.stop()
                instance.wait_until_stopped()

                spinner.ok("✅ ")
            except:
                spinner.fail("💥 ")
                raise

    def delete_server(self, server: ServerDTO) -> None:
        with yaspin(text=f"Deleting server {server.name}", color="yellow") as spinner:
            try:
                instance = self.instance.get_by_id(server.id)
                instance.terminate()
                instance.wait_until_terminated()
                spinner.write("> Deleted instance")

                key_pair = self.key_pair.get_by_server_id(server.id)

                if key_pair is not None:
                    key_pair.delete()
                    self.key_pair.delete_key_file(key_pair.name)
                    spinner.write("> Deleted key pair")

                sg = self.security_group.get_by_server_id(server.id)

                if sg is not None:
                    sg.delete()
                    spinner.write("> Deleted security group")

                if (
                    instance.iam_instance_profile is not None
                    and "holy-role" in instance.iam_instance_profile["Arn"]
                ):
                    self.iam.delete(server.id)
                    spinner.write("> Deleted IAM role")

                spinner.ok("✅ ")
            except:
                spinner.fail("💥 ")
                raise

    def change_port(
        self, server: ServerDTO, port: int, action: str, ip_source: Optional[str]
    ) -> None:
        instance = self.instance.get_by_id(server.id)
        self.security_group.change_port(server.id, port, action, ip_source)

    @classmethod
    def load_from_cli(cls, region: Optional[str], profile: Optional[str]) -> AWSActions:
        global_config = GlobalConfig()
        config = Config(global_config, region, profile)

        return cls(config)

src/holy_cli/cloud/aws/actions.py
holy-cli/cli
from typing import List, Optional, Sequence

from boto3 import Session
from mypy_boto3_ec2 import EC2ServiceResource
from mypy_boto3_ec2.literals import ResourceTypeType
from mypy_boto3_ec2.type_defs import TagSpecificationTypeDef, TagTypeDef

from holy_cli.config import Config
from holy_cli.log import getLogger

from . import AWS_OS_USER_MAPPING, AWS_TAG_KEY, AWS_TAG_VALUE


class BaseWrapper:
    def __init__(self, config: Config) -> None:
        self.config = config
        self.log = getLogger()
        self.ec2 = self.init_ec2()

    def init_ec2(self) -> EC2ServiceResource:
        session = self.init_boto3_session()
        return session.resource("ec2")

    def init_boto3_session(self) -> Session:
        return Session(
            region_name=self.config.aws_region, profile_name=self.config.aws_profile
        )

    def get_tags_for_resource(
        self,
        resource_type: ResourceTypeType,
        resource_name: Optional[str] = None,
        additional_tags: Optional[dict] = None,
    ) -> Sequence[TagSpecificationTypeDef]:
        return [
            {
                "ResourceType": resource_type,
                "Tags": self.get_tags(resource_name, additional_tags),
            }
        ]

    def get_tags(
        self,
        resource_name: Optional[str] = None,
        additional_tags: Optional[dict] = None,
    ) -> Sequence[TagTypeDef]:
        tags: Sequence[TagTypeDef] = [{"Key": AWS_TAG_KEY, "Value": AWS_TAG_VALUE}]

        if resource_name is not None:
            name_tag: TagTypeDef = {"Key": "Name", "Value": resource_name}
            tags.append(name_tag)

        if additional_tags is not None:
            for key in additional_tags.keys():
                additional: TagTypeDef = {
                    "Key": key,
                    "Value": additional_tags[key],
                }
                tags.append(additional)

        return tags

    def get_tag_value(self, tags: List[TagTypeDef], tag_key: str) -> Optional[str]:
        tags = list(tags)
        return next((tag["Value"] for tag in tags if tag["Key"] == tag_key), None)

src/holy_cli/cloud/aws/base.py
holy-cli/cli
import json

from mypy_boto3_iam.client import IAMClient
from mypy_boto3_iam.service_resource import IAMServiceResource, InstanceProfile

from holy_cli.config import Config

from .base import BaseWrapper

POLICY_PATH_PREFIX = "/holy/"
POLICY_INLINE_NAME = "holy-custom-policy"


class IAMWrapper(BaseWrapper):
    """Encapsulates Amazon IAM actions."""

    def __init__(self, config: Config) -> None:
        super().__init__(config)
        session = self.init_boto3_session()
        self.iam: IAMServiceResource = session.resource("iam")

    def create_instance_profile(self, server_id: str, actions: str) -> InstanceProfile:
        role_name = f"holy-role-{server_id}"
        tags = self.get_tags(None, {"holy-cli:server": server_id})

        role = self.iam.create_role(
            RoleName=role_name,
            Path=POLICY_PATH_PREFIX,
            AssumeRolePolicyDocument=self._get_trust_ec2_policy(),
            Tags=tags,  # type: ignore
        )

        role_policy = self.iam.RolePolicy(role.name, POLICY_INLINE_NAME)
        role_policy.put(PolicyDocument=self._get_custom_policy(actions))

        profile = self.iam.create_instance_profile(
            InstanceProfileName=role.name, Tags=tags  # type: ignore
        )
        profile.add_role(RoleName=role.name)

        return profile

    def _get_trust_ec2_policy(self) -> str:
        return json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {"Service": "ec2.amazonaws.com"},
                        "Action": "sts:AssumeRole",
                    }
                ],
            }
        )

    def _get_custom_policy(self, actions: str) -> str:
        actions_list = [action.strip() for action in actions.split(",")]

        return json.dumps(
            {
                "Version": "2012-10-17",
                "Statement": [
                    {"Effect": "Allow", "Action": actions_list, "Resource": "*"}
                ],
            }
        )

    def teardown(self) -> None:
        roles = list(self.iam.roles.filter(PathPrefix=POLICY_PATH_PREFIX))

        for role in roles:
            for profile in list(role.instance_profiles.all()):
                profile.remove_role(RoleName=role.name)
                profile.delete()

            for policy in list(role.policies.all()):
                policy.delete()

            role.delete()

    def delete(self, server_id: str) -> None:
        session = self.init_boto3_session()
        client: IAMClient = session.client("iam")
        role_name = f"holy-role-{server_id}"

        client.remove_role_from_instance_profile(
            InstanceProfileName=role_name, RoleName=role_name
        )
        client.delete_instance_profile(InstanceProfileName=role_name)
        client.delete_role_policy(RoleName=role_name, PolicyName=POLICY_INLINE_NAME)
        client.delete_role(RoleName=role_name)

src/holy_cli/cloud/aws/iam.py
holy-cli/cli
from datetime import datetime
from typing import Optional

from mypy_boto3_ec2.service_resource import Image
from mypy_boto3_ssm.client import SSMClient

from holy_cli.config import Config
from holy_cli.exceptions import AbortError

from .base import BaseWrapper


class ImageWrapper(BaseWrapper):
    """Encapsulates Amazon EC2 AMI image actions."""

    def __init__(self, config: Config) -> None:
        super().__init__(config)
        session = self.init_boto3_session()
        self.ssm: SSMClient = session.client("ssm")

    def find_image_choices(self, os: str, architecture: str) -> Image:
        if os == "amazon-linux":
            return self._find_amazon_linux_image(architecture)

        if os.startswith("ubuntu"):
            return self._find_ubuntu_image(os, architecture)

        if os.startswith("rhel"):
            return self._find_redhat_image(os, architecture)

        raise AbortError("OS not recognised")

    def get_image_by_id(self, image_id: str) -> Optional[Image]:
        results = list(self.ec2.images.filter(ImageIds=[image_id]))

        if len(results) > 0:
            return results[0]

    def _find_amazon_linux_image(self, architecture: str) -> Image:
        param_path = "/aws/service/ami-amazon-linux-latest"
        ami_paginator = self.ssm.get_paginator("get_parameters_by_path")
        ami_options = []

        self.log.debug(f"Looking up SSM param path {param_path}")
        for page in ami_paginator.paginate(Path=param_path):
            ami_options += page["Parameters"]

        image_ids = [
            opt["Value"]
            for opt in ami_options
            if "default" in opt["Name"] and "minimal" not in opt["Name"]
        ]

        images = list(
            self.ec2.images.filter(
                ImageIds=image_ids,
                Filters=[{"Name": "architecture", "Values": [architecture]}],
            )
        )
        self.log.debug(f"Found {len(images)} amazon linux images")

        if len(images) == 0:
            raise AbortError("Could not find default Amazon Linux image")

        image = images[0]
        self.log.debug(f"Chose {image.name} - {image.description}")

        return image

    def _find_ubuntu_image(self, os: str, architecture: str) -> Image:
        if architecture == "x86_64":
            architecture = "amd64"

        version = os.split(":")[1]
        version = f"{version}.04"  # todo: update when minor version changes

        param_name = f"/aws/service/canonical/ubuntu/server/{version}/stable/current/{architecture}/hvm/ebs-gp2/ami-id"
        self.log.debug(f"Looking up SSM param name {param_name}")

        try:
            result = self.ssm.get_parameter(Name=param_name)
        except:
            raise AbortError("Could not find default Ubuntu image")

        image = self.get_image_by_id(result["Parameter"]["Value"])

        if image is None:
            raise AbortError("Could not find default Ubuntu image")

        self.log.debug(f"Chose {image.name} - {image.description}")

        return image

    def _find_redhat_image(self, os: str, architecture: str) -> Image:
        version = os.split(":")[1]

        images = list(
            self.ec2.images.filter(
                Owners=["309956199498"],
                Filters=[
                    {"Name": "name", "Values": [f"RHEL-{version}.*"]},
                    {"Name": "virtualization-type", "Values": ["hvm"]},
                    {"Name": "architecture", "Values": [architecture]},
                ],
            )
        )
        self.log.debug(f"Found {len(images)} RHEL images")

        if len(images) == 0:
            raise AbortError("Could not find default Red Hat image")

        images = [
            image
            for image in sorted(
                images,
                key=lambda image: datetime.strptime(
                    image.creation_date, "%Y-%m-%dT%H:%M:%S.%f%z"
                ),
                reverse=True,
            )
            if "BETA" not in image.name
        ]

        if len(images) == 0:
            raise AbortError("Could not find default Red Hat image")

        image = images[0]
        self.log.debug(f"Chose {image.name} - {image.description}")

        return image

src/holy_cli/cloud/aws/image.py
holy-cli/cli
from typing import List, Optional

from mypy_boto3_ec2.client import EC2Client
from mypy_boto3_ec2.service_resource import Instance, Volume
from mypy_boto3_ec2.type_defs import IamInstanceProfileSpecificationTypeDef

from holy_cli.exceptions import AbortError

from .base import AWS_TAG_KEY, AWS_TAG_VALUE, BaseWrapper


class InstanceWrapper(BaseWrapper):
    """Encapsulates Amazon EC2 instance actions."""

    def create(
        self,
        server_id: str,
        server_name: str,
        os: Optional[str],
        subnet_id: str,
        image_id: str,
        root_device_name: str,
        instance_type: str,
        key_pair_name: str,
        security_group_id: str,
        disk_size: int,
        script_file: Optional[str],
        iam_profile: Optional[str],
    ) -> Instance:
        user_data = self._get_script_file(script_file) if script_file else ""
        additional_tags = {"holy-cli:server": server_id}

        if os is not None:
            additional_tags["holy-cli:os"] = os

        iam_instance_profile: IamInstanceProfileSpecificationTypeDef = {}

        if iam_profile is not None:
            if iam_profile.startswith("arn:"):
                iam_instance_profile["Arn"] = iam_profile
            else:
                iam_instance_profile["Name"] = iam_profile

        instance = self.ec2.create_instances(
            ImageId=image_id,
            InstanceType=instance_type,  # type: ignore
            KeyName=key_pair_name,
            MinCount=1,
            MaxCount=1,
            IamInstanceProfile=iam_instance_profile,
            TagSpecifications=self.get_tags_for_resource(
                "instance", server_name, additional_tags
            ),
            BlockDeviceMappings=[
                {
                    "DeviceName": root_device_name,
                    "Ebs": {"DeleteOnTermination": True, "VolumeSize": disk_size},
                }
            ],
            NetworkInterfaces=[
                {
                    "SubnetId": subnet_id,
                    "DeviceIndex": 0,
                    "AssociatePublicIpAddress": True,
                    "Groups": [security_group_id],
                }
            ],
            UserData=user_data,
        )[0]

        return instance

    def get_by_id(self, server_id: str) -> Instance:
        results = list(
            self.ec2.instances.filter(
                Filters=[{"Name": "tag:holy-cli:server", "Values": [server_id]}]
            )
        )

        if len(results) > 0:
            return results[0]

        raise AbortError("Could not find server")

    def exists(self, server_id: str) -> bool:
        try:
            instance = self.get_by_id(server_id)

            return instance.state["Name"] != "terminated"
        except AbortError:
            return False

    def get_all(self) -> List[Instance]:
        return list(
            self.ec2.instances.filter(
                Filters=[{"Name": f"tag:{AWS_TAG_KEY}", "Values": [AWS_TAG_VALUE]}]
            )
        )

    def get_volume(self, volume_id: str) -> Optional[Volume]:
        results = list(self.ec2.volumes.filter(VolumeIds=[volume_id]))

        if len(results) > 0:
            return results[0]

    def associate_iam_instance_profile(
        self, instance_id: str, profile_arn: str
    ) -> None:
        session = self.init_boto3_session()
        client: EC2Client = session.client("ec2")
        client.associate_iam_instance_profile(
            InstanceId=instance_id, IamInstanceProfile={"Arn": profile_arn}
        )

    def teardown(self) -> None:
        instances = self.get_all()

        for instance in instances:
            self.log.debug(f"Deleting instance {instance.id}")
            instance.terminate()

    def _get_script_file(self, file: str) -> str:
        with open(file, "r") as f:
            contents = f.read()

        contents = contents.strip()

        if contents[0:2] != "#!":
            contents = "#!/bin/bash\n" + contents

        return contents

src/holy_cli/cloud/aws/instance.py
holy-cli/cli
import os
from typing import Optional, Tuple

from mypy_boto3_ec2.service_resource import KeyPair, KeyPairInfo

from .base import AWS_TAG_KEY, AWS_TAG_VALUE, BaseWrapper


class KeyPairWrapper(BaseWrapper):
    """Encapsulates Amazon EC2 key pair actions."""

    def teardown(self) -> None:
        results = list(
            self.ec2.key_pairs.filter(
                Filters=[{"Name": f"tag:{AWS_TAG_KEY}", "Values": [AWS_TAG_VALUE]}]
            )
        )

        for key_pair in results:
            self.log.debug(f"Deleting key pair {key_pair.name}")
            key_pair.delete()
            self.delete_key_file(key_pair.name)

    def get_name_and_path(self, server_id: str) -> Tuple[str, str]:
        key_name = f"holy-kp-{server_id}"
        key_file_path = self._get_path(key_name)

        return key_name, key_file_path

    def create(self, server_id: str) -> KeyPair:
        key_name, key_file_path = self.get_name_and_path(server_id)

        self.log.debug(f"Creating key pair {key_name}")
        key_pair = self.ec2.create_key_pair(
            KeyName=key_name,
            TagSpecifications=self.get_tags_for_resource(
                "key-pair", None, {"holy-cli:server": server_id}
            ),
        )

        with open(key_file_path, "w") as key_file:
            key_file.write(key_pair.key_material)

        try:
            os.chmod(key_file_path, 0o400)
        except:
            self.log.warning(f"Could not CHMOD key pair file - {key_file_path}")

        return key_pair

    def get_by_server_id(self, server_id: str) -> Optional[KeyPairInfo]:
        results = list(
            self.ec2.key_pairs.filter(
                Filters=[{"Name": "tag:holy-cli:server", "Values": [server_id]}]
            )
        )

        if len(results) > 0:
            return results[0]

    def delete_key_file(self, key_name: str) -> None:
        key_file_path = self._get_path(key_name)

        if os.path.exists(key_file_path):
            os.remove(key_file_path)

    def _get_path(self, key_name: str) -> str:
        return os.path.join(self.config.global_config.keys_dir, f"{key_name}.pem")

src/holy_cli/cloud/aws/key_pair.py
holy-cli/cli
from typing import Optional, Sequence

from mypy_boto3_ec2.service_resource import SecurityGroup
from mypy_boto3_ec2.type_defs import IpPermissionTypeDef

from holy_cli.exceptions import AbortError

from .base import AWS_TAG_KEY, AWS_TAG_VALUE, BaseWrapper


class SecurityGroupWrapper(BaseWrapper):
    """Encapsulates Amazon Elastic Compute Cloud (Amazon EC2) security group actions."""

    def teardown(self) -> None:
        results = list(
            self.ec2.security_groups.filter(
                Filters=[{"Name": f"tag:{AWS_TAG_KEY}", "Values": [AWS_TAG_VALUE]}]
            )
        )

        for sg in results:
            self.log.debug(f"Deleting security group {sg.group_name}")
            sg.delete()

    def create(
        self, vpc_id: str, server_id: str, server_name: str, ports: Optional[str]
    ) -> SecurityGroup:
        group_name = f"holy-sg-{server_id}"
        self.log.debug(f"Creating security group {group_name}")

        security_group = self.ec2.create_security_group(
            VpcId=vpc_id,
            GroupName=group_name,
            Description=f"Security group for {server_name} created with holy-cli",
            TagSpecifications=self.get_tags_for_resource(
                "security-group", group_name, {"holy-cli:server": server_id}
            ),
        )

        ip_permissions: Sequence[IpPermissionTypeDef] = []

        if ports:
            for port in ports.split(","):
                port = int(port.strip())

                if port == 0:
                    continue

                ip_permissions.append(
                    {
                        "IpProtocol": "tcp",
                        "FromPort": port,
                        "ToPort": port,
                        "IpRanges": [{"CidrIp": "0.0.0.0/0"}],
                    }
                )

        if len(ip_permissions) > 0:
            security_group.authorize_ingress(IpPermissions=ip_permissions)

        return security_group

    def get_by_server_id(self, server_id: str) -> Optional[SecurityGroup]:
        results = list(
            self.ec2.security_groups.filter(
                Filters=[{"Name": "tag:holy-cli:server", "Values": [server_id]}]
            )
        )

        if len(results) > 0:
            return results[0]

    def change_port(
        self, server_id: str, port: int, action: str, ip_source: Optional[str]
    ) -> None:
        sg = self.get_by_server_id(server_id)

        if sg is None:
            raise AbortError("Could not find security group")

        if ip_source is None:
            ip_source = "0.0.0.0/0"
        elif "/" not in ip_source:
            ip_source = f"{ip_source}/32"

        for perm in list(sg.ip_permissions):
            if perm["FromPort"] == port:
                for ip_range in perm["IpRanges"]:
                    if ip_range["CidrIp"] == ip_source:
                        sg.revoke_ingress(
                            CidrIp=ip_range["CidrIp"],
                            FromPort=perm["FromPort"],
                            ToPort=perm["ToPort"],
                            IpProtocol="tcp",
                        )

        if action == "open":
            sg.authorize_ingress(
                CidrIp=ip_source, FromPort=port, ToPort=port, IpProtocol="tcp"
            )

src/holy_cli/cloud/aws/security_group.py
holy-cli/cli
import os
import platform
import subprocess
import sys
from typing import Optional

from mypy_boto3_ec2.service_resource import Instance

from holy_cli.exceptions import AbortError

from .base import AWS_OS_USER_MAPPING, BaseWrapper


class SSHWrapper(BaseWrapper):
    def ssh_into_instance(
        self, instance: Instance, key_file_path: str, username: Optional[str]
    ) -> None:
        if instance.state["Name"] != "running":
            raise AbortError("Server is not running")

        if not os.path.exists(key_file_path):
            raise AbortError(
                f"Key file is missing, you may need to re-create the server"
            )

        user = username or self._get_instance_username(instance)
        host = instance.public_ip_address

        try:
            result = self._run_ssh_cmd(key_file_path, user, host)
        except FileNotFoundError:
            raise AbortError(
                f"Could not find SSH tool, please try run manually:\n\n/path/to/ssh -i {key_file_path} {user}@{host}"
            )

        if result.returncode:
            err = result.stderr.decode("utf-8")
            self.log.error(f"SSH error response: {err}")
            debug_cmd = f"ssh -v -i {key_file_path} {user}@{host}"

            if "Permission denied" in err:
                raise AbortError(
                    f"Could not SSH into server (permission denied), to debug run:\n\n{debug_cmd}\n\nNote: In some cases the username may not be correct, check your AMI usage instructions and pass in with the --username flag"
                )

            if "Could not resolve" in err:
                raise AbortError(
                    f"Could not SSH into server (hostname could not be resolved), to debug run:\n\n{debug_cmd}"
                )

            if "Connection refused" in err:
                raise AbortError(
                    f"Could not SSH into server (connection refused), if the server has just started try again in a minute or to debug run:\n\n{debug_cmd}"
                )

    def save_to_file(
        self, instance: Instance, key_file_path: str, username: Optional[str]
    ) -> None:
        if instance.state["Name"] != "running":
            raise AbortError("Server is not running")

        if not os.path.exists(key_file_path):
            raise AbortError(
                f"Key file is missing, you may need to re-create the server"
            )

        user = username or self._get_instance_username(instance)
        host = instance.public_ip_address
        name = self.get_tag_value(instance.tags, "Name")
        ssh_config_file_path = os.path.expanduser("~/.ssh/config")
        ssh_contents = f"\nHost {name}\n\tUser {user}\n\tHostName {host}\n\tIdentityFile {key_file_path}"

        with open(ssh_config_file_path, "a") as f:
            f.write(ssh_contents)

    def _get_instance_username(self, instance: Instance) -> str:
        os = self.get_tag_value(instance.tags, "holy-cli:os")
        user = "ec2-user"

        if os is None:
            self.log.warning("Could not find OS for instance")
        else:
            if os in AWS_OS_USER_MAPPING:
                user = AWS_OS_USER_MAPPING[os]
            else:
                self.log.warning("Could not find OS in username mapping")

        return user

    def _run_ssh_cmd(
        self, key_path: str, user: str, host: str
    ) -> subprocess.CompletedProcess:
        cmd = [
            self._get_ssh_path(),
            "-i",
            key_path,
            "-o",
            "StrictHostKeyChecking=no",
            "-o",
            "UserKnownHostsFile=/dev/null",
            "-o",
            "LogLevel=ERROR",
            f"{user}@{host}",
        ]

        self.log.debug(f"Trying SSH command: {' '.join(cmd)}")
        return subprocess.run(cmd, stderr=subprocess.PIPE)

    def _get_ssh_path(self) -> str:
        if sys.platform == "win32":
            system32 = os.path.join(
                os.environ["SystemRoot"],
                "SysNative" if platform.architecture()[0] == "32bit" else "System32",
            )
            ssh_path = os.path.join(system32, "OpenSSH\\ssh.exe")

            if not os.path.exists(ssh_path):
                ssh_path = os.path.join(os.environ["ProgramFiles"], "OpenSSH\\ssh.exe")

            if not os.path.exists(ssh_path) and os.environ.get("ProgramFiles(x86)"):
                ssh_path = os.path.join(
                    os.environ["ProgramFiles(x86)"], "OpenSSH\\ssh.exe"
                )

            return ssh_path
        else:
            if "SSH_AUTH_SOCK" not in os.environ:
                raise AbortError(
                    "SSH agent is not running, please try running: eval $(ssh-agent -s)"
                )

            return "ssh"

src/holy_cli/cloud/aws/ssh.py
holy-cli/cli
from typing import Optional

from mypy_boto3_ec2.service_resource import Subnet, Vpc

from .base import AWS_TAG_KEY, AWS_TAG_VALUE, BaseWrapper


class VPCWrapper(BaseWrapper):
    """Encapsulates Amazon EC2 VPC actions."""

    def teardown(self) -> None:
        vpc = self.get_vpc()

        if vpc is None:
            return

        for subnet in list(vpc.subnets.all()):
            self.log.debug(f"Deleting subnet {subnet.id}")
            subnet.delete()

        for ig in list(vpc.internet_gateways.all()):
            self.log.debug(f"Deleting internet gateway {ig.id}")
            ig.detach_from_vpc(VpcId=vpc.id)
            ig.delete()

        self.log.debug(f"Deleting VPC {vpc.id}")
        vpc.delete()

    def get_vpc(self) -> Optional[Vpc]:
        results = list(
            self.ec2.vpcs.filter(
                Filters=[{"Name": f"tag:{AWS_TAG_KEY}", "Values": [AWS_TAG_VALUE]}]
            )
        )

        if len(results) > 0:
            return results[0]

    def get_subnet_by_id(self, subnet_id: str) -> Optional[Subnet]:
        results = list(self.ec2.subnets.filter(SubnetIds=[subnet_id]))

        if len(results) > 0:
            return results[0]

    def create(self) -> Vpc:
        vpc_cidr_block = "10.0.0.0/16"

        # Create VPC
        vpc = self.ec2.create_vpc(
            CidrBlock=vpc_cidr_block,
            TagSpecifications=self.get_tags_for_resource("vpc", "holy-vpc"),
        )
        self.log.debug(f"VPC {vpc.id} created")

        # Create Internet Gateway
        igw = self.ec2.create_internet_gateway(
            TagSpecifications=self.get_tags_for_resource(
                "internet-gateway", "holy-internet-gateway"
            ),
        )
        self.log.debug(f"Internet Gateway {igw.id} created")

        # Attach Internet Gateway to VPC
        vpc.attach_internet_gateway(InternetGatewayId=igw.id)
        self.log.debug("Internet Gateway attached to VPC")

        # Modify VPC to enable DNS hostnames
        vpc.modify_attribute(EnableDnsHostnames={"Value": True})
        self.log.debug("Enabled DNS hostnames on VPC")

        # Create Public Subnet
        subnet = vpc.create_subnet(
            CidrBlock=vpc_cidr_block,
            TagSpecifications=self.get_tags_for_resource("subnet", "holy-subnet"),
        )
        self.log.debug(f"Public Subnet {subnet.id} created")

        # Add route to Internet Gateway
        route_tables = list(vpc.route_tables.all())
        route_tables[0].create_route(DestinationCidrBlock="0.0.0.0/0", GatewayId=igw.id)
        self.log.debug("Added route to Internet Gateway")

        # Modify Subnet to enable auto-assign public IPv4 addresses
        # https://github.com/boto/boto3/issues/276
        subnet.meta.client.modify_subnet_attribute(SubnetId=subnet.id, MapPublicIpOnLaunch={"Value": True})  # type: ignore
        self.log.debug("Public IP auto-assign enabled for Subnet")

        return vpc

src/holy_cli/cloud/aws/vpc.py
holy-cli/cli
from __future__ import annotations

from typing import Optional

from holy_cli.cloud.aws import AWS_ARCHITECTURE_VALUES, AWS_OS_USER_MAPPING
from holy_cli.exceptions import AbortError
from holy_cli.util import get_random_name, hash_server_name


class ServerDTO:
    def __init__(self, name: str) -> None:
        self.name = name.strip()
        self.id = hash_server_name(self.name.lower())


class CreateServerOptions(ServerDTO):
    def __init__(
        self,
        name: str,
        os: Optional[str],
        architecture: Optional[str],
        image_id: Optional[str],
        type: str,
        disk_size: int,
        ports: Optional[str],
        actions: Optional[str],
        script_file: Optional[str],
        iam_profile: Optional[str],
        subnet_id: Optional[str],
    ) -> None:
        super().__init__(name)
        self.os = os
        self.architecture = architecture
        self.image_id = image_id
        self.type = type
        self.disk_size = disk_size
        self.ports = ports
        self.actions = actions
        self.script_file = script_file
        self.iam_profile = iam_profile
        self.subnet_id = subnet_id

    @classmethod
    def load_from_cli(cls, **kwargs) -> CreateServerOptions:
        image_id = kwargs.get("image_id")

        if image_id is None:
            os = kwargs.get("os")
            architecture = kwargs.get("architecture")

            if os not in AWS_OS_USER_MAPPING:
                raise AbortError(
                    "Invalid OS value, must be one of: "
                    + ", ".join(AWS_OS_USER_MAPPING.keys())
                )

            if architecture not in AWS_ARCHITECTURE_VALUES:
                raise AbortError(
                    "Invalid architecture value, must be one of: "
                    + ", ".join(AWS_ARCHITECTURE_VALUES)
                )
        else:
            os = None
            architecture = None

        return cls(
            name=(kwargs.get("name") or get_random_name()),
            os=os,
            architecture=architecture,
            image_id=image_id,
            type=kwargs["type"],
            disk_size=int(kwargs["disk_size"]),
            ports=kwargs.get("ports"),
            actions=kwargs.get("actions"),
            script_file=kwargs.get("script"),
            iam_profile=kwargs.get("iam_profile"),
            subnet_id=kwargs.get("subnet_id"),
        )

src/holy_cli/cloud/options.py
holy-cli/cli
import os
from typing import Optional

from .exceptions import AbortError


class GlobalConfig:
    def __init__(self) -> None:
        self.root_dir = os.path.expanduser("~/.holy")
        self.keys_dir = os.path.join(self.root_dir, "keys")
        self._check_root_dir()

    def _check_root_dir(self):
        if not os.path.isdir(self.root_dir):
            try:
                os.mkdir(self.root_dir)
                os.mkdir(self.keys_dir)
            except:
                raise AbortError(f"Could not create directory: {self.root_dir}")


class Config:
    def __init__(
        self, global_config: GlobalConfig, region: Optional[str], profile: Optional[str]
    ) -> None:
        self.global_config = global_config
        self.aws_region = region
        self.aws_profile = profile

src/holy_cli/config.py
holy-cli/cli
from click import ClickException


class AbortError(ClickException):
    """There was an exception that requires the program to abort (will return an exit code of 1)"""

src/holy_cli/exceptions.py
holy-cli/cli
import logging

import boto3


class ColorPercentStyle(logging.PercentStyle):
    grey = "38"
    green = "32"
    yellow = "33"
    red = "31"

    def _get_color_fmt(self, color_code, bold=False):
        if bold:
            return "\x1b[" + color_code + ";1m" + self._fmt + "\x1b[0m"
        return "\x1b[" + color_code + ";20m" + self._fmt + "\x1b[0m"

    def _get_fmt(self, levelno):
        colors = {
            logging.DEBUG: self._get_color_fmt(self.grey),
            logging.INFO: self._get_color_fmt(self.green),
            logging.WARNING: self._get_color_fmt(self.yellow),
            logging.ERROR: self._get_color_fmt(self.red),
            logging.CRITICAL: self._get_color_fmt(self.red, bold=True),
        }

        return colors.get(levelno, self._get_color_fmt(self.grey))

    def _format(self, record):
        return self._get_fmt(record.levelno) % record.__dict__


class CustomFormatter(logging.Formatter):
    def formatMessage(self, record):
        return ColorPercentStyle(self._fmt).format(record)


def getLogger() -> logging.Logger:
    logger = logging.getLogger("holy-cli")

    if not logger.hasHandlers():
        logger.addHandler(boto3.NullHandler())

    return logger


def setLoggerToStream() -> None:
    logger = getLogger()
    logger.setLevel(logging.DEBUG)

    format_string = "[%(name)s %(levelname)s] %(message)s"
    formatter = CustomFormatter(fmt=format_string)

    stream = logging.StreamHandler()
    stream.setFormatter(formatter)
    logger.addHandler(stream)

    boto3.set_stream_logger("boto3.resources", logging.DEBUG, format_string)

src/holy_cli/log.py
holy-cli/cli
from .hash import hash_server_name
from .names import get_random_name
from .version_check import version_up_to_date

src/holy_cli/util/__init__.py
holy-cli/cli
import hashlib


def hash_server_name(name: str) -> str:
    return hashlib.md5(name.encode("utf-8")).hexdigest()[0:16]

src/holy_cli/util/hash.py
holy-cli/cli
# The adjectives and names were sourced from https://github.com/moby/moby
# Copyright The Moby Project (Apache License 2.0)

import random

ADJECTIVES = [
    "admiring",
    "adoring",
    "affectionate",
    "agitated",
    "amazing",
    "angry",
    "awesome",
    "beautiful",
    "blissful",
    "bold",
    "boring",
    "brave",
    "busy",
    "charming",
    "clever",
    "compassionate",
    "competent",
    "condescending",
    "confident",
    "cool",
    "cranky",
    "crazy",
    "dazzling",
    "determined",
    "distracted",
    "dreamy",
    "eager",
    "ecstatic",
    "elastic",
    "elated",
    "elegant",
    "eloquent",
    "epic",
    "exciting",
    "fervent",
    "festive",
    "flamboyant",
    "focused",
    "friendly",
    "frosty",
    "funny",
    "gallant",
    "gifted",
    "goofy",
    "gracious",
    "great",
    "happy",
    "hardcore",
    "heuristic",
    "hopeful",
    "hungry",
    "infallible",
    "inspiring",
    "intelligent",
    "interesting",
    "jolly",
    "jovial",
    "keen",
    "kind",
    "laughing",
    "loving",
    "lucid",
    "magical",
    "modest",
    "musing",
    "mystifying",
    "naughty",
    "nervous",
    "nice",
    "nifty",
    "nostalgic",
    "objective",
    "optimistic",
    "peaceful",
    "pedantic",
    "pensive",
    "practical",
    "priceless",
    "quirky",
    "quizzical",
    "recursing",
    "relaxed",
    "reverent",
    "romantic",
    "sad",
    "serene",
    "sharp",
    "silly",
    "sleepy",
    "stoic",
    "strange",
    "stupefied",
    "suspicious",
    "sweet",
    "tender",
    "thirsty",
    "trusting",
    "unruffled",
    "upbeat",
    "vibrant",
    "vigilant",
    "vigorous",
    "wizardly",
    "wonderful",
    "xenodochial",
    "youthful",
    "zealous",
    "zen",
]

NAMES = [
    "agnesi",
    "albattani",
    "allen",
    "almeida",
    "antonelli",
    "archimedes",
    "ardinghelli",
    "aryabhata",
    "austin",
    "babbage",
    "banach",
    "banzai",
    "bardeen",
    "bartik",
    "bassi",
    "beaver",
    "bell",
    "benz",
    "bhabha",
    "bhaskara",
    "black",
    "blackburn",
    "blackwell",
    "bohr",
    "booth",
    "borg",
    "bose",
    "bouman",
    "boyd",
    "brahmagupta",
    "brattain",
    "brown",
    "buck",
    "burnell",
    "cannon",
    "carson",
    "cartwright",
    "carver",
    "cerf",
    "chandrasekhar",
    "chaplygin",
    "chatelet",
    "chatterjee",
    "chaum",
    "chebyshev",
    "clarke",
    "cohen",
    "colden",
    "cori",
    "cray",
    "curie",
    "curran",
    "darwin",
    "davinci",
    "dewdney",
    "dhawan",
    "diffie",
    "dijkstra",
    "dirac",
    "driscoll",
    "dubinsky",
    "easley",
    "edison",
    "einstein",
    "elbakyan",
    "elgamal",
    "elion",
    "ellis",
    "engelbart",
    "euclid",
    "euler",
    "faraday",
    "feistel",
    "fermat",
    "fermi",
    "feynman",
    "franklin",
    "gagarin",
    "galileo",
    "galois",
    "ganguly",
    "gates",
    "gauss",
    "germain",
    "goldberg",
    "goldstine",
    "goldwasser",
    "golick",
    "goodall",
    "gould",
    "greider",
    "grothendieck",
    "haibt",
    "hamilton",
    "haslett",
    "hawking",
    "heisenberg",
    "hellman",
    "hermann",
    "herschel",
    "hertz",
    "heyrovsky",
    "hodgkin",
    "hofstadter",
    "hoover",
    "hopper",
    "hugle",
    "hypatia",
    "ishizaka",
    "jackson",
    "jang",
    "jemison",
    "jennings",
    "jepsen",
    "johnson",
    "joliot",
    "jones",
    "kalam",
    "kapitsa",
    "kare",
    "keldysh",
    "keller",
    "kepler",
    "khayyam",
    "khorana",
    "kilby",
    "kirch",
    "knuth",
    "kowalevski",
    "lalande",
    "lamarr",
    "lamport",
    "leakey",
    "leavitt",
    "lederberg",
    "lehmann",
    "lewin",
    "lichterman",
    "liskov",
    "lovelace",
    "lumiere",
    "mahavira",
    "margulis",
    "matsumoto",
    "maxwell",
    "mayer",
    "mccarthy",
    "mcclintock",
    "mclaren",
    "mclean",
    "mcnulty",
    "meitner",
    "mendel",
    "mendeleev",
    "meninsky",
    "merkle",
    "mestorf",
    "mirzakhani",
    "montalcini",
    "moore",
    "morse",
    "moser",
    "murdock",
    "napier",
    "nash",
    "neumann",
    "newton",
    "nightingale",
    "nobel",
    "noether",
    "northcutt",
    "noyce",
    "panini",
    "pare",
    "pascal",
    "pasteur",
    "payne",
    "perlman",
    "pike",
    "poincare",
    "poitras",
    "proskuriakova",
    "ptolemy",
    "raman",
    "ramanujan",
    "rhodes",
    "ride",
    "ritchie",
    "robinson",
    "roentgen",
    "rosalind",
    "rubin",
    "saha",
    "sammet",
    "sanderson",
    "satoshi",
    "shamir",
    "shannon",
    "shaw",
    "shirley",
    "shockley",
    "shtern",
    "sinoussi",
    "snyder",
    "solomon",
    "spence",
    "stonebraker",
    "sutherland",
    "swanson",
    "swartz",
    "swirles",
    "taussig",
    "tesla",
    "tharp",
    "thompson",
    "torvalds",
    "tu",
    "turing",
    "varahamihira",
    "vaughan",
    "villani",
    "visvesvaraya",
    "volhard",
    "wescoff",
    "wilbur",
    "wiles",
    "williams",
    "williamson",
    "wilson",
    "wing",
    "wozniak",
    "wright",
    "wu",
    "yalow",
    "yonath",
    "zhukovsky",
]


def get_random_name() -> str:
    return random.choice(ADJECTIVES) + "_" + random.choice(NAMES)

src/holy_cli/util/names.py
holy-cli/cli
import json
from urllib.error import HTTPError
from urllib.request import Request, urlopen

from holy_cli import __version__
from holy_cli.exceptions import AbortError
from holy_cli.log import getLogger

LATEST_RELEASE_API_URL = "https://api.github.com/repos/holy-cli/cli/releases/latest"

log = getLogger()


def version_up_to_date() -> bool:
    tag = _get_latest_tag()
    log.debug(f"Tag name is {tag}")

    return tag == f"v{__version__}"


def _get_latest_tag() -> str:
    request = Request(
        LATEST_RELEASE_API_URL,
        headers={
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        },
    )

    try:
        with urlopen(request) as response:
            body = json.loads(response.read())
            return body["tag_name"]
    except HTTPError as err:
        log.error(err)
        raise AbortError("Failed to check for latest version")

src/holy_cli/util/version_check.py
holy-cli/cli
from click.testing import CliRunner

from holy_cli import __version__
from holy_cli.cli import cli


def test_version_cmd():
    runner = CliRunner()
    result = runner.invoke(cli, ["--version"])
    assert result.exit_code == 0
    assert result.output == f"holy-cli, version {__version__}\n"


def test_server_cmd():
    runner = CliRunner()
    result = runner.invoke(cli, ["server"])
    assert result.exit_code == 0

tests/test_cli.py
holy-cli/cli