[thirdparty/pdns.git] / regression-tests.dnsdist / test_API.py

#!/usr/bin/env python
import os.path

import json
import requests
from dnsdisttests import DNSDistTest

class TestAPIBasics(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    # paths accessible using the API key only
    _apiOnlyPaths = ['/api/v1/servers/localhost/config', '/api/v1/servers/localhost/config/allow-from', '/api/v1/servers/localhost/statistics']
    # paths accessible using an API key or basic auth
    _statsPaths = [ '/jsonstat?command=stats', '/jsonstat?command=dynblocklist', '/api/v1/servers/localhost']
    # paths accessible using basic auth only (list not exhaustive)
    _basicOnlyPaths = ['/', '/index.html']
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
    _config_template = """
    setACL({"127.0.0.1/32", "::1/128"})
    newServer{address="127.0.0.1:%s"}
    webserver("127.0.0.1:%s", "%s", "%s")
    """

    def testBasicAuth(self):
        """
        API: Basic Authentication
        """
        for path in self._basicOnlyPaths + self._statsPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
            self.assertTrue(r)
            self.assertEquals(r.status_code, 200)

    def testXAPIKey(self):
        """
        API: X-Api-Key
        """
        headers = {'x-api-key': self._webServerAPIKey}
        for path in self._apiOnlyPaths + self._statsPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, headers=headers, timeout=self._webTimeout)
            self.assertTrue(r)
            self.assertEquals(r.status_code, 200)

    def testBasicAuthOnly(self):
        """
        API: Basic Authentication Only
        """
        headers = {'x-api-key': self._webServerAPIKey}
        for path in self._basicOnlyPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, headers=headers, timeout=self._webTimeout)
            self.assertEquals(r.status_code, 401)

    def testAPIKeyOnly(self):
        """
        API: API Key Only
        """
        for path in self._apiOnlyPaths:
            url = 'http://127.0.0.1:' + str(self._webServerPort) + path
            r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
            self.assertEquals(r.status_code, 401)

    def testServersLocalhost(self):
        """
        API: /api/v1/servers/localhost
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        self.assertEquals(content['daemon_type'], 'dnsdist')

        rule_groups = ['response-rules', 'cache-hit-response-rules', 'self-answered-response-rules']
        for key in ['version', 'acl', 'local', 'rules', 'servers', 'frontends', 'pools'] + rule_groups:
            self.assertIn(key, content)

        for rule in content['rules']:
            for key in ['id', 'matches', 'rule', 'action', 'uuid']:
                self.assertIn(key, rule)
            for key in ['id', 'matches']:
                self.assertTrue(rule[key] >= 0)

        for rule_group in rule_groups:
            for rule in content[rule_group]:
                for key in ['id', 'matches', 'rule', 'action', 'uuid']:
                    self.assertIn(key, rule)
                for key in ['id', 'matches']:
                    self.assertTrue(rule[key] >= 0)

        for server in content['servers']:
            for key in ['id', 'latency', 'name', 'weight', 'outstanding', 'qpsLimit',
                        'reuseds', 'state', 'address', 'pools', 'qps', 'queries', 'order', 'sendErrors',
                        'dropRate']:
                self.assertIn(key, server)

            for key in ['id', 'latency', 'weight', 'outstanding', 'qpsLimit', 'reuseds',
                        'qps', 'queries', 'order']:
                self.assertTrue(server[key] >= 0)

            self.assertTrue(server['state'] in ['up', 'down', 'UP', 'DOWN'])

        for frontend in content['frontends']:
            for key in ['id', 'address', 'udp', 'tcp', 'queries']:
                self.assertIn(key, frontend)

            for key in ['id', 'queries']:
                self.assertTrue(frontend[key] >= 0)

        for pool in content['pools']:
            for key in ['id', 'name', 'cacheSize', 'cacheEntries', 'cacheHits', 'cacheMisses', 'cacheDeferredInserts', 'cacheDeferredLookups', 'cacheLookupCollisions', 'cacheInsertCollisions', 'cacheTTLTooShorts']:
                self.assertIn(key, pool)

            for key in ['id', 'cacheSize', 'cacheEntries', 'cacheHits', 'cacheMisses', 'cacheDeferredInserts', 'cacheDeferredLookups', 'cacheLookupCollisions', 'cacheInsertCollisions', 'cacheTTLTooShorts']:
                self.assertTrue(pool[key] >= 0)

    def testServersIDontExist(self):
        """
        API: /api/v1/servers/idontexist (should be 404)
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/idontexist'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertEquals(r.status_code, 404)

    def testServersLocalhostConfig(self):
        """
        API: /api/v1/servers/localhost/config
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        values = {}
        for entry in content:
            for key in ['type', 'name', 'value']:
                self.assertIn(key, entry)

            self.assertEquals(entry['type'], 'ConfigSetting')
            values[entry['name']] = entry['value']

        for key in ['acl', 'control-socket', 'ecs-override', 'ecs-source-prefix-v4',
                    'ecs-source-prefix-v6', 'fixup-case', 'max-outstanding', 'server-policy',
                    'stale-cache-entries-ttl', 'tcp-recv-timeout', 'tcp-send-timeout',
                    'truncate-tc', 'verbose', 'verbose-health-checks']:
            self.assertIn(key, values)

        for key in ['max-outstanding', 'stale-cache-entries-ttl', 'tcp-recv-timeout',
                    'tcp-send-timeout']:
            self.assertTrue(values[key] >= 0)

        self.assertTrue(values['ecs-source-prefix-v4'] >= 0 and values['ecs-source-prefix-v4'] <= 32)
        self.assertTrue(values['ecs-source-prefix-v6'] >= 0 and values['ecs-source-prefix-v6'] <= 128)

    def testServersLocalhostConfigAllowFrom(self):
        """
        API: /api/v1/servers/localhost/config/allow-from
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        for key in ['type', 'name', 'value']:
            self.assertIn(key, content)

        self.assertEquals(content['name'], 'allow-from')
        self.assertEquals(content['type'], 'ConfigSetting')
        acl = content['value']
        expectedACL = ["127.0.0.1/32", "::1/128"]
        acl.sort()
        expectedACL.sort()
        self.assertEquals(acl, expectedACL)

    def testServersLocalhostConfigAllowFromPut(self):
        """
        API: PUT /api/v1/servers/localhost/config/allow-from (should be refused)

        The API is read-only by default, so this should be refused
        """
        newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
        payload = json.dumps({"name": "allow-from",
                              "type": "ConfigSetting",
                              "value": newACL})
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
        self.assertFalse(r)
        self.assertEquals(r.status_code, 405)

    def testServersLocalhostStatistics(self):
        """
        API: /api/v1/servers/localhost/statistics
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/statistics'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        values = {}
        for entry in content:
            self.assertIn('type', entry)
            self.assertIn('name', entry)
            self.assertIn('value', entry)
            self.assertEquals(entry['type'], 'StatisticItem')
            values[entry['name']] = entry['value']

        expected = ['responses', 'servfail-responses', 'queries', 'acl-drops',
                    'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
                    'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
                    'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
                    'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
                    'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
                    'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
                    'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
                    'dyn-block-nmg-size', 'rule-servfail']

        for key in expected:
            self.assertIn(key, values)
            self.assertTrue(values[key] >= 0)

        for key in values:
            self.assertIn(key, expected)

    def testJsonstatStats(self):
        """
        API: /jsonstat?command=stats
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=stats'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        expected = ['responses', 'servfail-responses', 'queries', 'acl-drops',
                    'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
                    'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
                    'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
                    'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
                    'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
                    'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
                    'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
                    'dyn-block-nmg-size', 'packetcache-hits', 'packetcache-misses', 'over-capacity-drops',
                    'too-old-drops']

        for key in expected:
            self.assertIn(key, content)
            self.assertTrue(content[key] >= 0)

    def testJsonstatDynblocklist(self):
        """
        API: /jsonstat?command=dynblocklist
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=dynblocklist'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)

        content = r.json()

        if content:
            for key in ['reason', 'seconds', 'blocks', 'action']:
                self.assertIn(key, content)

            for key in ['blocks']:
                self.assertTrue(content[key] >= 0)

class TestAPIServerDown(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    # paths accessible using the API key
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
    _config_template = """
    setACL({"127.0.0.1/32", "::1/128"})
    newServer{address="127.0.0.1:%s"}
    getServer(0):setDown()
    webserver("127.0.0.1:%s", "%s", "%s")
    """

    def testServerDownNoLatencyLocalhost(self):
        """
        API: /api/v1/servers/localhost, no latency for a down server
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        self.assertEquals(content['servers'][0]['latency'], None)

