Expose - Expose a web service on localhost to public internet using AWS EC2

Read Me:

Expose - Main Module

class expose.main.Tunnel(logger: Optional[Logger] = None)

Initiates Tunnel object to spin up an EC2 instance with a pre-configured AMI which acts as a tunnel.

>>> Tunnel

Instantiates all required AWS resources.

Parameters:

logger – Bring your own logger.

init(start: bool) → None

Initializer function.

Parameters:

start – Boolean flag to indicate if its startup or shutdown.

create_key_pair() → bool

Creates a KeyPair of type RSA stored as a PEM file to use with OpenSSH.

Returns:

Boolean flag to indicate the calling function if a KeyPair was created.

Return type:

bool

get_vpc_id() → Optional[str]

Fetches the default VPC id.

Returns:

Default VPC id.

Return type:

Union[str, None]

authorize_security_group(security_group_id: str, public_ip: str) → bool

Authorizes the security group, to allow ingress and egress traffic via VPC on certain ports.

Parameters:

  • security_group_id – Takes the SecurityGroup ID as an argument.

  • public_ip – Public IP address of the ec2 instance.

See also

Ports allowed:

  • 22: Only for ec2 and host machine’s public IP with CIDR notation 32.

  • 80: HTTP port for self.

  • 443: HTTPS port for self.

Apart from the above, the SG is authorized for the port number requested to forward.

Returns:

Boolean flag to indicate the calling function whether the security group was authorized successfully.

Return type:

bool

create_security_group() → Optional[str]

Gets VPC id and creates a security group for the ec2 instance.

Warning

Deletes and re-creates the SG, in case an SG exists with the same name already.

Returns:

SecurityGroup ID

Return type:

Union[str, None]

create_ec2_instance() → Optional[Tuple[str, str]]

Creates an EC2 instance with a pre-configured AMI id.

Returns:

Instance ID, SecurityGroup ID if successful.

Return type:

Union[Tuple[str, str], None]

delete_key_pair() → bool

Deletes the KeyPair created to access the ec2 instance.

Returns:

Boolean flag to indicate the calling function if the KeyPair was deleted successfully.

Return type:

bool

disassociate_security_group(security_group_id: str, instance: Optional[object] = None, instance_id: Optional[str] = None) → bool

Disassociates an SG from the ec2 instance by assigning it to the default security group.

Parameters:

  • security_group_id – Security group ID

  • instance – Instance object.

  • instance_id – Instance ID if object is unavailable.

Returns:

Boolean flag to indicate the calling function whether the disassociation was successful.

Return type:

bool

delete_security_group(security_group_id: str) → bool

Deletes the security group.

Parameters:

security_group_id – Takes the SecurityGroup ID as an argument.

Returns:

Boolean flag to indicate the calling function whether the SecurityGroup was deleted.

Return type:

bool

terminate_ec2_instance(instance_id: Optional[str] = None, instance: Optional[object] = None) → ServiceResource

Terminates the requested instance.

Parameters:

  • instance_id – Takes instance ID as an argument.

  • instance – Takes the instance object as an optional argument.

Returns:

Boolean flag to indicate the calling function whether the instance was terminated.

Return type:

bool

start(purge: bool = False) → None

Starts the reverse proxy server using nginx to initiate port forwarding.

Parameters:

purge – Boolean flag to delete all AWS resource if initial configuration fails.

See also

Automatic purge works only during initial setup, and not during re-configuration.

References

server_copy(source: str, destination: str) → None

Reads a file in localhost and writes it within the ec2 instance using SSH connection.

Parameters:

  • source – Name of the source file in localhost.

  • destination – Name of the destination file in the server.

get_config(filename: str, server: str) → str

Loads the configuration file and returns the data.

Parameters:

  • filename – Name of the file that has to be downloaded.

  • server – Custom server names to be added in configuration files.

Returns:

Returns the data of the file as a string.

Return type:

str

config_requirements(common_name: str, custom_servers: str, san_list: List[str]) → Dict[str, str]

Tries to create a self-signed SSL certificate and loads all the required configuration into a dictionary.

Parameters:

  • common_name – DNS entry for SSL creation.

  • custom_servers – Server names to be loaded in the nginx config files.

  • san_list – Subject Alternative Names to validate the certificate for.

Returns:

Dictionary of config file name and the configuration data.

Return type:

Dict[str, str]

configure_vm(public_dns: str, public_ip: str, disposal: bool = False) → None

Configures the ec2 instance to take traffic from localhost and initiates tunneling.

