[Openvpn-devel] [L] Change in openvpn[master]: Add python scripts that demonstrate using the mangement interface

plaisthos has uploaded this change for review. ( http://gerrit.openvpn.net/c/openvpn/+/1859?usp=email )


Change subject: Add python scripts that demonstrate using the mangement interface
......................................................................

Add python scripts that demonstrate using the mangement interface

Change-Id: Ib09bafabf612e9501ea757647c86a450c73228a5
Signed-off-by: Arne Schwabe <[email protected]>
---
A sample/management/acc_sysinfo_client_demo.py
A sample/management/acc_sysinfo_server_demo.py
A sample/management/omi.py
A sample/management/omi_tools.py
4 files changed, 774 insertions(+), 0 deletions(-)



  git pull ssh://gerrit.openvpn.net:29418/openvpn refs/changes/59/1859/1

diff --git a/sample/management/acc_sysinfo_client_demo.py b/sample/management/acc_sysinfo_client_demo.py
new file mode 100755
index 0000000..7a7f015
--- /dev/null
+++ b/sample/management/acc_sysinfo_client_demo.py
@@ -0,0 +1,174 @@
+#!/opt/homebrew/bin/python3
+# Copyright (c) 2026 OpenVPN Inc <[email protected]>
+# Copyright (c) 2026 Arne Schwabe <[email protected]>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+# This implement a small demonstration of using the
+# OpenVPN Access Server Management Interface (OMI) and an app Custom Control
+# protocol (info_sample) to retrieve a number of system information from the
+# client and push them to the server. To use this use on the server
+
+# --app-custom-control info_sample
+# --management /path/to/acc_sysinfo_server_demo.py unix-script
+
+# and on the client
+# --app-custom-control info_sample
+# --management /path/to/acc_sysinfo_client_demo.py unix-script
+
+# Note this is just a demonstration and not a production-ready solution.
+# When modifying this code, keep the security in mind. It is easy to
+# introduce a security problem when executing commands on the client.
+
+import asyncio
+import base64
+import json
+import logging
+import os
+import platform
+import socket
+import subprocess
+
+from omi import OmiProtocol
+from omi_tools import run_main, start_management_protocol, ConnectData
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+
+class OmiSysInfoClientProtocol(OmiProtocol):
+    def __init__(self, finish_future, connect_data: ConnectData):
+        super().__init__(finish_future, connect_data)
+        self._acc_msg = ""
+
+    def recv_notify_ACC(self, args):
+        protocol, fragment, b64msg = args.split(",", 2)
+        if protocol != "sys_info":
+            logging.info(f"Received unknown ACC protocol: {protocol}")
+            return
+
+        msg = base64.decodebytes(b64msg.encode())
+        self._acc_msg += msg.decode()
+
+        # Message is not yet complete. Wait for more fragments.
+        if fragment != "0":
+            return
+
+        try:
+            self.parse_sysinfo_message(self._acc_msg)
+        except Exception as e:
+            logging.error(f"Error parsing ACC message: {e}", exc_info=True)
+        self._acc_msg = ""
+
+    def parse_sysinfo_message(self, accmsg):
+        msg = json.loads(accmsg)
+
+        logging.info(f"Received sys_info message: {accmsg}")
+
+        asyncio.create_task(self.send_info_response(msg))
+
+    async def send_info_response(self, msg):
+        response = {}
+        if msg.get("uname", False):
+            response['uname'] = platform.uname()._asdict()
+
+        if msg.get("username", False):
+            response['username'] = os.getlogin()
+
+        if msg.get("hostname", False):
+            response['hostname'] = platform.node()
+
+        if msg.get("dns-google.com", False):
+            try:
+                addrs = [x[4][0] for x in socket.getaddrinfo('google.com', 443)]
+                response["dns-google.com"] = addrs
+
+            except Exception as e:
+                logging.error(f"Error resolving DNS for google.com: {e}", exc_info=True)
+                response["dns-google.com"] = None
+
+        if msg.get("route", False):
+            response['route'] = await self.get_routes(False)
+
+        if msg.get("route6", False):
+            response['route6'] = await self.get_routes(True)
+
+        # TODO implement fragmentation
+
+        msg = json.dumps(response).encode()
+        # 980 chars after base64 encoding
+        fragment_length = 980*6//8
+
+        for part_start in range(0, len(msg), fragment_length):
+
+            msg_part = msg[part_start:part_start + fragment_length]
+            b64msg = base64.encodebytes(msg_part).decode()
+
+            protocol = "sys_info"
+
+            flags = "6"
+            if part_start + fragment_length < len(msg):
+                flags = "6:F"
+
+            command = f"acc-msg\n{protocol}\n{flags}\n{b64msg}\nEND"
+            self.queue_command(command)
+
+    async def get_routes(self, v6):
+        if platform.system() == 'linux':
+            cmd = '/usr/sbin/ip'
+            if v6:
+                args = ['-6', 'route', 'show']
+            else:
+                args = ['-4', 'route', 'show']
+        elif platform.system() == 'Darwin' or os.platform() == 'FreeBSD':
+            # FreeBSD and macOS
+            if os.path.exists('/usr/sbin/netstat'):
+                cmd = '/usr/sbin/netstat'
+            else:
+                cmd = '/usr/bin/netstat'
+
+            if v6:
+                args = [ '-f', 'inet6', '-rn']
+            else:
+
+                args = [ '-f', 'inet', '-rn']
+        else:
+            return "Unsupported platform"
+
+        proc = await asyncio.subprocess.create_subprocess_exec(cmd, *args, stdout=subprocess.PIPE)
+        output, _ = await proc.communicate()
+        return output.decode()
+
+
+async def main(connect_data: ConnectData):
+    transport, protocol, finish_future = await start_management_protocol(connect_data, OmiSysInfoClientProtocol)
+    while True:
+        await asyncio.sleep(1)
+        print(".", end="", flush=True)
+
+    management_version = await protocol.management_ready()
+    print(f"Management interface version: {management_version}")
+
+    await protocol.set_bytecount_interval(5)
+
+    await finish_future
+
+
+if __name__ == '__main__':
+    run_main(main)
diff --git a/sample/management/acc_sysinfo_server_demo.py b/sample/management/acc_sysinfo_server_demo.py
new file mode 100755
index 0000000..139d2f9
--- /dev/null
+++ b/sample/management/acc_sysinfo_server_demo.py
@@ -0,0 +1,135 @@
+#!/opt/homebrew/bin/python3
+# Copyright (c) 2026 OpenVPN Inc <[email protected]>
+# Copyright (c) 2026 Arne Schwabe <[email protected]>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import asyncio
+# This implement a small demonstration of using the
+# OpenVPN Access Server Management Interface (OMI) and an app Custom Control
+# protocol (info_sample) to retrieve a number of system information from the
+# client and push them to the server. To use this use on the server
+
+# --app-custom-control info_sample
+# --management /path/to/acc_sysinfo_server_demo.py unix-script
+
+# and on the client
+# --app-custom-control info_sample
+# --management /path/to/acc_sysinfo_client_demo.py unix-script
+
+import base64
+import json
+import logging
+import random
+from pprint import pprint, pformat
+
+from omi import OMIServerProtocol, ConnectedClient
+from omi_tools import run_main, start_management_protocol, ConnectData
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+class SysInoConnectedClient(ConnectedClient):
+    accmsg = ""
+
+class OMISysInfoServerProtocol(OMIServerProtocol):
+    def __init__(self, finish_future, connect_data: ConnectData = None):
+        super().__init__(finish_future, connect_data)
+        self.accmsg = ""
+        self._connected_client_class = SysInoConnectedClient
+
+
+    def client_event_ACC(self,client, kid, env, extra):
+        protocol, fragment, b64msg = extra.split(",", 2)
+        if protocol != "sys_info":
+            logging.info(f"Received unknown ACC protocol: {protocol}")
+            return
+
+        try:
+
+            msg = base64.decodebytes(b64msg.encode())
+            client.accmsg += msg.decode()
+
+            # Message is not yet complete. Wait for more fragments.
+            if fragment != "0":
+                return
+
+            self.parse_sys_info_reply(client.accmsg)
+            client.accmsg = ""
+
+        except Exception as e:
+            logger.error(f"Error parsing sys_info message: {e}", exc_info=True)
+
+    def parse_sys_info_reply(self, accmsg):
+        logger.debug(f"Received sys_info message: {accmsg!r}")
+        client_info = json.loads(accmsg)
+
+        logger.info(f"Received sys_info response: {pformat(client_info)}")
+
+    def client_event_ESTABLISHED(self, client, kid, env, extra):
+        print(f"CLient connected: {client}")
+        pprint(env)
+
+        # TODO: track kid and don't assume we haven't done a renegotiation
+        kid = 1
+        asyncio.create_task(self._send_sys_info_command(client.cid, kid))
+
+    def client_event_CONNECT(self, client, kid, env, extra):
+        print(f"Authenticating client {client.cid}, {kid}")
+        self.queue_command(f"client-auth-nt {client.cid} {kid}")
+
+    async def _send_sys_info_command(self, cid, kid):
+        # Give the client some time to settle its routes
+        await asyncio.sleep(5)
+
+        info = ["route", "route6", "dns-google.com", "username", "hostname", "uname"]
+        # Pick 3 random infos from the client
+
+        infos = random.choices(info, k=3)
+        request = {k: True for k in infos}
+
+        for i in info:
+            request = {i: True}
+            req_str = json.dumps(request).encode()
+
+            logger.info(f"Sending sysinfo request for client {cid} with infos: {infos}")
+
+            req_base64 = base64.encodebytes(req_str).decode()
+            acc_protocol = "sys_info"
+
+            # TODO: Implement fragmentation
+            cmd = f"client-acc-msg {cid} {kid}\n{acc_protocol}\n6\n{req_base64}\nEND"
+
+            await self.queue_command(cmd)
+
+
+async def main(connect_data: ConnectData):
+    transport, protocol, finish_future = await start_management_protocol(connect_data, OMISysInfoServerProtocol)
+
+    management_version = await protocol.management_ready()
+    print(f"Management interface version: {management_version}")
+
+    await protocol.enable_log()
+
+    await protocol.set_bytecount_interval(5)
+
+    await finish_future
+
+
+if __name__ == '__main__':
+    run_main(main)
diff --git a/sample/management/omi.py b/sample/management/omi.py
new file mode 100644
index 0000000..2a7f90e
--- /dev/null
+++ b/sample/management/omi.py
@@ -0,0 +1,372 @@
+#!/opt/homebrew/bin/python3
+# Copyright (c) 2026 OpenVPN Inc <[email protected]>
+# Copyright (c) 2026 Arne Schwabe <[email protected]>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+# This is an example implementation in python and asyncio to drive OpenVPN's
+# management interface (OMI). The permissive license allows you to use this
+# code in other projects.
+
+"""
+Implementation of the OpenVPN management interface (OMI) protocol.
+"""
+import asyncio
+import logging
+import re
+from asyncio import Future
+from dataclasses import dataclass
+from datetime import datetime
+from typing import override, List
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+@dataclass
+class OmiCommandResult:
+    command: str
+    result: List[str]
+    error: bool
+    status_text: str
+
+
+@dataclass
+class OmiSendCommand:
+    command: str
+    result: Future[OmiCommandResult]
+
+
+@dataclass
+class ConnectionProperties:
+    """
+    Properties for connecting to the management interface.
+    """
+    password: str
+
+
+class OmiProtocol(asyncio.Protocol):
+    def __init__(self, finish_future, connect_data: ConnectionProperties):
+        super().__init__()
+        self.bytes_received = None
+        self.bytes_sent = None
+        self.version = -1
+        self.recvBuffer = ""
+        self.finish_future = finish_future
+        self._can_send = asyncio.Event()
+        self._can_send.clear()
+
+        self._management_ready = asyncio.Event()
+        self._management_ready.clear()
+        self._response = []
+
+        self._response_ready = asyncio.Event()
+        self._response_ready.clear()
+
+        self._send_task = asyncio.create_task(self._send_loop())
+        self._send_queue = asyncio.Queue()
+        self._connect_data: ConnectionProperties = connect_data
+
+    @override
+    def eof_received(self):
+        logger.info('EOF received. Shutting down.')
+        self._stop()
+
+    @override
+    def connection_lost(self, excp):
+        if excp:
+            logger.info(f'Connection lost ({excp}). Shutting down.')
+        else:
+            logger.info(f'Connection lost. Shutting down.')
+        self._stop()
+
+    def _stop(self):
+        if not self.finish_future.done():
+            self._send_task.cancel()
+            self.finish_future.set_result(True)
+
+    async def _send_loop(self):
+        while True:
+            command: OmiSendCommand = await self._send_queue.get()
+            await self._can_send.wait()
+            self.send_line(command.command)
+
+            response = []
+            response_complete = False
+
+            # This is a bit brittle as it will loop forever if not getting a
+            # line that starts with ERROR or SUCCESS. But this is the OMI protocol
+            while not response_complete:
+                await self._response_ready.wait()
+
+                line = self._response.pop(0)
+                response.append(line)
+
+                if not self._response:
+                    self._response_ready.clear()
+
+                lastline = response[-1]
+                if lastline.startswith('ERROR:'):
+                    status = False
+                    response_complete = True
+                    _, status_text = lastline.split(":", 1)
+
+                elif lastline.startswith('SUCCESS:'):
+                    status = True
+                    response_complete = True
+                    _, status_text = lastline.split(":", 1)
+                elif lastline == "END":
+                    status = True
+                    response_complete = True
+                    status_text = None
+                else:
+                    pass
+
+            # logger.debug(f'Result received command: {response}')
+
+            result = OmiCommandResult(command.command, response[:-1], status, status_text)
+
+            command.result.set_result(result)
+            logger.debug(f"Command done.")
+
+    @override
+    def pause_writing(self):
+        self._can_send.clear()
+
+    @override
+    def resume_writing(self):
+        self._can_send.set()
+
+    @override
+    def connection_made(self, transport):
+        peername = transport.get_extra_info('peername')
+        logging.info('Connection from {}'.format(peername))
+        self.transport = transport
+        self._can_send.set()
+
+        # OMI expects us to authenticate with the password just on
+        # its own on a line
+        if self._connect_data.password:
+            self.transport.write(f"{self._connect_data.password}\r\n".encode())
+
+    @override
+    def data_received(self, data):
+        # OMI protocol is pure text so decode everything as UTF-8
+        message = data.decode()
+        self.recvBuffer += message
+
+        parts = self.recvBuffer.split("\r\n")
+
+        # pass complete lines to recvLine
+        for part in parts[:-1]:
+            logger.debug(f'Line received: {part!r}')
+            self.recv_line(part)
+
+        # keep the last incomplete line for the next call
+        self.recvBuffer = parts[-1]
+
+    def recv_line(self, line):
+        if line.startswith(">"):
+            self.recv_notify(line)
+        else:
+            self._response.append(line)
+            self._response_ready.set()
+
+    def recv_notify(self, line):
+        if not ":" in line:
+            logger.warning("Invalid line received: " + line)
+            return
+
+        command, args = line[1:].split(":", 1)
+
+        # do dynamic dispatch based on command to call a handler
+        if hasattr(self, f"recv_notify_{command}"):
+            handler = getattr(self, f"recv_notify_{command}")
+            handler(args)
+            return
+
+        logger.info("Unknown notify line received: " + line)
+
+    def recv_notify_INFO(self, args):
+        m = re.match(r"OpenVPN Management Interface Version (?P<version>\d+)", args)
+        if m:
+            self.version = int(m.group("version"))
+            self._management_ready.set()
+        else:
+            logger.warning(f"Unknown INFO line received: {args}")
+
+    def recv_notify_BYTECOUNT(self, args):
+        bytes_sent, bytes_received = args.split(",", 1)
+        self.bytes_sent = int(bytes_sent)
+        self.bytes_received = int(bytes_received)
+
+    async def _release_hold_task(self, hold_time: int = 0):
+        await asyncio.sleep(hold_time)
+        await self.queue_command("hold release")
+
+    def recv_notify_HOLD(self, args):
+        _text, hold_time = args.split(":", 1)
+        hold_time = int(hold_time)
+
+        asyncio.create_task(self._release_hold_task(hold_time))
+
+    def queue_command(self, command) -> Future[OmiCommandResult]:
+        cmd = OmiSendCommand(command, Future())
+        self._send_queue.put_nowait(cmd)
+        return cmd.result
+
+    async def management_ready(self) -> int:
+        """
+        Waits until the management interface is ready. Returns the management interface version.
+        """
+        await self._management_ready.wait()
+        return self.version
+
+    def send_line(self, line):
+        logger.debug(f"Sending line: {line}")
+        self.transport.write(f"{line}\n".encode())
+
+    """
+    Set the bytecount interval. Use 0 to disable
+    """
+
+    async def set_bytecount_interval(self, interval: int):
+        return await self.queue_command(f"bytecount {interval}")
+
+    async def enable_log(self):
+        return await self.queue_command("log on")
+
+    def recv_notify_LOG(self, args):
+        # >LOG:1786453401,D,MANAGEMENT: CMD 'bytecount 30'
+        timestamp, level, message = args.split(",", 2)
+
+        timestamp = datetime.fromtimestamp(int(timestamp))
+        level = level.strip()
+        message = message.strip()
+        self.log_message(timestamp, level, message)
+
+    def log_message(self, time, level, message):
+        """
+        A new log message has been received. This method is intended
+        to be overridden by a subclass to handle the log messages.
+        :param time:    time in local time format
+        :param level:   level of the log message
+        :param message: the log message itself
+        :return:
+        """
+        logger.debug(f"Received log message: {time}, {level}, {message}")
+
+
+@dataclass
+class ConnectedClient:
+    cid: int
+    bytes_sent: int = 0
+    bytes_received: int = 0
+
+
+class OMIServerProtocol(OmiProtocol):
+    """
+    This class implement common method for the OMI when OpenVPN is running
+    as a server.
+    """
+
+    # Connected client class, can be overridden by subclasses to store
+    # more information about the connected client
+    _connected_client_class = ConnectedClient
+
+    def __init__(self, finish_future, connect_data: ConnectionProperties = None):
+        super().__init__(finish_future, connect_data)
+        self._reset_pending_client_event()
+        self._connected_clients = {}
+
+    def _reset_pending_client_event(self):
+        self._pending_client_event = None
+        self._pending_client_event_cid = -1
+        self._pending_client_event_kid = None
+        self._pending_client_event_extra_args = None
+        self._client_env = {}
+
+    def recv_notify_CLIENT(self, args: str):
+        event, args = args.split(",", 1)
+
+        if event == "ENV":
+            if args == "END":
+                # Client environment is finished, trigger previous client
+                # event that contained an ENV
+                self._trigger_client_event(self._pending_client_event, self._pending_client_event_cid,
+                                           self._pending_client_event_kid, self._client_env,
+                                           self._pending_client_event_extra_args)
+                self._reset_pending_client_event()
+            else:
+                key, value = args.split("=", 1)
+                self._client_env[key] = value
+
+        elif event in ("ESTABLISHED", "DISCONNECT"):
+            # This event is followed by a client environment. Wait for
+            # the client evironment to be completed before triggering the event
+            cid = int(args)
+
+            self._pending_client_event = event
+            self._pending_client_event_cid = int(cid)
+            self._pending_client_event_kid = None
+            if event == "ESTABLISHED":
+                self._add_client(cid)
+
+        elif event in ("CONNECT", "REAUTH", "CR_RESPONSE"):
+            # This event is similar to other event but also carries the key-id
+            cid, kid = args.split(",", 1)
+            cid, kid = int(cid), int(kid)
+            self._add_client(cid)
+
+            self._pending_client_event = event
+            self._pending_client_event_cid = int(cid)
+            self._pending_client_event_kid = int(kid)
+        elif event in ("ACC",):
+            cid, kid, acc_args = args.split(",", 2)
+            self._trigger_client_event(event, int(cid), int(kid), None, acc_args)
+        else:
+            logger.debug(f"Client event {event} not handled.")
+
+    def recv_notify_BYTECOUNT_CLI(self, args):
+        cid, bytes_sent, bytes_received = args.split(",", 2)
+
+        client = self._connected_clients.get(int(cid))
+        if client:
+            client.bytes_sent = int(bytes_sent)
+            client.bytes_received = int(bytes_received)
+
+    def _trigger_client_event(self, event, cid, kid, env, extra):
+        handler_name = f"client_event_{event}"
+        if hasattr(self, handler_name):
+            handler = getattr(self, handler_name)
+            client = self._connected_clients.get(cid, None)
+            handler(client, kid, env, extra)
+        else:
+            logger.debug(f"Client event {event} not handled. Method {handler_name} does not exist.")
+
+    def _add_client(self, cid):
+        if not cid in self._connected_clients:
+            self._connected_clients[cid] = self._connected_client_class(cid = cid)
+
+    def _remove_client(self, cid):
+        del self._connected_clients[cid]
+
+    async def get_client_status(self) -> OmiCommandResult:
+        cmd_result = self.queue_command("status 3")
+        return await cmd_result
diff --git a/sample/management/omi_tools.py b/sample/management/omi_tools.py
new file mode 100644
index 0000000..afa0e46
--- /dev/null
+++ b/sample/management/omi_tools.py
@@ -0,0 +1,93 @@
+#! /usr/bin/python3
+# Copyright (c) 2026 OpenVPN Inc <[email protected]>
+# Copyright (c) 2026 Arne Schwabe <[email protected]>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import asyncio
+import os
+import sys
+from dataclasses import dataclass
+
+from omi import OmiProtocol, ConnectionProperties
+
+
+@dataclass
+class ConnectData(ConnectionProperties):
+    socket: str
+
+
+def get_unix_script_environment() -> ConnectData | None:
+    password = os.environ.get("MANAGEMENT_PASSWORD", None)
+    socket = os.environ.get("MANAGEMENT_SOCKET", None)
+
+    if password and socket:
+        return ConnectData(password=password, socket=socket)
+    else:
+        return None
+
+
+async def start_management_protocol(connect_data: ConnectData, protocol=OmiProtocol):
+    # Get a reference to the event loop as we plan to use
+    # low-level APIs.
+    loop = asyncio.get_running_loop()
+
+    finish_future = loop.create_future()
+
+    print(f"Connecting to management socket {connect_data.socket} with passwd {connect_data.password[:4]}...")
+    transport, protocol = await loop.create_unix_connection(lambda: protocol(finish_future, connect_data),
+                                                            connect_data.socket)
+
+    return transport, protocol, finish_future
+
+
+async def main():
+    transport, protocol, finish_future = await start_management_protocol()
+
+    management_version = await protocol.management_ready()
+    await protocol.enable_log()
+    print(f"Management interface version: {management_version}")
+
+    await protocol.set_bytecount_interval(5)
+
+    await finish_future
+
+
+def run_main(main_func, connect_parameters: ConnectData = None):
+    """
+    Example function to run a main function to be either using the
+    environment variable for the unix-script or to use. If the unix-script
+    environment are detected, the function will fork and run the main_func
+    with the ConnectData from the environment. If the environment variables
+    are not available, it will not fork will fall back to the connect_parameters
+    if present.
+    :param connect_parameters:  fallback connect parameters if unix-script is not available
+    :param main_func:           The function to call.
+    :return:
+    """
+    unix_script_parameters = get_unix_script_environment()
+    if unix_script_parameters:
+        asyncio.run(main_func(unix_script_parameters))
+    elif connect_parameters:
+        asyncio.run(main_func(connect_parameters))
+    else:
+        print(f"Must be called via unix-script")
+
+
+if __name__ == '__main__':
+    run_main(main)

-- 
To view, visit http://gerrit.openvpn.net/c/openvpn/+/1859?usp=email
To unsubscribe, or for help writing mail filters, visit http://gerrit.openvpn.net/settings?usp=email

Gerrit-MessageType: newchange
Gerrit-Project: openvpn
Gerrit-Branch: master
Gerrit-Change-Id: Ib09bafabf612e9501ea757647c86a450c73228a5
Gerrit-Change-Number: 1859
Gerrit-PatchSet: 1
Gerrit-Owner: plaisthos <[email protected]>
Gerrit-CC: openvpn-devel <[email protected]>

_______________________________________________
Openvpn-devel mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/openvpn-devel