diff --git a/.freeipa-pr-ci.yaml b/.freeipa-pr-ci.yaml index f956c0770..a16b38869 100644 --- a/.freeipa-pr-ci.yaml +++ b/.freeipa-pr-ci.yaml @@ -87,6 +87,18 @@ jobs: timeout: 3600 topology: *master_1repl_1client + fedora-27/test_ipa_cli: + requires: [fedora-27/build] + priority: 50 + job: + class: RunPytest + args: + build_url: '{fedora-27/build_url}' + test_suite: test_integration/test_ipa_cli.py + template: *ci-master-f27 + timeout: 3600 + topology: *master_1repl + fedora-27/test_kerberos_flags: requires: [fedora-27/build] priority: 50 diff --git a/ipaclient/plugins/cert.py b/ipaclient/plugins/cert.py index a8d1e2127..6d453d0d6 100644 --- a/ipaclient/plugins/cert.py +++ b/ipaclient/plugins/cert.py @@ -27,7 +27,7 @@ from ipaclient.frontend import MethodOverride from ipalib import errors from ipalib import x509 from ipalib import util -from ipalib.parameters import File, Flag, Str +from ipalib.parameters import BinaryFile, File, Flag, Str from ipalib.plugable import Registry from ipalib.text import _ @@ -196,7 +196,7 @@ class cert_remove_hold(MethodOverride): @register(override=True, no_fail=True) class cert_find(MethodOverride): takes_options = ( - File( + BinaryFile( 'file?', label=_("Input filename"), doc=_('File to load the certificate from.'), diff --git a/ipaclient/plugins/certmap.py b/ipaclient/plugins/certmap.py index 981ba292f..31db99197 100644 --- a/ipaclient/plugins/certmap.py +++ b/ipaclient/plugins/certmap.py @@ -4,7 +4,7 @@ from ipaclient.frontend import MethodOverride from ipalib import errors, x509 -from ipalib.parameters import File +from ipalib.parameters import BinaryFile from ipalib.plugable import Registry from ipalib.text import _ @@ -14,7 +14,7 @@ register = Registry() @register(override=True, no_fail=True) class certmap_match(MethodOverride): takes_args = ( - File( + BinaryFile( 'file?', label=_("Input file"), doc=_("File to load the certificate from"), diff --git a/ipalib/cli.py b/ipalib/cli.py index 3fde58197..99f9c5512 100644 --- a/ipalib/cli.py +++ b/ipalib/cli.py @@ -56,7 +56,7 @@ from ipalib.errors import (PublicError, CommandError, HelpError, InternalError, NoSuchNamespaceError, ValidationError, NotFound, NotConfiguredError, PromptFailed) from ipalib.constants import CLI_TAB, LDAP_GENERALIZED_TIME_FORMAT -from ipalib.parameters import File, Str, Enum, Any, Flag +from ipalib.parameters import File, BinaryFile, Str, Enum, Any, Flag from ipalib.text import _ from ipalib import api # pylint: disable=unused-import from ipapython.dnsutil import DNSName @@ -1333,7 +1333,7 @@ class cli(backend.Executioner): 3) the webUI will use a different way of loading files """ for p in cmd.params(): - if isinstance(p, File): + if isinstance(p, (File, BinaryFile)): # FIXME: this only reads the first file raw = None if p.name in kw: @@ -1342,9 +1342,8 @@ class cli(backend.Executioner): else: fname = kw[p.name] try: - f = open(fname, 'r') - raw = f.read() - f.close() + with open(fname, p.open_mode) as f: + raw = f.read() except IOError as e: raise ValidationError( name=to_cli(p.cli_name), @@ -1352,14 +1351,22 @@ class cli(backend.Executioner): ) elif p.stdin_if_missing: try: - raw = sys.stdin.read() + if six.PY3 and p.type is bytes: + # pylint: disable=no-member + raw = sys.stdin.buffer.read() + # pylint: enable=no-member + else: + raw = sys.stdin.read() except IOError as e: raise ValidationError( name=to_cli(p.cli_name), error=e.args[1] ) if raw: - kw[p.name] = self.Backend.textui.decode(raw) + if p.type is bytes: + kw[p.name] = raw + else: + kw[p.name] = self.Backend.textui.decode(raw) elif p.required: raise ValidationError( name=to_cli(p.cli_name), error=_('No file to read') diff --git a/ipalib/parameters.py b/ipalib/parameters.py index 902ae3401..62d555129 100644 --- a/ipalib/parameters.py +++ b/ipalib/parameters.py @@ -1753,17 +1753,29 @@ class Any(Param): class File(Str): - """ - File parameter type. + """Text file parameter type. Accepts file names and loads their content into the parameter value. """ + open_mode = 'r' kwargs = Data.kwargs + ( # valid for CLI, other backends (e.g. webUI) can ignore this ('stdin_if_missing', bool, False), ('noextrawhitespace', bool, False), ) + +class BinaryFile(Bytes): + """Binary file parameter type + """ + open_mode = 'rb' + kwargs = Data.kwargs + ( + # valid for CLI, other backends (e.g. webUI) can ignore this + ('stdin_if_missing', bool, False), + ('noextrawhitespace', bool, False), + ) + + class DateTime(Param): """ DateTime parameter type. diff --git a/ipaserver/plugins/certmap.py b/ipaserver/plugins/certmap.py index 843a0fd08..8705e4290 100644 --- a/ipaserver/plugins/certmap.py +++ b/ipaserver/plugins/certmap.py @@ -22,6 +22,8 @@ import logging import dbus import six +from cryptography import x509 as crypto_x509 + from ipalib import api, errors, x509 from ipalib.crud import Search from ipalib.frontend import Object @@ -432,18 +434,23 @@ class _sssd(object): Call Users.ListByCertificate interface and return a dict with key = domain, value = list of uids corresponding to the users matching the provided cert - :param cert: DER cert + + :param cert: DER cert, Certificate instances (IPACertificate) :raise RemoteRetrieveError: if DBus error occurs """ - try: + if isinstance(cert, crypto_x509.Certificate): + cert_pem = cert.public_bytes(x509.Encoding.PEM) + else: cert_obj = x509.load_der_x509_certificate(cert) + cert_pem = cert_obj.public_bytes(x509.Encoding.PEM) + + try: # bug 3306 in sssd returns 0 entry when max_entries = 0 # Temp workaround is to use a non-null value, not too high # to avoid reserving unneeded memory max_entries = dbus.UInt32(100) user_paths = self._users_iface.ListByCertificate( - cert_obj.public_bytes(x509.Encoding.PEM), - max_entries) + cert_pem, max_entries) users = dict() for user_path in user_paths: user_obj = self._bus.get_object(DBUS_SSSD_NAME, user_path) diff --git a/ipatests/test_cmdline/test_cli.py b/ipatests/test_cmdline/test_cli.py index e1305b9be..2b9d859d4 100644 --- a/ipatests/test_cmdline/test_cli.py +++ b/ipatests/test_cmdline/test_cli.py @@ -3,12 +3,14 @@ import os import shlex import subprocess import sys +import tempfile import unittest import six from six import StringIO from ipatests import util +from ipatests.test_ipalib.test_x509 import goodcert_headers from ipalib import api, errors import pytest @@ -312,6 +314,16 @@ class TestCLIParsing(object): if not adtrust_is_enabled: mockldap.del_entry(adtrust_dn) + def test_certfind(self): + with tempfile.NamedTemporaryFile() as f: + f.write(goodcert_headers) + f.flush() + self.check_command( + 'cert_find --file={}'.format(f.name), + 'cert_find', + file=goodcert_headers + ) + def test_cli_fsencoding(): # https://pagure.io/freeipa/issue/5887 diff --git a/ipatests/test_integration/test_ipa_cli.py b/ipatests/test_integration/test_ipa_cli.py new file mode 100644 index 000000000..6538433f5 --- /dev/null +++ b/ipatests/test_integration/test_ipa_cli.py @@ -0,0 +1,67 @@ +# +# Copyright (C) 2018 FreeIPA Contributors see COPYING for license +# +"""Misc test for 'ipa' CLI regressions +""" +from __future__ import absolute_import + +import base64 +import ssl + + +from ipaplatform.paths import paths + +from ipatests.test_integration.base import IntegrationTest +from ipatests.pytest_plugins.integration import tasks + + +class TestIPACommand(IntegrationTest): + topology = 'line' + + def get_cert_base64(self, host, path): + """Retrieve cert and return content as single line, base64 encoded + """ + cacrt = host.get_file_contents(path, encoding='ascii') + cader = ssl.PEM_cert_to_DER_cert(cacrt) + return base64.b64encode(cader).decode('ascii') + + def test_certmap_match_issue7520(self): + # https://pagure.io/freeipa/issue/7520 + tasks.kinit_admin(self.master) + result = self.master.run_command( + ['ipa', 'certmap-match', paths.IPA_CA_CRT], + raiseonerr=False + ) + assert result.returncode == 1 + assert not result.stderr_text + assert "0 users matched" in result.stdout_text + + cab64 = self.get_cert_base64(self.master, paths.IPA_CA_CRT) + result = self.master.run_command( + ['ipa', 'certmap-match', '--certificate', cab64], + raiseonerr=False + ) + assert result.returncode == 1 + assert not result.stderr_text + assert "0 users matched" in result.stdout_text + + def test_cert_find_issue7520(self): + # https://pagure.io/freeipa/issue/7520 + tasks.kinit_admin(self.master) + subject = 'CN=Certificate Authority,O={}'.format( + self.master.domain.realm) + + # by cert file + result = self.master.run_command( + ['ipa', 'cert-find', '--file', paths.IPA_CA_CRT] + ) + assert subject in result.stdout_text + assert '1 certificate matched' in result.stdout_text + + # by base64 cert + cab64 = self.get_cert_base64(self.master, paths.IPA_CA_CRT) + result = self.master.run_command( + ['ipa', 'cert-find', '--certificate', cab64] + ) + assert subject in result.stdout_text + assert '1 certificate matched' in result.stdout_text diff --git a/tox.ini b/tox.ini index 2e44f359c..f401f16f3 100644 --- a/tox.ini +++ b/tox.ini @@ -52,3 +52,8 @@ deps= ipatests commands= {envpython} -m pytest {toxinidir}/pypi/test_placeholder.py + +[pycodestyle] +# E402 module level import not at top of file +ignore = E402 +