Parameters:

  • public_dns – Public DNS name of the EC2 that was created.

  • public_ip – Public IP of the EC2 that was created.

  • disposal – Boolean flag to delete all AWS resources on failed configuration.

stop(instance_id: Optional[str] = None, security_group_id: Optional[str] = None, public_ip: Optional[str] = None) → None

Disables tunnelling by removing all AWS resources acquired.

Parameters:

  • instance_id – Instance that has to be terminated.

  • security_group_id – Security group that has to be removed.

  • public_ip – Public IP address to delete the A record from route53.

See also

Doesn’t require any argument, as long as the JSON dump is neither removed nor modified by hand.

References

Expose - Auxiliary

expose.models.auxiliary.write_screen(text: Any) → None

Write text on screen that can be cleared later.

Parameters:

text – Text to be written.

expose.models.auxiliary.flush_screen() → None

Flushes the screen output.

See also

Writes new set of empty strings for the size of the terminal if ran using one.

Expose - Certificates

expose.models.cert._get_serial() → bytes

Generates a serial number for the self-signed SSL.

See also

  • This function is not called, but it is here only as a just in case measure to insert serial number manually.

  • Serial Number is a unique identifier assigned by the CA which issued the certificate.

Returns:

Encoded serial number for the certificate.

Return type:

bytes

expose.models.cert._generate_serial_hash(byte_size: int = 18, int_size: int = 36) → int

Generates a hashed serial number.

Parameters:

  • byte_size – Size of the bytes object containing random bytes.

  • int_size – Size of the base int.

Returns:

Returns the hashed serial.

Return type:

int

expose.models.cert.generate_cert(common_name: str, san_list: List[str], country_name: str = 'US', locality_name: str = 'Springfield', state_or_province_name: str = 'Missouri', organization_unit_name: str = 'Information Technology', key_file: str = 'private.pem', cert_file: str = 'public.pem', key_size: int = 2048) → None

Creates a self-signed certificate.

Parameters:

  • common_name – DNS name of the origin.

  • country_name – Name of the country. Defaults to US

  • locality_name – Name of the city. Defaults to New York

  • state_or_province_name – Name of the state/province. Defaults to New York

  • organization_unit_name – Name of the organization unit. Defaults to Information Technology

  • key_file – Name of the key file.

  • cert_file – Name of the certificate.

  • key_size – Size of the public key. Defaults to 2048.

  • san_list – List of Subject Alternative Names (SANs). Defaults to None.

See also

Use openssl x509 -inform pem -in public.pem -noout -text to look at the generated cert using openssl.

Expose - Configuration

class expose.models.config.AMIBase(pydantic.BaseModel)

Default values to fetch AMI image ID.

>>> AMIBase

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

_BASE_URL: str = ModelPrivateAttr(default='https://aws.amazon.com/marketplace/server/configuration?productId={productId}')
_BASE_SSM: str = ModelPrivateAttr(default='/aws/service/marketplace/prod-{path}')
PRODUCT_PAGE: Url
NAME: str
ALIAS: str
PRODUCT_CODE: str
S_PRODUCT_PAGE: Url
S_NAME: str
S_ALIAS: str
S_PRODUCT_CODE: str
model_post_init(__context: Any) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • __context – The context.

class expose.models.config.EnvConfig(pydantic_settings.BaseSettings)

Env configuration.

>>> EnvConfig

References

https://docs.pydantic.dev/2.3/migration/#required-optional-and-nullable-fields

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

port: int
open_port: bool
channel_timeout: int
image_id: Optional[str]
instance_type: str
aws_region_name: str
key_pair: str
security_group: str
key_file: str
cert_file: str
server_info: str
hosted_zone: Optional[str]
subdomain: Optional[str]
aws_access_key: Optional[str]
aws_secret_key: Optional[str]
email_address: EmailStr
organization: Optional[str]
class Config

Extra config for .env file and extra.

extra = 'allow'
env_file = '.env'
class expose.models.config.Settings(pydantic.BaseModel)

Wrapper for AWS settings.

>>> Settings

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

interactive: bool
current_dir: Path
ssh_home: str
key_pair_file: Path
configuration: Path
ami_deprecation: int
entrypoint: str
nginx_config_commands: Dict[str, bool]

Expose - Exceptions

exception expose.models.exceptions.NotImplementedWarning

Custom implementation warning.

exception expose.models.exceptions.DuplicateResource

Custom validation error for duplicate AWS resources.

exception expose.models.exceptions.AWSResourceError(status_code: int, error_msg: str)

