diff --git a/.gitignore b/.gitignore
index ae80164e502726efaa193dfbe2ddf3455a095011..d2c630c329ec00d89b5ed8026a41458fd8b04945 100644
--- a/.gitignore
+++ b/.gitignore
@@ -22,6 +22,7 @@ bin
*/Gemfile.lock
test/dummy/log/*
test/dummy/tmp/*
+*.pyc
# ignore all deploy specific configuration
config/couchdb.yml
diff --git a/test/nagios/soledad_sync.py b/test/nagios/soledad_sync.py
index 94679b1096de06892a6a06b3b181ec9ad2d42a37..617dd3a8578667df4339f83ee2bc6ecc8c2b3945 100755
--- a/test/nagios/soledad_sync.py
+++ b/test/nagios/soledad_sync.py
@@ -7,12 +7,12 @@
import tempfile
-import requests
import os
-import srp._pysrp as srp
import shutil
import u1db
-import webapp_login
+from support.api import Api
+from support.config import Config
+from support.user import User
from u1db.remote.http_target import HTTPSyncTarget
@@ -32,14 +32,6 @@ HTTPSyncTarget.set_token_credentials = set_token_credentials
HTTPSyncTarget._sign_request = _sign_request
-def fail(reason):
- print '2 soledad_sync - CRITICAL - ' + reason
- exit(2)
-
-# monkey patch webapp_login's fail function to report as soledad
-webapp_login.fail = fail
-
-
# The following function could fetch all info needed to sync using soledad.
# Despite that, we won't use all that info because we are instead faking a
# Soledad sync by using U1DB slightly modified syncing capabilities. Part of
@@ -47,58 +39,39 @@ webapp_login.fail = fail
# to actually use the Soledad client in the future.
def get_soledad_info(config, tempdir):
- # get login and get user info
- user = config['user']
- api = config['api']
- usr = srp.User( user['username'], user['password'], srp.SHA256, srp.NG_1024 )
- try:
- auth = webapp_login.parse(webapp_login.authenticate(api, usr))
- except requests.exceptions.ConnectionError:
- fail('no connection to server')
- # get soledad server url
- service_url = 'https://%s:%d/%d/config/soledad-service.json' % \
- (api['domain'], api['port'], api['version'])
- soledad_hosts = requests.get(service_url).json['hosts']
- host = soledad_hosts.keys()[0]
- server_url = 'https://%s:%d/user-%s' % \
- (soledad_hosts[host]['hostname'], soledad_hosts[host]['port'],
- auth['id'])
- # get provider ca certificate
- #ca_cert = requests.get('https://127.0.0.1/ca.crt', verify=False).text
- #cert_file = os.path.join(tempdir, 'ca.crt')
- cert_file = None # not used for now
- #with open(cert_file, 'w') as f:
- # f.write(ca_cert)
- return auth['id'], user['password'], server_url, cert_file, auth['token']
-
-
-def run_tests():
- tempdir = tempfile.mkdtemp()
- uuid, password, server_url, cert_file, token = \
- get_soledad_info(webapp_login.read_config(), tempdir)
- exc = None
- try:
- # in the future, we can replace the following by an actual Soledad
- # client sync, if needed
- db = u1db.open(os.path.join(tempdir, '%s.db' % uuid), True)
- creds = {'token': {'uuid': uuid, 'token': token}}
- db.sync(server_url, creds=creds, autocreate=False)
- except Exception as e:
- exc = e
- shutil.rmtree(tempdir)
- exit(report(exc))
-
-
-def report(exc):
- if exc is None:
- print '0 soledad_sync - OK - can sync soledad fine'
- return 0
- if isinstance(exc, u1db.errors.U1DBError):
- print '2 soledad_sync - CRITICAL - ' + exc.message
- else:
- print '2 soledad_sync - CRITICAL - ' + str(exc)
- return 2
-
+ # get login and get user info
+ user = User(config)
+ api = Api(config, verify=False)
+ auth = user.login(api)
+ # get soledad server url
+ soledad_hosts = api.get('config/soledad-service.json')['hosts']
+ host = soledad_hosts.keys()[0]
+ server_url = 'https://%s:%d/user-%s' % \
+ (soledad_hosts[host]['hostname'], soledad_hosts[host]['port'],
+ auth['id'])
+ # get provider ca certificate
+ #ca_cert = requests.get('https://127.0.0.1/ca.crt', verify=False).text
+ #cert_file = os.path.join(tempdir, 'ca.crt')
+ cert_file = None # not used for now
+ #with open(cert_file, 'w') as f:
+ # f.write(ca_cert)
+ return auth['id'], server_url, cert_file, auth['token']
+
+
+def can_sync_soledad_fine():
+ tempdir = tempfile.mkdtemp()
+ try:
+ uuid, server_url, cert_file, token = \
+ get_soledad_info(Config(), tempdir)
+ # in the future, we can replace the following by an actual Soledad
+ # client sync, if needed
+ db = u1db.open(os.path.join(tempdir, '%s.db' % uuid), True)
+ creds = {'token': {'uuid': uuid, 'token': token}}
+ db.sync(server_url, creds=creds, autocreate=False)
+ finally:
+ shutil.rmtree(tempdir)
if __name__ == '__main__':
- run_tests()
+ from support import nagios_test
+ exit_code = nagios_test.run(can_sync_soledad_fine)
+ exit(exit_code)
diff --git a/test/nagios/support/__init__.py b/test/nagios/support/__init__.py
new file mode 100644
index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
diff --git a/test/nagios/support/api.py b/test/nagios/support/api.py
new file mode 100644
index 0000000000000000000000000000000000000000..ec1af990d932220c87e4d7d6d99a47b8914aa014
--- /dev/null
+++ b/test/nagios/support/api.py
@@ -0,0 +1,39 @@
+import requests
+import json
+
+class Api():
+ def __init__(self, config, verify=True):
+ self.config = config.api
+ self.session = requests.session()
+ self.verify = verify
+
+ def api_url(self, path):
+ return self.api_root() + path
+
+ def api_root(self):
+ return "https://{domain}:{port}/{version}/".format(**self.config)
+
+ def get(self, path, **args):
+ response = self.session.get(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return self.parse_json(response)
+
+ def post(self, path, **args):
+ response = self.session.post(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return self.parse_json(response)
+
+ def put(self, path, **args):
+ response = self.session.put(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return self.parse_json(response)
+
+ def parse_json(self, response):
+ try:
+ return response.json()
+ except TypeError:
+ return response.json # older versions of requests
+
diff --git a/test/nagios/support/config.py b/test/nagios/support/config.py
new file mode 100644
index 0000000000000000000000000000000000000000..afb4464ed380b907c81453413815ced3b5eb2a69
--- /dev/null
+++ b/test/nagios/support/config.py
@@ -0,0 +1,14 @@
+import yaml
+
+class Config():
+ def __init__(self, filename="/etc/leap/hiera.yaml"):
+ with open("/etc/leap/hiera.yaml", 'r') as stream:
+ config = yaml.load(stream)
+ self.user = config['webapp']['nagios_test_user']
+ if 'username' not in self.user:
+ raise Exception('nagios test user lacks username')
+ if 'password' not in self.user:
+ raise Exception('nagios test user lacks password')
+ self.api = config['api']
+ self.api['version'] = config['webapp']['api_version']
+
diff --git a/test/nagios/support/nagios_report.py b/test/nagios/support/nagios_report.py
new file mode 100644
index 0000000000000000000000000000000000000000..13cd5515f20cf1e7138e1f6f61ebf91d403a12a8
--- /dev/null
+++ b/test/nagios/support/nagios_report.py
@@ -0,0 +1,24 @@
+def functions_for_system(under_test):
+ """
+ returns a set of functions to use for nagios reporting:
+ >>> ok, warn, critical, unknown = functions_for_system("tested system")
+
+ each of them will print a nagios line with its argument and
+ return the exit code:
+ >>> warn("that looks strange")
+ 1 tested system - WARNING - that looks strange
+ 1
+ """
+ def report_function(code):
+ return lambda message : report(under_test, code, message)
+ return map(report_function, [0,1,2,3])
+
+def report(system, code, message):
+ codes = {0: 'OK', 1: 'WARNING', 2: 'CRITICAL', 3: 'UNKNOWN'}
+ print "%d %s - %s - %s" % \
+ (code, system, codes[code], message)
+ return code
+
+if __name__ == "__main__":
+ import doctest
+ doctest.testmod()
diff --git a/test/nagios/support/nagios_test.py b/test/nagios/support/nagios_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..3eb8d55690382420de2b2dc775c45633a768f369
--- /dev/null
+++ b/test/nagios/support/nagios_test.py
@@ -0,0 +1,49 @@
+import __main__ as main
+import os
+import sys
+import nagios_report
+
+def run(test):
+ """
+ run takes a function and tries it out.
+ If it returns nothing or 0 everything is fine and run prints an OK message
+ with the function name.
+ >>> def this_works_fine(): return
+ >>> run(this_works_fine)
+ 0 nagios_test.py - OK - this_works_fine
+ 0
+ >>> def this_also_works_fine(): return 0
+ >>> run(this_also_works_fine)
+ 0 nagios_test.py - OK - this_also_works_fine
+ 0
+
+ If the function returns something else it will be printed as a warning.
+ >>> run(lambda : "this is a warning")
+ 1 nagios_test.py - WARNING - this is a warning
+ 1
+
+ Errors raised will result in a CRITICAL nagios string.
+ >>> def failure(): raise Exception("something went wrong")
+ >>> run(failure)
+ 2 nagios_test.py - CRITICAL - something went wrong
+ 2
+ """
+ try:
+ name = os.path.basename(main.__file__)
+ except AttributeError:
+ name = sys.argv[0]
+ ok, warn, fail, unknown = nagios_report.functions_for_system(name)
+ try:
+ warning = test()
+ if warning and warning != 0:
+ code = warn(warning)
+ else:
+ code = ok(test.__name__)
+ except Exception as exc:
+ code = fail(exc.message or str(exc))
+ return code
+
+
+if __name__ == "__main__":
+ import doctest
+ doctest.testmod()
diff --git a/test/nagios/support/user.py b/test/nagios/support/user.py
new file mode 100644
index 0000000000000000000000000000000000000000..9bf1d0a098b23897a3198ff3fc56b4c2b67ce8a6
--- /dev/null
+++ b/test/nagios/support/user.py
@@ -0,0 +1,64 @@
+import srp._pysrp as srp
+import binascii
+import string
+import random
+
+safe_unhexlify = lambda x: binascii.unhexlify(x) if (
+ len(x) % 2 == 0) else binascii.unhexlify('0' + x)
+
+# let's have some random name and password
+def id_generator(size=6, chars=string.ascii_lowercase + string.digits):
+ return ''.join(random.choice(chars) for x in range(size))
+
+class User():
+ def __init__(self, config = None):
+ if config and config.user:
+ self.username = config.user["username"]
+ self.password = config.user["password"]
+ else:
+ self.username = 'test_' + id_generator()
+ self.password = id_generator() + id_generator()
+ self.srp_user = srp.User(self.username, self.password, srp.SHA256, srp.NG_1024)
+
+ def signup(self, api):
+ salt, vkey = srp.create_salted_verification_key( self.username, self.password, srp.SHA256, srp.NG_1024 )
+ user_params = {
+ 'user[login]': self.username,
+ 'user[password_verifier]': binascii.hexlify(vkey),
+ 'user[password_salt]': binascii.hexlify(salt)
+ }
+ return api.post('users.json', data = user_params)
+
+ def login(self, api):
+ init=self.init_authentication(api)
+ if ('errors' in init):
+ raise Exception('test user not found')
+ auth=self.authenticate(api, init)
+ if ('errors' in auth):
+ raise Exception('srp password auth failed')
+ self.verify_server(auth)
+ if not self.is_authenticated():
+ raise Exception('user is not authenticated')
+ return auth
+
+ def init_authentication(self, api):
+ uname, A = self.srp_user.start_authentication()
+ params = {
+ 'login': uname,
+ 'A': binascii.hexlify(A)
+ }
+ return api.post('sessions', data=params)
+
+ def authenticate(self, api, init):
+ M = self.srp_user.process_challenge(
+ safe_unhexlify(init['salt']), safe_unhexlify(init['B']))
+ auth = api.put('sessions/' + self.username,
+ data={'client_auth': binascii.hexlify(M)})
+ return auth
+
+ def verify_server(self, auth):
+ self.srp_user.verify_session(safe_unhexlify(auth["M2"]))
+
+ def is_authenticated(self):
+ return self.srp_user.authenticated()
+
diff --git a/test/nagios/webapp_login.py b/test/nagios/webapp_login.py
index 171123892abc159b251f5562adfe4a53b6de825a..77413257b68a8f3b92a281012ed2a93e658c882b 100755
--- a/test/nagios/webapp_login.py
+++ b/test/nagios/webapp_login.py
@@ -2,85 +2,17 @@
# Test Authentication with the webapp API works.
-import requests
-import json
-import string
-import random
-import srp._pysrp as srp
-import binascii
-import yaml
+from support.api import Api
+from support.config import Config
+from support.user import User
-
-safe_unhexlify = lambda x: binascii.unhexlify(x) if (
- len(x) % 2 == 0) else binascii.unhexlify('0' + x)
-
-
-def read_config():
- with open("/etc/leap/hiera.yaml", 'r') as stream:
- config = yaml.load(stream)
- user = config['webapp']['nagios_test_user']
- if 'username' not in user:
- fail('nagios test user lacks username')
- if 'password' not in user:
- fail('nagios test user lacks password')
- api = config['api']
- api['version'] = config['webapp']['api_version']
- return {'api': api, 'user': user}
-
-
-def run_tests(config):
- user = config['user']
- api = config['api']
- usr = srp.User(user['username'], user['password'], srp.SHA256, srp.NG_1024)
- try:
- auth = parse(authenticate(api, usr))
- except requests.exceptions.ConnectionError:
- fail('no connection to server')
- exit(report(auth, usr))
-
-# parse the server responses
-
-
-def parse(response):
- request = response.request
- try:
- return json.loads(response.text)
- except ValueError:
- return None
-
-
-def authenticate(api, usr):
- api_url = "https://{domain}:{port}/{version}".format(**api)
- session = requests.session()
- uname, A = usr.start_authentication()
- params = {
- 'login': uname,
- 'A': binascii.hexlify(A)
- }
- init = parse(
- session.post(api_url + '/sessions', data=params, verify=False))
- if ('errors' in init):
- fail('test user not found')
- M = usr.process_challenge(
- safe_unhexlify(init['salt']), safe_unhexlify(init['B']))
- return session.put(api_url + '/sessions/' + uname, verify=False,
- data={'client_auth': binascii.hexlify(M)})
-
-
-def report(auth, usr):
- if ('errors' in auth):
- fail('srp password auth failed')
- usr.verify_session(safe_unhexlify(auth["M2"]))
- if usr.authenticated():
- print '0 webapp_login - OK - can login to webapp fine'
- return 0
- print '1 webapp_login - WARNING - failed to verify webapp server'
- return 1
-
-
-def fail(reason):
- print '2 webapp_login - CRITICAL - ' + reason
- exit(2)
+def login_successfully():
+ config = Config()
+ user = User(config)
+ api = Api(config, verify=False)
+ user.login(api)
if __name__ == '__main__':
- run_tests(read_config())
+ from support import nagios_test
+ exit_code = nagios_test.run(login_successfully)
+ exit(exit_code)
diff --git a/test/nagios/webapp_signup.py b/test/nagios/webapp_signup.py
new file mode 100755
index 0000000000000000000000000000000000000000..3e7283e5a3bc60751310659fb098127615cbf193
--- /dev/null
+++ b/test/nagios/webapp_signup.py
@@ -0,0 +1,19 @@
+#!/usr/bin/env python
+
+# Test Signup and Login with the webapp API works.
+
+from support.api import Api
+from support.config import Config
+from support.user import User
+
+def signup_successfully():
+ config = Config()
+ user = User()
+ api = Api(config, verify=False)
+ user.signup(api)
+ user.login(api)
+
+if __name__ == '__main__':
+ from support import nagios_test
+ exit_code = nagios_test.run(signup_successfully)
+ exit(exit_code)