Add tests for http and ari ACL support

tests/http_server/acl/configs/ast1/http.conf

@@ -0,0 +1,20 @@
+[general]
+servername=Peabody
+enabled=yes
+bindaddr=127.0.0.1
+bindport=8088
+enablestatic=yes
+
+[/httpstatus1]
+type=restriction
+deny=0.0.0.0/0
+permit=127.0.0.1
+
+[/httpstatus2]
+type=restriction
+deny=0.0.0.0/0
+
+[/httpstatus3]
+type=restriction
+deny=0.0.0.0/0
+permit=127.0.0.2

tests/http_server/acl/run-test

@@ -0,0 +1,91 @@
+#!/usr/bin/env python
+"""
+Copyright (C) 2015, Digium, Inc.
+Ashley Sanders <asanders@digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+"""
+
+
+import logging
+import requests
+import sys
+import re
+
+sys.path.append("lib/python")
+
+from twisted.internet import reactor
+from asterisk.test_case import TestCase
+
+LOGGER = logging.getLogger(__name__)
+
+BASE_URL   = "http://127.0.0.1:8088/httpstatus%d"
+
+EXPECTED_STATUS_CODES = [404, 403, 403]
+
+class HttpClientTest(TestCase):
+    """Responsible for making a request to the http-server and then parsing
+    response headers to verify that the value returned for 'Server' matches
+    the expected value."""
+
+    def __init__(self):
+        """Constructor."""
+
+        TestCase.__init__(self)
+        self.create_asterisk(1)
+
+    def run(self):
+        """Runs the test."""
+
+        TestCase.run(self)
+
+        for i in range(0, 3):
+            status = EXPECTED_STATUS_CODES[i]
+
+            url = BASE_URL % (i+1)
+
+            self.run_scenario(url, status)
+            if self.passed == False:
+                break
+
+        LOGGER.debug("Stopping reactor...")
+        self.stop_reactor()
+
+    def run_scenario(self, url, expected_status):
+        """Runs an individual test scenario.
+
+        Keyword Arguments:
+        url             --  The url to use for making the server request
+        expected_name   --  The name that the server is expected to return
+                            in its response headers
+        """
+
+        LOGGER.debug("Running %s", self)
+
+        resp = requests.get(url)
+        LOGGER.debug("[resp]: [%r]", resp)
+
+        status = resp.status_code
+        LOGGER.debug("[status]: [%d]", status)
+
+        if status != expected_status:
+            LOGGER.debug("[%d] != [%d]", status, expected_status)
+            self.set_passed(False)
+            return
+
+        self.set_passed(True)
+        return
+
+def main():
+    """Main entry-point for the test."""
+
+    test = HttpClientTest()
+    reactor.run()
+
+    if test.passed:
+        return 0
+    return 1
+
+if __name__ == "__main__":
+    sys.exit(main() or 0)

tests/http_server/acl/test-config.yaml

@@ -0,0 +1,20 @@
+testinfo:
+    summary: Tests the http-server acl configuration.
+    description: |
+        This test connects as a client to a running instance of the asterisk
+        http-server. We then check to make sure that the expected ACL rules
+        are being applied correctly. This test is run three times, each time
+        testing a different base url and ACL configuration.
+        The three iterations verify that:
+        A blanket deny with a matching permit allows
+        A blanket deny with no permit denies
+        A blanket deny with a non-matching permit denies
+
+properties:
+    dependencies:
+        - python : twisted
+        - python : starpy
+        - python : requests
+    tags:
+        - HTTP_SERVER
+        - configuration

tests/http_server/tests.yaml

@@ -1,3 +1,5 @@
 # Enter tests here in the order they should be considered for execution:
 tests:
     - test: 'servername'
+    - test: 'acl'
+    - test: 'acl_ari_user'

tests/rest_api/authentication_user_acl/configs/ast1/ari.conf