Custom resource error for AWS resources.

Expose - ImageFactory

expose.models.image_factory.deprecation_warning(image_id: str, deprecation_time: str) → None

Raises a deprecation warning if the chosen AMI is nearing (value is set in config) its DeprecationTime.

class expose.models.image_factory.ImageFactory(session: Session, logger: Logger)

Handles retrieving AMI ID from multiple sources.

>>> ImageFactory

Instantiates the ImageFactory object.

Parameters:

  • session – boto3 session instantiated in the origin class.

  • logger – Custom logger.

get_ami_id_ssm() → str

Retrieve AMI ID using Ami Alias.

get_ami_id_product_code() → str

Retrieve AMI ID using Product Code.

get_ami_id_name() → str

Retrieve AMI ID using Ami Name.

get_image_id() → str

Tries to get image id from multiple sources, sequentially.

See also

Executes in sequence as fastest first. - Alias on SSM points to a single parameter that possibly contains a single AMI ID as its value. - Lookup AMI with image name is specific and possibly points to a single AMI ID. - Lookup AMI with product code will possibly return many images, so grabs the most recently created one.

Returns:

AMI image ID.

Return type:

str

Raises:

Expose - LOGGER

Custom logging setup shared across modules.

Expose - Route53

expose.models.route_53.get_zone_id(client: client, logger: Logger, dns: str, init: bool = False) → str

Gets the zone ID of a DNS name registered in route53.

Parameters:

  • client – Pre-instantiated boto3 client.

  • logger – Custom logger.

  • dns – Hosted zone name.

  • init – Boolean flag to raise an error in case of missing zone ID.

Returns:

Returns the zone ID.

Return type:

str or None

expose.models.route_53.change_record_set(client: client, source: str, destination: str, logger: Logger, zone_id: str, action: str) → Optional[Dict]

Changes a record set within an existing hosted zone.

Parameters:

  • client – Pre-instantiated boto3 client.

  • source – Source DNS name.

  • destination – Destination hostname or IP address.

  • logger – Custom logger.

  • zone_id – Hosted zone ID.

  • action – Action to perform. Example: UPSERT or DELETE

Returns:

ChangeSet response from AWS.

Return type:

dict or None

Expose - Server Configuration

expose.models.server.join(value: Union[tuple, list, str], separator: str = ':') → str

Uses .join to squash a list or tuple using a separator.

Parameters:

  • value – Value to be squashed.

  • separator – Separator to be used to squash.

Returns:

A squashed string.

Return type:

str

expose.models.server.print_warning() → None

Prints a message on screen to run an app or api on the specific port.

See also

  • This is a safety net to improve latency.

  • This function will block instantiating channel transport until the port is locked for tunneling.

class expose.models.server.Server(hostname: str, logger: Logger, username: str = 'ubuntu', timeout: int = 30)

Initiates Server object to create an SSH session to configure the server and intiate the tunneling.

>>> Server

Reverse SSH Port Forwarding

Specifies that the given port on the remote server host is to be forwarded to the given host and port on the local side. So, instead of your machine doing a simple SSH, the server does an SSH and through the port forwarding makes sure that you can SSH back to the server machine.

Instantiates the session using RSAKey generated from the ec2’s keypair PEM file.

Parameters:

  • hostname – Hostname of the server to connect to.

  • username – Takes the username of the server to authenticate.

  • timeout – Connection timeout for SSH server.

run_ondemand_ssh(ondemand: str, command: str) → bool

Runs on demand SSH commands in case of recoverable errors.

Parameters:

  • ondemand – Recovery command that needs to run.

  • command – Original command that failed.

Returns:

To indicate if recovery failed.

Return type:

bool

run_interactive_ssh() → bool

Authenticates remote server using a PEM file and runs interactive ssh commands.

Returns:

Boolean flag to indicate the calling function if all the commands were executed successfully.

Return type:

bool

server_write(data: dict) → None

Writes configuration data into dedicated files using FTP.

Parameters:

data – Takes a dictionary of filename and content as key-value pairs.

_handler(channel: Channel, port: int) → None

Creates a socket and handles TCP IO on the channel created.

Parameters:

  • channel – Channel for Transport.

  • port – Port number on which the socket should connect.

stop_tunnel(transport: Transport, threads: List[Thread]) → None

Stops port forwarding.

Parameters:

  • transport – Transport object that creates the channel

  • threads – Daemon threads handling connections.

initiate_tunnel() → None

Initiates port forwarding using Transport which creates a channel.

Indices and tables