How to build your own proxy IP pool using ExpressVPN

Supported platforms:

Windows 10
Ubuntu 20.04+
CentOS 7.5

Step 1: Purchase an ExpressVPN account and obtain a 23-character activation code.

Step 2: Install Docker and pull the latest image of "pikadoramon/expressvpn".

Step 3: Run the Docker container with the following command:

echo "~> Run testing container..."
docker run \
    --env=ACTIVATION_CODE=${ACTIVATION_CODE} \
    --env=SERVICE=full_groxy \
    --cap-add=NET_ADMIN \
    --device=/dev/net/tun \
    --privileged \
    --detach=true \
    --tty=true \
    --name=expressvpn-proxy \
    -p 50100:50100 -p 50101:50101 -p 50099:50099 \
    pikadoramon/expressvpn:latest
echo ""

Step 4: Test if the image is successfully launched by running:

curl http://127.0.0.1:50099/ok

Here is a Python wrapper for managing VPN

-*- coding: utf-8 -*-
"""
-------------------------------------------------
   File Name：     proxy_admin
   Description :
   Author :       pikadoramon
   date：          2023/8/4
-------------------------------------------------
   Change Activity:
                   2023/8/4:
-------------------------------------------------
"""
__author__ = 'pikadoramon'
import json
import time



import random
import requests
import hashlib
from urllib.parse import urlencode

logger = logging.getLogger(__name__)

def md5(s):
    m = hashlib.md5()
    m.update(s.encode())
    return m.hexdigest()


def generator_headers(data, user, secret):
    request_time = str(int(time.time()))
    md5_str = "reqtime={}|body={}|secret={}20230803796431".format(request_time, data, secret)
    calculate_sign = md5(md5_str)
    return {
        "Spider-Request-Time": request_time,
        "Spider-Sign": calculate_sign,
        "Spider-User": user
    }


class UnExpectError(Exception):
    pass


class ExpressVPNProxyAdmin:

    def __init__(self, addr):

        self._session = requests.Session()

        self.addr = addr
        self._uuid = "20230803796431"
        self.user_name = None
        self._secret = None
        self._vpn_status = None
        self._connect_country = None
        self._acode = None

    def set_admin_account(self, user, secret):
        self.user_name = user
        self._secret = secret

    def _request(self, method, url, headers, body, timeout, retry):
        if retry is None:
            retry = 3
        if timeout is None or timeout < 3:
            timeout = 3

        err = None
        while retry > 0:
            retry -= 1
            try:
                logger.debug("method={} url={} body={}".format(method, url, body))
                resp = self._session.request(method, url,
                                             headers=headers,
                                             json=body,
                                             timeout=timeout)
                return resp
            except Exception as e:
                logger.error("request url fail {}, reason: {}".format(headers, e))
                err = e
        raise err

    def vpn_status(self):
        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/status?rnd=" + rnd
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3).json()
        if resp["code"] != 401 and resp["code"] != 400:
            self._connect_country = resp["data"]["Country"]
            self._vpn_status = resp["data"]["CurrentStatus"]
            self._acode = resp["data"]["ACode"]
        return resp

    def vpn_country_list(self, show_all: bool = False):
        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/list?rnd=" + rnd
        if show_all:
            url += "&num=all"
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3)
        return resp.json()

    def vpn_disconnect(self):
        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/disconnect?rnd=" + rnd
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3)
        return resp.json()

    def vpn_connect(self):

        resp_json = self.vpn_status()
        current_status = resp_json["data"]["CurrentStatus"]
        if current_status == "Not Activated":
            self.vpn_activate()
        elif current_status.startswith("Connected") or current_status.startswith("Connected"):
            self.vpn_disconnect()

        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/connect?rnd=" + rnd
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3)
        resp_json = resp.json()
        self.vpn_status()
        return resp_json

    def vpn_activate(self):
        resp_json = self.vpn_status()
        current_status = resp_json["data"]["CurrentStatus"]
        if current_status == "Not Activated":
            rnd = "%s" % random.random()
            url = self.addr + "/v1/expressvpn/activate?rnd=" + rnd
            headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
            resp = self._request("GET", url, headers, None, 5, 3)
            resp_json = resp.json()
            current_status = resp_json["data"]["CurrentStatus"]

        if current_status.startswith("Not Activated"):
            raise UnExpectError("activate fail, unexpect status: %s" % current_status)
        return resp_json

    def vpn_update(self, country="smart", light_way_cipher="auto", preferred_protocol="auto", stop_cron=False):
        params = dict(coutry=country,
                      lightWayCipher=light_way_cipher,
                      preferredProtocol=preferred_protocol,
                      stopCron=str(stop_cron).lower())
        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/update?" + urlencode(params) + "&rnd=" + rnd
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3)
        resp_json = resp.json()
        return resp_json

    def vpn_reload(self, code, country="smart", light_way_cipher="auto", preferred_protocol="auto"):
        if not code and not isinstance(code, str) and len(code) < 23:
            raise UnExpectError("激活号应为长度大于等于23个字符的字符串, 而你的是 %s" % code)
        data = dict(code=code,
                    coutry=country,
                    lightWayCipher=light_way_cipher,
                    preferredProtocol=preferred_protocol)

        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/reload?" + "&rnd=" + rnd
        headers = generator_headers(json.dumps(data).replace(" ", ""), self.user_name, self._secret, self._uuid)
        headers["Content-Type"] = "application/json"
        resp = self._request("POST", url, headers, data, 5, 3)
        resp_json = resp.json()
        return resp_json