class TestAPIWritable(DNSDistTest):

    _webTimeout = 2.0
    _webServerPort = 8083
    _webServerBasicAuthPassword = 'secret'
    _webServerAPIKey = 'apisecret'
    _APIWriteDir = '/tmp'
    _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey', '_APIWriteDir']
    _config_template = """
    setACL({"127.0.0.1/32", "::1/128"})
    newServer{address="127.0.0.1:%s"}
    webserver("127.0.0.1:%s", "%s", "%s")
    setAPIWritable(true, "%s")
    """

    def testSetACL(self):
        """
        API: Set ACL
        """
        headers = {'x-api-key': self._webServerAPIKey}
        url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        acl = content['value']
        expectedACL = ["127.0.0.1/32", "::1/128"]
        acl.sort()
        expectedACL.sort()
        self.assertEquals(acl, expectedACL)

        newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
        payload = json.dumps({"name": "allow-from",
                              "type": "ConfigSetting",
                              "value": newACL})
        r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        self.assertEquals(content['value'], newACL)

        r = requests.get(url, headers=headers, timeout=self._webTimeout)
        self.assertTrue(r)
        self.assertEquals(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()
        self.assertEquals(content['value'], newACL)

        configFile = self._APIWriteDir + '/' + 'acl.conf'
        self.assertTrue(os.path.isfile(configFile))
        fileContent = None
        with open(configFile, 'rt') as f:
            fileContent = f.read()

        self.assertEquals(fileContent, """-- Generated by the REST API, DO NOT EDIT
setACL({"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"})
""")
Commit	Line	Data
02bbf9eb	1	#!/usr/bin/env python
56d68fad	2	import os.path
02bbf9eb	3
56d68fad	4	import json
02bbf9eb RG	5	import requests
	6	from dnsdisttests import DNSDistTest
	7
56d68fad	8	class TestAPIBasics(DNSDistTest):
02bbf9eb RG	9
	10	_webTimeout = 2.0
	11	_webServerPort = 8083
	12	_webServerBasicAuthPassword = 'secret'
	13	_webServerAPIKey = 'apisecret'
55afa518 RG	14	# paths accessible using the API key only
	15	_apiOnlyPaths = ['/api/v1/servers/localhost/config', '/api/v1/servers/localhost/config/allow-from', '/api/v1/servers/localhost/statistics']
	16	# paths accessible using an API key or basic auth
	17	_statsPaths = [ '/jsonstat?command=stats', '/jsonstat?command=dynblocklist', '/api/v1/servers/localhost']
02bbf9eb RG	18	# paths accessible using basic auth only (list not exhaustive)
	19	_basicOnlyPaths = ['/', '/index.html']
	20	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
	21	_config_template = """
56d68fad	22	setACL({"127.0.0.1/32", "::1/128"})
02bbf9eb RG	23	newServer{address="127.0.0.1:%s"}
	24	webserver("127.0.0.1:%s", "%s", "%s")
	25	"""
	26
	27	def testBasicAuth(self):
	28	"""
	29	API: Basic Authentication
	30	"""
55afa518	31	for path in self._basicOnlyPaths + self._statsPaths:
02bbf9eb RG	32	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	33	r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
	34	self.assertTrue(r)
	35	self.assertEquals(r.status_code, 200)
	36
	37	def testXAPIKey(self):
	38	"""
	39	API: X-Api-Key
	40	"""
	41	headers = {'x-api-key': self._webServerAPIKey}
55afa518	42	for path in self._apiOnlyPaths + self._statsPaths:
02bbf9eb RG	43	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	44	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	45	self.assertTrue(r)
	46	self.assertEquals(r.status_code, 200)
	47
	48	def testBasicAuthOnly(self):
	49	"""
	50	API: Basic Authentication Only
	51	"""
	52	headers = {'x-api-key': self._webServerAPIKey}
	53	for path in self._basicOnlyPaths:
	54	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	55	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	56	self.assertEquals(r.status_code, 401)
	57
55afa518 RG	58	def testAPIKeyOnly(self):
	59	"""
	60	API: API Key Only
	61	"""
	62	for path in self._apiOnlyPaths:
	63	url = 'http://127.0.0.1:' + str(self._webServerPort) + path
	64	r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
	65	self.assertEquals(r.status_code, 401)
	66
02bbf9eb RG	67	def testServersLocalhost(self):
	68	"""
	69	API: /api/v1/servers/localhost
	70	"""
	71	headers = {'x-api-key': self._webServerAPIKey}
	72	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
	73	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	74	self.assertTrue(r)
	75	self.assertEquals(r.status_code, 200)
	76	self.assertTrue(r.json())
	77	content = r.json()
	78
	79	self.assertEquals(content['daemon_type'], 'dnsdist')
	80
2d4783a8	81	rule_groups = ['response-rules', 'cache-hit-response-rules', 'self-answered-response-rules']
d18eab67	82	for key in ['version', 'acl', 'local', 'rules', 'servers', 'frontends', 'pools'] + rule_groups:
02bbf9eb RG	83	self.assertIn(key, content)
	84
	85	for rule in content['rules']:
4d5959e6	86	for key in ['id', 'matches', 'rule', 'action', 'uuid']:
02bbf9eb RG	87	self.assertIn(key, rule)
	88	for key in ['id', 'matches']:
	89	self.assertTrue(rule[key] >= 0)
46e8b49e	90
d18eab67 CH	91	for rule_group in rule_groups:
	92	for rule in content[rule_group]:
	93	for key in ['id', 'matches', 'rule', 'action', 'uuid']:
	94	self.assertIn(key, rule)
	95	for key in ['id', 'matches']:
	96	self.assertTrue(rule[key] >= 0)
4ace9fe8	97
02bbf9eb RG	98	for server in content['servers']:
02bbf9eb RG	99	for key in ['id', 'latency', 'name', 'weight', 'outstanding', 'qpsLimit',
6a78f305 PL	100	'reuseds', 'state', 'address', 'pools', 'qps', 'queries', 'order', 'sendErrors',
6a78f305 PL	101	'dropRate']:
02bbf9eb RG	102	self.assertIn(key, server)
	103
	104	for key in ['id', 'latency', 'weight', 'outstanding', 'qpsLimit', 'reuseds',
	105	'qps', 'queries', 'order']:
	106	self.assertTrue(server[key] >= 0)
	107
	108	self.assertTrue(server['state'] in ['up', 'down', 'UP', 'DOWN'])
	109
	110	for frontend in content['frontends']:
	111	for key in ['id', 'address', 'udp', 'tcp', 'queries']:
	112	self.assertIn(key, frontend)
	113
	114	for key in ['id', 'queries']:
	115	self.assertTrue(frontend[key] >= 0)
	116
4ace9fe8 RG	117	for pool in content['pools']:
	118	for key in ['id', 'name', 'cacheSize', 'cacheEntries', 'cacheHits', 'cacheMisses', 'cacheDeferredInserts', 'cacheDeferredLookups', 'cacheLookupCollisions', 'cacheInsertCollisions', 'cacheTTLTooShorts']:
	119	self.assertIn(key, pool)
	120
	121	for key in ['id', 'cacheSize', 'cacheEntries', 'cacheHits', 'cacheMisses', 'cacheDeferredInserts', 'cacheDeferredLookups', 'cacheLookupCollisions', 'cacheInsertCollisions', 'cacheTTLTooShorts']:
	122	self.assertTrue(pool[key] >= 0)
	123
00566cbf PL	124	def testServersIDontExist(self):
	125	"""
	126	API: /api/v1/servers/idontexist (should be 404)
	127	"""
	128	headers = {'x-api-key': self._webServerAPIKey}
	129	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/idontexist'
	130	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	131	self.assertEquals(r.status_code, 404)
	132
02bbf9eb RG	133	def testServersLocalhostConfig(self):
	134	"""
	135	API: /api/v1/servers/localhost/config
	136	"""
	137	headers = {'x-api-key': self._webServerAPIKey}
	138	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config'
	139	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	140	self.assertTrue(r)
	141	self.assertEquals(r.status_code, 200)
	142	self.assertTrue(r.json())
	143	content = r.json()
	144	values = {}
	145	for entry in content:
	146	for key in ['type', 'name', 'value']:
	147	self.assertIn(key, entry)
	148
	149	self.assertEquals(entry['type'], 'ConfigSetting')
	150	values[entry['name']] = entry['value']
	151
	152	for key in ['acl', 'control-socket', 'ecs-override', 'ecs-source-prefix-v4',
	153	'ecs-source-prefix-v6', 'fixup-case', 'max-outstanding', 'server-policy',
	154	'stale-cache-entries-ttl', 'tcp-recv-timeout', 'tcp-send-timeout',
	155	'truncate-tc', 'verbose', 'verbose-health-checks']:
	156	self.assertIn(key, values)
	157
	158	for key in ['max-outstanding', 'stale-cache-entries-ttl', 'tcp-recv-timeout',
	159	'tcp-send-timeout']:
	160	self.assertTrue(values[key] >= 0)
	161
	162	self.assertTrue(values['ecs-source-prefix-v4'] >= 0 and values['ecs-source-prefix-v4'] <= 32)
	163	self.assertTrue(values['ecs-source-prefix-v6'] >= 0 and values['ecs-source-prefix-v6'] <= 128)
	164
56d68fad RG	165	def testServersLocalhostConfigAllowFrom(self):
	166	"""
	167	API: /api/v1/servers/localhost/config/allow-from
	168	"""
	169	headers = {'x-api-key': self._webServerAPIKey}
	170	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	171	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	172	self.assertTrue(r)
	173	self.assertEquals(r.status_code, 200)
	174	self.assertTrue(r.json())
	175	content = r.json()
	176	for key in ['type', 'name', 'value']:
	177	self.assertIn(key, content)
	178
	179	self.assertEquals(content['name'], 'allow-from')
	180	self.assertEquals(content['type'], 'ConfigSetting')
078efd26 RG	181	acl = content['value']
	182	expectedACL = ["127.0.0.1/32", "::1/128"]
	183	acl.sort()
	184	expectedACL.sort()
	185	self.assertEquals(acl, expectedACL)
56d68fad RG	186
	187	def testServersLocalhostConfigAllowFromPut(self):
	188	"""
	189	API: PUT /api/v1/servers/localhost/config/allow-from (should be refused)
	190
	191	The API is read-only by default, so this should be refused
	192	"""
	193	newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
	194	payload = json.dumps({"name": "allow-from",
	195	"type": "ConfigSetting",
	196	"value": newACL})
	197	headers = {'x-api-key': self._webServerAPIKey}
	198	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	199	r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
	200	self.assertFalse(r)
	201	self.assertEquals(r.status_code, 405)
	202
02bbf9eb RG	203	def testServersLocalhostStatistics(self):
	204	"""
	205	API: /api/v1/servers/localhost/statistics
	206	"""
	207	headers = {'x-api-key': self._webServerAPIKey}
	208	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/statistics'
	209	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	210	self.assertTrue(r)
	211	self.assertEquals(r.status_code, 200)
	212	self.assertTrue(r.json())
	213	content = r.json()
	214	values = {}
	215	for entry in content:
	216	self.assertIn('type', entry)
	217	self.assertIn('name', entry)
	218	self.assertIn('value', entry)
	219	self.assertEquals(entry['type'], 'StatisticItem')
	220	values[entry['name']] = entry['value']
	221
0c369ddc	222	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops',
dd46e5e3	223	'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
02bbf9eb RG	224	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
	225	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
	226	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
	227	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
	228	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
	229	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
8fbade0d	230	'dyn-block-nmg-size', 'rule-servfail']
02bbf9eb RG	231
	232	for key in expected:
	233	self.assertIn(key, values)
	234	self.assertTrue(values[key] >= 0)
	235
