openstack/anchor
Code Issues Proposed changes

Simplifying the validator config

Browse Source 
Less nesting, removed redundent lists

Change-Id: Ib223deca5bfd5fc67560fd030813020a07e054ce
This commit is contained in:
Tim Kelsey
2015-04-24 15:18:45 +01:00
parent e6e091f761
commit abd2fef7a8
8 changed files with 163 additions and 448 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
anchor/app.py
View File

@@ -15,6 +15,7 @@
import logging
import os
import stat
import sys
import paste
from paste import translogger # noqa
@@ -30,16 +31,14 @@ class ConfigValidationException(Exception):
pass
def config_check_domains(conf):
# gc.validators[0]['steps'][0][1]['allowed_domains']
for validator in conf.validators:
for step in validator['steps']:
if 'allowed_domains' in step[1]:
for domain in step[1]['allowed_domains']:
if not domain.startswith('.'):
raise ConfigValidationException(
"Domain that does not start with "
"a '.' <%s>", domain)
def config_check_domains(validator_set):
for name, step in validator_set.iteritems():
if 'allowed_domains' in step:
for domain in step['allowed_domains']:
if not domain.startswith('.'):
raise ConfigValidationException(
"Domain that does not start with "
"a '.' <{}>".format(domain))
def _check_file_permissions(path):
@@ -60,6 +59,8 @@ def _check_file_exists(path):
def validate_config(conf):
logger = logging.getLogger("anchor")
if not hasattr(conf, "auth") or not conf.auth:
raise ConfigValidationException("No authentication configured")
@@ -86,28 +87,21 @@ def validate_config(conf):
if not hasattr(conf, "validators"):
raise ConfigValidationException("No validators configured")
for i, validators_list in enumerate(conf.validators):
name = validators_list.get("name")
if not name:
raise ConfigValidationException("Validator set <%d> is missing a "
"name" % (i + 1))
logger.info("Found {} validator sets.".format(len(conf.validators)))
for name, validator_set in conf.validators.iteritems():
logger.info("Checking validator set <{}> ....".format(name))
if len(validator_set) == 0:
raise ConfigValidationException(
"Validator set <{}> is empty".format(name))
if not validators_list.get("steps"):
raise ConfigValidationException("Validator set <%s> is missing "
"validation steps" % name)
for step in validator_set.keys():
if not hasattr(validators, step):
raise ConfigValidationException(
"Validator set <{}> contains an "
"unknown validator <{}>".format(name, step))
for step in validators_list["steps"]:
if len(step) == 0:
raise ConfigValidationException("Validator set <%s> contains "
"a step with no validator "
"name" % name)
if not hasattr(validators, step[0]):
raise ConfigValidationException("Validator set <%s> contains "
"an unknown validator <%s>" %
(name, step[0]))
config_check_domains(conf)
config_check_domains(validator_set)
logger.info("Validator set OK")
def check_default_auth(conf):
@@ -140,11 +134,14 @@ def load_config():
config_path = user_config_path
elif os.path.isfile(sys_config_path):
config_path = sys_config_path
print("using config: {}".format(config_path)) # no logger yet
logger = logging.getLogger("anchor")
logger.info("using config: {}".format(config_path))
jsonloader.conf.load_file_data(config_path)
def setup_app(config):
# initial logging, will be re-configured later
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
app_conf = dict(config.app)
load_config()

32
anchor/certificate_ops.py
View File

