mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2025-02-25 18:55:28 -06:00
csrgen: Add code to generate scripts that generate CSRs
Adds a library that uses jinja2 to format a script that, when run, will build a CSR. Also adds a CLI command, 'cert-get-requestdata', that uses this library and builds the script for a given principal. The rules are read from json files in /usr/share/ipa/csr, but the rule provider is a separate class so that it can be replaced easily. https://fedorahosted.org/freeipa/ticket/4899 Reviewed-By: Jan Cholasta <jcholast@redhat.com>
This commit is contained in:
Ben Lipton
Jan Cholasta
319
ipaclient/csrgen.py
Normal file
319
ipaclient/csrgen.py
Normal file
@@ -0,0 +1,319 @@
|
||||
#
|
||||
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
|
||||
#
|
||||
|
||||
import collections
|
||||
import json
|
||||
import os.path
|
||||
import pipes
|
||||
import traceback
|
||||
|
||||
import jinja2
|
||||
import jinja2.ext
|
||||
import jinja2.sandbox
|
||||
import six
|
||||
|
||||
from ipalib import api
|
||||
from ipalib import errors
|
||||
from ipalib.text import _
|
||||
from ipaplatform.paths import paths
|
||||
from ipapython.ipa_log_manager import log_mgr
|
||||
|
||||
if six.PY3:
|
||||
unicode = str
|
||||
|
||||
__doc__ = _("""
|
||||
Routines for constructing certificate signing requests using IPA data and
|
||||
stored templates.
|
||||
""")
|
||||
|
||||
logger = log_mgr.get_logger(__name__)
|
||||
|
||||
|
||||
class IndexableUndefined(jinja2.Undefined):
|
||||
def __getitem__(self, key):
|
||||
return jinja2.Undefined(
|
||||
hint=self._undefined_hint, obj=self._undefined_obj,
|
||||
name=self._undefined_name, exc=self._undefined_exception)
|
||||
|
||||
|
||||
class IPAExtension(jinja2.ext.Extension):
|
||||
"""Jinja2 extension providing useful features for CSR generation rules."""
|
||||
|
||||
def __init__(self, environment):
|
||||
super(IPAExtension, self).__init__(environment)
|
||||
|
||||
environment.filters.update(
|
||||
quote=self.quote,
|
||||
required=self.required,
|
||||
)
|
||||
|
||||
def quote(self, data):
|
||||
return pipes.quote(data)
|
||||
|
||||
def required(self, data, name):
|
||||
if not data:
|
||||
raise errors.CSRTemplateError(
|
||||
reason=_('Required CSR generation rule %(name)s is missing data') %
|
||||
{'name': name})
|
||||
return data
|
||||
|
||||
|
||||
class Formatter(object):
|
||||
"""
|
||||
Class for processing a set of CSR generation rules into a template.
|
||||
|
||||
The template can be rendered with user and database data to produce a
|
||||
script, which generates a CSR when run.
|
||||
|
||||
Subclasses of Formatter should set the value of base_template_name to the
|
||||
filename of a base template with spaces for the processed rules.
|
||||
Additionally, they should override the _get_template_params method to
|
||||
produce the correct output for the base template.
|
||||
"""
|
||||
base_template_name = None
|
||||
|
||||
def __init__(self, csr_data_dir=paths.CSR_DATA_DIR):
|
||||
self.jinja2 = jinja2.sandbox.SandboxedEnvironment(
|
||||
loader=jinja2.FileSystemLoader(
|
||||
os.path.join(csr_data_dir, 'templates')),
|
||||
extensions=[jinja2.ext.ExprStmtExtension, IPAExtension],
|
||||
keep_trailing_newline=True, undefined=IndexableUndefined)
|
||||
|
||||
self.passthrough_globals = {}
|
||||
self._define_passthrough('ipa.syntaxrule')
|
||||
self._define_passthrough('ipa.datarule')
|
||||
|
||||
def _define_passthrough(self, call):
|
||||
|
||||
def passthrough(caller):
|
||||
return u'{%% call %s() %%}%s{%% endcall %%}' % (call, caller())
|
||||
|
||||
parts = call.split('.')
|
||||
current_level = self.passthrough_globals
|
||||
for part in parts[:-1]:
|
||||
if part not in current_level:
|
||||
current_level[part] = {}
|
||||
current_level = current_level[part]
|
||||
current_level[parts[-1]] = passthrough
|
||||
|
||||
def build_template(self, rules):
|
||||
"""
|
||||
Construct a template that can produce CSR generator strings.
|
||||
|
||||
:param rules: list of FieldMapping to use to populate the template.
|
||||
|
||||
:returns: jinja2.Template that can be rendered to produce the CSR data.
|
||||
"""
|
||||
syntax_rules = []
|
||||
for description, syntax_rule, data_rules in rules:
|
||||
data_rules_prepared = [
|
||||
self._prepare_data_rule(rule) for rule in data_rules]
|
||||
syntax_rules.append(self._prepare_syntax_rule(
|
||||
syntax_rule, data_rules_prepared, description))
|
||||
|
||||
template_params = self._get_template_params(syntax_rules)
|
||||
base_template = self.jinja2.get_template(
|
||||
self.base_template_name, globals=self.passthrough_globals)
|
||||
|
||||
try:
|
||||
combined_template_source = base_template.render(**template_params)
|
||||
except jinja2.UndefinedError:
|
||||
logger.debug(traceback.format_exc())
|
||||
raise errors.CSRTemplateError(reason=_(
|
||||
'Template error when formatting certificate data'))
|
||||
|
||||
logger.debug(
|
||||
'Formatting with template: %s' % combined_template_source)
|
||||
combined_template = self.jinja2.from_string(combined_template_source)
|
||||
|
||||
return combined_template
|
||||
|
||||
def _wrap_rule(self, rule, rule_type):
|
||||
template = '{%% call ipa.%srule() %%}%s{%% endcall %%}' % (
|
||||
rule_type, rule)
|
||||
|
||||
return template
|
||||
|
||||
def _wrap_required(self, rule, description):
|
||||
template = '{%% filter required("%s") %%}%s{%% endfilter %%}' % (
|
||||
description, rule)
|
||||
|
||||
return template
|
||||
|
||||
def _prepare_data_rule(self, data_rule):
|
||||
return self._wrap_rule(data_rule.template, 'data')
|
||||
|
||||
def _prepare_syntax_rule(self, syntax_rule, data_rules, description):
|
||||
logger.debug('Syntax rule template: %s' % syntax_rule.template)
|
||||
template = self.jinja2.from_string(
|
||||
syntax_rule.template, globals=self.passthrough_globals)
|
||||
is_required = syntax_rule.options.get('required', False)
|
||||
try:
|
||||
rendered = template.render(datarules=data_rules)
|
||||
except jinja2.UndefinedError:
|
||||
logger.debug(traceback.format_exc())
|
||||
raise errors.CSRTemplateError(reason=_(
|
||||
'Template error when formatting certificate data'))
|
||||
|
||||
prepared_template = self._wrap_rule(rendered, 'syntax')
|
||||
if is_required:
|
||||
prepared_template = self._wrap_required(
|
||||
prepared_template, description)
|
||||
|
||||
return prepared_template
|
||||
|
||||
def _get_template_params(self, syntax_rules):
|
||||
"""
|
||||
Package the syntax rules into fields expected by the base template.
|
||||
|
||||
:param syntax_rules: list of prepared syntax rules to be included in
|
||||
the template.
|
||||
|
||||
:returns: dict of values needed to render the base template.
|
||||
"""
|
||||
raise NotImplementedError('Formatter class must be subclassed')
|
||||
|
||||
|
||||
class OpenSSLFormatter(Formatter):
|
||||
"""Formatter class supporting the openssl command-line tool."""
|
||||
|
||||
base_template_name = 'openssl_base.tmpl'
|
||||
|
||||
# Syntax rules are wrapped in this data structure, to keep track of whether
|
||||
# each goes in the extension or the root section
|
||||
SyntaxRule = collections.namedtuple(
|
||||
'SyntaxRule', ['template', 'is_extension'])
|
||||
|
||||
def __init__(self):
|
||||
super(OpenSSLFormatter, self).__init__()
|
||||
self._define_passthrough('openssl.section')
|
||||
|
||||
def _get_template_params(self, syntax_rules):
|
||||
parameters = [rule.template for rule in syntax_rules
|
||||
if not rule.is_extension]
|
||||
extensions = [rule.template for rule in syntax_rules
|
||||
if rule.is_extension]
|
||||
|
||||
return {'parameters': parameters, 'extensions': extensions}
|
||||
|
||||
def _prepare_syntax_rule(self, syntax_rule, data_rules, description):
|
||||
"""Overrides method to pull out whether rule is an extension or not."""
|
||||
prepared_template = super(OpenSSLFormatter, self)._prepare_syntax_rule(
|
||||
syntax_rule, data_rules, description)
|
||||
is_extension = syntax_rule.options.get('extension', False)
|
||||
return self.SyntaxRule(prepared_template, is_extension)
|
||||
|
||||
|
||||
class CertutilFormatter(Formatter):
|
||||
base_template_name = 'certutil_base.tmpl'
|
||||
|
||||
def _get_template_params(self, syntax_rules):
|
||||
return {'options': syntax_rules}
|
||||
|
||||
|
||||
# FieldMapping - representation of the rules needed to construct a complete
|
||||
# certificate field.
|
||||
# - description: str, a name or description of this field, to be used in
|
||||
# messages
|
||||
# - syntax_rule: Rule, the rule defining the syntax of this field
|
||||
# - data_rules: list of Rule, the rules that produce data to be stored in this
|
||||
# field
|
||||
FieldMapping = collections.namedtuple(
|
||||
'FieldMapping', ['description', 'syntax_rule', 'data_rules'])
|
||||
Rule = collections.namedtuple(
|
||||
'Rule', ['name', 'template', 'options'])
|
||||
|
||||
|
||||
class RuleProvider(object):
|
||||
def rules_for_profile(self, profile_id, helper):
|
||||
"""
|
||||
Return the rules needed to build a CSR using the given profile.
|
||||
|
||||
:param profile_id: str, name of the CSR generation profile to use
|
||||
:param helper: str, name of tool (e.g. openssl, certutil) that will be
|
||||
used to create CSR
|
||||
|
||||
:returns: list of FieldMapping, filled out with the appropriate rules
|
||||
"""
|
||||
raise NotImplementedError('RuleProvider class must be subclassed')
|
||||
|
||||
|
||||
class FileRuleProvider(RuleProvider):
|
||||
def __init__(self, csr_data_dir=paths.CSR_DATA_DIR):
|
||||
self.rules = {}
|
||||
self.csr_data_dir = csr_data_dir
|
||||
|
||||
def _rule(self, rule_name, helper):
|
||||
if (rule_name, helper) not in self.rules:
|
||||
rule_path = os.path.join(self.csr_data_dir, 'rules',
|
||||
'%s.json' % rule_name)
|
||||
try:
|
||||
with open(rule_path) as rule_file:
|
||||
ruleset = json.load(rule_file)
|
||||
except IOError:
|
||||
raise errors.NotFound(
|
||||
reason=_('Ruleset %(ruleset)s does not exist.') %
|
||||
{'ruleset': rule_name})
|
||||
|
||||
matching_rules = [r for r in ruleset['rules']
|
||||
if r['helper'] == helper]
|
||||
if len(matching_rules) == 0:
|
||||
raise errors.EmptyResult(
|
||||
reason=_('No transformation in "%(ruleset)s" rule supports'
|
||||
' helper "%(helper)s"') %
|
||||
{'ruleset': rule_name, 'helper': helper})
|
||||
elif len(matching_rules) > 1:
|
||||
raise errors.RedundantMappingRule(
|
||||
ruleset=rule_name, helper=helper)
|
||||
rule = matching_rules[0]
|
||||
|
||||
options = {}
|
||||
if 'options' in ruleset:
|
||||
options.update(ruleset['options'])
|
||||
if 'options' in rule:
|
||||
options.update(rule['options'])
|
||||
self.rules[(rule_name, helper)] = Rule(
|
||||
rule_name, rule['template'], options)
|
||||
return self.rules[(rule_name, helper)]
|
||||
|
||||
def rules_for_profile(self, profile_id, helper):
|
||||
profile_path = os.path.join(self.csr_data_dir, 'profiles',
|
||||
'%s.json' % profile_id)
|
||||
with open(profile_path) as profile_file:
|
||||
profile = json.load(profile_file)
|
||||
|
||||
field_mappings = []
|
||||
for field in profile:
|
||||
syntax_rule = self._rule(field['syntax'], helper)
|
||||
data_rules = [self._rule(name, helper) for name in field['data']]
|
||||
field_mappings.append(FieldMapping(
|
||||
syntax_rule.name, syntax_rule, data_rules))
|
||||
return field_mappings
|
||||
|
||||
|
||||
class CSRGenerator(object):
|
||||
FORMATTERS = {
|
||||
'openssl': OpenSSLFormatter,
|
||||
'certutil': CertutilFormatter,
|
||||
}
|
||||
|
||||
def __init__(self, rule_provider):
|
||||
self.rule_provider = rule_provider
|
||||
|
||||
def csr_script(self, principal, profile_id, helper):
|
||||
config = api.Command.config_show()['result']
|
||||
render_data = {'subject': principal, 'config': config}
|
||||
|
||||
formatter = self.FORMATTERS[helper]()
|
||||
rules = self.rule_provider.rules_for_profile(profile_id, helper)
|
||||
template = formatter.build_template(rules)
|
||||
|
||||
try:
|
||||
script = template.render(render_data)
|
||||
except jinja2.UndefinedError:
|
||||
logger.debug(traceback.format_exc())
|
||||
raise errors.CSRTemplateError(reason=_(
|
||||
'Template error when formatting certificate data'))
|
||||
|
||||
return script
|
||||
114
ipaclient/plugins/csrgen.py
Normal file
114
ipaclient/plugins/csrgen.py
Normal file
@@ -0,0 +1,114 @@
|
||||
#
|
||||
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
|
||||
#
|
||||
|
||||
import six
|
||||
|
||||
from ipaclient.csrgen import CSRGenerator, FileRuleProvider
|
||||
from ipalib import api
|
||||
from ipalib import errors
|
||||
from ipalib import output
|
||||
from ipalib import util
|
||||
from ipalib.frontend import Local, Str
|
||||
from ipalib.parameters import Principal
|
||||
from ipalib.plugable import Registry
|
||||
from ipalib.text import _
|
||||
|
||||
if six.PY3:
|
||||
unicode = str
|
||||
|
||||
register = Registry()
|
||||
|
||||
__doc__ = _("""
|
||||
Commands to build certificate requests automatically
|
||||
""")
|
||||
|
||||
|
||||
@register()
|
||||
class cert_get_requestdata(Local):
|
||||
__doc__ = _('Gather data for a certificate signing request.')
|
||||
|
||||
takes_options = (
|
||||
Principal(
|
||||
'principal',
|
||||
label=_('Principal'),
|
||||
doc=_('Principal for this certificate (e.g.'
|
||||
' HTTP/test.example.com)'),
|
||||
),
|
||||
Str(
|
||||
'profile_id',
|
||||
label=_('Profile ID'),
|
||||
doc=_('CSR Generation Profile to use'),
|
||||
),
|
||||
Str(
|
||||
'helper',
|
||||
label=_('Name of CSR generation tool'),
|
||||
doc=_('Name of tool (e.g. openssl, certutil) that will be used to'
|
||||
' create CSR'),
|
||||
),
|
||||
Str(
|
||||
'out?',
|
||||
doc=_('Write CSR generation script to file'),
|
||||
),
|
||||
)
|
||||
|
||||
has_output = (
|
||||
output.Output(
|
||||
'result',
|
||||
type=dict,
|
||||
doc=_('Dictionary mapping variable name to value'),
|
||||
),
|
||||
)
|
||||
|
||||
has_output_params = (
|
||||
Str(
|
||||
'script',
|
||||
label=_('Generation script'),
|
||||
)
|
||||
)
|
||||
|
||||
def execute(self, *args, **options):
|
||||
if 'out' in options:
|
||||
util.check_writable_file(options['out'])
|
||||
|
||||
principal = options.get('principal')
|
||||
profile_id = options.get('profile_id')
|
||||
helper = options.get('helper')
|
||||
|
||||
if self.api.env.in_server:
|
||||
backend = self.api.Backend.ldap2
|
||||
else:
|
||||
backend = self.api.Backend.rpcclient
|
||||
if not backend.isconnected():
|
||||
backend.connect()
|
||||
|
||||
try:
|
||||
if principal.is_host:
|
||||
principal_obj = api.Command.host_show(
|
||||
principal.hostname, all=True)
|
||||
elif principal.is_service:
|
||||
principal_obj = api.Command.service_show(
|
||||
unicode(principal), all=True)
|
||||
elif principal.is_user:
|
||||
principal_obj = api.Command.user_show(
|
||||
principal.username, all=True)
|
||||
except errors.NotFound:
|
||||
raise errors.NotFound(
|
||||
reason=_("The principal for this request doesn't exist."))
|
||||
principal_obj = principal_obj['result']
|
||||
|
||||
generator = CSRGenerator(FileRuleProvider())
|
||||
|
||||
script = generator.csr_script(
|
||||
principal_obj, profile_id, helper)
|
||||
|
||||
result = {}
|
||||
if 'out' in options:
|
||||
with open(options['out'], 'wb') as f:
|
||||
f.write(script)
|
||||
else:
|
||||
result = dict(script=script)
|
||||
|
||||
return dict(
|
||||
result=result
|
||||
)
|
||||
@@ -47,6 +47,7 @@ if __name__ == '__main__':
|
||||
"cryptography",
|
||||
"ipalib",
|
||||
"ipapython",
|
||||
"jinja2",
|
||||
"python-nss",
|
||||
"python-yubico",
|
||||
"pyusb",
|
||||
|
||||
Reference in New Issue
Block a user