mirror of
https://github.com/samba-team/samba.git
synced 2025-02-25 17:57:42 +03:00
python: Add credentials cache test
Test that we can use a credentials cache with a user's service ticket obtained with our Python code to connect to a service using the normal credentials system backed on to MIT/Heimdal Kerberos 5 libraries. This will allow us to validate the output of the MIT/Heimdal libraries in the future. Signed-off-by: Joseph Sutton <josephsutton@catalyst.net.nz> Reviewed-by: Andrew Bartlett <abartlet@samba.org> Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
This commit is contained in:
Joseph Sutton
Jeremy Allison
parent
2d88a6ff3d
commit
c15f26ec40
@ -25,7 +25,7 @@ import random
|
||||
|
||||
import samba.tests
|
||||
from samba.credentials import Credentials
|
||||
from samba.tests import TestCase
|
||||
from samba.tests import TestCaseInTempDir
|
||||
import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
|
||||
import samba.tests.krb5.kcrypto as kcrypto
|
||||
|
||||
@ -178,11 +178,11 @@ class Krb5EncryptionKey(object):
|
||||
return EncryptionKey_obj
|
||||
|
||||
|
||||
class RawKerberosTest(TestCase):
|
||||
class RawKerberosTest(TestCaseInTempDir):
|
||||
"""A raw Kerberos Test case."""
|
||||
|
||||
def setUp(self):
|
||||
super(RawKerberosTest, self).setUp()
|
||||
super().setUp()
|
||||
self.do_asn1_print = False
|
||||
self.do_hexdump = False
|
||||
|
||||
@ -192,7 +192,7 @@ class RawKerberosTest(TestCase):
|
||||
|
||||
def tearDown(self):
|
||||
self._disconnect("tearDown")
|
||||
super(TestCase, self).tearDown()
|
||||
super().tearDown()
|
||||
|
||||
def _disconnect(self, reason):
|
||||
if self.s is None:
|
||||
|
||||
@ -46,6 +46,7 @@ KDC_ERR_SKEW = 37
|
||||
# Name types
|
||||
NT_UNKNOWN = int(krb5_asn1.NameTypeValues('kRB5-NT-UNKNOWN'))
|
||||
NT_PRINCIPAL = int(krb5_asn1.NameTypeValues('kRB5-NT-PRINCIPAL'))
|
||||
NT_SRV_HST = int(krb5_asn1.NameTypeValues('kRB5-NT-SRV-HST'))
|
||||
NT_SRV_INST = int(krb5_asn1.NameTypeValues('kRB5-NT-SRV-INST'))
|
||||
NT_ENTERPRISE_PRINCIPAL = int(krb5_asn1.NameTypeValues(
|
||||
'kRB5-NT-ENTERPRISE-PRINCIPAL'))
|
||||
|
||||
127
python/samba/tests/krb5/test_ccache.py
Executable file
127
python/samba/tests/krb5/test_ccache.py
Executable file
@ -0,0 +1,127 @@
|
||||
#!/usr/bin/env python3
|
||||
# Unix SMB/CIFS implementation.
|
||||
# Copyright (C) Stefan Metzmacher 2020
|
||||
# Copyright (C) 2021 Catalyst.Net Ltd
|
||||
#
|
||||
# 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/>.
|
||||
#
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
from ldb import SCOPE_SUBTREE
|
||||
from samba import gensec
|
||||
from samba.auth import AuthContext
|
||||
from samba.dcerpc import security
|
||||
from samba.ndr import ndr_unpack
|
||||
|
||||
from samba.tests.krb5.kdc_base_test import KDCBaseTest
|
||||
|
||||
sys.path.insert(0, "bin/python")
|
||||
os.environ["PYTHONUNBUFFERED"] = "1"
|
||||
|
||||
global_asn1_print = False
|
||||
global_hexdump = False
|
||||
|
||||
|
||||
class CcacheTests(KDCBaseTest):
|
||||
"""Test for authentication using Kerberos credentials stored in a
|
||||
credentials cache file.
|
||||
"""
|
||||
|
||||
def test_ccache(self):
|
||||
# Create a user account and a machine account, along with a Kerberos
|
||||
# credentials cache file where the service ticket authenticating the
|
||||
# user are stored.
|
||||
|
||||
user_name = "ccacheusr"
|
||||
mach_name = "ccachemac"
|
||||
|
||||
# Create the user account.
|
||||
(user_credentials, _) = self.create_account(user_name)
|
||||
|
||||
# Create the machine account.
|
||||
(mach_credentials, _) = self.create_account(mach_name,
|
||||
machine_account=True)
|
||||
|
||||
# Talk to the KDC to obtain the service ticket, which gets placed into
|
||||
# the cache. The machine account name has to match the name in the
|
||||
# ticket, to ensure that the krbtgt ticket doesn't also need to be
|
||||
# stored.
|
||||
(creds, cachefile) = self.create_ccache_with_user(user_credentials,
|
||||
mach_name)
|
||||
|
||||
# Authenticate in-process to the machine account using the user's
|
||||
# cached credentials.
|
||||
|
||||
settings = {}
|
||||
settings["lp_ctx"] = self.lp
|
||||
settings["target_hostname"] = mach_name
|
||||
|
||||
gensec_client = gensec.Security.start_client(settings)
|
||||
gensec_client.set_credentials(creds)
|
||||
gensec_client.want_feature(gensec.FEATURE_SEAL)
|
||||
gensec_client.start_mech_by_sasl_name("GSSAPI")
|
||||
|
||||
auth_context = AuthContext(lp_ctx=self.lp, ldb=self.ldb, methods=[])
|
||||
|
||||
gensec_server = gensec.Security.start_server(settings, auth_context)
|
||||
gensec_server.set_credentials(mach_credentials)
|
||||
|
||||
gensec_server.start_mech_by_sasl_name("GSSAPI")
|
||||
|
||||
client_finished = False
|
||||
server_finished = False
|
||||
server_to_client = b''
|
||||
|
||||
# Operate as both the client and the server to verify the user's
|
||||
# credentials.
|
||||
while not client_finished or not server_finished:
|
||||
if not client_finished:
|
||||
print("running client gensec_update")
|
||||
(client_finished, client_to_server) = gensec_client.update(
|
||||
server_to_client)
|
||||
if not server_finished:
|
||||
print("running server gensec_update")
|
||||
(server_finished, server_to_client) = gensec_server.update(
|
||||
client_to_server)
|
||||
|
||||
# Ensure that the first SID contained within the obtained security
|
||||
# token is the SID of the user we created.
|
||||
|
||||
# Retrieve the user account's SID.
|
||||
ldb_res = self.ldb.search(scope=SCOPE_SUBTREE,
|
||||
expression="(sAMAccountName=%s)" % user_name,
|
||||
attrs=["objectSid"])
|
||||
self.assertEqual(1, len(ldb_res))
|
||||
sid = ndr_unpack(security.dom_sid, ldb_res[0]["objectSid"][0])
|
||||
|
||||
# Retrieve the SIDs from the security token.
|
||||
session = gensec_server.session_info()
|
||||
token = session.security_token
|
||||
token_sids = token.sids
|
||||
self.assertGreater(len(token_sids), 0)
|
||||
|
||||
# Ensure that they match.
|
||||
self.assertEqual(sid, token_sids[0])
|
||||
|
||||
# Remove the cached credentials file.
|
||||
os.remove(cachefile.name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
global_asn1_print = True
|
||||
global_hexdump = True
|
||||
import unittest
|
||||
unittest.main()
|
||||
@ -96,6 +96,7 @@ EXCLUDE_USAGE = {
|
||||
'python/samba/tests/krb5/kdc_tests.py',
|
||||
'python/samba/tests/krb5/kdc_base_test.py',
|
||||
'python/samba/tests/krb5/kdc_tgs_tests.py',
|
||||
'python/samba/tests/krb5/test_ccache.py',
|
||||
'python/samba/tests/krb5/ms_kile_client_principal_lookup_tests.py',
|
||||
}
|
||||
|
||||
|
||||
@ -835,6 +835,8 @@ planoldpythontestsuite("ad_dc_default", "samba.tests.dsdb_dns")
|
||||
|
||||
planoldpythontestsuite("fl2008r2dc:local", "samba.tests.krb5.xrealm_tests")
|
||||
|
||||
planoldpythontestsuite("ad_dc_default", "samba.tests.krb5.test_ccache")
|
||||
|
||||
for env in ["ad_dc", smbv1_disabled_testenv]:
|
||||
planoldpythontestsuite(env, "samba.tests.smb", extra_args=['-U"$USERNAME%$PASSWORD"'])
|
||||
planoldpythontestsuite(env + ":local", "samba.tests.ntacls_backup",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user