GSS-API / SSPI authenticated Diffie-Hellman Key Exchange and user

authentication with Python 3 support

Add Python 3 support for the GSS-API / SSPI authenticated Diffie-Hellman
Key Exchange and user authentication. This patch supersedes pull request
#250.

3 files changed, 446 insertions, 0 deletions

diff --git a/tests/test_gssapi.py b/tests/test_gssapi.py
new file mode 100644
index 00000000..e9ef99a9
--- /dev/null
+++ b/tests/test_gssapi.py
@@ -0,0 +1,167 @@
+# Copyright (C) 2013-2014 science + computing ag
+# Author: Sebastian Deiss <sebastian.deiss@t-online.de>
+#
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# Paramiko 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 Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Test the used APIs for GSS-API / SSPI authentication
+
+@author: Sebastian Deiss
+@contact: U{https://github.com/SebastianDeiss/paramiko/issues}
+@organization: science + computing ag
+               (U{EMail<mailto:a.kruis@science-computing.de>})
+@copyright: (C) 2013-2014 U{science + computing ag
+            <https://www.science-computing.de>}
+@license: GNU Lesser General Public License (LGPL)
+
+Created on 04.12.2013
+"""
+
+import unittest
+import socket
+
+
+class GSSAPITest(unittest.TestCase):
+
+    def init(hostname=None, srv_mode=False):
+        global krb5_mech, targ_name, server_mode
+        krb5_mech = "1.2.840.113554.1.2.2"
+        targ_name = hostname
+        server_mode = srv_mode
+
+    init = staticmethod(init)
+
+    def test_1_pyasn1(self):
+        """
+        Test the used methods of pyasn1.
+        """
+        from pyasn1.type.univ import ObjectIdentifier
+        from pyasn1.codec.der import encoder, decoder
+        oid = encoder.encode(ObjectIdentifier(krb5_mech))
+        mech, __ = decoder.decode(oid)
+        self.assertEquals(krb5_mech, mech.__str__())
+
+    def test_2_gssapi_sspi(self):
+        """
+        Test the used methods of python-gssapi or sspi, sspicon from pywin32.
+        """
+        _API = "MIT"
+        try:
+            import gssapi
+        except ImportError:
+            import sspicon
+            import sspi
+            _API = "SSPI"
+
+        c_token = None
+        gss_ctxt_status = False
+        mic_msg = b"G'day Mate!"
+
+        if _API == "MIT":
+            if server_mode:
+                gss_flags = (gssapi.C_PROT_READY_FLAG,
+                             gssapi.C_INTEG_FLAG,
+                             gssapi.C_MUTUAL_FLAG,
+                             gssapi.C_DELEG_FLAG)
+            else:
+                gss_flags = (gssapi.C_PROT_READY_FLAG,
+                             gssapi.C_INTEG_FLAG,
+                             gssapi.C_DELEG_FLAG)
+            """
+            Initialize a GSS-API context.
+            """
+            ctx = gssapi.Context()
+            ctx.flags = gss_flags
+            krb5_oid = gssapi.OID.mech_from_string(krb5_mech)
+            target_name = gssapi.Name("host@" + targ_name,
+                                      gssapi.C_NT_HOSTBASED_SERVICE)
+            gss_ctxt = gssapi.InitContext(peer_name=target_name,
+                                          mech_type=krb5_oid,
+                                          req_flags=ctx.flags)
+            if server_mode:
+                c_token = gss_ctxt.step(c_token)
+                gss_ctxt_status = gss_ctxt.established
+                self.assertEquals(False, gss_ctxt_status)
+                """
+                Accept a GSS-API context.
+                """
+                gss_srv_ctxt = gssapi.AcceptContext()
+                s_token = gss_srv_ctxt.step(c_token)
+                gss_ctxt_status = gss_srv_ctxt.established
+                self.assertNotEquals(None, s_token)
+                self.assertEquals(True, gss_ctxt_status)
+                """
+                Establish the client context
+                """
+                c_token = gss_ctxt.step(s_token)
+                self.assertEquals(None, c_token)
+            else:
+                while not gss_ctxt.established:
+                    c_token = gss_ctxt.step(c_token)
+                self.assertNotEquals(None, c_token)
+            """
+            Build MIC
+            """
+            mic_token = gss_ctxt.get_mic(mic_msg)
+
+            if server_mode:
+                """
+                Check MIC
+                """
+                status = gss_srv_ctxt.verify_mic(mic_msg, mic_token)
+                self.assertEquals(0, status)
+        else:
+            gss_flags = sspicon.ISC_REQ_INTEGRITY |\
+                        sspicon.ISC_REQ_MUTUAL_AUTH |\
+                        sspicon.ISC_REQ_DELEGATE
+            """
+            Initialize a GSS-API context.
+            """
+            target_name = "host/" + socket.getfqdn(targ_name)
+            gss_ctxt = sspi.ClientAuth("Kerberos",
+                                        scflags=gss_flags,
+                                        targetspn=target_name)
+            if server_mode:
+                error, token = gss_ctxt.authorize(c_token)
+                c_token = token[0].Buffer
+                self.assertEquals(0, error)
+                """
+                Accept a GSS-API context.
+                """
+                gss_srv_ctxt = sspi.ServerAuth("Kerberos", spn=target_name)
+                error, token = gss_srv_ctxt.authorize(c_token)
+                s_token = token[0].Buffer
+                """
+                Establish the context.
+                """
+                error, token = gss_ctxt.authorize(s_token)
+                c_token = token[0].Buffer
+                self.assertEquals(None, c_token)
+                self.assertEquals(0, error)
+                """
+                Build MIC
+                """
+                mic_token = gss_ctxt.sign(mic_msg)
+                """
+                Check MIC
+                """
+                gss_srv_ctxt.verify(mic_msg, mic_token)
+            else:
+                error, token = gss_ctxt.authorize(c_token)
+                c_token = token[0].Buffer
+                self.assertNotEquals(0, error)
diff --git a/tests/test_kex_gss.py b/tests/test_kex_gss.py
new file mode 100644
index 00000000..e160eb35
--- /dev/null
+++ b/tests/test_kex_gss.py
@@ -0,0 +1,143 @@
+# Copyright (C) 2003-2007  Robey Pointer <robeypointer@gmail.com>
+# Copyright (C) 2013-2014 science + computing ag
+# Author: Sebastian Deiss <sebastian.deiss@t-online.de>
+#
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# Paramiko 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 Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Unit Tests for the GSS-API / SSPI SSHv2 Diffie-Hellman Key Exchange and user
+authentication
+
+@author: Sebastian Deiss
+@contact: U{https://github.com/SebastianDeiss/paramiko/issues}
+@organization: science + computing ag
+               (U{EMail<mailto:a.kruis@science-computing.de>})
+@copyright: (C) 2003-2009  Robey Pointer, (C) 2013-2014 U{science + computing ag
+            <https://www.science-computing.de>}
+@license: GNU Lesser General Public License (LGPL)
+
+Created on 08.01.2014
+"""
+
+
+import socket
+import threading
+import unittest
+
+import paramiko
+
+
+class NullServer (paramiko.ServerInterface):
+
+    def get_allowed_auths(self, username):
+        return 'gssapi-keyex'
+
+    def check_auth_gssapi_keyex(self, username,
+                                gss_authenticated=paramiko.AUTH_FAILED,
+                                cc_file=None):
+        if gss_authenticated == paramiko.AUTH_SUCCESSFUL:
+            return paramiko.AUTH_SUCCESSFUL
+        return paramiko.AUTH_FAILED
+
+    def enable_auth_gssapi(self):
+        UseGSSAPI = True
+        return UseGSSAPI
+
+    def check_channel_request(self, kind, chanid):
+        return paramiko.OPEN_SUCCEEDED
+
+    def check_channel_exec_request(self, channel, command):
+        if command != 'yes':
+            return False
+        return True
+
+
+class GSSKexTest(unittest.TestCase):
+
+    def init(username, hostname):
+        global krb5_principal, targ_name
+        krb5_principal = username
+        targ_name = hostname
+
+    init = staticmethod(init)
+
+    def setUp(self):
+        self.username = krb5_principal
+        self.hostname = socket.getfqdn(targ_name)
+        self.sockl = socket.socket()
+        self.sockl.bind((targ_name, 0))
+        self.sockl.listen(1)
+        self.addr, self.port = self.sockl.getsockname()
+        self.event = threading.Event()
+        thread = threading.Thread(target=self._run)
+        thread.start()
+
+    def tearDown(self):
+        for attr in "tc ts socks sockl".split():
+            if hasattr(self, attr):
+                getattr(self, attr).close()
+
+    def _run(self):
+        self.socks, addr = self.sockl.accept()
+        self.ts = paramiko.Transport(self.socks, True)
+        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+        self.ts.add_server_key(host_key)
+        self.ts.set_gss_host(targ_name)
+        try:
+            self.ts.load_server_moduli()
+        except:
+            print ('(Failed to load moduli -- gex will be unsupported.)')
+        server = NullServer()
+        self.ts.start_server(self.event, server)
+
+    def test_1_gsskex_and_auth(self):
+        """
+        Verify that Paramiko can handle SSHv2 GSS-API / SSPI authenticated
+        Diffie-Hellman Key Exchange and user authentication with the GSS-API
+        context created during key exchange.
+        """
+        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = paramiko.RSAKey(data=host_key.asbytes())
+
+        self.tc = paramiko.SSHClient()
+        self.tc.get_host_keys().add('[%s]:%d' % (self.hostname, self.port),
+                                    'ssh-rsa', public_host_key)
+        self.tc.connect(self.hostname, self.port, username=self.username,
+                        gss_auth=True, gss_kex=True)
+
+        self.event.wait(1.0)
+        self.assert_(self.event.isSet())
+        self.assert_(self.ts.is_active())
+        self.assertEquals(self.username, self.ts.get_username())
+        self.assertEquals(True, self.ts.is_authenticated())
+
+        stdin, stdout, stderr = self.tc.exec_command('yes')
+        schan = self.ts.accept(1.0)
+
+        schan.send('Hello there.\n')
+        schan.send_stderr('This is on stderr.\n')
+        schan.close()
+
+        self.assertEquals('Hello there.\n', stdout.readline())
+        self.assertEquals('', stdout.readline())
+        self.assertEquals('This is on stderr.\n', stderr.readline())
+        self.assertEquals('', stderr.readline())
+
+        stdin.close()
+        stdout.close()
+        stderr.close()
diff --git a/tests/test_ssh_gss.py b/tests/test_ssh_gss.py
new file mode 100644
index 00000000..98e280ec
--- /dev/null
+++ b/tests/test_ssh_gss.py
@@ -0,0 +1,136 @@
+# Copyright (C) 2003-2007  Robey Pointer <robeypointer@gmail.com>
+# Copyright (C) 2013-2014 science + computing ag
+# Author: Sebastian Deiss <sebastian.deiss@t-online.de>
+#
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# Paramiko 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 Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
+
+"""
+Unit Tests for the GSS-API / SSPI SSHv2 Authentication (gssapi-with-mic)
+
+@author: Sebastian Deiss
+@contact: U{https://github.com/SebastianDeiss/paramiko/issues}
+@organization: science + computing ag
+               (U{EMail<mailto:a.kruis@science-computing.de>})
+@copyright: (C) 2003-2007  Robey Pointer, (C) 2013-2014 U{science + computing ag
+            <https://www.science-computing.de>}
+@license: GNU Lesser General Public License (LGPL)
+
+Created on 04.12.2013
+"""
+
+import socket
+import threading
+import unittest
+
+import paramiko
+
+
+class NullServer (paramiko.ServerInterface):
+
+    def get_allowed_auths(self, username):
+        return 'gssapi-with-mic'
+
+    def check_auth_gssapi_with_mic(self, username,
+                                   gss_authenticated=paramiko.AUTH_FAILED,
+                                   cc_file=None):
+        if gss_authenticated == paramiko.AUTH_SUCCESSFUL:
+            return paramiko.AUTH_SUCCESSFUL
+        return paramiko.AUTH_FAILED
+
+    def enable_auth_gssapi(self):
+        UseGSSAPI = True
+        GSSAPICleanupCredentials = True
+        return UseGSSAPI
+
+    def check_channel_request(self, kind, chanid):
+        return paramiko.OPEN_SUCCEEDED
+
+    def check_channel_exec_request(self, channel, command):
+        if command != 'yes':
+            return False
+        return True
+
+
+class GSSAuthTest(unittest.TestCase):
+
+    def init(username, hostname):
+        global krb5_principal, targ_name
+        krb5_principal = username
+        targ_name = hostname
+
+    init = staticmethod(init)
+
+    def setUp(self):
+        self.username = krb5_principal
+        self.hostname = socket.getfqdn(targ_name)
+        self.sockl = socket.socket()
+        self.sockl.bind((targ_name, 0))
+        self.sockl.listen(1)
+        self.addr, self.port = self.sockl.getsockname()
+        self.event = threading.Event()
+        thread = threading.Thread(target=self._run)
+        thread.start()
+
+    def tearDown(self):
+        for attr in "tc ts socks sockl".split():
+            if hasattr(self, attr):
+                getattr(self, attr).close()
+
+    def _run(self):
+        self.socks, addr = self.sockl.accept()
+        self.ts = paramiko.Transport(self.socks)
+        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+        self.ts.add_server_key(host_key)
+        server = NullServer()
+        self.ts.start_server(self.event, server)
+
+    def test_1_gss_auth(self):
+        """
+        Verify that Paramiko can handle SSHv2 GSS-API / SSPI authentication
+        (gssapi-with-mic) in client and server mode.
+        """
+        host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+        public_host_key = paramiko.RSAKey(data=host_key.asbytes())
+
+        self.tc = paramiko.SSHClient()
+        self.tc.get_host_keys().add('[%s]:%d' % (self.hostname, self.port),
+                                    'ssh-rsa', public_host_key)
+        self.tc.connect(self.hostname, self.port, username=self.username,
+                        gss_auth=True)
+
+        self.event.wait(1.0)
+        self.assert_(self.event.isSet())
+        self.assert_(self.ts.is_active())
+        self.assertEquals(self.username, self.ts.get_username())
+        self.assertEquals(True, self.ts.is_authenticated())
+
+        stdin, stdout, stderr = self.tc.exec_command('yes')
+        schan = self.ts.accept(1.0)
+
+        schan.send('Hello there.\n')
+        schan.send_stderr('This is on stderr.\n')
+        schan.close()
+
+        self.assertEquals('Hello there.\n', stdout.readline())
+        self.assertEquals('', stdout.readline())
+        self.assertEquals('This is on stderr.\n', stderr.readline())
+        self.assertEquals('', stderr.readline())
+
+        stdin.close()
+        stdout.close()
+        stderr.close()