Welcome to VPN Server’s documentation!¶

Read Me:

VPN Server¶

class vpn.main.VPNServer(**kwargs: Unpack)¶

Initiates VPNServer object to spin up an EC2 instance with a pre-configured AMI which serves as a VPN server.

>>> VPNServer

Unpacks the kwargs and loads it as Envconfig, the subclass for BaseSettings validated using pydantic.

Parameters:

kwargs – Dictionary of keyword arguments for the following key-value pairs.

vpn_username - Username to access VPN client.
vpn_password - Password to access VPN client.
vpn_port - Port number for OpenVPN web interface access.
aws_profile_name - Name of the AWS profile to initiate a session.
aws_access_key - Access token for AWS account.
aws_secret_key - Secret ID for AWS account.
aws_region_name - Region where the instance should live. Defaults to AWS profile default.
image_id - AMI ID using which the instance should be created.
instance_type - Instance type to use in AWS to host the OpenVPN Access Server.
key_pair - Name of the key pair for the ec2 and the file stored for SSH connection.
security_group - Name of the security group to be created in AWS.
vpn_info - Name of the JSON file to dump the instance and connection information.
domain - Domain name for the hosted zone.
hosted_zone - Name of the hosted zone in route53.
subdomain - Name of the subdomain to be created in the hosted zone.
logger - Bring your own logger.

See also

Access point to the VPN server will be the combination of subdomain and hosted_zone.

Examples:

If you have a hosted zone in AWS, named example.com and want the entrypoint to be

vpn.example.com then,
- hosted_zone should be example.com
- subdomain should be vpn

Notes:

Please note that the hosted zone should also be a registered domain in AWS.
You cannot simply create a hosted zone named google.com and expect it to work.
The alias record will fail as duplicate at DNS name resolution level unless a domain is

registered with the same name.

_authorize_security_group(security_group_id: str) → bool¶

Authorizes the security group for certain ingress list.

Parameters:: security_group_id – Takes the SecurityGroup ID as an argument.

TCP 22 — SSH access.
TCP 443 — Web interface access and OpenVPN TCP connections.
TCP 943 — Web interface access (can be dynamic)
TCP 945 — Cluster control channel.
UDP 1194 — OpenVPN UDP connections.

Returns:: Flag to indicate the calling function, whether the security group was authorized successfully.
Return type:: bool

_configure_vpn(public_dns: str) → 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.

_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]

_create_key_pair() → bool¶

Creates a KeyPair of type RSA stores as a PEM file for SSH connection.

Returns:: Flag to indicate the calling function, if a KeyPair was created 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]

_delete_key_pair() → bool¶

Deletes the KeyPair created to access the ec2 instance.

Returns:: Flag to indicate the calling function, if the KeyPair was deleted successfully.
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:: Flag to indicate the calling function, whether the SecurityGroup was deleted successfully.
Return type:: bool

