--- /dev/null
+#!/usr/bin/env python3
+# Unix SMB/CIFS implementation.
+# Copyright (C) Stefan Metzmacher 2020
+# Copyright (C) 2022 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 samba.dcerpc import security
+
+from samba.tests.krb5.kdc_tgs_tests import KdcTgsBaseTests
+from samba.tests.krb5.rfc4120_constants import (
+ AES256_CTS_HMAC_SHA1_96,
+ ARCFOUR_HMAC_MD5,
+ KDC_ERR_ETYPE_NOSUPP,
+)
+
+sys.path.insert(0, "bin/python")
+os.environ["PYTHONUNBUFFERED"] = "1"
+
+global_asn1_print = False
+global_hexdump = False
+
+
+class EtypeTests(KdcTgsBaseTests):
+ def setUp(self):
+ super().setUp()
+ self.do_asn1_print = global_asn1_print
+ self.do_hexdump = global_hexdump
+
+ # Perform an AS-REQ for a service ticket, specifying AES. The request
+ # should fail with an error.
+ def test_as_aes_requested(self):
+ creds = self.get_mach_creds()
+ target_creds = self.get_service_creds()
+
+ self._as_req(creds, expected_error=KDC_ERR_ETYPE_NOSUPP,
+ target_creds=target_creds,
+ etype=(AES256_CTS_HMAC_SHA1_96,))
+
+ # Perform an AS-REQ for a service ticket, specifying RC4. The resulting
+ # ticket should be encrypted with RC4, with an RC4 session key.
+ def test_as_rc4_requested(self):
+ creds = self.get_mach_creds()
+ target_creds = self.get_service_creds()
+
+ ticket = self._as_req(creds, expected_error=0,
+ target_creds=target_creds,
+ etype=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+ # Perform an AS-REQ for a service ticket, specifying AES, when the target
+ # service only supports AES. The resulting ticket should be encrypted with
+ # AES, with an AES session key.
+ def test_as_aes_supported_aes_requested(self):
+ creds = self.get_mach_creds()
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_AES256_CTS_HMAC_SHA1_96,
+ })
+
+ ticket = self._as_req(creds, expected_error=0,
+ target_creds=target_creds,
+ etype=(AES256_CTS_HMAC_SHA1_96,))
+
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.decryption_key.etype)
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.session_key.etype)
+
+ # Perform an AS-REQ for a service ticket, specifying RC4, when the target
+ # service only supports AES. The resulting ticket should be encrypted with
+ # AES, with an RC4 session key.
+ def test_as_aes_supported_rc4_requested(self):
+ creds = self.get_mach_creds()
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_AES256_CTS_HMAC_SHA1_96,
+ })
+
+ ticket = self._as_req(creds, expected_error=0,
+ target_creds=target_creds,
+ etype=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+ # Perform an AS-REQ for a service ticket, specifying AES, when the target
+ # service only supports RC4. The request should fail with an error.
+ def test_as_rc4_supported_aes_requested(self):
+ creds = self.get_mach_creds()
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_RC4_HMAC_MD5,
+ })
+
+ self._as_req(creds, expected_error=KDC_ERR_ETYPE_NOSUPP,
+ target_creds=target_creds,
+ etype=(AES256_CTS_HMAC_SHA1_96,))
+
+ # Perform an AS-REQ for a service ticket, specifying RC4, when the target
+ # service only supports RC4. The resulting ticket should be encrypted with
+ # RC4, with an RC4 session key.
+ def test_as_rc4_supported_rc4_requested(self):
+ creds = self.get_mach_creds()
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_RC4_HMAC_MD5,
+ })
+
+ ticket = self._as_req(creds, expected_error=0,
+ target_creds=target_creds,
+ etype=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+ # Perform a TGS-REQ for a service ticket, specifying AES. The request
+ # should fail with an error.
+ def test_tgs_aes_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_mach_creds()
+
+ self._tgs_req(tgt, expected_error=KDC_ERR_ETYPE_NOSUPP,
+ target_creds=target_creds,
+ etypes=(AES256_CTS_HMAC_SHA1_96,))
+
+ # Perform a TGS-REQ for a service ticket, specifying RC4. The resulting
+ # ticket should be encrypted with RC4, with an RC4 session key.
+ def test_tgs_rc4_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_mach_creds()
+
+ ticket = self._tgs_req(tgt, expected_error=0,
+ target_creds=target_creds,
+ etypes=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+ # Perform a TGS-REQ for a service ticket, specifying AES, when the target
+ # service only supports AES. The resulting ticket should be encrypted with
+ # AES, with an AES session key.
+ def test_tgs_aes_supported_aes_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_AES256_CTS_HMAC_SHA1_96,
+ })
+
+ ticket = self._tgs_req(tgt, expected_error=0,
+ target_creds=target_creds,
+ etypes=(AES256_CTS_HMAC_SHA1_96,))
+
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.decryption_key.etype)
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.session_key.etype)
+
+ # Perform a TGS-REQ for a service ticket, specifying RC4, when the target
+ # service only supports AES. The resulting ticket should be encrypted with
+ # AES, with an RC4 session key.
+ def test_tgs_aes_supported_rc4_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_AES256_CTS_HMAC_SHA1_96,
+ })
+
+ ticket = self._tgs_req(tgt, expected_error=0,
+ target_creds=target_creds,
+ etypes=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(AES256_CTS_HMAC_SHA1_96, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+ # Perform a TGS-REQ for a service ticket, specifying AES, when the target
+ # service only supports RC4. The request should fail with an error.
+ def test_tgs_rc4_supported_aes_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_RC4_HMAC_MD5,
+ })
+
+ self._tgs_req(tgt, expected_error=KDC_ERR_ETYPE_NOSUPP,
+ target_creds=target_creds,
+ etypes=(AES256_CTS_HMAC_SHA1_96,))
+
+ # Perform a TGS-REQ for a service ticket, specifying RC4, when the target
+ # service only supports RC4. The resulting ticket should be encrypted with
+ # RC4, with an RC4 session key.
+ def test_tgs_rc4_supported_rc4_requested(self):
+ creds = self.get_mach_creds()
+ tgt = self.get_tgt(creds)
+
+ target_creds = self.get_cached_creds(
+ account_type=self.AccountType.COMPUTER,
+ opts={
+ 'supported_enctypes':
+ security.KERB_ENCTYPE_RC4_HMAC_MD5,
+ })
+
+ ticket = self._tgs_req(tgt, expected_error=0,
+ target_creds=target_creds,
+ etypes=(ARCFOUR_HMAC_MD5,))
+
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.decryption_key.etype)
+ self.assertEqual(ARCFOUR_HMAC_MD5, ticket.session_key.etype)
+
+
+if __name__ == "__main__":
+ global_asn1_print = False
+ global_hexdump = False
+ import unittest
+ unittest.main()
KDC_ERR_MODIFIED,
KDC_ERR_NOT_US,
KDC_ERR_POLICY,
+ KDC_ERR_PREAUTH_REQUIRED,
KDC_ERR_C_PRINCIPAL_UNKNOWN,
KDC_ERR_S_PRINCIPAL_UNKNOWN,
KDC_ERR_TGT_REVOKED,
class KdcTgsBaseTests(KDCBaseTest):
+ def _as_req(self,
+ creds,
+ expected_error,
+ target_creds,
+ etype):
+ user_name = creds.get_username()
+ cname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
+ names=user_name.split('/'))
+
+ target_name = target_creds.get_username()
+ sname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
+ names=['host', target_name[:-1]])
+
+ if expected_error:
+ expected_sname = sname
+ else:
+ expected_sname = self.PrincipalName_create(name_type=NT_PRINCIPAL,
+ names=[target_name])
+
+ realm = creds.get_realm()
+ salt = creds.get_salt()
+
+ till = self.get_KerberosTime(offset=36000)
+
+ ticket_decryption_key = (
+ self.TicketDecryptionKey_from_creds(target_creds))
+ expected_etypes = target_creds.tgs_supported_enctypes
+
+ kdc_options = ('forwardable,'
+ 'renewable,'
+ 'canonicalize,'
+ 'renewable-ok')
+ kdc_options = krb5_asn1.KDCOptions(kdc_options)
+
+ if expected_error:
+ initial_error = (KDC_ERR_PREAUTH_REQUIRED, expected_error)
+ else:
+ initial_error = KDC_ERR_PREAUTH_REQUIRED
+
+ rep, kdc_exchange_dict = self._test_as_exchange(
+ cname=cname,
+ realm=realm,
+ sname=sname,
+ till=till,
+ client_as_etypes=etype,
+ expected_error_mode=initial_error,
+ expected_crealm=realm,
+ expected_cname=cname,
+ expected_srealm=realm,
+ expected_sname=sname,
+ expected_salt=salt,
+ expected_supported_etypes=expected_etypes,
+ etypes=etype,
+ padata=None,
+ kdc_options=kdc_options,
+ preauth_key=None,
+ ticket_decryption_key=ticket_decryption_key)
+ self.assertIsNotNone(rep)
+ self.assertEqual(KRB_ERROR, rep['msg-type'])
+ error_code = rep['error-code']
+ if expected_error:
+ self.assertIn(error_code, initial_error)
+ if error_code == expected_error:
+ return
+ else:
+ self.assertEqual(initial_error, error_code)
+
+ etype_info2 = kdc_exchange_dict['preauth_etype_info2']
+
+ preauth_key = self.PasswordKey_from_etype_info2(creds,
+ etype_info2[0],
+ creds.get_kvno())
+
+ ts_enc_padata = self.get_enc_timestamp_pa_data_from_key(preauth_key)
+
+ padata = [ts_enc_padata]
+
+ expected_realm = realm.upper()
+
+ rep, kdc_exchange_dict = self._test_as_exchange(
+ cname=cname,
+ realm=realm,
+ sname=sname,
+ till=till,
+ client_as_etypes=etype,
+ expected_error_mode=expected_error,
+ expected_crealm=expected_realm,
+ expected_cname=cname,
+ expected_srealm=expected_realm,
+ expected_sname=expected_sname,
+ expected_salt=salt,
+ expected_supported_etypes=expected_etypes,
+ etypes=etype,
+ padata=padata,
+ kdc_options=kdc_options,
+ preauth_key=preauth_key,
+ ticket_decryption_key=ticket_decryption_key,
+ expect_edata=False)
+ if expected_error:
+ self.check_error_rep(rep, expected_error)
+ return None
+
+ self.check_as_reply(rep)
+ return kdc_exchange_dict['rep_ticket_creds']
+
def _tgs_req(self, tgt, expected_error, target_creds,
armor_tgt=None,
kdc_options='0',
projects / thirdparty / samba.git / commitdiff
