python/samba/tests/gensec.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2009
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 """Tests for GENSEC.
  19
  20 Note that this just tests the bindings work. It does not intend to test
  21 the functionality, that's already done in other tests.
  22 """
  23
  24 from samba.credentials import Credentials
  25 from samba import gensec, auth
  26 import samba.tests
  27
  28 class GensecTests(samba.tests.TestCase):
  29
  30     def setUp(self):
  31         super(GensecTests, self).setUp()
  32         self.settings = {}
  33         self.settings["lp_ctx"] = self.lp_ctx = samba.tests.env_loadparm()
  34         self.settings["target_hostname"] = self.lp_ctx.get("netbios name")
  35         """This is just for the API tests"""
  36         self.gensec = gensec.Security.start_client(self.settings)
  37
  38     def test_start_mech_by_unknown_name(self):
  39         self.assertRaises(RuntimeError, self.gensec.start_mech_by_name, "foo")
  40
  41     def test_start_mech_by_name(self):
  42         self.gensec.start_mech_by_name("spnego")
  43
  44     def test_info_uninitialized(self):
  45         self.assertRaises(RuntimeError, self.gensec.session_info)
  46
  47     def test_update(self):
  48         """Test GENSEC by doing an exchange with ourselves using GSSAPI against a KDC"""
  49
  50         """Start up a client and server GENSEC instance to test things with"""
  51
  52         self.gensec_client = gensec.Security.start_client(self.settings)
  53         self.gensec_client.set_credentials(self.get_credentials())
  54         self.gensec_client.want_feature(gensec.FEATURE_SEAL)
  55         self.gensec_client.start_mech_by_sasl_name("GSSAPI")
  56
  57         self.gensec_server = gensec.Security.start_server(settings=self.settings,
  58                                                           auth_context=auth.AuthContext(lp_ctx=self.lp_ctx))
  59         creds = Credentials()
  60         creds.guess(self.lp_ctx)
  61         creds.set_machine_account(self.lp_ctx)
  62         self.gensec_server.set_credentials(creds)
  63
  64         self.gensec_server.want_feature(gensec.FEATURE_SEAL)
  65         self.gensec_server.start_mech_by_sasl_name("GSSAPI")
  66
  67         client_finished = False
  68         server_finished = False
  69         server_to_client = ""
  70
  71         """Run the actual call loop"""
  72         while not client_finished and not server_finished:
  73             if not client_finished:
  74                 print "running client gensec_update"
  75                 (client_finished, client_to_server) = self.gensec_client.update(server_to_client)
  76             if not server_finished:
  77                 print "running server gensec_update"
  78                 (server_finished, server_to_client) = self.gensec_server.update(client_to_server)
  79         session_info = self.gensec_server.session_info()
  80
  81         test_string = "Hello Server"
  82         test_wrapped = self.gensec_client.wrap(test_string)
  83         test_unwrapped = self.gensec_server.unwrap(test_wrapped)
  84         self.assertEqual(test_string, test_unwrapped)
  85         test_string = "Hello Client"
  86         test_wrapped = self.gensec_server.wrap(test_string)
  87         test_unwrapped = self.gensec_client.unwrap(test_wrapped)
  88         self.assertEqual(test_string, test_unwrapped)
  89
  90         client_session_key = self.gensec_client.session_key()
  91         server_session_key = self.gensec_server.session_key()
  92         self.assertEqual(client_session_key, server_session_key)
  93
  94     def test_max_update_size(self):
  95         """Test GENSEC by doing an exchange with ourselves using GSSAPI against a KDC"""
  96
  97         """Start up a client and server GENSEC instance to test things with"""
  98
  99         self.gensec_client = gensec.Security.start_client(self.settings)
 100         self.gensec_client.set_credentials(self.get_credentials())
 101         self.gensec_client.want_feature(gensec.FEATURE_SIGN)
 102         self.gensec_client.set_max_update_size(5)
 103         self.gensec_client.start_mech_by_name("spnego")
 104
 105         self.gensec_server = gensec.Security.start_server(settings=self.settings,
 106                                                           auth_context=auth.AuthContext(lp_ctx=self.lp_ctx))
 107         creds = Credentials()
 108         creds.guess(self.lp_ctx)
 109         creds.set_machine_account(self.lp_ctx)
 110         self.gensec_server.set_credentials(creds)
 111         self.gensec_server.want_feature(gensec.FEATURE_SIGN)
 112         self.gensec_server.set_max_update_size(5)
 113         self.gensec_server.start_mech_by_name("spnego")
 114
 115         client_finished = False
 116         server_finished = False
 117         server_to_client = ""
 118
 119         """Run the actual call loop"""
 120         i = 0
 121         while not client_finished or not server_finished:
 122             i += 1
 123             if not client_finished:
 124                 print "running client gensec_update: %d: %r" % (len(server_to_client), server_to_client)
 125                 (client_finished, client_to_server) = self.gensec_client.update(server_to_client)
 126             if not server_finished:
 127                 print "running server gensec_update: %d: %r" % (len(client_to_server), client_to_server)
 128                 (server_finished, server_to_client) = self.gensec_server.update(client_to_server)
 129
 130         """Here we expect a lot more than the typical 1 or 2 roundtrips"""
 131         self.assertTrue(i > 10)
 132
 133         session_info = self.gensec_server.session_info()
 134
 135         test_string = "Hello Server"
 136         test_wrapped = self.gensec_client.wrap(test_string)
 137         test_unwrapped = self.gensec_server.unwrap(test_wrapped)
 138         self.assertEqual(test_string, test_unwrapped)
 139         test_string = "Hello Client"
 140         test_wrapped = self.gensec_server.wrap(test_string)
 141         test_unwrapped = self.gensec_client.unwrap(test_wrapped)
 142         self.assertEqual(test_string, test_unwrapped)
 143
 144         client_session_key = self.gensec_client.session_key()
 145         server_session_key = self.gensec_server.session_key()
 146         self.assertEqual(client_session_key, server_session_key)