openstack/anchor
Code Issues Proposed changes

Fixing several issues in Anchor startup

Browse Source 
co-authored-by: Robert Clark <robert.clark@hp.com>

Change-Id: Ic170b372de3da421eef87d2b6faeed6bcebd368d
This commit is contained in:
Tim Kelsey 2015-03-04 16:31:47 +00:00
parent e37fd557b8
commit 7b6c4ba40a
8 changed files with 63 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
anchor/X509/certificate.py
View File

@ -112,7 +112,7 @@ class X509Certificate(object):
:param path: Output file path
"""
bio = self._lib.BIO_new_file(path, "w")
bio = self._lib.BIO_new_file(path.encode('ascii', 'ignore'), "w")
ret = self._lib.PEM_write_bio_X509(bio, self._certObj)
self._lib.BIO_free(bio)

11
anchor/app.py
View File

@ -16,6 +16,8 @@ from paste import translogger # noqa
import pecan
import validators
from anchor import jsonloader
class ConfigValidationException(Exception):
pass
@ -52,11 +54,6 @@ def validate_config(conf):
"validation steps", name)
for step in validators_list["steps"]:
if not isinstance(step, tuple):
raise ConfigValidationException("Validator set <%s> contains "
"a step that's <%s> and not a "
"tuple", name, step)
if len(step) == 0:
raise ConfigValidationException("Validator set <%s> contains "
"a step with no validator "
@ -73,11 +70,11 @@ def validate_config(conf):
def setup_app(config):
app_conf = dict(config.app)
validate_config(config)
validate_config(jsonloader.conf)
app = pecan.make_app(
app_conf.pop('root'),
logging=getattr(config, 'logging', {}),
logging=config.logging,
**app_conf
)
return paste.translogger.TransLogger(app, setup_console_handler=False)

33
anchor/certificate_ops.py
View File

@ -60,30 +60,20 @@ def parse_csr(csr, encoding):
pecan.abort(400, "CSR cannot be parsed")
def _run_validator(tup, args):
def _run_validator(validator_step, args):
"""Parse the validator tuple, call the validator, and return result.
:param tup: validator tuple directly from the config
:param validator_step: validator tuple directly from the config
:param args: additional arguments to pass to the validator function
:return: True on success, else False
"""
# make sure that tup is actually a tuple
if type(tup) is not tuple:
logger.error("_run_validator: validator not a tuple {}".format(tup))
return False
# extract the args from the tuple
if len(tup) == 1:
name, params = tup[0], {}
elif len(tup) == 2:
name, params = tup
else:
logger.error("_run_validator: validator malformed {}".format(tup))
return False
function_name = validator_step[0]
params = validator_step[1]
# make sure the requested validator exists
if not hasattr(validators, name):
logger.error("_run_validator: no validator method {}".format(tup))
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
@ -91,12 +81,13 @@ def _run_validator(tup, args):
new_kwargs.update(params)
# perform the actual check
logger.debug("_run_validator: checking {}".format(name))
logger.debug("_run_validator: checking {}".format(function_name))
try:
getattr(validators, name)(**new_kwargs)
validator = getattr(validators, function_name)
validator(**new_kwargs)
return True # validator passed b/c no exceptions
except validators.ValidationError as e:
logger.debug("_run_validator: validation failed %s", e)
logger.error("_run_validator: validation failed %s", e)
return False
@ -111,6 +102,8 @@ def validate_csr(auth_result, csr, request):
:param csr: CSR value from certificate_ops.parse_csr
:param request: pecan request object associated with this action
"""
# TODO(tkelsey): make this more robust
args = {'auth_result': auth_result,
'csr': csr,
'conf': jsonloader.conf,

8
anchor/jsonloader.py
View File

@ -24,11 +24,6 @@ logger = logging.getLogger(__name__)
class AnchorConf():
ca = None
_config = None
_logger = None
_settings = dict()
def __init__(self, logger, config_file):
'''Attempt to initialize a config dictionary from a yaml file.
@ -39,6 +34,7 @@ class AnchorConf():
'''
self._logger = logger
self._config = {}
try:
f = open(config_file, 'r')
@ -46,7 +42,7 @@ class AnchorConf():
logger.error("could not open config file: %s" % config_file)
sys.exit(2)
else:
# yaml parser does its own exception handling
# JSON parser does its own exception handling
self._config = json.load(f)
@property

6
anchor/validators.py
View File

@ -44,6 +44,11 @@ def check_networks(domain, allowed_networks):
at least one of the IP addresses is listed in allowed_networks for the
deployment.
"""
if len(allowed_networks) == 0:
# no valid networks were provided, so we can't make any assertions
logger.warning("No valid network IP ranges were given, skipping")
return True
try:
networks = socket.gethostbyname_ex(domain)
except socket.gaierror:
@ -86,7 +91,6 @@ def common_name(csr, allowed_domains=[], allowed_networks=[], **kwargs):
entries, or the domain does not match the list of known suffixes
or network ranges.
"""
alt_present = any(ext.get_name() == "subjectAltName"
for ext in csr.get_extensions())

28
config.json
View File

@ -12,34 +12,6 @@
"signing_hash": "sha1",
"valid_hours": 24
},
"logging": {
"formatters": {
"simple": {
"format": "%(asctime)s %(levelname)-5.5s [%(name)s][%(process)d/%(threadName)s] %(message)s"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "simple",
"level": "DEBUG"
}
},
"loggers": {
"anchor": {
"level": "DEBUG"
},
"wsgi": {
"level": "INFO"
}
},
"root": {
"handlers": [
"console"
],
"level": "INFO"
}
},
"validators": [
{
"name": "default",

30
config.py
View File

@ -13,5 +13,33 @@ app = {
'errors': {
'404': '/error/404',
'__force_dict__': True
}
},
}
logging = {
"formatters": {
"simple": {
"format": ("%(asctime)s %(levelname)-5.5s [%(name)s][%(process)d/"
"%(threadName)s] %(message)s")
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "simple",
"level": "DEBUG"
},
},
"loggers": {
"anchor": {
"level": "DEBUG"
},
"wsgi": {
"level": "INFO"
},
},
"root": {
"handlers": ["console"],
"level": "INFO"
},
}

44
tests/test_certificate_ops.py
View File

@ -108,50 +108,24 @@ class CertificateOpsTests(unittest.TestCase):
with mock.patch.dict(config, data):
certificate_ops.validate_csr(None, csr_obj, None)
def test_validate_csr_fail1(self):
def test_validate_csr_bypass(self):
"""Test empty validator set for validate_csr."""
csr_obj = certificate_ops.parse_csr(self.csr, 'pem')
config = "anchor.jsonloader.conf._config"
data = {'validators': []}
with mock.patch.dict(config, data):
# this should work, it allows people to bypass validation
certificate_ops.validate_csr(None, None, None)
certificate_ops.validate_csr(None, csr_obj, None)
def test_validate_csr_fail2(self):
"""Test invalid validator set (no tuples) for validate_csr."""
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'common', 'steps': [True]}]
data = {'validators': validators}
with mock.patch.dict(config, data):
with self.assertRaises(http_status.HTTPClientError):
certificate_ops.validate_csr(None, None, None)
def test_validate_csr_fail3(self):
"""Test invalid validator set (tuple too long) for validate_csr."""
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'common', 'steps': [(1, 2, 3)]}]
data = {'validators': validators}
with mock.patch.dict(config, data):
with self.assertRaises(http_status.HTTPClientError):
certificate_ops.validate_csr(None, None, None)
def test_validate_csr_fail4(self):
"""Test invalid validator set (bogus validator) for validate_csr."""
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'common', 'steps': [('no_such_method')]}]
data = {'validators': validators}
with mock.patch.dict(config, data):
with self.assertRaises(http_status.HTTPClientError):
certificate_ops.validate_csr(None, None, None)
def test_validate_csr_fail5(self):
"""Test validate_csr with a validator that should fail."""
def test_validate_csr_fail(self):
"""Test failure path for validate_csr."""
csr_obj = certificate_ops.parse_csr(self.csr, 'pem')
config = "anchor.jsonloader.conf._config"
validators = [{'name': 'common', 'steps': [('common_name')]}]
validators = [{'name': 'name',
'steps': [
('common_name', {'allowed_domains':
['.testing.com']})]}]
data = {'validators': validators}
with mock.patch.dict(config, data):