VPN status

VPN network status indicates whether the network is connected to the VPN server properly. Only when the network status is normal can the VPN forwarding service run correctly. Otherwise, the proxy service will return a 501 status or close the connection.

def vpn_status(self):
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/status?rnd=" + rnd
    headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
    resp = self._request("GET", url, headers, None, 5, 3).json()
    if resp["code"] != 401 and resp["code"] != 400:
        self._connect_country = resp["data"]["Country"]
        self._vpn_status = resp["data"]["CurrentStatus"]
        self._acode = resp["data"]["ACode"]
    return resp

Get List of Available VPN Connection Countries

num is an optional parameter. When num is set to 'all', it returns a list of all available countries. Otherwise, it returns a list of countries with the highest network connection speed.

def vpn_country_list(self, show_all: bool = False):
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/list?rnd=" + rnd
    if show_all:
        url += "&num=all"
    headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
    resp = self._request("GET", url, headers, None, 5, 3)
    return resp.json()

Connect VPN

def vpn_connect(self):
 
    resp_json = self.vpn_status()
    current_status = resp_json["data"]["CurrentStatus"]
    if current_status == "Not Activated":
        self.vpn_activate()
    elif current_status.startswith("Connected") or current_status.startswith("Connected"):
        self.vpn_disconnect()
 
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/connect?rnd=" + rnd
    headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
    resp = self._request("GET", url, headers, None, 5, 3)
    resp_json = resp.json()
    self.vpn_status()
    return resp_json

Disconnect VPN

def vpn_disconnect(self):
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/disconnect?rnd=" + rnd
    headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
    resp = self._request("GET", url, headers, None, 5, 3)
    return resp.json()

Activate VPN

def vpn_activate(self):
    resp_json = self.vpn_status()
    current_status = resp_json["data"]["CurrentStatus"]
    if current_status == "Not Activated":
        rnd = "%s" % random.random()
        url = self.addr + "/v1/expressvpn/activate?rnd=" + rnd
        headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
        resp = self._request("GET", url, headers, None, 5, 3)
        resp_json = resp.json()
        current_status = resp_json["data"]["CurrentStatus"]
 
    if current_status.startswith("Not Activated"):
        raise UnExpectError("activate fail, unexpect status: %s" % current_status)
    return resp_json

Update Current VPN Network Parameters (Without Restarting)

Update the connection country, encryption method, forwarding protocol, and whether to use a timer.

country: VPN connection country, abbreviated as returned by vpn_country_list
light_way_cipher: encryption method, select "auto" here
preferred_protocol: forwarding network protocol, select "auto" here, which represents TCP and UDP
stop_cron: whether to stop the timer. If set to True, it means not to check the VPN availability periodically.

Note: After updating the VPN network parameters, if you want them to take effect, you need to execute this update interface (update), and then execute the connection (connect).

def vpn_update(self, country="smart", light_way_cipher="auto", preferred_protocol="auto", stop_cron=False):
    params = dict(coutry=country,
                  lightWayCipher=light_way_cipher,
                  preferredProtocol=preferred_protocol,
                  stopCron=str(stop_cron).lower())
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/update?" + urlencode(params) + "&rnd=" + rnd
    headers = generator_headers(rnd, self.user_name, self._secret, self._uuid)
    resp = self._request("GET", url, headers, None, 5, 3)
    resp_json = resp.json()
    return resp_json

Reset Current VPN Network Parameters (Without Automatic Offline Processing)

Update the activation code, connection country, encryption method, forwarding protocol, and whether to use a timer.

code: activation code used for VPN
country: VPN connection country, abbreviated as returned by vpn_country_list
light_way_cipher: encryption method, select "auto" here
preferred_protocol: forwarding network protocol, select "auto" here, which represents TCP and UDP
stop_cron: whether to stop the timer. If set to True, it means not to check the VPN availability periodically.

Note: Before resetting the VPN network parameters, you need to manually disconnect the VPN (disconnect), and then execute this reset interface (reload). After that, you can activate (activate) or connect (connect) again.

def vpn_reload(self, code, country="smart", light_way_cipher="auto", preferred_protocol="auto"):
    if not code and not isinstance(code, str) and len(code) < 23:
        raise UnExpectError("The activation code should be a string with a length greater than 32 characters, but yours is %s" % code)
    data = dict(code=code,
                coutry=country,
                lightWayCipher=light_way_cipher,
                preferredProtocol=preferred_protocol)
 
    rnd = "%s" % random.random()
    url = self.addr + "/v1/expressvpn/reload?" + "&rnd=" + rnd
    headers = generator_headers(json.dumps(data).replace(" ", ""), self.user_name, self._secret, self._uuid)
    headers["Content-Type"] = "application/json"
    resp = self._request("POST", url, headers, data, 5, 3)
    resp_json = resp.json()
    return resp_json

Blog

How to build your own proxy IP pool using ExpressVPN?

pikadoramon