Skip to content

S
slapos.toolbox
Commit a38d2035 authored by Eric Zheng's avatar Eric Zheng
Browse files
Options

move HTTP auth check into check_url_available 

It seems like the HTTP authentication check would fit naturally in the
existing check_url_available promise, since the logic is so similar.
parent 1af096b5
Hide whitespace changes
Inline Side-by-side
Showing with 192 additions and 77 deletions
slapos/promise/plugin/check_http_authentication.py deleted 100644 → 0
View file @ 1af096b5
from zope.interface import implementer
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
import requests
@implementer(interface.IPromise)
class RunPromise(GenericPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(float(self.getConfig('frequency', 2)))
def sense(self):
"""
Check basic HTTP authentication for a service. You should
probably run check_url_available.py first.
"""
url = self.getConfig('url')
username = self.getConfig('username')
password = self.getConfig('password')
try:
result = requests.get(url, auth=(username, password))
except requests.ConnectionError as _:
self.logger.error(
'ERROR connection not possible while accessing %r' % url)
return
except Exception as e:
self.logger.error('ERROR %r' % e)
return
credentials = '(%r, %r)' % (username, password)
if result.ok:
self.logger.info('%r authenticated with %s' % (url, credentials))
else:
self.logger.error('ERROR could not authenticate %r with %s' % \
(url, credentials))
def anomaly(self):
return self._test(result_count=3, failure_amount=3)
slapos/promise/plugin/check_url_available.py
View file @ a38d2035
"""
Some notable parameters:
promise-timeout:
Optional timeout (in seconds) for promise.
timeout:
Optional timeout (in seconds) for HTTP request.
verify, ca-cert-file, cert-file, key-file:
Optional SSL information. (See Python requests documentation.)
check-secure:
(default 0) If set, treat a 401 (forbidden) response as a
success. You probably don't want this if you're specifying a
username and password.
http_code:
(default 200) The expected response HTTP code.
ignore-code:
(default 1) Ignore the response HTTP code.
username, password:
If supplied, enables basic HTTP authentication.
require-auth:
(default 0) If set, check that the server responds with a 401
when receiving a request with no credentials. (Redundant if
you don't specify a username and password.)
"""
from zope.interface import implementer
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
......@@ -12,9 +37,52 @@ class RunPromise(GenericPromise):
# SR can set custom periodicity
self.setPeriodicity(float(self.getConfig('frequency', 2)))
def request_and_check_code(self, url, expected_http_code=None, **kwargs):
"""
Wrapper around GET requests, to make multiple requests easier. If
no expected code is given, use the http_code configuration
parameter, finally defaulting to 200.
Note: if you specify an expected_http_code here, ignore-code is
automatically overridden.
"""
if expected_http_code == None:
expected_http_code = int(self.getConfig('http_code', '200'))
ignore_code = int(self.getConfig('ignore-code', 0))
else:
ignore_code = 0
try:
result = requests.get(url, allow_redirects=True, **kwargs)
except requests.exceptions.SSLError as e:
if 'certificate verify failed' in str(e):
self.logger.error(
"ERROR SSL verify failed while accessing %r" % (url,))
else:
self.logger.error(
"ERROR Unknown SSL error %r while accessing %r" % (e, url))
except requests.ConnectionError as e:
self.logger.error(
"ERROR connection not possible while accessing %r" % (url, ))
except Exception as e:
self.logger.error("ERROR: %s" % (e,))
else:
# Check that the returned status code is what we expected
http_code = result.status_code
check_secure = int(self.getConfig('check-secure', 0))
if http_code == 401 and check_secure == 1:
self.logger.info("%r is protected (returned %s)." % (url, http_code))
elif not ignore_code and http_code != expected_http_code:
self.logger.error("%r is not available (returned %s, expected %s)." % (
url, http_code, expected_http_code))
else:
self.logger.info("%r is available" % (url,))
def sense(self):
"""
Check if frontend URL is available
Check if frontend URL is available.
"""
url = self.getConfig('url')
......@@ -23,11 +91,13 @@ class RunPromise(GenericPromise):
default_timeout = max(
1, min(5, int(self.getConfig('promise-timeout', 20)) - 1))
timeout = int(self.getConfig('timeout', default_timeout))
expected_http_code = int(self.getConfig('http_code', '200'))
ca_cert_file = self.getConfig('ca-cert-file')
cert_file = self.getConfig('cert-file')
key_file = self.getConfig('key-file')
verify = int(self.getConfig('verify', 0))
username = self.getConfig('username')
password = self.getConfig('password')
require_auth = int(self.getConfig('require-auth', 0))
if ca_cert_file:
verify = ca_cert_file
......@@ -41,36 +111,20 @@ class RunPromise(GenericPromise):
else:
cert = None
try:
result = requests.get(
url, verify=verify, allow_redirects=True, timeout=timeout, cert=cert)
except requests.exceptions.SSLError as e:
if 'certificate verify failed' in str(e):
self.logger.error(
"ERROR SSL verify failed while accessing %r" % (url,))
else:
self.logger.error(
"ERROR Unknown SSL error %r while accessing %r" % (e, url))
return
except requests.ConnectionError as e:
self.logger.error(
"ERROR connection not possible while accessing %r" % (url, ))
return
except Exception as e:
self.logger.error("ERROR: %s" % (e,))
return
if username and password:
credentials = (username, password)
else:
credentials = None
http_code = result.status_code
check_secure = int(self.getConfig('check-secure', 0))
ignore_code = int(self.getConfig('ignore-code', 0))
self.request_and_check_code(url, verify=verify, timeout=timeout,
cert=cert, auth=credentials)
if http_code == 401 and check_secure == 1:
self.logger.info("%r is protected (returned %s)." % (url, http_code))
elif not ignore_code and http_code != expected_http_code:
self.logger.error("%r is not available (returned %s, expected %s)." % (
url, http_code, expected_http_code))
else:
self.logger.info("%r is available" % (url,))
# If require-auth is set, verify that we get a 401 when requesting
# without credentials
if require_auth == 1:
self.request_and_check_code(url, expected_http_code=401,
verify=verify, timeout=timeout,
cert=cert, auth=None)
def anomaly(self):
return self._test(result_count=3, failure_amount=3)
slapos/test/promise/plugin/test_check_url_available.py
View file @ a38d2035
......@@ -37,6 +37,7 @@ from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from six.moves import BaseHTTPServer
from base64 import b64encode
import datetime
import ipaddress
import json
......@@ -53,6 +54,11 @@ SLAPOS_TEST_IPV4 = os.environ.get('SLAPOS_TEST_IPV4', '127.0.0.1')
SLAPOS_TEST_IPV4_PORT = 57965
HTTPS_ENDPOINT = "https://%s:%s/" % (SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV4_PORT)
# Good and bad username/password for HTTP authentication tests.
TEST_GOOD_USERNAME = 'good username'
TEST_GOOD_PASSWORD = 'good password'
TEST_BAD_USERNAME = 'bad username'
TEST_BAD_PASSWORD = 'bad password'
def createKey():
key = rsa.generate_private_key(
......@@ -135,6 +141,16 @@ class CertificateAuthority(object):
class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def do_GET(self):
path = self.path.split('/')[-1]
# This is a bit of a hack, but to ensure compatibility with previous
# tests, prepend an '!' to the path if you want the server to check
# for authentication.
if path[0] == '!':
require_auth = True
path = path[1:]
else:
require_auth = False
if '_' in path:
response, timeout = path.split('_')
response = int(response)
......@@ -143,15 +159,30 @@ class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
timeout = 0
response = int(path)
# The encoding/decoding trick is necessary for compatibility with
# Python 2 and 3.
key = b64encode(('%s:%s' % (TEST_GOOD_USERNAME,
TEST_GOOD_PASSWORD).encode())).decode()
try:
authorization = self.headers['Authorization']
except KeyError:
authorization = None
time.sleep(timeout)
self.send_response(response)
if require_auth and authorization != 'Basic ' + key:
self.send_response(401)
self.send_header('WWW-Authenticate', 'Basic realm="test"')
self.end_headers()
self.wfile.write('bad credentials\n'.encode())
else:
self.send_response(response)
self.send_header("Content-type", "application/json")
self.end_headers()
response = {
'Path': self.path,
}
self.wfile.write(str2bytes(json.dumps(response, indent=2)))
self.send_header("Content-type", "application/json")
self.end_headers()
response = {
'Path': self.path,
}
self.wfile.write(str2bytes(json.dumps(response, indent=2)))
class CheckUrlAvailableMixin(TestPromisePluginMixin):
......@@ -259,6 +290,17 @@ extra_config_dict = {
'ignore-code': %(ignore_code)s,
'http_code': %(http_code)s
}
"""
self.base_content_authenticate = """from slapos.promise.plugin.check_url_available import RunPromise
extra_config_dict = {
'url': '%(url)s',
'timeout': %(timeout)s,
'username': '%(username)s',
'password': '%(password)s',
'require-auth': %(require_auth)s
}
"""
def tearDown(self):
......@@ -484,6 +526,65 @@ class TestCheckUrlAvailable(CheckUrlAvailableMixin):
"%r is available" % (url,)
)
# Test normal authentication success.
def test_check_authenticate_success(self):
url = HTTPS_ENDPOINT + '!200'
content = self.base_content_authenticate % {
'url': url,
'username': TEST_GOOD_USERNAME,
'password': TEST_GOOD_PASSWORD,
'require_auth': 1
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertEqual(
result['result']['message'],
"%r is available" % (url,)
)
# Test authentication failure due to bad password.
def test_check_authenticate_bad_password(self):
url = HTTPS_ENDPOINT + '!200'
content = self.base_content_authenticate % {
'url': url,
'username': TEST_BAD_USERNAME,
'password': TEST_BAD_PASSWORD,
'require_auth': 1
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(
result['result']['message'],
"%r is not available (returned 401, expected 200)." % (url,)
)
# Test authentication failure due to the server not requiring any
# authentication.
def test_check_authenticate_no_password(self):
url = HTTPS_ENDPOINT + '200'
content = self.base_content_authenticate % {
'url': url,
'username': TEST_GOOD_USERNAME,
'password': TEST_GOOD_PASSWORD,
'require_auth': 1
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(
result['result']['message'],
"%r is not available (returned 200, expected 401)." % (url,)
)
class TestCheckUrlAvailableTimeout(CheckUrlAvailableMixin):
def test_check_200_timeout(self):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7