@@ -60,37 +60,27 @@ def parse_csr(csr, encoding):
pecan.abort(400, "CSR cannot be parsed")
def _run_validator(validator_step, args):
def _run_validator(name, body, args):
"""Parse the validator tuple, call the validator, and return result.
:param validator_step: validator tuple directly from the config
:param name: the validator name
:param body: validator body, directly from config
:param args: additional arguments to pass to the validator function
:return: True on success, else False
"""
function_name = validator_step[0]
params = validator_step[1]
# make sure the requested validator exists
if not hasattr(validators, function_name):
logger.error("_run_validator: no validator method {}"
.format(function_name))
return False
# careful to not modify the master copy of args with local params
new_kwargs = args.copy()
new_kwargs.update(params)
new_kwargs.update(body)
# perform the actual check
logger.debug("_run_validator: checking <%s> with rules: %s",
function_name, params)
logger.debug("_run_validator: checking <%s> with rules: %s", name, body)
try:
validator = getattr(validators, function_name)
validator = getattr(validators, name)
validator(**new_kwargs)
logger.debug("_run_validator: success: <%s> ", function_name)
logger.debug("_run_validator: success: <%s> ", name)
return True # validator passed b/c no exceptions
except validators.ValidationError as e:
logger.error("_run_validator: FAILED: <%s> - %s",
function_name, e)
logger.error("_run_validator: FAILED: <%s> - %s", name, e)
return False
@@ -116,10 +106,10 @@ def validate_csr(auth_result, csr, request):
# so we set the initial state to valid.
valid = True
for vset in jsonloader.conf.validators:
logger.debug("validate_csr: checking {}".format(vset.get("name")))
for name, vset in jsonloader.conf.validators.iteritems():
logger.debug("validate_csr: checking {}".format(name))
results = [_run_validator(x, args) for x in vset['steps']]
results = [_run_validator(x, y, args) for x, y in vset.iteritems()]
results.append(valid) # track previous failures
valid = all(results)

162
config.json
View File

