Use OpenSSH-style public keys as the preferred format of SSH public keys.

Public keys in the old format (raw RFC 4253 blob) are automatically
converted to OpenSSH-style public keys. OpenSSH-style public keys are now
stored in LDAP.

Changed sshpubkeyfp to be an output parameter, as that is what it actually
is.

Allow parameter normalizers to be used on values of any type, not just
unicode, so that public key blobs (which are str) can be normalized to
OpenSSH-style public keys.

ticket 2932, 2935

API.txt

@@ -1669,7 +1669,7 @@ option: Str('userpassword', attribute=True, cli_name='password', multivalue=Fals
 option: Flag('random', attribute=False, autofill=True, cli_name='random', default=False, multivalue=False, required=False)
 option: Bytes('usercertificate', attribute=True, cli_name='certificate', multivalue=False, required=False)
 option: Str('macaddress', attribute=True, cli_name='macaddress', csv=True, multivalue=True, pattern='^([a-fA-F0-9]{2}[:|\\-]?){5}[a-fA-F0-9]{2}$', required=False)
-option: Bytes('ipasshpubkey', attribute=True, cli_name='sshpubkey', multivalue=True, required=False)
+option: Str('ipasshpubkey', attribute=True, cli_name='sshpubkey', csv=True, multivalue=True, required=False)
 option: Str('setattr*', cli_name='setattr', exclude='webui')
 option: Str('addattr*', cli_name='addattr', exclude='webui')
 option: Flag('force', autofill=True, default=False)
@@ -1754,7 +1754,7 @@ option: Str('userpassword', attribute=True, autofill=False, cli_name='password',
 option: Flag('random', attribute=False, autofill=True, cli_name='random', default=False, multivalue=False, required=False)
 option: Bytes('usercertificate', attribute=True, autofill=False, cli_name='certificate', multivalue=False, required=False)
 option: Str('macaddress', attribute=True, autofill=False, cli_name='macaddress', csv=True, multivalue=True, pattern='^([a-fA-F0-9]{2}[:|\\-]?){5}[a-fA-F0-9]{2}$', required=False)
-option: Bytes('ipasshpubkey', attribute=True, autofill=False, cli_name='sshpubkey', multivalue=True, required=False)
+option: Str('ipasshpubkey', attribute=True, autofill=False, cli_name='sshpubkey', csv=True, multivalue=True, required=False)
 option: Str('setattr*', cli_name='setattr', exclude='webui')
 option: Str('addattr*', cli_name='addattr', exclude='webui')
 option: Str('delattr*', cli_name='delattr', exclude='webui')
@@ -3291,7 +3291,7 @@ option: Str('title', attribute=True, cli_name='title', multivalue=False, require
 option: Str('manager', attribute=True, cli_name='manager', multivalue=False, required=False)
 option: Str('carlicense', attribute=True, cli_name='carlicense', multivalue=False, required=False)
 option: Bool('nsaccountlock', attribute=True, cli_name='nsaccountlock', multivalue=False, required=False)
-option: Bytes('ipasshpubkey', attribute=True, cli_name='sshpubkey', multivalue=True, required=False)
+option: Str('ipasshpubkey', attribute=True, cli_name='sshpubkey', csv=True, multivalue=True, required=False)
 option: Str('setattr*', cli_name='setattr', exclude='webui')
 option: Str('addattr*', cli_name='addattr', exclude='webui')
 option: Flag('noprivate', autofill=True, cli_name='noprivate', default=False)
@@ -3400,7 +3400,7 @@ option: Str('title', attribute=True, autofill=False, cli_name='title', multivalu
 option: Str('manager', attribute=True, autofill=False, cli_name='manager', multivalue=False, required=False)
 option: Str('carlicense', attribute=True, autofill=False, cli_name='carlicense', multivalue=False, required=False)
 option: Bool('nsaccountlock', attribute=True, autofill=False, cli_name='nsaccountlock', multivalue=False, required=False)
-option: Bytes('ipasshpubkey', attribute=True, autofill=False, cli_name='sshpubkey', multivalue=True, required=False)
+option: Str('ipasshpubkey', attribute=True, autofill=False, cli_name='sshpubkey', csv=True, multivalue=True, required=False)
 option: Str('setattr*', cli_name='setattr', exclude='webui')
 option: Str('addattr*', cli_name='addattr', exclude='webui')
 option: Str('delattr*', cli_name='delattr', exclude='webui')

VERSION

@@ -79,4 +79,4 @@ IPA_DATA_VERSION=20100614120000
 #                                                      #
 ########################################################
 IPA_API_VERSION_MAJOR=2
-IPA_API_VERSION_MINOR=42
+IPA_API_VERSION_MINOR=43

ipa-client/ipa-install/ipa-client-install

@@ -29,7 +29,6 @@ try:
     from ipapython.ipa_log_manager import *
     import tempfile
     import getpass
-    from base64 import b64decode
     from ipaclient import ipadiscovery
     import ipaclient.ipachangeconf
     import ipaclient.ntpconf
@@ -42,6 +41,7 @@ try:
     from ipapython.config import IPAOptionParser
     from ipalib import api, errors
     from ipapython.dn import DN
+    from ipapython.ssh import SSHPublicKey
     import SSSDConfig
     from ConfigParser import RawConfigParser
     from optparse import SUPPRESS_HELP, OptionGroup
@@ -1112,29 +1112,23 @@ def update_ssh_keys(server, hostname, ssh_dir, create_sshfp):
             continue
 
         for line in f:
-            line = line[:-1]
+            line = line[:-1].lstrip()
-            if line.startswith('#'):
+            if not line or line.startswith('#'):
-                continue
-            parts = line.split()
-            if len(parts) < 2:
                 continue
             try:
-                pubkey = b64decode(parts[1])
+                pubkey = SSHPublicKey(line)
-            except TypeError:
+            except ValueError, UnicodeDecodeError:
-                continue
-            try:
-                algo, data, fp = ipautil.decode_ssh_pubkey(pubkey)
-            except ValueError:
-                continue
-            if parts[0] != algo:
                 continue
             root_logger.info("Adding SSH public key from %s", filename)
-            pubkeys.append(unicode(parts[1]))
+            pubkeys.append(pubkey)
 
         f.close()
 
     try:
-        result = api.Command['host_mod'](unicode(hostname), ipasshpubkey=pubkeys, updatedns=False)
+        result = api.Command['host_mod'](unicode(hostname),
+            ipasshpubkey=[pk.openssh() for pk in pubkeys],
+            updatedns=False
+        )
     except errors.EmptyModlist:
         pass
     except StandardError, e:
@@ -1148,8 +1142,7 @@ def update_ssh_keys(server, hostname, ssh_dir, create_sshfp):
 
         update_txt = 'zone %s.\nupdate delete %s. IN SSHFP\nsend\n' % (zone, hostname)
         for pubkey in pubkeys:
-            pubkey = b64decode(pubkey)
+            sshfp = pubkey.fingerprint_dns_sha1()
-            sshfp = ipautil.make_sshfp(pubkey)
             if sshfp is not None:
                 update_txt += 'update add %s. %s IN SSHFP %s\n' % (hostname, ttl, sshfp)
         update_txt += 'send\n'

ipalib/parameters.py

@@ -772,8 +772,6 @@ class Param(ReadOnly):
 
         This method is called once for each value in a multivalue.
         """
-        if type(value) is not unicode:
-            return value
         if self.normalizer is None:
             return value
         try:

ipalib/plugins/host.py

@@ -39,8 +39,10 @@ from ipalib.plugins.dns import get_reverse_zone
 from ipalib import _, ngettext
 from ipalib import x509
 from ipalib.request import context
-from ipalib.util import validate_sshpubkey, output_sshpubkey
+from ipalib.util import (normalize_sshpubkey, validate_sshpubkey_no_options,
-from ipapython.ipautil import ipa_generate_password, CheckedIPAddress, make_sshfp
+    convert_sshpubkey_post)
+from ipapython.ipautil import ipa_generate_password, CheckedIPAddress
+from ipapython.ssh import SSHPublicKey
 from ipapython.dn import DN
 
 __doc__ = _("""
@@ -131,7 +133,10 @@ def update_sshfp_record(zone, record, entry_attrs):
     pubkeys = entry_attrs['ipasshpubkey'] or ()
     sshfps=[]
     for pubkey in pubkeys:
-        sshfp = unicode(make_sshfp(pubkey))
+        try:
+            sshfp = SSHPublicKey(pubkey).fingerprint_dns_sha1()
+        except ValueError, UnicodeDecodeError:
+            continue
         if sshfp is not None:
             sshfps.append(sshfp)
 
@@ -180,6 +185,9 @@ host_output_params = (
     Str('managedby',
         label=_('Failed managedby'),
     ),
+    Str('sshpubkeyfp*',
+        label=_('SSH public key fingerprint'),
+    ),
 )
 
 def validate_ipaddr(ugettext, ipaddr):
@@ -216,7 +224,6 @@ class host(LDAPObject):
         'fqdn', 'description', 'l', 'nshostlocation', 'krbprincipalname',
         'nshardwareplatform', 'nsosversion', 'usercertificate', 'memberof',
         'managedby', 'memberindirect', 'memberofindirect', 'macaddress',
-        'sshpubkeyfp',
     ]
     uuid_attribute = 'ipauniqueid'
     attribute_members = {
@@ -303,15 +310,13 @@ class host(LDAPObject):
             label=_('MAC address'),
             doc=_('Hardware MAC address(es) on this host'),
         ),
-        Bytes('ipasshpubkey*', validate_sshpubkey,
+        Str('ipasshpubkey*', validate_sshpubkey_no_options,
             cli_name='sshpubkey',
-            label=_('Base-64 encoded SSH public key'),
+            label=_('SSH public key'),
+            normalizer=normalize_sshpubkey,
+            csv=True,
             flags=['no_search'],
         ),
-        Str('sshpubkeyfp*',
-            label=_('SSH public key fingerprint'),
-            flags=['virtual_attribute', 'no_create', 'no_update', 'no_search'],
-        ),
     )
 
     def get_dn(self, *keys, **options):
@@ -472,7 +477,7 @@ class host_add(LDAPCreate):
             # fetched anywhere.
             entry_attrs['has_keytab'] = False
 
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
 
         return dn
 
@@ -717,7 +722,7 @@ class host_mod(LDAPUpdate):
 
         self.obj.suppress_netgroup_memberof(entry_attrs)
 
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
 
         return dn
 
@@ -802,7 +807,7 @@ class host_find(LDAPSearch):
             if options.get('all', False):
                 entry_attrs['managing'] = self.obj.get_managed_hosts(entry[0])
 
-            output_sshpubkey(ldap, dn, entry_attrs)
+            convert_sshpubkey_post(ldap, dn, entry_attrs)
 
         return truncated
 
@@ -836,7 +841,7 @@ class host_show(LDAPRetrieve):
 
         self.obj.suppress_netgroup_memberof(entry_attrs)
 
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
 
         return dn
 

ipalib/plugins/user.py

@@ -30,7 +30,8 @@ from ipalib import output
 from ipapython.ipautil import ipa_generate_password
 from ipapython.ipavalidate import Email
 import posixpath
-from ipalib.util import validate_sshpubkey, output_sshpubkey
+from ipalib.util import (normalize_sshpubkey, validate_sshpubkey,
+    convert_sshpubkey_post)
 if api.env.in_server and api.env.context in ['lite', 'server']:
     from ipaserver.plugins.ldap2 import ldap2
     import os
@@ -86,6 +87,9 @@ user_output_params = (
     Flag('has_keytab',
         label=_('Kerberos keys available'),
     ),
+    Str('sshpubkeyfp*',
+        label=_('SSH public key fingerprint'),
+    ),
    )
 
 status_output_params = (
@@ -200,7 +204,7 @@ class user(LDAPObject):
         'uid', 'givenname', 'sn', 'homedirectory', 'loginshell',
         'uidnumber', 'gidnumber', 'mail', 'ou',
         'telephonenumber', 'title', 'memberof', 'nsaccountlock',
-        'memberofindirect', 'sshpubkeyfp',
+        'memberofindirect',
     ]
     search_display_attributes = [
         'uid', 'givenname', 'sn', 'homedirectory', 'loginshell',
@@ -357,15 +361,13 @@ class user(LDAPObject):
             label=_('Account disabled'),
             flags=['no_option'],
         ),
-        Bytes('ipasshpubkey*', validate_sshpubkey,
+        Str('ipasshpubkey*', validate_sshpubkey,
             cli_name='sshpubkey',
-            label=_('Base-64 encoded SSH public key'),
+            label=_('SSH public key'),
+            normalizer=normalize_sshpubkey,
+            csv=True,
             flags=['no_search'],
         ),
-        Str('sshpubkeyfp*',
-            label=_('SSH public key fingerprint'),
-            flags=['virtual_attribute', 'no_create', 'no_update', 'no_search'],
-        ),
     )
 
     def _normalize_and_validate_email(self, email, config=None):
@@ -567,7 +569,7 @@ class user_add(LDAPCreate):
 
         self.obj.get_password_attributes(ldap, dn, entry_attrs)
 
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
 
         return dn
 
@@ -636,7 +638,7 @@ class user_mod(LDAPUpdate):
         convert_nsaccountlock(entry_attrs)
         self.obj._convert_manager(entry_attrs, **options)
         self.obj.get_password_attributes(ldap, dn, entry_attrs)
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
         return dn
 
 api.register(user_mod)
@@ -678,7 +680,7 @@ class user_find(LDAPSearch):
             self.obj._convert_manager(attrs, **options)
             self.obj.get_password_attributes(ldap, dn, attrs)
             convert_nsaccountlock(attrs)
-            output_sshpubkey(ldap, dn, attrs)
+            convert_sshpubkey_post(ldap, dn, attrs)
         return truncated
 
     msg_summary = ngettext(
@@ -698,7 +700,7 @@ class user_show(LDAPRetrieve):
         convert_nsaccountlock(entry_attrs)
         self.obj._convert_manager(entry_attrs, **options)
         self.obj.get_password_attributes(ldap, dn, entry_attrs)
-        output_sshpubkey(ldap, dn, entry_attrs)
+        convert_sshpubkey_post(ldap, dn, entry_attrs)
         return dn
 
 api.register(user_show)

ipalib/util.py

@@ -34,7 +34,7 @@ from dns.exception import DNSException
 
 from ipalib import errors
 from ipalib.text import _
-from ipapython.ipautil import decode_ssh_pubkey
+from ipapython.ssh import SSHPublicKey
 from ipapython.dn import DN, RDN
 
 
@@ -266,37 +266,55 @@ def validate_hostname(hostname, check_fqdn=True, allow_underscore=False):
     else:
         validate_domain_name(hostname,allow_underscore)
 
-def validate_sshpubkey(ugettext, pubkey):
+def normalize_sshpubkey(value):
+    return SSHPublicKey(value).openssh()
+
+def validate_sshpubkey(ugettext, value):
     try:
-        algo, data, fp = decode_ssh_pubkey(pubkey)
+        SSHPublicKey(value)
-    except ValueError:
+    except ValueError, UnicodeDecodeError:
         return _('invalid SSH public key')
 
-def output_sshpubkey(ldap, dn, entry_attrs):
+def validate_sshpubkey_no_options(ugettext, value):
+    try:
+        pubkey = SSHPublicKey(value)
+    except ValueError, UnicodeDecodeError:
+        return _('invalid SSH public key')
+
+    if pubkey.has_options():
+        return _('options are not allowed')
+
+def convert_sshpubkey_post(ldap, dn, entry_attrs):
     if 'ipasshpubkey' in entry_attrs:
-        pubkeys = entry_attrs.get('ipasshpubkey')
+        pubkeys = entry_attrs.pop('ipasshpubkey')
     else:
-        entry = ldap.get_entry(dn, ['ipasshpubkey'])
+        old_entry_attrs = ldap.get_entry(dn, ['ipasshpubkey'])
-        pubkeys = entry[1].get('ipasshpubkey')
+        pubkeys = old_entry_attrs[1].get('ipasshpubkey')
-    if pubkeys is None:
+    if not pubkeys:
         return
 
+    newpubkeys = []
     fingerprints = []
     for pubkey in pubkeys:
         try:
-            algo, data, fp = decode_ssh_pubkey(pubkey)
+            pubkey = SSHPublicKey(pubkey)
-            fp = u':'.join([fp[j:j+2] for j in range(0, len(fp), 2)])
+        except ValueError, UnicodeDecodeError:
-            fingerprints.append(u'%s (%s)' % (fp, algo))
+            continue
-        except ValueError:
+
-            pass
+        fp = pubkey.fingerprint_hex_md5()
+        comment = pubkey.comment()
+        if comment:
+            fp = u'%s %s' % (fp, comment)
+        fp = u'%s (%s)' % (fp, pubkey.keytype())
+
+        newpubkeys.append(pubkey.openssh())
+        fingerprints.append(fp)
+
+    if newpubkeys:
+        entry_attrs['ipasshpubkey'] = newpubkeys
     if fingerprints:
         entry_attrs['sshpubkeyfp'] = fingerprints
 
-def normalize_sshpubkeyfp(value):
-    value = value.split()[0]
-    value = unicode(c for c in value if c in '0123456789ABCDEFabcdef')
-    return value
-
 class cachedproperty(object):
     """
     A property-like attribute that caches the return value of a method call.

ipapython/ipautil.py

@@ -64,7 +64,6 @@ except ImportError:
             self.cmd = cmd
         def __str__(self):
             return "Command '%s' returned non-zero exit status %d" % (self.cmd, self.returncode)
-from ipapython.compat import sha1, md5
 
 def get_domain_name():
     try:
@@ -1022,25 +1021,6 @@ def backup_config_and_replace_variables(
 
     return old_values
 
-def decode_ssh_pubkey(data, fptype=md5):
-    try:
-        (algolen,) = struct.unpack('>I', data[:4])
-        if algolen > 0 and algolen <= len(data) - 4:
-            return (data[4:algolen+4], data[algolen+4:], fptype(data).hexdigest().upper())
-    except struct.error:
-        pass
-    raise ValueError('not a SSH public key')
-
-def make_sshfp(key):
-    algo, data, fp = decode_ssh_pubkey(key, fptype=sha1)
-    if algo == 'ssh-rsa':
-        algo = 1
-    elif algo == 'ssh-dss':
-        algo = 2
-    else:
-        return
-    return '%d 1 %s' % (algo, fp)
-
 
 def utf8_encode_value(value):
     if isinstance(value, unicode):

ipapython/ssh.py

@@ -0,0 +1,199 @@
+# Authors:
+#   Jan Cholasta <jcholast@redhat.com>
+#
+# Copyright (C) 2012  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+"""
+SSH utilities.
+"""
+
+import base64
+import re
+import struct
+
+from ipapython.compat import md5, sha1
+
+__all__ = ['SSHPublicKey']
+
+OPENSSH_BASE_REGEX = re.compile(r'^[\t ]*(?P<keytype>[^\x00\n\r]+?) [\t ]*(?P<key>[^\x00\n\r]+?)(?:[\t ]+(?P<comment>[^\x00\n\r]*?)[\t ]*)?$')
+OPENSSH_OPTIONS_REGEX = re.compile(r'(?P<name>[-0-9A-Za-z]+)(?:="(?P<value>(?:\\"|[^\x00\n\r"])*)")?')
+
+class SSHPublicKey(object):
+    """
+    SSH public key object.
+    """
+
+    __slots__ = ('_key', '_keytype', '_comment', '_options')
+
+    def __init__(self, key, comment=None, options=None, encoding='utf-8'):
+        if isinstance(key, SSHPublicKey):
+            self._key = key._key
+            self._keytype = key._keytype
+            self._comment = key._comment
+            self._options = key._options
+            return
+
+        if not isinstance(key, (str, unicode)):
+            raise TypeError("argument must be str or unicode, got %s" % type(key).__name__)
+
+        # All valid public key blobs start with 3 null bytes (see RFC 4253
+        # section 6.6, RFC 4251 section 5 and RFC 4250 section 4.6)
+        if isinstance(key, str) and key[:3] != '\0\0\0':
+            key = key.decode(encoding)
+
+        valid = self._parse_raw(key) or self._parse_base64(key) or self._parse_openssh(key)
+        if not valid:
+            raise ValueError("not a valid SSH public key")
+
+        if comment is not None:
+            self._comment = comment
+        if options is not None:
+            self._options = options
+
+    def _parse_raw(self, key):
+        if not isinstance(key, str):
+            return False
+
+        try:
+            (ktlen,) = struct.unpack('>I', key[:4])
+        except struct.error:
+            return False
+
+        if ktlen < 1 or ktlen > len(key) - 4:
+            return False
+
+        try:
+            keytype = key[4:ktlen+4].decode('ascii')
+        except UnicodeDecodeError:
+            return False
+
+        self._key = key
+        self._keytype = keytype
+        self._options = {}
+        self._comment = None
+
+        return True
+
+    def _parse_base64(self, key):
+        if not isinstance(key, unicode):
+            return False
+
+        try:
+            key = base64.b64decode(key)
+        except TypeError:
+            return False
+
+        return self._parse_raw(key)
+
+    def _parse_openssh_without_options(self, key):
+        match = OPENSSH_BASE_REGEX.match(key)
+        if not match:
+            return False
+
+        if not self._parse_base64(match.group('key')):
+            return False
+
+        if self._keytype != match.group('keytype'):
+            return False
+
+        self._comment = match.group('comment')
+
+        return True
+
+    def _parse_openssh_with_options(self, key):
+        key = key.lstrip('\t ')
+
+        options = {}
+        while True:
+            match = OPENSSH_OPTIONS_REGEX.match(key)
+            if not match:
+                return False
+
+            name = match.group('name').lower()
+            value = match.group('value')
+            if value:
+                value = value.replace('\\"', '"')
+
+            options[name] = value
+
+            key = key[len(match.group(0)):]
+            key0, key = key[:1], key[1:]
+
+            if key0 != ',':
+                break
+
+        if not self._parse_openssh_without_options(key):
+            return False
+
+        self._options = options
+
+        return True
+
+    def _parse_openssh(self, key):
+        if not isinstance(key, unicode):
+            return False
+
+        if self._parse_openssh_without_options(key):
+            return True
+        else:
+            return self._parse_openssh_with_options(key)
+
+    def keytype(self):
+        return self._keytype
+
+    def comment(self):
+        return self._comment
+
+    def has_options(self):
+        return bool(self._options)
+
+    def openssh(self):
+        out = u'%s %s' % (self._keytype, base64.b64encode(self._key))
+
+        if self._options:
+            options = []
+            for name in sorted(self._options):
+                value = self._options[name]
+                if value is None:
+                    options.append(name)
+                else:
+                    value = value.replace('"', '\\"')
+                    options.append(u'%s="%s"' % (name, value))
+            options = u','.join(options)
+
+            out = u'%s %s' % (options, out)
+
+        if self._comment:
+            out = u'%s %s' % (out, self._comment)
+
+        return out
+
+    def fingerprint_hex_md5(self):
+        fp = md5(self._key).hexdigest().upper()
+        fp = u':'.join([fp[j:j+2] for j in range(0, len(fp), 2)])
+        return fp
+
+    def fingerprint_dns_sha1(self):
+        if self._keytype == 'ssh-rsa':
+            keytype = 1
+        elif self._keytype == 'ssh-dss':
+            keytype = 2
+        else:
+            return
+        fp = sha1(self._key).hexdigest().upper()
+        return '%d 1 %s' % (keytype, fp)

tests/test_ipapython/test_ssh.py

@@ -0,0 +1,76 @@
+# Authors:
+#   Jan Cholasta <jcholast@redhat.com>
+#
+# Copyright (C) 2011  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+Test the `ipapython/ssh.py` module.
+"""
+
+import base64
+import nose
+
+from ipapython import ssh
+
+class CheckPublicKey:
+    def __init__(self, pk):
+        self.description = "Test SSH public key parsing (%s)" % repr(pk)
+
+    def __call__(self, pk, out):
+        try:
+            parsed = ssh.SSHPublicKey(pk)
+            assert parsed.openssh() == out
+        except Exception, e:
+            assert type(e) is out
+
+def test_public_key_parsing():
+    b64 = 'AAAAB3NzaC1yc2EAAAADAQABAAABAQDGAX3xAeLeaJggwTqMjxNwa6XHBUAikXPGMzEpVrlLDCZtv00djsFTBi38PkgxBJVkgRWMrcBsr/35lq7P6w8KGIwA8GI48Z0qBS2NBMJ2u9WQ2hjLN6GdMlo77O0uJY3251p12pCVIS/bHRSq8kHO2No8g7KA9fGGcagPfQH+ee3t7HUkpbQkFTmbPPN++r3V8oVUk5LxbryB3UIIVzNmcSIn3JrXynlvui4MixvrtX6zx+O/bBo68o8/eZD26QrahVbA09fivrn/4h3TM019Eu/c2jOdckfU3cHUV/3Tno5d6JicibyaoDDK7S/yjdn5jhaz8MSEayQvFkZkiF0L'
+    raw = base64.b64decode(b64)
+    openssh = 'ssh-rsa %s' % b64
+
+    pks = [
+        ('\xff',                    UnicodeDecodeError),
+
+        (raw,                       openssh),
+        ('\0\0\0\x04none',          u'none AAAABG5vbmU='),
+        ('\0\0\0',                  ValueError),
+        ('\0\0\0\0',                ValueError),
+        ('\0\0\0\x01',              ValueError),
+        ('\0\0\0\x01\xff',          ValueError),
+
+        (b64,                       openssh),
+        (unicode(b64),              openssh),
+        (u'\n%s\n\n' % b64,         openssh),
+        (u'AAAABG5vbmU=',           u'none AAAABG5vbmU='),
+        (u'AAAAB',                  ValueError),
+
+        (openssh,                   openssh),
+        (unicode(openssh),          openssh),
+        (u'none AAAABG5vbmU=',      u'none AAAABG5vbmU='),
+        (u'\t \t ssh-rsa \t \t%s\t \tthis is a comment\t \t ' % b64,
+                                    u'%s this is a comment' % openssh),
+        (u'opt3,opt2="\tx ",opt1,opt2="\\"x " %s comment ' % openssh,
+                                    u'opt1,opt2="\\"x ",opt3 %s comment' % openssh),
+        (u'ssh-rsa\n%s' % b64,      ValueError),
+        (u'ssh-rsa\t%s' % b64,      ValueError),
+        (u'vanitas %s' % b64,       ValueError),
+        (u'@opt %s' % openssh,      ValueError),
+        (u'opt=val %s' % openssh,   ValueError),
+        (u'opt, %s' % openssh,      ValueError),
+    ]
+
+    for pk in pks:
+        yield (CheckPublicKey(pk[0]),) + pk

tests/test_xmlrpc/test_host_plugin.py

@@ -62,6 +62,9 @@ servercert = ''.join(servercert)
 servercert = x509.strip_header(servercert)
 fd.close()
 
+sshpubkey = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDGAX3xAeLeaJggwTqMjxNwa6XHBUAikXPGMzEpVrlLDCZtv00djsFTBi38PkgxBJVkgRWMrcBsr/35lq7P6w8KGIwA8GI48Z0qBS2NBMJ2u9WQ2hjLN6GdMlo77O0uJY3251p12pCVIS/bHRSq8kHO2No8g7KA9fGGcagPfQH+ee3t7HUkpbQkFTmbPPN++r3V8oVUk5LxbryB3UIIVzNmcSIn3JrXynlvui4MixvrtX6zx+O/bBo68o8/eZD26QrahVbA09fivrn/4h3TM019Eu/c2jOdckfU3cHUV/3Tno5d6JicibyaoDDK7S/yjdn5jhaz8MSEayQvFkZkiF0L public key test'
+sshpubkeyfp = u'13:67:6B:BF:4E:A2:05:8E:AE:25:8B:A1:31:DE:6F:1B public key test (ssh-rsa)'
+
 class test_host(Declarative):
 
     cleanup_commands = [
@@ -541,6 +544,45 @@ class test_host(Declarative):
         ),
 
 
+        dict(
+            desc='Add SSH public key to %r' % fqdn1,
+            command=('host_mod', [fqdn1], dict(ipasshpubkey=[sshpubkey])),
+            expected=dict(
+                value=fqdn1,
+                summary=u'Modified host "%s"' % fqdn1,
+                result=dict(
+                    description=[u'Updated host 1'],
+                    fqdn=[fqdn1],
+                    l=[u'Undisclosed location 1'],
+                    krbprincipalname=[u'host/%s@%s' % (fqdn1, api.env.realm)],
+                    managedby_host=[u'%s' % fqdn1],
+                    usercertificate=[base64.b64decode(servercert)],
+                    valid_not_before=fuzzy_date,
+                    valid_not_after=fuzzy_date,
+                    subject=DN(('CN',api.env.host),x509.subject_base()),
+                    serial_number=fuzzy_digits,
+                    serial_number_hex=fuzzy_hex,
+                    md5_fingerprint=fuzzy_hash,
+                    sha1_fingerprint=fuzzy_hash,
+                    issuer=fuzzy_issuer,
+                    macaddress=[u'00:50:56:30:F6:5F', u'00:50:56:2C:8D:82'],
+                    ipasshpubkey=[sshpubkey],
+                    sshpubkeyfp=[sshpubkeyfp],
+                    has_keytab=False,
+                    has_password=False,
+                ),
+            ),
+        ),
+
+
+        dict(
+            desc='Add an illegal SSH public key to %r' % fqdn1,
+            command=('host_mod', [fqdn1], dict(ipasshpubkey=[u'no-pty %s' % sshpubkey])),
+            expected=errors.ValidationError(name='sshpubkey',
+                error=u'options are not allowed'),
+        ),
+
+
         dict(
             desc='Delete %r' % fqdn1,
             command=('host_del', [fqdn1], {}),

tests/test_xmlrpc/test_user_plugin.py

@@ -40,6 +40,9 @@ admins_group=u'admins'
 invaliduser1=u'+tuser1'
 invaliduser2=u'tuser1234567890123456789012345678901234567890'
 
+sshpubkey = u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDGAX3xAeLeaJggwTqMjxNwa6XHBUAikXPGMzEpVrlLDCZtv00djsFTBi38PkgxBJVkgRWMrcBsr/35lq7P6w8KGIwA8GI48Z0qBS2NBMJ2u9WQ2hjLN6GdMlo77O0uJY3251p12pCVIS/bHRSq8kHO2No8g7KA9fGGcagPfQH+ee3t7HUkpbQkFTmbPPN++r3V8oVUk5LxbryB3UIIVzNmcSIn3JrXynlvui4MixvrtX6zx+O/bBo68o8/eZD26QrahVbA09fivrn/4h3TM019Eu/c2jOdckfU3cHUV/3Tno5d6JicibyaoDDK7S/yjdn5jhaz8MSEayQvFkZkiF0L public key test'
+sshpubkeyfp = u'13:67:6B:BF:4E:A2:05:8E:AE:25:8B:A1:31:DE:6F:1B public key test (ssh-rsa)'
+
 def get_user_dn(uid):
     return DN(('uid', uid), api.env.container_user, api.env.basedn)
 
@@ -562,6 +565,64 @@ class test_user(Declarative):
         ),
 
 
+        dict(
+            desc='Create "%s" with SSH public key' % user1,
+            command=(
+                'user_add', [user1], dict(givenname=u'Test', sn=u'User1', ipasshpubkey=[sshpubkey])
+            ),
+            expected=dict(
+                value=user1,
+                summary=u'Added user "%s"' % user1,
+                result=dict(
+                    gecos=[u'Test User1'],
+                    givenname=[u'Test'],
+                    homedirectory=[u'/home/tuser1'],
+                    krbprincipalname=[u'tuser1@' + api.env.realm],
+                    loginshell=[u'/bin/sh'],
+                    objectclass=objectclasses.user,
+                    sn=[u'User1'],
+                    uid=[user1],
+                    uidnumber=[fuzzy_digits],
+                    gidnumber=[fuzzy_digits],
+                    displayname=[u'Test User1'],
+                    cn=[u'Test User1'],
+                    initials=[u'TU'],
+                    mail=[u'%s@%s' % (user1, api.env.domain)],
+                    ipasshpubkey=[sshpubkey],
+                    sshpubkeyfp=[sshpubkeyfp],
+                    ipauniqueid=[fuzzy_uuid],
+                    krbpwdpolicyreference=[DN(('cn','global_policy'),('cn',api.env.realm),
+                                              ('cn','kerberos'),api.env.basedn)],
+                    mepmanagedentry=[get_group_dn(user1)],
+                    memberof_group=[u'ipausers'],
+                    has_keytab=False,
+                    has_password=False,
+                    dn=get_user_dn(user1),
+                ),
+            ),
+            extra_check = upg_check,
+        ),
+
+
+        dict(
+            desc='Add an illegal SSH public key to "%r"' % user1,
+            command=('user_mod', [user1], dict(ipasshpubkey=[u"anal nathrach orth' bhais's bethad do che'l de'nmha"])),
+            expected=errors.ValidationError(name='sshpubkey',
+                error=u'invalid SSH public key'),
+        ),
+
+
+        dict(
+            desc='Delete "%s"' % user1,
+            command=('user_del', [user1], {}),
+            expected=dict(
+                result=dict(failed=u''),
+                summary=u'Deleted user "%s"' % user1,
+                value=user1,
+            ),
+        ),
+
+
         dict(
             desc='Create "%s"' % user1,
             command=(