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.

See also

Firewall configuration ports to be open:

  • 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

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

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:
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:

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.

Indices and tables