rest_api: Add tests for outbound websocket.

* Added AriServerFactory and AriServerProtocol to lib/python/asterisk/ari.py.
* Created tests/rest_api/websocket.
* Moved existing tests/rest_api/websocket_requests/* tests to
  tests/rest_api/websocket/rest_over_websocket.
* Updated the rest_over_websocket tests to use reactor.run() properly.
* Added test outbound_websocket/persistent.
* Added test outbound_websocket/per_call.
* Added test outbound_websocket/local_ari_user_read_only.

Author: gtjoseph

lib/python/asterisk/ari.py

@@ -26,10 +26,12 @@
 from twisted.internet import reactor
 try:
     from autobahn.websocket import WebSocketClientFactory, \
-        WebSocketClientProtocol, connectWS
+        WebSocketClientProtocol, connectWS, WebSocketServerFactory, \
+        WebSocketServerProtocol
 except:
     from autobahn.twisted.websocket import WebSocketClientFactory, \
-        WebSocketClientProtocol, connectWS
+        WebSocketClientProtocol, connectWS, WebSocketServerFactory, \
+        WebSocketServerProtocol
 
 LOGGER = logging.getLogger(__name__)
 
@@ -444,6 +446,99 @@ def sendRequest(self, method, uri, **kwargs):
         self.sendMessage(msg.encode('utf-8'))
         return uuidstr
 
+class AriServerFactory(WebSocketServerFactory):
+    """Twisted protocol factory for building ARI WebSocket clients."""
+
+    def __init__(self, receiver, uri, protocols, server_name, reactor):
+        """Constructor
+
+        :param receiver The object that will receive events from the protocol
+        :param uri: URI to be served.
+        :param protocols: List of protocols to accept.
+        :param reactor: The twisted reactor.
+        """
+        try:
+            WebSocketServerFactory.__init__(self, uri, protocols, server_name,
+                                            reactor=reactor)
+        except TypeError:
+            WebSocketServerFactory.__init__(self, uri, protocols=['ari'])
+        self.attempts = 0
+        self.start = None
+        self.receiver = receiver
+
+    def buildProtocol(self, addr):
+        """Make the protocol"""
+        return AriServerProtocol(self.receiver, self)
+
+class AriServerProtocol(WebSocketServerProtocol):
+    """Twisted protocol for handling a ARI WebSocket connection."""
+
+    def __init__(self, receiver, factory):
+        """Constructor.
+
+        :param receiver The event receiver
+        """
+        try:
+            super(AriServerProtocol, self).__init__()
+        except TypeError as te:
+            # Older versions of Autobahn use old style classes with no initializer.
+            # Newer versions must have their initializer called by derived
+            # implementations.
+            LOGGER.debug("AriServerProtocol: TypeError thrown in init: {0}".format(te))
+        LOGGER.debug("Made me a client protocol!")
+        self.receiver = receiver
+        self.factory = factory
+
+    def onConnect(self, request):
+        """Called back when connection is open."""
+        LOGGER.debug("New WebSocket Connected")
+        self.receiver.on_ws_connect(request)
+
+    def onOpen(self):
+        """Called back when connection is open."""
+        LOGGER.debug("WebSocket Open")
+        self.receiver.on_ws_open(self)
+
+    def onClose(self, wasClean, code, reason):
+        """Called back when connection is closed."""
+        LOGGER.debug("WebSocket closed(%r, %d, %s)", wasClean, code, reason)
+        self.receiver.on_ws_closed(self)
+
+    def onMessage(self, msg, binary):
+        """Called back when message is received.
+
+        :param msg: Received text message.
+        """
+        LOGGER.debug("rxed: %s", msg)
+        msg = json.loads(msg)
+        self.receiver.on_ws_event(msg)
+
+    def sendRequest(self, method, uri, **kwargs):
+        """Send a REST Request over Websocket.
+
+        :param method: Method.
+        :param path: Resource URI without query string.
+        :param kwargs: Additional request parameters
+        :returns: Request UUID
+        """
+        uuidstr = kwargs.pop('request_id', str(uuid.uuid4()))
+        req = {
+            'type': 'RESTRequest',
+            'request_id': uuidstr,
+            'method': method,
+            'uri': uri
+        }
+
+        for k,v in kwargs.items():
+            req[k] = v
+
+        msg = json.dumps(req)
+        LOGGER.info("Sending request message: %s", msg)
+        self.sendMessage(msg.encode('utf-8'))
+        return uuidstr
+
+
+
 class ARI(object):
     """Bare bones object for an ARI interface."""
 

tests/rest_api/tests.yaml

@@ -1,6 +1,6 @@
 # Enter tests here in the order they should be considered for execution:
 tests:
-    - dir:  'websocket_requests'
+    - dir:  'websocket'
     - test: 'continue'
     - test: 'authentication'
     - test: 'CORS'

tests/rest_api/websocket/outbound_websocket/local_ari_user_read_only/configs/ast1/ari.conf

@@ -0,0 +1,24 @@
+[general]
+enabled = yes
+
+[user1]
+type = user
+password = userpass
+read_only = yes
+
+[connection1]
+type = outbound_websocket
+uri = ws://localhost:9000/ari
+apps = testapp
+subscribe_all = no
+protocols = ari
+; base64(myuser:mypass) = bXl1c2VyOm15cGFzcw==
+username = myuser
+password = mypass
+local_ari_user = user1
+connection_type = per_call_config
+connection_timeout = 500
+reconnect_interval = 1000
+reconnect_attempts = 5
+tls_enabled = no
+

tests/rest_api/websocket/outbound_websocket/local_ari_user_read_only/configs/ast1/extensions.conf

@@ -0,0 +1,8 @@
+[default]
+exten = echo,1,NoOp(Echo test)
+same  = n,Answer()
+same  = n,Echo()
+
+exten = stasis,1,NoOp(Stasis test)
+same  = n,Stasis(testapp)
+same  = n,Verbose(Return from stasis: ${STASISSTATUS})

tests/rest_api/websocket_requests/bad_requests/configs/ast1/http.conf renamed to tests/rest_api/websocket/outbound_websocket/local_ari_user_read_only/configs/ast1/http.conf

@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+
+"""A test that creates a websocket server and waits for events
+
+Copyright (C) 2025, Sangoma Technologies Corporation
+George Joseph <[email protected]>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+import sys
+import logging
+import json
+import os
+import time
+from twisted.internet import reactor
+
+sys.path.append("lib/python")
+
+from asterisk.test_case import TestCase
+from asterisk.ari import AriServerFactory
+
+LOGGER = logging.getLogger(__name__)
+
+class PerCallWebsocketTest(TestCase):
+
+    def __init__(self):
+        TestCase.__init__(self)
+        self.protocol = None
+        self.connection_count = 0
+        self.peer = ""
+        self.auth_header = ""
+        self.auth_failed = False
+        self.channel_id = ""
+
+        self.reactor_timeout = 10
+        if self.test_config.config and 'test-object-config' in self.test_config.config:
+            self.reactor_timeout = self.test_config.config['test-object-config'].get('reactor-timeout', 10)
+
+        self.factory = AriServerFactory(self, "ws://localhost:9000/ari",
+                                        ['ari'], "localhost", reactor)
+        def on_startup(ast):
+            LOGGER.info("Asterisk started")
+            # We want the stasis extension to answer first
+            self.ast[0].cli_exec("channel originate Local/stasis@default extension echo@default")
+
+        LOGGER.info("Registering startup observer")
+        self.register_start_observer(on_startup)
+        self.create_asterisk()
+
+    def run(self):
+        """Entry point for the twisted reactor"""
+        super(PerCallWebsocketTest, self).run()
+        LOGGER.info("Binding websocket server")
+        reactor.listenTCP(9000, self.factory)
+
+    def on_reactor_timeout(self):
+        LOGGER.info("on_reactor_timeout")
+        self.passed = False
+        self.stop_reactor()
+
+    def on_ws_connect(self, request):
+        self.connection_count +=1
+        LOGGER.info("WebSocket connection %d from %s attempted" % (self.connection_count, request.peer))
+        self.peer = request.peer
+        """
+        Normally authentication would be done by the web server
+        but trying to integrate that here would be too complicated.
+        Instead we simply check the raw value of the Authorization
+        header.
+        """
+        if 'authorization' in request.headers:
+            auth_header = request.headers['authorization']
+            # base64(myuser:mypass) = bXl1c2VyOm15cGFzcw==
+            if auth_header != "Basic bXl1c2VyOm15cGFzcw==":
+                LOGGER.error("Authentication failed")
+                self.auth_failed = True
+        return "ari"
+
+    def on_ws_open(self, protocol):
+        LOGGER.info("WebSocket connection %d from %s opened" % (self.connection_count, self.peer))
+        self.protocol = protocol
+
+    def on_ws_closed(self, protocol):
+        LOGGER.info("WebSocket connection %d from %s closed.  Stopping reactor" % (self.connection_count, self.peer))
+        self.stop_reactor()
+
+    def on_ws_event(self, message):
+        if self.auth_failed:
+            self.passed = False
+            self.ast[0].cli_exec("ari shut sessions")
+            return
+        msg_type = message.get('type')
+        LOGGER.info("Received message: %s" % json.dumps(message))
+
+        def closer():
+            self.protocol.sendClose()
+
+        if msg_type == "StasisStart":
+            self.channel_id = message['channel']['id']
+            self.protocol.sendRequest("POST", "channels/%s/answer" % self.channel_id)
+        elif msg_type == "RESTResponse":
+            if message['status_code'] == 403:
+                LOGGER.info("Received '403 Forbidden' (good)")
+                self.passed = True
+            else:
+                LOGGER.error("Received '%d %s' instead of '403 Forbidden'" % (message['status_code'], message['reason_phrase']))
+                self.passed = False
+            LOGGER.info("Hanging up channel")
+            self.ast[0].cli_exec("channel request hangup all")
+
+def main():
+    test = PerCallWebsocketTest()
+    reactor.run()
+
+    if not test.passed:
+        return 1
+
+    return 0
+
+if __name__ == "__main__":
+    sys.exit(main() or 0)

tests/rest_api/websocket/outbound_websocket/local_ari_user_read_only/run-test

@@ -0,0 +1,33 @@
+testinfo:
+    summary: Test that local_ari_user read-only is respected
+    description: |
+        Test Steps
+         * The test script originates a call between the 'stasis' and 'echo'
+           extensions. This causes asterisk to open a websocket connection to the
+           test script and send ApplicationRegistered and StasisStart.
+         * The test script verifies that the Authorization header is present and correct.
+         * The test script detects the first ApplicationRegistered which it ignores.
+         * The test script detects StasisStart and answers the call using a
+           REST over Websocket request.
+         * Asterisk determines that the configured local_ari_user has only read-only
+           permissions and responds with a 403 Forbidden in a RESTResponse.
+         * The test script detects the 403 and marks the test as passed then issues
+           a 'channel request hangup all' CLI command.
+         * Asterisk hangs up the call, sends a StasisEnd and closes teh websocket.
+         * The test script detects the websocket close and stops the reactor.
+
+test-object-config:
+  reactor-timeout: 10
+
+properties:
+    dependencies:
+        - python : autobahn
+        - python : requests
+        - python : twisted
+        - asterisk : res_ari
+        - asterisk : res_http_websocket
+        - asterisk : res_ari_events
+        - asterisk : res_ari_asterisk
+        - asterisk : chan_pjsip
+    tags:
+        - ARI

tests/rest_api/websocket/outbound_websocket/local_ari_user_read_only/test-config.yaml

@@ -0,0 +1,23 @@
+[general]
+enabled = yes
+
+[user1]
+type = user
+password = userpass
+
+[connection1]
+type = outbound_websocket
+uri = ws://localhost:9000/ari
+apps = testapp
+subscribe_all = no
+protocols = ari
+; base64(myuser:mypass) = bXl1c2VyOm15cGFzcw==
+username = myuser
+password = mypass
+local_ari_user = user1
+connection_type = per_call_config
+connection_timeout = 500
+reconnect_interval = 1000
+reconnect_attempts = 5
+tls_enabled = no
+

tests/rest_api/websocket/outbound_websocket/per_call/configs/ast1/ari.conf

@@ -0,0 +1,8 @@
+[default]
+exten = echo,1,NoOp(Echo test)
+same  = n,Answer()
+same  = n,Echo()
+
+exten = stasis,1,NoOp(Stasis test)
+same  = n,Stasis(testapp)
+same  = n,Verbose(Return from stasis: ${STASISSTATUS})