dd46e5e3 RG	236	for key in values:
	237	self.assertIn(key, expected)
	238
02bbf9eb RG	239	def testJsonstatStats(self):
	240	"""
	241	API: /jsonstat?command=stats
	242	"""
	243	headers = {'x-api-key': self._webServerAPIKey}
	244	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=stats'
	245	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	246	self.assertTrue(r)
	247	self.assertEquals(r.status_code, 200)
	248	self.assertTrue(r.json())
	249	content = r.json()
	250
0c369ddc	251	expected = ['responses', 'servfail-responses', 'queries', 'acl-drops',
dd46e5e3	252	'rule-drop', 'rule-nxdomain', 'rule-refused', 'self-answered', 'downstream-timeouts',
02bbf9eb RG	253	'downstream-send-errors', 'trunc-failures', 'no-policy', 'latency0-1',
	254	'latency1-10', 'latency10-50', 'latency50-100', 'latency100-1000',
	255	'latency-slow', 'latency-avg100', 'latency-avg1000', 'latency-avg10000',
	256	'latency-avg1000000', 'uptime', 'real-memory-usage', 'noncompliant-queries',
	257	'noncompliant-responses', 'rdqueries', 'empty-queries', 'cache-hits',
	258	'cache-misses', 'cpu-user-msec', 'cpu-sys-msec', 'fd-usage', 'dyn-blocked',
dd46e5e3 RG	259	'dyn-block-nmg-size', 'packetcache-hits', 'packetcache-misses', 'over-capacity-drops',
dd46e5e3 RG	260	'too-old-drops']
02bbf9eb RG	261
	262	for key in expected:
	263	self.assertIn(key, content)
	264	self.assertTrue(content[key] >= 0)
	265
	266	def testJsonstatDynblocklist(self):
	267	"""
	268	API: /jsonstat?command=dynblocklist
	269	"""
	270	headers = {'x-api-key': self._webServerAPIKey}
	271	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/jsonstat?command=dynblocklist'
	272	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	273	self.assertTrue(r)
	274	self.assertEquals(r.status_code, 200)
	275
	276	content = r.json()
	277
	278	if content:
477c86a0	279	for key in ['reason', 'seconds', 'blocks', 'action']:
02bbf9eb RG	280	self.assertIn(key, content)
	281
	282	for key in ['blocks']:
	283	self.assertTrue(content[key] >= 0)
56d68fad	284
36927800 RG	285	class TestAPIServerDown(DNSDistTest):
	286
	287	_webTimeout = 2.0
	288	_webServerPort = 8083
	289	_webServerBasicAuthPassword = 'secret'
	290	_webServerAPIKey = 'apisecret'
	291	# paths accessible using the API key
	292	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
	293	_config_template = """
	294	setACL({"127.0.0.1/32", "::1/128"})
	295	newServer{address="127.0.0.1:%s"}
	296	getServer(0):setDown()
	297	webserver("127.0.0.1:%s", "%s", "%s")
	298	"""
	299
	300	def testServerDownNoLatencyLocalhost(self):
	301	"""
	302	API: /api/v1/servers/localhost, no latency for a down server
	303	"""
	304	headers = {'x-api-key': self._webServerAPIKey}
	305	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost'
	306	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	307	self.assertTrue(r)
	308	self.assertEquals(r.status_code, 200)
	309	self.assertTrue(r.json())
	310	content = r.json()
	311
	312	self.assertEquals(content['servers'][0]['latency'], None)
	313
