IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Remove support for csrgen

Browse Source 
This was never feature complete and currently has issues and
we lack the resources to maintain it.

Drop it for now. It can be revived from git history in the
future if we see the need.

https://pagure.io/freeipa/issue/8669

Signed-off-by: Rob Crittenden <rcritten@redhat.com>
Reviewed-By: Alexander Bokovoy <abokovoy@redhat.com>
This commit is contained in:
Rob Crittenden 2021-01-18 14:38:27 -05:00 committed by Florence Blanc-Renaud
parent 6518a600b4
commit 767232b478
23 changed files with 0 additions and 1206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
freeipa.spec.in
View File

@ -1580,17 +1580,6 @@ fi
%dir %{python3_sitelib}/ipaclient/remote_plugins/2_*
%{python3_sitelib}/ipaclient/remote_plugins/2_*/*.py
%{python3_sitelib}/ipaclient/remote_plugins/2_*/__pycache__/*.py*
%if 0%{?rhel}
# RHEL spec file only: DELETED: Remove csrgen
%else
%dir %{python3_sitelib}/ipaclient/csrgen
%dir %{python3_sitelib}/ipaclient/csrgen/profiles
%{python3_sitelib}/ipaclient/csrgen/profiles/*.json
%dir %{python3_sitelib}/ipaclient/csrgen/rules
%{python3_sitelib}/ipaclient/csrgen/rules/*.json
%dir %{python3_sitelib}/ipaclient/csrgen/templates
%{python3_sitelib}/ipaclient/csrgen/templates/*.tmpl
%endif
%{python3_sitelib}/ipaclient-*.egg-info

488
ipaclient/csrgen.py
View File

@ -1,488 +0,0 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
import base64
import collections
import errno
import json
import logging
import os
import os.path
import pipes
import subprocess
import traceback
import codecs
import pkg_resources
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import (
load_pem_private_key, Encoding, PublicFormat)
from cryptography.x509 import load_pem_x509_certificate
import jinja2
import jinja2.ext
import jinja2.sandbox
from pyasn1.codec.der import decoder, encoder
from pyasn1.type import univ
from pyasn1_modules import rfc2314
import six
from ipalib import api
from ipalib import errors
from ipalib.text import _
if six.PY3:
unicode = str
__doc__ = _("""
Routines for constructing certificate signing requests using IPA data and
stored templates.
""")
logger = logging.getLogger(__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:
"""
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
config, which specifies how to build a CSR.
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=None):
# chain loaders:
# 1) csr_data_dir/templates
# 2) /etc/ipa/csrgen/templates
# 3) ipaclient/csrgen/templates
loaders = []
if csr_data_dir is not None:
loaders.append(jinja2.FileSystemLoader(
os.path.join(csr_data_dir, 'templates'))
)
loaders.append(jinja2.FileSystemLoader(
os.path.join(api.env.confdir, 'csrgen/templates'))
)
loaders.append(jinja2.PackageLoader('ipaclient', 'csrgen/templates'))
self.jinja2 = jinja2.sandbox.SandboxedEnvironment(
loader=jinja2.ChoiceLoader(loaders),
extensions=[jinja2.ext.ExprStmtExtension, IPAExtension],
keep_trailing_newline=True, undefined=IndexableUndefined)
self.passthrough_globals = {}
def _define_passthrough(self, call):
"""Some macros are meant to be interpreted during the final render, not
when data rules are interpolated into syntax rules. This method allows
those macros to be registered so that calls to them are passed through
to the prepared rule rather than interpreted.
"""
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 field_mapping in rules:
data_rules_prepared = [
self._prepare_data_rule(rule)
for rule in field_mapping.data_rules]
data_sources = []
for xrule in field_mapping.data_rules:
data_source = xrule.options.get('data_source')
if data_source:
data_sources.append(data_source)
syntax_rules.append(self._prepare_syntax_rule(
field_mapping.syntax_rule, data_rules_prepared,
field_mapping.description, data_sources))
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_conditional(self, rule, condition):
rule = '{%% if %s %%}%s{%% endif %%}' % (condition, rule)
return rule
def _wrap_required(self, rule, description):
template = '{%% filter required("%s") %%}%s{%% endfilter %%}' % (
description, rule)
return template
def _prepare_data_rule(self, data_rule):
template = data_rule.template
data_source = data_rule.options.get('data_source')
if data_source:
template = self._wrap_conditional(template, data_source)
return template
def _prepare_syntax_rule(
self, syntax_rule, data_rules, description, data_sources):
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:
prepared_template = template.render(datarules=data_rules)
except jinja2.UndefinedError:
logger.debug(traceback.format_exc())
raise errors.CSRTemplateError(reason=_(
'Template error when formatting certificate data'))
if data_sources:
combinator = ' %s ' % syntax_rule.options.get(
'data_source_combinator', 'or')
condition = combinator.join(data_sources)
prepared_template = self._wrap_conditional(
prepared_template, condition)
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 generating the openssl config-file format."""
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, *args, **kwargs):
super(OpenSSLFormatter, self).__init__(*args, **kwargs)
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, data_sources):
"""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, data_sources)
is_extension = syntax_rule.options.get('extension', False)
return self.SyntaxRule(prepared_template, is_extension)
class FieldMapping:
"""Representation of the rules needed to construct a complete cert field.
Attributes:
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
"""
__slots__ = ['description', 'syntax_rule', 'data_rules']
def __init__(self, description, syntax_rule, data_rules):
self.description = description
self.syntax_rule = syntax_rule
self.data_rules = data_rules
class Rule:
__slots__ = ['name', 'template', 'options']
def __init__(self, name, template, options):
self.name = name
self.template = template
self.options = options
class RuleProvider:
def rules_for_profile(self, profile_id):
"""
Return the rules needed to build a CSR using the given profile.
:param profile_id: str, name of the CSR generation profile to use
: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=None):
self.rules = {}
self._csrgen_data_dirs = []
if csr_data_dir is not None:
self._csrgen_data_dirs.append(csr_data_dir)
self._csrgen_data_dirs.append(
os.path.join(api.env.confdir, 'csrgen')
)
self._csrgen_data_dirs.append(
pkg_resources.resource_filename('ipaclient', 'csrgen')
)
def _open(self, subdir, filename):
for data_dir in self._csrgen_data_dirs:
path = os.path.join(data_dir, subdir, filename)
try:
return open(path)
except IOError as e:
if e.errno != errno.ENOENT:
raise
raise IOError(
errno.ENOENT,
"'{}' not found in {}".format(
os.path.join(subdir, filename),
", ".join(self._csrgen_data_dirs)
)
)
def _rule(self, rule_name):
if rule_name not in self.rules:
try:
with self._open('rules', '%s.json' % rule_name) as f:
ruleconf = json.load(f)
except IOError:
raise errors.NotFound(
reason=_('No generation rule %(rulename)s found.') %
{'rulename': rule_name})
try:
rule = ruleconf['rule']
except KeyError:
raise errors.EmptyResult(
reason=_('Generation rule "%(rulename)s" is missing the'
' "rule" key') % {'rulename': rule_name})
options = ruleconf.get('options', {})
self.rules[rule_name] = Rule(
rule_name, rule['template'], options)
return self.rules[rule_name]
def rules_for_profile(self, profile_id):
try:
with self._open('profiles', '%s.json' % profile_id) as f:
profile = json.load(f)
except IOError:
raise errors.NotFound(
reason=_('No CSR generation rules are defined for profile'
' %(profile_id)s') % {'profile_id': profile_id})
field_mappings = []
for field in profile:
syntax_rule = self._rule(field['syntax'])
data_rules = [self._rule(name) for name in field['data']]
field_mappings.append(FieldMapping(
syntax_rule.name, syntax_rule, data_rules))
return field_mappings
class CSRGenerator:
def __init__(self, rule_provider, formatter_class=OpenSSLFormatter):
self.rule_provider = rule_provider
self.formatter = formatter_class()
def csr_config(self, principal, config, profile_id):
render_data = {'subject': principal, 'config': config}
rules = self.rule_provider.rules_for_profile(profile_id)
template = self.formatter.build_template(rules)
try:
config = template.render(render_data)
except jinja2.UndefinedError:
logger.debug(traceback.format_exc())
raise errors.CSRTemplateError(reason=_(
'Template error when formatting certificate data'))
return config
class CSRLibraryAdaptor:
def get_subject_public_key_info(self):
raise NotImplementedError('Use a subclass of CSRLibraryAdaptor')
def sign_csr(self, certification_request_info):
"""Sign a CertificationRequestInfo.
:returns: bytes, a DER-encoded signed CSR.
"""
raise NotImplementedError('Use a subclass of CSRLibraryAdaptor')
class OpenSSLAdaptor:
def __init__(self, key=None, key_filename=None, password_filename=None):
"""
Must provide either ``key_filename`` or ``key``.
"""
if key_filename is not None:
with open(key_filename, 'rb') as key_file:
key_bytes = key_file.read()
password = None
if password_filename is not None:
with open(password_filename, 'rb') as password_file:
password = password_file.read().strip()
self._key = load_pem_private_key(
key_bytes, password, default_backend())
elif key is not None:
self._key = key
else:
raise ValueError("Must provide 'key' or 'key_filename'")
def key(self):
return self._key
def get_subject_public_key_info(self):
pubkey_info = self.key().public_key().public_bytes(
Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
return pubkey_info
def sign_csr(self, certification_request_info):
reqinfo = decoder.decode(
certification_request_info, rfc2314.CertificationRequestInfo())[0]
csr = rfc2314.CertificationRequest()
csr.setComponentByName('certificationRequestInfo', reqinfo)
algorithm = rfc2314.SignatureAlgorithmIdentifier()
algorithm.setComponentByName(
'algorithm', univ.ObjectIdentifier(
'1.2.840.113549.1.1.11')) # sha256WithRSAEncryption
csr.setComponentByName('signatureAlgorithm', algorithm)
signature = self.key().sign(
certification_request_info,
padding.PKCS1v15(),
hashes.SHA256()
)
asn1sig = univ.BitString("'{sig}'H".format(
sig=codecs.encode(signature, 'hex')
.decode('ascii'))
)
csr.setComponentByName('signature', asn1sig)
return encoder.encode(csr)
class NSSAdaptor:
def __init__(self, database, password_filename):
self.database = database
self.password_filename = password_filename
self.nickname = base64.b32encode(os.urandom(40))
def get_subject_public_key_info(self):
temp_cn = base64.b32encode(os.urandom(40)).decode('ascii')
password_args = []
if self.password_filename is not None:
password_args = ['-f', self.password_filename]
subprocess.check_call(
['certutil', '-S', '-n', self.nickname, '-s', 'CN=%s' % temp_cn,
'-x', '-t', ',,', '-d', self.database] + password_args)
cert_pem = subprocess.check_output(
['certutil', '-L', '-n', self.nickname, '-a',
'-d', self.database] + password_args)
cert = load_pem_x509_certificate(cert_pem, default_backend())
pubkey_info = cert.public_key().public_bytes(
Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
return pubkey_info
def sign_csr(self, certification_request_info):
raise NotImplementedError('NSS is not yet supported')

15
ipaclient/csrgen/profiles/caIPAserviceCert.json
View File

@ -1,15 +0,0 @@
[
{
"syntax": "syntaxSubject",
"data": [
"dataHostCN",
"dataSubjectBase"
]
},
{
"syntax": "syntaxSAN",
"data": [
"dataDNS"
]
}
]

15
ipaclient/csrgen/profiles/userCert.json
View File

@ -1,15 +0,0 @@
[
{
"syntax": "syntaxSubject",
"data": [
"dataUsernameCN",
"dataSubjectBase"
]
},
{
"syntax": "syntaxSAN",
"data": [
"dataEmail"
]
}
]

8
ipaclient/csrgen/rules/dataDNS.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "DNS = {{subject.krbprincipalname.0.partition('/')[2].partition('@')[0]}}"
},
"options": {
"data_source": "subject.krbprincipalname.0.partition('/')[2].partition('@')[0]"
}
}

8
ipaclient/csrgen/rules/dataEmail.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "email = {{subject.mail.0}}"
},
"options": {
"data_source": "subject.mail.0"
}
}

8
ipaclient/csrgen/rules/dataHostCN.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "CN={{subject.krbprincipalname.0.partition('/')[2].partition('@')[0]}}"
},
"options": {
"data_source": "subject.krbprincipalname.0.partition('/')[2].partition('@')[0]"
}
}

8
ipaclient/csrgen/rules/dataSubjectBase.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "{{config.ipacertificatesubjectbase.0}}"
},
"options": {
"data_source": "config.ipacertificatesubjectbase.0"
}
}

8
ipaclient/csrgen/rules/dataUsernameCN.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "CN={{subject.uid.0}}"
},
"options": {
"data_source": "subject.uid.0"
}
}

8
ipaclient/csrgen/rules/syntaxSAN.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "subjectAltName = @{% call openssl.section() %}{{ datarules|join('\n') }}{% endcall %}"
},
"options": {
"extension": true
}
}

9
ipaclient/csrgen/rules/syntaxSubject.json
View File

@ -1,9 +0,0 @@
{
"rule": {
"template": "distinguished_name = {% call openssl.section() %}{{ datarules|reverse|join('\n') }}{% endcall %}"
},
"options": {
"required": true,
"data_source_combinator": "and"
}
}

17
ipaclient/csrgen/templates/openssl_base.tmpl
View File

@ -1,17 +0,0 @@
{% raw -%}
{% import "openssl_macros.tmpl" as openssl -%}
{% endraw -%}
[ req ]
prompt = no
encrypt_key = no
{{ parameters|join('\n') }}
{% raw %}{% set rendered_extensions -%}{% endraw %}
{{ extensions|join('\n') }}
{% raw -%}
{%- endset -%}
{% if rendered_extensions -%}
req_extensions = {% call openssl.section() %}{{ rendered_extensions }}{% endcall %}
{% endif %}
{{ openssl.openssl_sections|join('\n\n') }}
{%- endraw %}

29
ipaclient/csrgen/templates/openssl_macros.tmpl
View File

@ -1,29 +0,0 @@
{# List containing rendered sections to be included at end #}
{% set openssl_sections = [] %}
{#
List containing one entry for each section name allocated. Because of
scoping rules, we need to use a list so that it can be a "per-render global"
that gets updated in place. Real globals are shared by all templates with the
same environment, and variables defined in the macro don't persist after the
macro invocation ends.
#}
{% set openssl_section_num = [] %}
{% macro section() -%}
{% set name -%}
sec{{ openssl_section_num|length -}}
{% endset -%}
{% do openssl_section_num.append('') -%}
{% set contents %}{{ caller() }}{% endset -%}
{% if contents -%}
{% set sectiondata = formatsection(name, contents) -%}
{% do openssl_sections.append(sectiondata) -%}
{% endif -%}
{{ name -}}
{% endmacro %}
{% macro formatsection(name, contents) -%}
[ {{ name }} ]
{{ contents -}}
{% endmacro %}

80
ipaclient/plugins/cert.py
View File

@ -21,8 +21,6 @@
import base64
import six
from ipaclient.frontend import MethodOverride
from ipalib import errors
from ipalib import x509
@ -31,9 +29,6 @@ from ipalib.parameters import BinaryFile, File, Flag, Str
from ipalib.plugable import Registry
from ipalib.text import _
if six.PY3:
unicode = str
register = Registry()
@ -73,87 +68,12 @@ class CertRetrieveOverride(MethodOverride):
@register(override=True, no_fail=True)
class cert_request(CertRetrieveOverride):
takes_options = CertRetrieveOverride.takes_options + (
Str(
'database?',
label=_('Path to NSS database'),
doc=_('Path to NSS database to use for private key'),
),
Str(
'private_key?',
label=_('Path to private key file'),
doc=_('Path to PEM file containing a private key'),
),
Str(
'password_file?',
label=_(
'File containing a password for the private key or database'),
),
Str(
'csr_profile_id?',
label=_('Name of CSR generation profile (if not the same as'
' profile_id)'),
),
)
def get_args(self):
for arg in super(cert_request, self).get_args():
if arg.name == 'csr':
arg = arg.clone_retype(arg.name, File, required=False)
yield arg
def forward(self, csr=None, **options):
database = options.pop('database', None)
private_key = options.pop('private_key', None)
csr_profile_id = options.pop('csr_profile_id', None)
password_file = options.pop('password_file', None)
if csr is None:
# Deferred import, ipaclient.csrgen is expensive to load.
# see https://pagure.io/freeipa/issue/7484
from ipaclient import csrgen
if database:
adaptor = csrgen.NSSAdaptor(database, password_file)
elif private_key:
adaptor = csrgen.OpenSSLAdaptor(
key_filename=private_key, password_filename=password_file)
else:
raise errors.InvocationError(
message=u"One of 'database' or 'private_key' is required")
pubkey_info = adaptor.get_subject_public_key_info()
pubkey_info_b64 = base64.b64encode(pubkey_info)
# If csr_profile_id is passed, that takes precedence.
# Otherwise, use profile_id. If neither are passed, the default
# in cert_get_requestdata will be used.
profile_id = csr_profile_id
if profile_id is None:
profile_id = options.get('profile_id')
response = self.api.Command.cert_get_requestdata(
profile_id=profile_id,
principal=options.get('principal'),
public_key_info=pubkey_info_b64)
req_info_b64 = response['result']['request_info']
req_info = base64.b64decode(req_info_b64)
csr = adaptor.sign_csr(req_info)
if not csr:
raise errors.CertificateOperationError(
error=(_('Generated CSR was empty')))
else:
if database is not None or private_key is not None:
raise errors.MutuallyExclusiveError(reason=_(
"Options 'database' and 'private_key' are not compatible"
" with 'csr'"))
return super(cert_request, self).forward(csr, **options)
@register(override=True, no_fail=True)
class cert_show(CertRetrieveOverride):

128
ipaclient/plugins/csrgen.py
View File

@ -1,128 +0,0 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
import base64
import six
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 Bytes, Principal
from ipalib.plugable import Registry
from ipalib.text import _
from ipapython import dogtag
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.')
NO_CLI = True
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'),
),
Bytes(
'public_key_info',
label=_('Subject Public Key Info'),
doc=_('DER-encoded SubjectPublicKeyInfo structure'),
),
Str(
'out?',
doc=_('Write CertificationRequestInfo to file'),
),
)
has_output = (
output.Output(
'result',
type=dict,
doc=_('Dictionary mapping variable name to value'),
),
)
has_output_params = (
Str(
'request_info',
label=_('CertificationRequestInfo structure'),
)
)
def execute(self, *args, **options):
# Deferred import, ipaclient.csrgen is expensive to load.
# see https://pagure.io/freeipa/issue/7484
from ipaclient import csrgen
from ipaclient import csrgen_ffi
if 'out' in options:
util.check_writable_file(options['out'])
principal = options.get('principal')
profile_id = options.get('profile_id')
if profile_id is None:
profile_id = dogtag.DEFAULT_PROFILE
public_key_info = options.get('public_key_info')
public_key_info = base64.b64decode(public_key_info)
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']
config = api.Command.config_show()['result']
generator = csrgen.CSRGenerator(csrgen.FileRuleProvider())
csr_config = generator.csr_config(principal_obj, config, profile_id)
request_info = base64.b64encode(csrgen_ffi.build_requestinfo(
csr_config.encode('utf8'), public_key_info))
result = {}
if 'out' in options:
with open(options['out'], 'wb') as f:
f.write(request_info)
else:
result = dict(request_info=request_info)
return dict(
result=result
)

8
ipaclient/setup.py
View File

@ -41,13 +41,6 @@ if __name__ == '__main__':
"ipaclient.remote_plugins.2_156",
"ipaclient.remote_plugins.2_164",
],
package_data={
'ipaclient': [
'csrgen/profiles/*.json',
'csrgen/rules/*.json',
'csrgen/templates/*.tmpl',
],
},
install_requires=[
"cryptography",
"ipalib",
@ -63,7 +56,6 @@ if __name__ == '__main__':
extras_require={
"install": ["ipaplatform"],
"otptoken_yubikey": ["python-yubico", "pyusb"],
"csrgen": ["cffi", "jinja2"],
"ldap": ["python-ldap"], # ipapython.ipaldap
},
zip_safe=False,

16
ipatests/test_ipaclient/data/test_csrgen/configs/caIPAserviceCert.conf
View File

@ -1,16 +0,0 @@
[ req ]
prompt = no
encrypt_key = no
distinguished_name = sec0
req_extensions = sec2
[ sec0 ]
O=DOMAIN.EXAMPLE.COM
CN=machine.example.com
[ sec1 ]
DNS = machine.example.com
[ sec2 ]
subjectAltName = @sec1

16
ipatests/test_ipaclient/data/test_csrgen/configs/userCert.conf
View File

@ -1,16 +0,0 @@
[ req ]
prompt = no
encrypt_key = no
distinguished_name = sec0
req_extensions = sec2
[ sec0 ]
O=DOMAIN.EXAMPLE.COM
CN=testuser
[ sec1 ]
email = testuser@example.com
[ sec2 ]
subjectAltName = @sec1

8
ipatests/test_ipaclient/data/test_csrgen/profiles/profile.json
View File

@ -1,8 +0,0 @@
[
{
"syntax": "basic",
"data": [
"options"
]
}
]

5
ipatests/test_ipaclient/data/test_csrgen/rules/basic.json
View File

@ -1,5 +0,0 @@
{
"rule": {
"template": "openssl_rule"
}
}

8
ipatests/test_ipaclient/data/test_csrgen/rules/options.json
View File

@ -1,8 +0,0 @@
{
"rule": {
"template": "openssl_rule"
},
"options": {
"rule_option": true
}
}

1
ipatests/test_ipaclient/data/test_csrgen/templates/identity_base.tmpl
View File

@ -1 +0,0 @@
{{ options|join(";") }}

304
ipatests/test_ipaclient/test_csrgen.py
View File

@ -1,304 +0,0 @@
#
# Copyright (C) 2016 FreeIPA Contributors see COPYING for license
#
import os
import pytest
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography import x509
from ipaclient import csrgen, csrgen_ffi
from ipalib import errors
BASE_DIR = os.path.dirname(__file__)
CSR_DATA_DIR = os.path.join(BASE_DIR, 'data', 'test_csrgen')
@pytest.fixture
def formatter():
return csrgen.Formatter(csr_data_dir=CSR_DATA_DIR)
@pytest.fixture
def rule_provider():
return csrgen.FileRuleProvider(csr_data_dir=CSR_DATA_DIR)
@pytest.fixture
def generator():
return csrgen.CSRGenerator(csrgen.FileRuleProvider())
class StubRuleProvider(csrgen.RuleProvider):
def __init__(self):
self.syntax_rule = csrgen.Rule(
'syntax', '{{datarules|join(",")}}', {})
self.data_rule = csrgen.Rule('data', 'data_template', {})
self.field_mapping = csrgen.FieldMapping(
'example', self.syntax_rule, [self.data_rule])
self.rules = [self.field_mapping]
def rules_for_profile(self, profile_id):
return self.rules
class IdentityFormatter(csrgen.Formatter):
base_template_name = 'identity_base.tmpl'
def __init__(self):
super(IdentityFormatter, self).__init__(csr_data_dir=CSR_DATA_DIR)
def _get_template_params(self, syntax_rules):
return {'options': syntax_rules}
class test_Formatter:
def test_prepare_data_rule_with_data_source(self, formatter):
data_rule = csrgen.Rule('uid', '{{subject.uid.0}}',
{'data_source': 'subject.uid.0'})
prepared = formatter._prepare_data_rule(data_rule)
assert prepared == '{% if subject.uid.0 %}{{subject.uid.0}}{% endif %}'
def test_prepare_data_rule_no_data_source(self, formatter):
"""Not a normal case, but we should handle it anyway"""
data_rule = csrgen.Rule('uid', 'static_text', {})
prepared = formatter._prepare_data_rule(data_rule)
assert prepared == 'static_text'
def test_prepare_syntax_rule_with_data_sources(self, formatter):
syntax_rule = csrgen.Rule(
'example', '{{datarules|join(",")}}', {})
data_rules = ['{{subject.field1}}', '{{subject.field2}}']
data_sources = ['subject.field1', 'subject.field2']
prepared = formatter._prepare_syntax_rule(
syntax_rule, data_rules, 'example', data_sources)
assert prepared == (
'{% if subject.field1 or subject.field2 %}{{subject.field1}},'
'{{subject.field2}}{% endif %}')
def test_prepare_syntax_rule_with_combinator(self, formatter):
syntax_rule = csrgen.Rule('example', '{{datarules|join(",")}}',
{'data_source_combinator': 'and'})
data_rules = ['{{subject.field1}}', '{{subject.field2}}']
data_sources = ['subject.field1', 'subject.field2']
prepared = formatter._prepare_syntax_rule(
syntax_rule, data_rules, 'example', data_sources)
assert prepared == (
'{% if subject.field1 and subject.field2 %}{{subject.field1}},'
'{{subject.field2}}{% endif %}')
def test_prepare_syntax_rule_required(self, formatter):
syntax_rule = csrgen.Rule('example', '{{datarules|join(",")}}',
{'required': True})
data_rules = ['{{subject.field1}}']
data_sources = ['subject.field1']
prepared = formatter._prepare_syntax_rule(
syntax_rule, data_rules, 'example', data_sources)
assert prepared == (
'{% filter required("example") %}{% if subject.field1 %}'
'{{subject.field1}}{% endif %}{% endfilter %}')
def test_prepare_syntax_rule_passthrough(self, formatter):
"""
Calls to macros defined as passthrough are still call tags in the final
template.
"""
formatter._define_passthrough('example.macro')
syntax_rule = csrgen.Rule(
'example',
'{% call example.macro() %}{{datarules|join(",")}}{% endcall %}',
{})
data_rules = ['{{subject.field1}}']
data_sources = ['subject.field1']
prepared = formatter._prepare_syntax_rule(
syntax_rule, data_rules, 'example', data_sources)
assert prepared == (
'{% if subject.field1 %}{% call example.macro() %}'
'{{subject.field1}}{% endcall %}{% endif %}')
def test_prepare_syntax_rule_no_data_sources(self, formatter):
"""Not a normal case, but we should handle it anyway"""
syntax_rule = csrgen.Rule(
'example', '{{datarules|join(",")}}', {})
data_rules = ['rule1', 'rule2']
data_sources = []
prepared = formatter._prepare_syntax_rule(
syntax_rule, data_rules, 'example', data_sources)
assert prepared == 'rule1,rule2'
class test_FileRuleProvider:
def test_rule_basic(self, rule_provider):
rule_name = 'basic'
rule = rule_provider._rule(rule_name)
assert rule.template == 'openssl_rule'
def test_rule_global_options(self, rule_provider):
rule_name = 'options'
rule = rule_provider._rule(rule_name)
assert rule.options['rule_option'] is True
def test_rule_nosuchrule(self, rule_provider):
with pytest.raises(errors.NotFound):
rule_provider._rule('nosuchrule')
def test_rules_for_profile_success(self, rule_provider):
rules = rule_provider.rules_for_profile('profile')
assert len(rules) == 1
field_mapping = rules[0]
assert field_mapping.syntax_rule.name == 'basic'
assert len(field_mapping.data_rules) == 1
assert field_mapping.data_rules[0].name == 'options'
def test_rules_for_profile_nosuchprofile(self, rule_provider):
with pytest.raises(errors.NotFound):
rule_provider.rules_for_profile('nosuchprofile')
class test_CSRGenerator:
def test_userCert_OpenSSL(self, generator):
principal = {
'uid': ['testuser'],
'mail': ['testuser@example.com'],
}
config = {
'ipacertificatesubjectbase': [
'O=DOMAIN.EXAMPLE.COM'
],
}
script = generator.csr_config(principal, config, 'userCert')
with open(os.path.join(
CSR_DATA_DIR, 'configs', 'userCert.conf')) as f:
expected_script = f.read()
assert script == expected_script
def test_caIPAserviceCert_OpenSSL(self, generator):
principal = {
'krbprincipalname': [
'HTTP/machine.example.com@DOMAIN.EXAMPLE.COM'
],
}
config = {
'ipacertificatesubjectbase': [
'O=DOMAIN.EXAMPLE.COM'
],
}
script = generator.csr_config(
principal, config, 'caIPAserviceCert')
with open(os.path.join(
CSR_DATA_DIR, 'configs', 'caIPAserviceCert.conf')) as f:
expected_script = f.read()
assert script == expected_script
def test_works_with_lowercase_attr_type_shortname(self, generator):
principal = {
'uid': ['testuser'],
'mail': ['testuser@example.com'],
}
template_env = {
'ipacertificatesubjectbase': [
'o=DOMAIN.EXAMPLE.COM' # lower-case attr type shortname
],
}
config = generator.csr_config(principal, template_env, 'userCert')
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend(),
)
adaptor = csrgen.OpenSSLAdaptor(key=key)
reqinfo = bytes(csrgen_ffi.build_requestinfo(
config.encode('utf-8'), adaptor.get_subject_public_key_info()))
csr_der = adaptor.sign_csr(reqinfo)
csr = x509.load_der_x509_csr(csr_der, default_backend())
assert (
csr.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
== [x509.NameAttribute(x509.NameOID.COMMON_NAME, u'testuser')]
)
assert (
csr.subject.get_attributes_for_oid(x509.NameOID.ORGANIZATION_NAME)
== [x509.NameAttribute(
x509.NameOID.ORGANIZATION_NAME, u'DOMAIN.EXAMPLE.COM')]
)
def test_unrecognised_attr_type_raises(self, generator):
principal = {
'uid': ['testuser'],
'mail': ['testuser@example.com'],
}
template_env = {
'ipacertificatesubjectbase': [
'X=DOMAIN.EXAMPLE.COM' # unrecognised attr type
],
}
config = generator.csr_config(principal, template_env, 'userCert')
key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048,
backend=default_backend(),
)
adaptor = csrgen.OpenSSLAdaptor(key=key)
with pytest.raises(
errors.CSRTemplateError,
match=r'^unrecognised attribute type: X$'):
csrgen_ffi.build_requestinfo(
config.encode('utf-8'), adaptor.get_subject_public_key_info())
class test_rule_handling:
def test_optionalAttributeMissing(self, generator):
principal = {'uid': 'testuser'}
rule_provider = StubRuleProvider()
rule_provider.data_rule.template = '{{subject.mail}}'
rule_provider.data_rule.options = {'data_source': 'subject.mail'}
generator = csrgen.CSRGenerator(
rule_provider, formatter_class=IdentityFormatter)
script = generator.csr_config(
principal, {}, 'example')
assert script == '\n'
def test_twoDataRulesOneMissing(self, generator):
principal = {'uid': 'testuser'}
rule_provider = StubRuleProvider()
rule_provider.data_rule.template = '{{subject.mail}}'
rule_provider.data_rule.options = {'data_source': 'subject.mail'}
rule_provider.field_mapping.data_rules.append(csrgen.Rule(
'data2', '{{subject.uid}}', {'data_source': 'subject.uid'}))
generator = csrgen.CSRGenerator(
rule_provider, formatter_class=IdentityFormatter)
script = generator.csr_config(principal, {}, 'example')
assert script == ',testuser\n'
def test_requiredAttributeMissing(self):
principal = {'uid': 'testuser'}
rule_provider = StubRuleProvider()
rule_provider.data_rule.template = '{{subject.mail}}'
rule_provider.data_rule.options = {'data_source': 'subject.mail'}
rule_provider.syntax_rule.options = {'required': True}
generator = csrgen.CSRGenerator(
rule_provider, formatter_class=IdentityFormatter)
with pytest.raises(errors.CSRTemplateError):
_script = generator.csr_config(
principal, {}, 'example')