@@ -5,6 +5,7 @@
"user": "myusername"
}
},
"ca": {
"cert_path": "CA/root-ca.crt",
"key_path": "CA/root-ca-unwrapped.key",
@@ -12,112 +13,63 @@
"signing_hash": "sha1",
"valid_hours": 24
},
"validators": [
{
"name": "default",
"steps": [
[
"common_name",
{
"allowed_domains": [
".example.com"
]
}
],
[
"alternative_names",
{
"allowed_domains": [
".example.com"
]
}
],
[
"server_group",
{
"group_prefixes": {
"bk": "Bock_Team",
"cs": "CS_Team",
"gl": "Glance_Team",
"mb": "MB_Team",
"nv": "Nova_Team",
"ops": "SysEng_Team",
"qu": "Neutron_Team",
"sw": "Swift_Team"
}
}
],
[
"extensions",
{
"allowed_extensions": [
"keyUsage",
"subjectAltName",
"basicConstraints",
"subjectKeyIdentifier"
]
}
],
[
"key_usage",
{
"allowed_usage": [
"Digital Signature",
"Key Encipherment",
"Non Repudiation"
]
}
],
[
"ca_status",
{
"ca_requested": false
}
],
[
"source_cidrs",
{
"cidrs": [
"127.0.0.0/8"
]
}
"validators": {
"default" : {
"common_name" : {
"allowed_domains": [
".example.com"
]
},
"alternative_names": {
"allowed_domains": [
".example.com"
]
]
},
"server_group": {
"group_prefixes": {
"bk": "Bock_Team",
"cs": "CS_Team",
"gl": "Glance_Team",
"mb": "MB_Team",
"nv": "Nova_Team",
"ops": "SysEng_Team",
"qu": "Neutron_Team",
"sw": "Swift_Team"
}
},
"extensions": {
"allowed_extensions": [
"keyUsage",
"subjectAltName",
"basicConstraints",
"subjectKeyIdentifier"
]
},
"key_usage": {
"allowed_usage": [
"Digital Signature",
"Key Encipherment",
"Non Repudiation"
]
},
"ca_status": {
"ca_requested": false
},
"source_cidrs": {
"cidrs": [
"127.0.0.0/8"
]
}
},
{
"name": "ip",
"steps": [
[
"common_name",
{
"allowed_networks": [
"127/8"
]
}
],
[
"alternative_names",
{
"allowed_networks": [
"127/8"
]
}
],
[
"ca_status",
{
"ca_requested": false
}
],
[
"source_cidrs",
{
"cidrs": [
"127.0.0.0/8"
]
}
]
]
"ip": {
"common_name": {
"allowed_networks": ["93.184.216.34"]
},
"alternative_names": {
"allowed_networks": ["93.184.216.34"]
}
}
]
}
}

96
tests/controllers/bad_config_domains.py
View File

@@ -1,96 +0,0 @@
server = {
'port': '5000',
'host': '0.0.0.0'
}
# Pecan Application Configurations
app = {
'root': 'anchor.controllers.RootController',
'modules': ['anchor'],
# 'static_root': '%(confdir)s/public',
# 'template_path': '%(confdir)s/${package}/templates',
'debug': True,
'errors': {
'404': '/error/404',
'__force_dict__': True
}
}
auth = {
'static': {
'user': 'woot',
'secret': 'woot',
},
}
validators = [
{
"name": "common",
"steps": [
# example.com should start with a '.'
('common_name', {'allowed_domains': ['example.com']}),
('alternative_names', {'allowed_domains': ['example.com']}),
('server_group', {'group_prefixes': {
'nv': 'Nova_Team',
'sw': 'Swift_Team',
'bk': 'Bock_Team',
'gl': 'Glance_Team',
'cs': 'CS_Team',
'mb': 'MB_Team',
'ops': 'SysEng_Team',
'qu': 'Neutron_Team',
}}),
('extensions', {'allowed_extensions': [
'keyUsage',
'subjectAltName',
'basicConstraints',
'subjectKeyIdentifier']}),
('key_usage', {'allowed_usage': [
'Digital Signature',
'Key Encipherment',
'Non Repudiation',
'Certificate Sign',
'CRL Sign']}),
('ca_status', {'ca_requested': False}),
('source_cidrs', {'cidrs': ["127.0.0.0/8"]}),
]
},
{
"name": "ip",
"steps": [
('common_name', {'allowed_networks': ['127/8']}),
('alternative_names', {'allowed_networks': ['127/8']}),
('ca_status', {'ca_requested': False}),
('source_cidrs', {'cidrs': ["127.0.0.0/8"]}),
]
},
]
ca = {
'cert_path': "tests/CA/root-ca.crt",
'key_path': "tests/CA/root-ca-unwrapped.key",
'output_path': "certs",
'valid_hours': 24,
'signing_hash': "sha1",
}
logging = {
'root': {'level': 'INFO', 'handlers': ['console']},
'loggers': {
'anchor': {'level': 'DEBUG'},
'wsgi': {'level': 'INFO'},
},
'handlers': {
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'simple'
}
},
'formatters': {
'simple': {
'format': ('%(asctime)s %(levelname)-5.5s [%(name)s]'
'[%(process)d/%(threadName)s] %(message)s')
}
}
}

103
tests/controllers/good_config_domains.py
View File

@@ -1,103 +0,0 @@
server = {
'port': '5000',
'host': '0.0.0.0'
}
# Pecan Application Configurations
app = {
'root': 'anchor.controllers.RootController',
'modules': ['anchor'],
# 'static_root': '%(confdir)s/public',
# 'template_path': '%(confdir)s/${package}/templates',
'debug': True,
'errors': {
'404': '/error/404',
'__force_dict__': True
}
}
auth = {
'static': {
'user': 'woot',
'secret': 'woot',
},
# 'ldap': {
# 'host': "ldap.host.com",
# 'domain': "host.com",
# 'base': "CN=Users,DC=host,DC=com",
# },
# 'keystone': {
# 'url': 'https://keystone.example.com:35357',
# },
}
validators = [
{
"name": "common",
"steps": [
('common_name', {'allowed_domains': ['.example.com']}),
('alternative_names', {'allowed_domains': ['.example.com']}),
('server_group', {'group_prefixes': {
'nv': 'Nova_Team',
'sw': 'Swift_Team',
'bk': 'Bock_Team',
'gl': 'Glance_Team',
'cs': 'CS_Team',
'mb': 'MB_Team',
'ops': 'SysEng_Team',
'qu': 'Neutron_Team',
}}),
('extensions', {'allowed_extensions': [
'keyUsage',
'subjectAltName',
'basicConstraints',
'subjectKeyIdentifier']}),
('key_usage', {'allowed_usage': [
'Digital Signature',
'Key Encipherment',
'Non Repudiation',
'Certificate Sign',
'CRL Sign']}),
('ca_status', {'ca_requested': False}),
('source_cidrs', {'cidrs': ["127.0.0.0/8"]}),
]
},
{
"name": "ip",
"steps": [
('common_name', {'allowed_networks': ['127/8']}),
('alternative_names', {'allowed_networks': ['127/8']}),
('ca_status', {'ca_requested': False}),
('source_cidrs', {'cidrs': ["127.0.0.0/8"]}),
]
},
]
ca = {
'cert_path': "tests/CA/root-ca.crt",
'key_path': "tests/CA/root-ca-unwrapped.key",
'output_path': "certs",
'valid_hours': 24,
'signing_hash': "sha1",
}
logging = {
'root': {'level': 'INFO', 'handlers': ['console']},
'loggers': {
'anchor': {'level': 'DEBUG'},
'wsgi': {'level': 'INFO'},
},
'handlers': {
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'simple'
}
},
'formatters': {
'simple': {
'format': ('%(asctime)s %(levelname)-5.5s [%(name)s]'
'[%(process)d/%(threadName)s] %(message)s')
}
}
}