56d68fad RG	314	class TestAPIWritable(DNSDistTest):
	315
	316	_webTimeout = 2.0
	317	_webServerPort = 8083
	318	_webServerBasicAuthPassword = 'secret'
	319	_webServerAPIKey = 'apisecret'
	320	_APIWriteDir = '/tmp'
	321	_config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey', '_APIWriteDir']
	322	_config_template = """
	323	setACL({"127.0.0.1/32", "::1/128"})
	324	newServer{address="127.0.0.1:%s"}
	325	webserver("127.0.0.1:%s", "%s", "%s")
	326	setAPIWritable(true, "%s")
	327	"""
	328
	329	def testSetACL(self):
	330	"""
	331	API: Set ACL
	332	"""
	333	headers = {'x-api-key': self._webServerAPIKey}
	334	url = 'http://127.0.0.1:' + str(self._webServerPort) + '/api/v1/servers/localhost/config/allow-from'
	335	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	336	self.assertTrue(r)
	337	self.assertEquals(r.status_code, 200)
	338	self.assertTrue(r.json())
	339	content = r.json()
078efd26 RG	340	acl = content['value']
	341	expectedACL = ["127.0.0.1/32", "::1/128"]
	342	acl.sort()
	343	expectedACL.sort()
	344	self.assertEquals(acl, expectedACL)
56d68fad RG	345
	346	newACL = ["192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"]
	347	payload = json.dumps({"name": "allow-from",
	348	"type": "ConfigSetting",
	349	"value": newACL})
	350	r = requests.put(url, headers=headers, timeout=self._webTimeout, data=payload)
	351	self.assertTrue(r)
	352	self.assertEquals(r.status_code, 200)
	353	self.assertTrue(r.json())
	354	content = r.json()
	355	self.assertEquals(content['value'], newACL)
	356
	357	r = requests.get(url, headers=headers, timeout=self._webTimeout)
	358	self.assertTrue(r)
	359	self.assertEquals(r.status_code, 200)
	360	self.assertTrue(r.json())
	361	content = r.json()
	362	self.assertEquals(content['value'], newACL)
	363
	364	configFile = self._APIWriteDir + '/' + 'acl.conf'
	365	self.assertTrue(os.path.isfile(configFile))
	366	fileContent = None
b4f23783	367	with open(configFile, 'rt') as f:
56d68fad RG	368	fileContent = f.read()
	369
	370	self.assertEquals(fileContent, """-- Generated by the REST API, DO NOT EDIT
	371	setACL({"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"})
	372	""")