Python script to retrieve OpenVPN clients states and enable/disable a specific connection

opnsense_clientvpn_toggle.py

# usage:
# opnsense_clientvpn_toggle.py returns a Json object with log and selected connection state
# opnsense_clientvpn_toggle.py to retrieve all OpenVPN clients states
# opnsense_clientvpn_toggle.py [ID] to retrieve specific OpenVPN client state in an additional "connected" Json field
# opnsense_clientvpn_toggle.py [ID] [(true|false)] to toggle a OpenVPN client state

import requests
import re
import sys
import json

OPNSENSE_VPN_CLIENT_PAGE = "https://[OPNSENSE_HOST]/vpn_openvpn_client.php"
OPNSENSE_USER = "[USERNAME]"
OPNSENSE_PASS = "[PASSWORD]"
OPNSENSE_CA_CERT = False  # Specify CA crt filepath or False


s = requests.Session()
s.verify = OPNSENSE_CA_CERT
if not s.verify:
    import urllib3
    urllib3.disable_warnings()

log = []
connected = None


def getPreLoginCSRF():
    # Prelogin to retrieve CSRF
    log.append("Loading %s to retrieve CSRF token" % OPNSENSE_VPN_CLIENT_PAGE)
    r = s.get(OPNSENSE_VPN_CLIENT_PAGE)
    hiddenInputMatch = re.search(
        'type="hidden" name="(?P<name>[^"]+)" value="(?P<value>[^"]+)"', r.text
    )

    return (hiddenInputMatch.group("name"), hiddenInputMatch.group("value"))


def getAllOpenVPNClientStatus(csrfData):
    log.append("OpenVPN status:")
    r = s.post(
        OPNSENSE_VPN_CLIENT_PAGE,
        data={
            "usernamefld": OPNSENSE_USER,
            "passwordfld": OPNSENSE_PASS,
            csrfData[0]: csrfData[1],
            "login": 1,
        },
    )

    if re.search("Wrong username or password", r.text) is not None:
        raise Exception("Wrong username or password")

    statusMatches = re.findall(
        'class="act_toggle" data-id="(?P<id>\d+).+title="(?P<status>Enable|Disable)"',
        r.text,
    )

    for match in statusMatches:
        id = match[0]
        enabled = match[1] == "Disable"
        log.append("- Id: %s, Connected: %s" % (match[0], enabled))

    csrfToken = re.search('"X-CSRFToken", "(?P<token>[^"]+)"', r.text).group("token")

    return (statusMatches, csrfToken)


def getOpenVPNClientStatus(id):
    match = next((x for x in statusMatches if x[0] == id), None)
    if match is None:
        log.append("No connection with id: %s" % id)
        return None
    else:
        return match[1] == "Disable"


def toggleOpenVPNClient(id, currentState, newState, csrfToken):
    if newState != currentState:
        log.append("Toggle VPN connection status")
        s.headers.update({"X-CSRFToken": csrfToken})
        r = s.post(
            OPNSENSE_VPN_CLIENT_PAGE,
            data={"act": "toggle", "id": id},
        )
        r.raise_for_status()
        return newState
    else:
        log.append("VPN connection has already expected status")
        return currentState


if __name__ == "__main__":
    csrfData = getPreLoginCSRF()
    (statusMatches, csrfToken) = getAllOpenVPNClientStatus(csrfData)
    if len(sys.argv) >= 2:
        connected = getOpenVPNClientStatus(sys.argv[1])
        if len(sys.argv) == 3:
            connected = toggleOpenVPNClient(
                sys.argv[1],
                connected,
                sys.argv[2].lower() in ["true", "1", "enable"],
                csrfToken,
            )

    jsonData = {"log": log}
    if connected is not None:
        jsonData.update({"connected": connected})

    print(json.dumps(jsonData))