129
tests/controllers/test_app.py
View File

@@ -18,8 +18,6 @@
import stat
import unittest
import bad_config_domains
import good_config_domains
import mock
from anchor import app
@@ -37,12 +35,54 @@ class TestValidDN(unittest.TestCase):
def test_self_test(self):
self.assertTrue(True)
def test_config_check_domains_good(self):
@mock.patch('anchor.app._check_file_exists')
@mock.patch('anchor.app._check_file_permissions')
def test_config_check_domains_good(self, a, b):
good_config_domains = jsonloader.AnchorConf(None)
good_config_domains._config = {
"auth": {"static": {}},
"ca": {
"cert_path": "no_cert_file",
"key_path": "no_key_file",
"output_path": "",
"signing_hash": "",
"valid_hours": ""
},
"validators": {
"steps": {
"common_name": {
"allowed_domains": [".test.com"]
}
}
}
}
config = {'return_value.st_mode': (stat.S_IRUSR | stat.S_IFREG)}
with mock.patch("os.stat", **config):
self.assertEqual(app.validate_config(good_config_domains), None)
def test_config_check_domains_bad(self):
@mock.patch('anchor.app._check_file_exists')
@mock.patch('anchor.app._check_file_permissions')
def test_config_check_domains_bad(self, a, b):
bad_config_domains = jsonloader.AnchorConf(None)
bad_config_domains._config = {
"auth": {"static": {}},
"ca": {
"cert_path": "no_cert_file",
"key_path": "no_key_file",
"output_path": "",
"signing_hash": "",
"valid_hours": ""
},
"validators": {
"steps": {
"common_name": {
"allowed_domains": ["error.test.com"]
}
}
}
}
config = {'return_value.st_mode': (stat.S_IRUSR | stat.S_IFREG)}
with mock.patch("os.stat", **config):
self.assertRaises(
@@ -119,26 +159,6 @@ class TestValidDN(unittest.TestCase):
"No validators configured",
app.validate_config, jsonloader.conf)
@mock.patch('os.path.isfile')
@mock.patch('os.access')
@mock.patch('os.stat')
def test_validate_config_no_validator_name(self, stat, access, isfile):
config = """{"auth" : { "static": {}},
"ca": { "cert_path":"no_cert_file",
"key_path":"no_key_file",
"output_path":"","signing_hash":"",
"valid_hours":""},
"validators": [ { "no_name" : ""} ]
}
"""
jsonloader.conf.load_str_data(config)
isfile.return_value = True
access.return_value = True
stat.return_value.st_mode = self.expected_key_permissions
self.assertRaisesRegexp(app.ConfigValidationException,
"Validator set <1> is missing a name",
app.validate_config, jsonloader.conf)
@mock.patch('os.path.isfile')
@mock.patch('os.access')
@mock.patch('os.stat')
@@ -148,44 +168,14 @@ class TestValidDN(unittest.TestCase):
"key_path":"no_key_file",
"output_path":"","signing_hash":"",
"valid_hours":""},
"validators": [ { "name" : "no_steps"} ]
}
"validators": { "no_steps" : {}}}
"""
jsonloader.conf.load_str_data(config)
isfile.return_value = True
access.return_value = True
stat.return_value.st_mode = self.expected_key_permissions
self.assertRaisesRegexp(app.ConfigValidationException,
"Validator set <no_steps> is missing "
"validation steps",
app.validate_config, jsonloader.conf)
@mock.patch('os.path.isfile')
@mock.patch('os.access')
@mock.patch('os.stat')
def test_validate_config_no_validator_step_name(self, stat, access,
isfile):
config = """{"auth" : { "static": {}},
"ca": { "cert_path":"no_cert_file",
"key_path":"no_key_file",
"output_path":"","signing_hash":"",
"valid_hours":""},
"validators": [
{ "name" : "has_steps",
"steps": [ [
] ]}
]
}
"""
jsonloader.conf.load_str_data(config)
isfile.return_value = True
access.return_value = True
stat.return_value.st_mode = self.expected_key_permissions
self.assertRaisesRegexp(app.ConfigValidationException,
"Validator set <has_steps> contains a step "
"with no validator name",
"Validator set <no_steps> is empty",
app.validate_config, jsonloader.conf)
@mock.patch('os.path.isfile')
@@ -197,22 +187,18 @@ class TestValidDN(unittest.TestCase):
"key_path":"no_key_file",
"output_path":"","signing_hash":"",
"valid_hours":""},
"validators": [
{ "name" : "has_steps",
"steps": [ [
"unknown_validator",
{}
] ]}
]
}
"validators": {
"steps": {
"unknown_validator": {}
}
}}
"""
jsonloader.conf.load_str_data(config)
isfile.return_value = True
access.return_value = True
stat.return_value.st_mode = self.expected_key_permissions
self.assertRaisesRegexp(app.ConfigValidationException,
"Validator set <has_steps> contains an "
"Validator set <steps> contains an "
"unknown validator <unknown_validator>",
app.validate_config, jsonloader.conf)
@@ -225,13 +211,12 @@ class TestValidDN(unittest.TestCase):
"key_path":"no_key_file",
"output_path":"","signing_hash":"",
"valid_hours":""},
"validators": [
{ "name" : "has_steps",
"steps": [ [
"common_name",
{ "allowed_domains": [
"validators": {
"steps": {
"common_name": {
"allowed_domains": [
".test.com" ]
} ] ]} ] } """
}}}}"""
jsonloader.conf.load_str_data(config)
isfile.return_value = True
access.return_value = True

12
tests/test_certificate_ops.py
View File

@@ -100,9 +100,7 @@ class CertificateOpsTests(unittest.TestCase):
"""Test basic success path for validate_csr."""
csr_obj = certificate_ops.parse_csr(self.csr, 'pem')
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'common',
'steps': [
('extensions', {'allowed_extensions': []})]}]
validators = {'steps': {'extensions': {'allowed_extensions': []}}}
data = {'validators': validators}
with mock.patch.dict(config, data):
@@ -112,7 +110,7 @@ class CertificateOpsTests(unittest.TestCase):
"""Test empty validator set for validate_csr."""
csr_obj = certificate_ops.parse_csr(self.csr, 'pem')
config = "anchor.jsonloader.conf._config"
data = {'validators': []}
data = {'validators': {}}
with mock.patch.dict(config, data):
# this should work, it allows people to bypass validation
@@ -122,10 +120,8 @@ class CertificateOpsTests(unittest.TestCase):
"""Test failure path for validate_csr."""
csr_obj = certificate_ops.parse_csr(self.csr, 'pem')
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'name',
'steps': [
('common_name', {'allowed_domains':
['.testing.com']})]}]
validators = {'steps': {'common_name': {'allowed_domains':
['.testing.com']}}}
data = {'validators': validators}
with mock.patch.dict(config, data):

18
tests/test_functional.py
View File

@@ -45,21 +45,15 @@ class TestFunctional(unittest.TestCase):
"signing_hash": "sha1",
"valid_hours": 24
},
"validators": [
{
"name": "default",
"steps": [
[
"common_name",
{
"allowed_domains": [
".test.com"
"validators": {
"default": {
"common_name": {
"allowed_domains": [
".test.com"
]
}
]
]
}
}
]
}
"""