@@ -0,0 +1,20 @@
+[general]
+enabled = yes
+
+[user1]
+type = user
+password = password
+deny = 0.0.0.0/0
+permit = 127.0.0.1
+
+[user2]
+type = user
+password = password
+deny = 0.0.0.0/0
+permit = 10.0.0.10
+
+[user3]
+type = user
+password = password
+deny = 127.0.0.1
+

tests/rest_api/authentication_user_acl/run-test

@@ -0,0 +1,94 @@
+#!/usr/bin/env python
+'''
+Copyright (C) 2013, Digium, Inc.
+David M. Lee, II <dlee@digium.com>
+
+This program is free software, distributed under the terms of
+the GNU General Public License Version 2.
+'''
+
+import logging
+import requests
+import sys
+
+from requests import codes
+from twisted.internet import reactor
+
+sys.path.append("lib/python")
+from asterisk.test_case import TestCase
+
+LOGGER = logging.getLogger(__name__)
+
+HOST='localhost'
+PORT=8088
+
+def build_url(*args):
+    return "http://%s:%d/%s" %\
+           (HOST, PORT, '/'.join([str(arg) for arg in args]))
+
+class Scenario(object):
+    def __init__(self, userpass, method, expected_response):
+        self.userpass = userpass
+        self.method = method
+        self.expected_response = expected_response
+
+    def __repr__(self):
+        return '{ userpass=%s, method=%s, expected=%d }' % (
+            self.userpass, self.method.__name__, self.expected_response)
+
+    def eval(self, str, test, resp):
+        if self.expected_response != resp.status_code:
+            LOGGER.error("Expected %d, got %d (%s). %s - %s" % (
+                self.expected_response, resp.status_code, resp.text, self, str))
+            test.passed = False
+
+    def run(self, test):
+        LOGGER.debug("Running %s" % self)
+        if self.userpass is None:
+            # Test no authentication
+            resp = self.method(build_url('ari', 'channels'))
+            self.eval('Unauthenticated', test, resp)
+            return
+
+        # Test basic auth
+        resp = self.method(build_url('ari', 'channels'),
+                           auth=self.userpass)
+        self.eval('Basic', test, resp)
+
+        # Test api_key auth
+        resp = self.method(build_url('ari', 'channels'),
+                           params={'api_key': "%s:%s" % self.userpass})
+        self.eval('api_key', test, resp)
+
+
+SCENARIOS=[
+    Scenario(('user1', 'password'), requests.get, codes.okay),
+    Scenario(('user2', 'password'), requests.get, codes.forbidden),
+    Scenario(('user3', 'password'), requests.get, codes.forbidden)
+]
+
+class ARIAuthenticationTest(TestCase):
+    def __init__(self):
+        TestCase.__init__(self)
+        self.passed = True
+        self.create_asterisk()
+
+    def run(self):
+        try:
+            for scenario in SCENARIOS:
+                scenario.run(self)
+        except:
+            logging.exception("Exception caught during test")
+            self.passed = False
+        finally:
+            self.stop_reactor()
+
+def main():
+    test = ARIAuthenticationTest()
+    reactor.run()
+    if test.passed:
+        return 0
+    return 1
+
+if __name__ == "__main__":
+    sys.exit(main() or 0)

tests/rest_api/authentication_user_acl/test-config.yaml

@@ -0,0 +1,18 @@
+testinfo:
+    summary: Test authentication for the RESTful API
+    description: |
+        This test attempts three sets of login attempts to the RESTful API via
+        each of its authentication mechanisms.
+        Tests include:
+        A blanket deny with a matching permit allows
+        A blanket deny with a non-matching permit denies
+        A blanket deny with no permit denies
+
+properties:
+    dependencies:
+        - python : twisted
+        - python : starpy
+        - python : requests
+        - asterisk : res_ari_channels
+    tags:
+        - ARI

tests/rest_api/tests.yaml

@@ -3,6 +3,7 @@ tests:
     - dir:  'websocket'
     - test: 'continue'
     - test: 'authentication'
+    - test: 'authentication_user_acl'
     - test: 'CORS'
     - dir:  'applications'
     - dir:  'asterisk'