summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--API.txt8
-rw-r--r--VERSION2
-rwxr-xr-xipa-client/ipa-install/ipa-client-install29
-rw-r--r--ipalib/parameters.py2
-rw-r--r--ipalib/plugins/host.py33
-rw-r--r--ipalib/plugins/user.py26
-rw-r--r--ipalib/util.py56
-rw-r--r--ipapython/ipautil.py20
-rw-r--r--ipapython/ssh.py199
-rw-r--r--tests/test_ipapython/test_ssh.py76
-rw-r--r--tests/test_xmlrpc/test_host_plugin.py42
-rw-r--r--tests/test_xmlrpc/test_user_plugin.py61
12 files changed, 464 insertions, 90 deletions
diff --git a/API.txt b/API.txt
index 6aee01ac5..1906e22fe 100644
--- a/API.txt
+++ b/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')
diff --git a/VERSION b/VERSION
index 25776435c..962d476e7 100644
--- a/VERSION
+++ b/VERSION
@@ -79,4 +79,4 @@ IPA_DATA_VERSION=20100614120000
# #
########################################################
IPA_API_VERSION_MAJOR=2
-IPA_API_VERSION_MINOR=42
+IPA_API_VERSION_MINOR=43
diff --git a/ipa-client/ipa-install/ipa-client-install b/ipa-client/ipa-install/ipa-client-install
index d87fcc2a6..03a8bd3e8 100755
--- a/ipa-client/ipa-install/ipa-client-install
+++ b/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]
- if line.startswith('#'):
- continue
- parts = line.split()
- if len(parts) < 2:
- continue
- try:
- pubkey = b64decode(parts[1])
- except TypeError:
+ line = line[:-1].lstrip()
+ if not line or line.startswith('#'):
continue
try:
- algo, data, fp = ipautil.decode_ssh_pubkey(pubkey)
- except ValueError:
- continue
- if parts[0] != algo:
+ pubkey = SSHPublicKey(line)
+ except ValueError, UnicodeDecodeError:
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 = ipautil.make_sshfp(pubkey)
+ sshfp = pubkey.fingerprint_dns_sha1()
if sshfp is not None:
update_txt += 'update add %s. %s IN SSHFP %s\n' % (hostname, ttl, sshfp)
update_txt += 'send\n'
diff --git a/ipalib/parameters.py b/ipalib/parameters.py
index 13f04b454..53756a80a 100644
--- a/ipalib/parameters.py
+++ b/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:
diff --git a/ipalib/plugins/host.py b/ipalib/plugins/host.py
index 91b3ce677..e1c07b53b 100644
--- a/ipalib/plugins/host.py
+++ b/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 ipapython.ipautil import ipa_generate_password, CheckedIPAddress, make_sshfp
+from ipalib.util import (normalize_sshpubkey, validate_sshpubkey_no_options,
+ 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
diff --git a/ipalib/plugins/user.py b/ipalib/plugins/user.py
index 3f0050917..84a63dfa6 100644
--- a/ipalib/plugins/user.py
+++ b/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)
diff --git a/ipalib/util.py b/ipalib/util.py
index 155d93294..ca71e78db 100644
--- a/ipalib/util.py
+++ b/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)
- except ValueError:
+ SSHPublicKey(value)
+ 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'])
- pubkeys = entry[1].get('ipasshpubkey')
- if pubkeys is None:
+ old_entry_attrs = ldap.get_entry(dn, ['ipasshpubkey'])
+ pubkeys = old_entry_attrs[1].get('ipasshpubkey')
+ if not pubkeys:
return
+ newpubkeys = []
fingerprints = []
for pubkey in pubkeys:
try:
- algo, data, fp = decode_ssh_pubkey(pubkey)
- fp = u':'.join([fp[j:j+2] for j in range(0, len(fp), 2)])
- fingerprints.append(u'%s (%s)' % (fp, algo))
- except ValueError:
- pass
+ pubkey = SSHPublicKey(pubkey)
+ except ValueError, UnicodeDecodeError:
+ continue
+
+ 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.
diff --git a/ipapython/ipautil.py b/ipapython/ipautil.py
index a212aa6ef..a3fd83e45 100644
--- a/ipapython/ipautil.py
+++ b/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):
diff --git a/ipapython/ssh.py b/ipapython/ssh.py
new file mode 100644
index 000000000..667d21e1e
--- /dev/null
+++ b/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)
diff --git a/tests/test_ipapython/test_ssh.py b/tests/test_ipapython/test_ssh.py
new file mode 100644
index 000000000..2640af50d
--- /dev/null
+++ b/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
diff --git a/tests/test_xmlrpc/test_host_plugin.py b/tests/test_xmlrpc/test_host_plugin.py
index 27d3adb04..b3eb3151e 100644
--- a/tests/test_xmlrpc/test_host_plugin.py
+++ b/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 = [
@@ -542,6 +545,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], {}),
expected=dict(
diff --git a/tests/test_xmlrpc/test_user_plugin.py b/tests/test_xmlrpc/test_user_plugin.py
index d374e02f0..15a195590 100644
--- a/tests/test_xmlrpc/test_user_plugin.py
+++ b/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)
@@ -563,6 +566,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=(
'user_add', [user1], dict(givenname=u'Test', sn=u'User1')