_disassociate_security_group(security_group_id: str, instance: ServiceResource = None, instance_id: 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:

Flag to indicate the calling function, whether the instance was disassociated from the SG successfully.

Return type:

bool

_get_vpc_id() → Optional[str]¶

Fetches the default VPC id.

Returns:: Default VPC id.
Return type:: Union[str, None]

_init(start: Union[bool, int]) → None¶

Initializer function.

Parameters:: start – Flag to indicate if its startup or shutdown.

_terminate_ec2_instance(instance_id: str = None, instance: ServiceResource = 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:

Flag to indicate the calling function, whether the instance was terminated successfully.

Return type:

bool

_test_get(host: str, timeout: Tuple = (3, 3), retries: int = 5) → Response¶

Test GET connection with multiple hostnames.

Parameters:

host – Public IP address or DNS name or alias record (entrypoint)
timeout – Tuple of connection timeout and read timeout.
retries – Number of times to retry in case of connection errors.

See also

Retries with exponential intervals between each attempt in case of a failure.

Returns:: Response object.
Return type:: Response

_tester(data: Dict[str, Union[str, int]]) → None¶

Tests GET and SSH connections on the existing server.

Parameters:: data – Takes the instance information in a dictionary format as an argument.

See also

All the tests run in parallel to improve runtime.

GET request against the public IP of the ec2 instance.
GET request against the public DNS of the ec2 instance.
SSH connection with the OpenVPN Access Server.
Test openvpnas service availability on the server.
Test alias record if values for hosted_zone and subdomain were provided

Raises:

AssertionError –
When any of the tests fail. –

create_vpn_server() → Optional[Dict[str, Union[str, int]]]¶

Creates an OpenVPN Access Server hosted on AWS ec2.

Returns:: VPN access server information.
Return type:: Dict[str, Union[str, int]] or None

delete_vpn_server(instance_id: str = None, security_group_id: str = None, public_ip: 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 manually.

References

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/instance/

wait_until_terminated.html

test_vpn() → None¶: Tests the GET and SSH connections to an existing VPN server.

Configuration¶

class vpn.models.config.ConfigurationSettings(pydantic.BaseModel)¶

OpenVPN’s configuration settings, for SSH interaction.

>>> ConfigurationSettings

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.

class vpn.models.config.AMIBase(pydantic.BaseModel)¶

Default values to fetch AMI image ID.

>>> AMIBase

See also

Subscription Home Page: https://{REGION}.console.aws.amazon.com/marketplace/home#/subscriptions/{_PRODUCT_ID}
Product ID: Found in the home page URL under Summary as ‘Product ID’
Product Code: Offer ID in the home page URL
AMI Alias: Found in configuration page (_BASE_URL) as ‘Ami Alias’
Product Code: Found in configuration page (_BASE_URL) as ‘Product Code’

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.

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 vpn.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.

class Config¶: Extra config for .env file and extra.

classmethod validate_instance_type(v: str) → str¶: Validate instance type to make sure it is not a nano.

classmethod validate_vpn_password(v: str) → str¶: Validates vpn_password as per the required regex.

class vpn.models.config.Settings(pydantic.BaseModel)¶

Wrapper for configuration 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.

Exceptions¶

exception vpn.models.exceptions.AWSResourceError(status_code: int, error_msg: str)¶: Custom resource error for AWS resources.

exception vpn.models.exceptions.NotImplementedWarning¶: Custom implementation warning.

ImageFactory¶

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

Handles retrieving AMI ID from multiple sources.

>>> ImageFactory

Creates ec2 and ssm client using an existing boto3 session.

Parameters:

session – boto3 session instantiated in the origin class.
logger – Custom logger.

get_ami_id_name() → str¶: Retrieve AMI ID using Ami Name.

get_ami_id_product_code() → str¶: Retrieve AMI ID using Product Code.

get_ami_id_ssm() → str¶: Retrieve AMI ID using Ami Alias.

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:

AWSResourceError –
If unable to fetch AMI ID from any source. –

vpn.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.

LOGGER¶

Loads a default logger with StreamHandler set to DEBUG mode.

>>> LOGGER

Route53¶

vpn.models.route53.change_record_set(client: client, source: str, destination: str, logger: Logger, zone_id: str, action: str) → bool¶

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:

Flag to indicate the calling function, whether the record was modified successfully.

Return type:

bool

vpn.models.route53.get_zone_id(client: client, logger: Logger, dns: str, init: bool = False) → Optional[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:

Union[str, None]

Raises:

AWSResourceError –
If unable to fetch the hosted zone ID by name. –

SSH Configuration¶

class vpn.models.server.Server(hostname: str, username: str, logger: Logger)¶

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

>>> Server

Instantiates the session using RSAKey generated from a ***.pem file.

Parameters:: hostname – Hostname of the server.

add_formatter() → None¶: Re-add any formatters that were removed during instantiation.

remove_formatter() → None¶: Remove any logging formatters to allow room for OpenVPN configuration interaction.

restart_service() → None¶: Restarts the openvpn service.

run_interactive_ssh(display: bool = True, timeout: int = 30) → None¶

Runs interactive ssh commands to configure the VPN server.

Parameters:

display – Boolean flag whether to display interaction data on screen.
timeout – Default interaction session timeout.

Returns:

Flag to indicate the calling function, whether the interactive session has completed successfully.

Return type:

bool

test_service(timeout: int, display: bool) → None¶

Check status of the service running on remote server.

Parameters:

timeout – Default interaction session timeout.
display – Boolean flag whether to display interaction data on screen.

Utilities¶

vpn.models.util.available_instance_types() → Generator[str]¶

Get all available EC2 instance types looping through describe instances API call.

Yields:: Generator[str] – Instance type.

vpn.models.util.available_regions() → Generator[str]¶

Get all available regions with describe regions API call.

Yields:: Generator[str] – Region name.

Welcome to VPN Server’s documentation!¶

VPN Server¶

Configuration¶

Exceptions¶

ImageFactory¶

LOGGER¶

Route53¶

SSH Configuration¶

Utilities¶

Indices and tables¶

Table of Contents

Next topic

This Page