mirror of
https://github.com/samba-team/samba.git
synced 2025-01-08 21:18:16 +03:00
selftest: add python S4U2Self tests including unkeyed checksums
To test the CRC32 I reverted the unkeyed-checksum fix (43958af1) and the weak-crypto fix (389d1b97). Note that the unkeyed-md5 still worked even with weak-crypto disabled, and that the unkeyed-sha1 never worked but I left it anyway. Signed-off-by: Isaac Boukris <iboukris@samba.org> Reviewed-by: Andrew Bartlett <abartlet@samba.org> Autobuild-User(master): Andrew Bartlett <abartlet@samba.org> Autobuild-Date(master): Fri May 15 12:25:40 UTC 2020 on sn-devel-184
This commit is contained in:
Isaac Boukris
Andrew Bartlett
parent
19875a3731
commit
8b5e764413
@ -51,6 +51,7 @@ os.environ["PYTHONUNBUFFERED"] = "1"
|
||||
from math import gcd
|
||||
from functools import reduce
|
||||
from struct import pack, unpack
|
||||
from binascii import crc32
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives import hmac
|
||||
from cryptography.hazmat.primitives.ciphers import algorithms as ciphers
|
||||
@ -533,6 +534,21 @@ class _MD5(_ChecksumProfile):
|
||||
return SIMPLE_HASH(text, hashes.MD5)
|
||||
|
||||
|
||||
class _SHA1(_ChecksumProfile):
|
||||
@classmethod
|
||||
def checksum(cls, key, keyusage, text):
|
||||
# This is unkeyed!
|
||||
return SIMPLE_HASH(text, hashes.SHA1)
|
||||
|
||||
|
||||
class _CRC32(_ChecksumProfile):
|
||||
@classmethod
|
||||
def checksum(cls, key, keyusage, text):
|
||||
# This is unkeyed!
|
||||
cksum = (~crc32(text, 0xffffffff)) & 0xffffffff
|
||||
return pack('<I', cksum)
|
||||
|
||||
|
||||
_enctype_table = {
|
||||
Enctype.DES3: _DES3CBC,
|
||||
Enctype.AES128: _AES128CTS,
|
||||
@ -547,6 +563,8 @@ _checksum_table = {
|
||||
Cksumtype.SHA1_AES256: _SHA1AES256,
|
||||
Cksumtype.HMAC_MD5: _HMACMD5,
|
||||
Cksumtype.MD5: _MD5,
|
||||
Cksumtype.SHA1: _SHA1,
|
||||
Cksumtype.CRC32: _CRC32,
|
||||
}
|
||||
|
||||
|
||||
@ -835,6 +853,73 @@ class KcrytoTest(TestCase):
|
||||
def test_md5_unkeyed_checksum_aes256_usage_50(self):
|
||||
return self._test_md5_unkeyed_checksum(Enctype.AES256, 50)
|
||||
|
||||
def _test_sha1_unkeyed_checksum(self, etype, usage):
|
||||
# SHA1 unkeyed checksum
|
||||
pw = b'password'
|
||||
salt = b'salt'
|
||||
key = string_to_key(etype, pw, salt)
|
||||
plain = b'twenty nineteen eighteen seventeen'
|
||||
cksum = h('381c870d8875d1913555de19af5c885fd27b7da9')
|
||||
verify_checksum(Cksumtype.SHA1, key, usage, plain, cksum)
|
||||
|
||||
def test_sha1_unkeyed_checksum_des3_usage_40(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.DES3, 40)
|
||||
|
||||
def test_sha1_unkeyed_checksum_des3_usage_50(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.DES3, 50)
|
||||
|
||||
def test_sha1_unkeyed_checksum_rc4_usage_40(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.RC4, 40)
|
||||
|
||||
def test_sha1_unkeyed_checksum_rc4_usage_50(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.RC4, 50)
|
||||
|
||||
def test_sha1_unkeyed_checksum_aes128_usage_40(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.AES128, 40)
|
||||
|
||||
def test_sha1_unkeyed_checksum_aes128_usage_50(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.AES128, 50)
|
||||
|
||||
def test_sha1_unkeyed_checksum_aes256_usage_40(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.AES256, 40)
|
||||
|
||||
def test_sha1_unkeyed_checksum_aes256_usage_50(self):
|
||||
return self._test_sha1_unkeyed_checksum(Enctype.AES256, 50)
|
||||
|
||||
def _test_crc32_unkeyed_checksum(self, etype, usage):
|
||||
# CRC32 unkeyed checksum
|
||||
pw = b'password'
|
||||
salt = b'salt'
|
||||
key = string_to_key(etype, pw, salt)
|
||||
plain = b'africa america asia australia europe'
|
||||
cksum = h('ce595a53')
|
||||
verify_checksum(Cksumtype.CRC32, key, usage, plain, cksum)
|
||||
|
||||
def test_crc32_unkeyed_checksum_des3_usage_40(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.DES3, 40)
|
||||
|
||||
def test_crc32_unkeyed_checksum_des3_usage_50(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.DES3, 50)
|
||||
|
||||
def test_crc32_unkeyed_checksum_rc4_usage_40(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.RC4, 40)
|
||||
|
||||
def test_crc32_unkeyed_checksum_rc4_usage_50(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.RC4, 50)
|
||||
|
||||
def test_crc32_unkeyed_checksum_aes128_usage_40(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.AES128, 40)
|
||||
|
||||
def test_crc32_unkeyed_checksum_aes128_usage_50(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.AES128, 50)
|
||||
|
||||
def test_crc32_unkeyed_checksum_aes256_usage_40(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.AES256, 40)
|
||||
|
||||
def test_crc32_unkeyed_checksum_aes256_usage_50(self):
|
||||
return self._test_crc32_unkeyed_checksum(Enctype.AES256, 50)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import unittest
|
||||
unittest.main()
|
||||
|
||||
@ -867,3 +867,26 @@ class RawKerberosTest(TestCase):
|
||||
if native_decoded_only:
|
||||
return decoded
|
||||
return decoded, obj
|
||||
|
||||
def PA_S4U2Self_create(self, name, realm, tgt_session_key, ctype=None):
|
||||
# PA-S4U2Self ::= SEQUENCE {
|
||||
# name [0] PrincipalName,
|
||||
# realm [1] Realm,
|
||||
# cksum [2] Checksum,
|
||||
# auth [3] GeneralString
|
||||
# }
|
||||
cksum_data = name['name-type'].to_bytes(4, byteorder='little')
|
||||
for n in name['name-string']:
|
||||
cksum_data += n.encode()
|
||||
cksum_data += realm.encode()
|
||||
cksum_data += "Kerberos".encode()
|
||||
cksum = self.Checksum_create(tgt_session_key, 17, cksum_data, ctype)
|
||||
|
||||
PA_S4U2Self_obj = {
|
||||
'name': name,
|
||||
'realm': realm,
|
||||
'cksum': cksum,
|
||||
'auth': "Kerberos",
|
||||
}
|
||||
pa_s4u2self = self.der_encode(PA_S4U2Self_obj, asn1Spec=krb5_asn1.PA_S4U2Self())
|
||||
return self.PA_DATA_create(129, pa_s4u2self)
|
||||
|
||||
@ -415,6 +415,14 @@ AD-AND-OR ::= SEQUENCE {
|
||||
|
||||
AD-MANDATORY-FOR-KDC ::= AuthorizationData
|
||||
|
||||
-- S4U
|
||||
|
||||
PA-S4U2Self ::= SEQUENCE {
|
||||
name [0] PrincipalName,
|
||||
realm [1] Realm,
|
||||
cksum [2] Checksum,
|
||||
auth [3] KerberosString
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
# Auto-generated by asn1ate v.0.6.1.dev0 from rfc4120.asn1
|
||||
# (last modified on 2020-03-26 10:28:24.346775)
|
||||
# (last modified on 2020-05-06 17:51:00.323318)
|
||||
|
||||
# KerberosV5Spec2
|
||||
from pyasn1.type import univ, char, namedtype, namedval, tag, constraint, useful
|
||||
@ -780,6 +780,18 @@ PA_ENC_TS_ENC.componentType = namedtype.NamedTypes(
|
||||
)
|
||||
|
||||
|
||||
class PA_S4U2Self(univ.Sequence):
|
||||
pass
|
||||
|
||||
|
||||
PA_S4U2Self.componentType = namedtype.NamedTypes(
|
||||
namedtype.NamedType('name', PrincipalName().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0))),
|
||||
namedtype.NamedType('realm', Realm().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1))),
|
||||
namedtype.NamedType('cksum', Checksum().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 2))),
|
||||
namedtype.NamedType('auth', KerberosString().subtype(explicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 3)))
|
||||
)
|
||||
|
||||
|
||||
class PADataTypeValues(univ.Integer):
|
||||
pass
|
||||
|
||||
|
||||
197
python/samba/tests/krb5/s4u_tests.py
Executable file
197
python/samba/tests/krb5/s4u_tests.py
Executable file
@ -0,0 +1,197 @@
|
||||
#!/usr/bin/env python3
|
||||
# Unix SMB/CIFS implementation.
|
||||
# Copyright (C) Stefan Metzmacher 2020
|
||||
#
|
||||
# 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
|
||||
|
||||
sys.path.insert(0, "bin/python")
|
||||
os.environ["PYTHONUNBUFFERED"] = "1"
|
||||
|
||||
from samba.tests import env_get_var_value
|
||||
from samba.tests.krb5.kcrypto import Cksumtype
|
||||
from samba.tests.krb5.raw_testcase import RawKerberosTest
|
||||
import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
|
||||
|
||||
global_asn1_print = False
|
||||
global_hexdump = False
|
||||
|
||||
class S4UKerberosTests(RawKerberosTest):
|
||||
|
||||
def setUp(self):
|
||||
super(S4UKerberosTests, self).setUp()
|
||||
self.do_asn1_print = global_asn1_print
|
||||
self.do_hexdump = global_hexdump
|
||||
|
||||
def _test_s4u2self(self, pa_s4u2self_ctype=None):
|
||||
service_creds = self.get_service_creds()
|
||||
service = service_creds.get_username()
|
||||
realm = service_creds.get_realm()
|
||||
|
||||
cname = self.PrincipalName_create(name_type=1, names=[service])
|
||||
sname = self.PrincipalName_create(name_type=2, names=["krbtgt", realm])
|
||||
|
||||
till = self.get_KerberosTime(offset=36000)
|
||||
|
||||
kdc_options = krb5_asn1.KDCOptions('forwardable')
|
||||
padata = None
|
||||
|
||||
etypes=(18,17,23)
|
||||
|
||||
req = self.AS_REQ_create(padata=padata,
|
||||
kdc_options=str(kdc_options),
|
||||
cname=cname,
|
||||
realm=realm,
|
||||
sname=sname,
|
||||
from_time=None,
|
||||
till_time=till,
|
||||
renew_time=None,
|
||||
nonce=0x7fffffff,
|
||||
etypes=etypes,
|
||||
addresses=None,
|
||||
EncAuthorizationData=None,
|
||||
EncAuthorizationData_key=None,
|
||||
additional_tickets=None)
|
||||
rep = self.send_recv_transaction(req)
|
||||
self.assertIsNotNone(rep)
|
||||
|
||||
self.assertEqual(rep['msg-type'], 30)
|
||||
self.assertEqual(rep['error-code'], 25)
|
||||
rep_padata = self.der_decode(rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA())
|
||||
|
||||
for pa in rep_padata:
|
||||
if pa['padata-type'] == 19:
|
||||
etype_info2 = pa['padata-value']
|
||||
break
|
||||
|
||||
etype_info2 = self.der_decode(etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
|
||||
|
||||
key = self.PasswordKey_from_etype_info2(service_creds, etype_info2[0])
|
||||
|
||||
(patime, pausec) = self.get_KerberosTimeWithUsec()
|
||||
pa_ts = self.PA_ENC_TS_ENC_create(patime, pausec)
|
||||
pa_ts = self.der_encode(pa_ts, asn1Spec=krb5_asn1.PA_ENC_TS_ENC())
|
||||
|
||||
enc_pa_ts_usage = 1
|
||||
pa_ts = self.EncryptedData_create(key, enc_pa_ts_usage, pa_ts)
|
||||
pa_ts = self.der_encode(pa_ts, asn1Spec=krb5_asn1.EncryptedData())
|
||||
|
||||
pa_ts = self.PA_DATA_create(2, pa_ts)
|
||||
|
||||
kdc_options = krb5_asn1.KDCOptions('forwardable')
|
||||
padata = [pa_ts]
|
||||
|
||||
req = self.AS_REQ_create(padata=padata,
|
||||
kdc_options=str(kdc_options),
|
||||
cname=cname,
|
||||
realm=realm,
|
||||
sname=sname,
|
||||
from_time=None,
|
||||
till_time=till,
|
||||
renew_time=None,
|
||||
nonce=0x7fffffff,
|
||||
etypes=etypes,
|
||||
addresses=None,
|
||||
EncAuthorizationData=None,
|
||||
EncAuthorizationData_key=None,
|
||||
additional_tickets=None)
|
||||
rep = self.send_recv_transaction(req)
|
||||
self.assertIsNotNone(rep)
|
||||
|
||||
msg_type = rep['msg-type']
|
||||
self.assertEqual(msg_type, 11)
|
||||
|
||||
usage = 3
|
||||
enc_part2 = key.decrypt(usage, rep['enc-part']['cipher'])
|
||||
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncASRepPart())
|
||||
|
||||
# S4U2Self Request
|
||||
sname = cname
|
||||
|
||||
for_user_name = env_get_var_value('FOR_USER')
|
||||
uname = self.PrincipalName_create(name_type=1, names=[for_user_name])
|
||||
|
||||
kdc_options = krb5_asn1.KDCOptions('forwardable')
|
||||
till = self.get_KerberosTime(offset=36000)
|
||||
ticket = rep['ticket']
|
||||
ticket_session_key = self.EncryptionKey_import(enc_part2['key'])
|
||||
pa_s4u = self.PA_S4U2Self_create(name=uname, realm=realm,
|
||||
tgt_session_key=ticket_session_key,
|
||||
ctype=pa_s4u2self_ctype)
|
||||
padata = [pa_s4u]
|
||||
|
||||
subkey = self.RandomKey(ticket_session_key.etype)
|
||||
subkey_usage = 9
|
||||
|
||||
(ctime, cusec) = self.get_KerberosTimeWithUsec()
|
||||
|
||||
req = self.TGS_REQ_create(padata=padata,
|
||||
cusec=cusec,
|
||||
ctime=ctime,
|
||||
ticket=ticket,
|
||||
kdc_options=str(kdc_options),
|
||||
cname=cname,
|
||||
realm=realm,
|
||||
sname=sname,
|
||||
from_time=None,
|
||||
till_time=till,
|
||||
renew_time=None,
|
||||
nonce=0x7ffffffe,
|
||||
etypes=etypes,
|
||||
addresses=None,
|
||||
EncAuthorizationData=None,
|
||||
EncAuthorizationData_key=None,
|
||||
additional_tickets=None,
|
||||
ticket_session_key=ticket_session_key,
|
||||
authenticator_subkey=subkey)
|
||||
rep = self.send_recv_transaction(req)
|
||||
self.assertIsNotNone(rep)
|
||||
|
||||
msg_type = rep['msg-type']
|
||||
if msg_type == 13:
|
||||
enc_part2 = subkey.decrypt(subkey_usage, rep['enc-part']['cipher'])
|
||||
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
|
||||
|
||||
return msg_type
|
||||
|
||||
# Using the checksum type from the tgt_session_key happens to work everywhere
|
||||
def test_s4u2self(self):
|
||||
msg_type = self._test_s4u2self()
|
||||
self.assertEqual(msg_type, 13)
|
||||
|
||||
# Per spec, the checksum of PA-FOR-USER is HMAC_MD5, see [MS-SFU] 2.2.1
|
||||
def test_s4u2self_hmac_md5_checksum(self):
|
||||
msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.HMAC_MD5)
|
||||
self.assertEqual(msg_type, 13)
|
||||
|
||||
def test_s4u2self_md5_unkeyed_checksum(self):
|
||||
msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.MD5)
|
||||
self.assertEqual(msg_type, 30)
|
||||
|
||||
def test_s4u2self_sha1_unkeyed_checksum(self):
|
||||
msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.SHA1)
|
||||
self.assertEqual(msg_type, 30)
|
||||
|
||||
def test_s4u2self_crc32_unkeyed_checksum(self):
|
||||
msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.CRC32)
|
||||
self.assertEqual(msg_type, 30)
|
||||
|
||||
if __name__ == "__main__":
|
||||
global_asn1_print = True
|
||||
global_hexdump = True
|
||||
import unittest
|
||||
unittest.main()
|
||||
@ -87,6 +87,7 @@ EXCLUDE_USAGE = {
|
||||
'python/samba/tests/dcerpc/raw_protocol.py',
|
||||
'python/samba/tests/krb5/kcrypto.py',
|
||||
'python/samba/tests/krb5/simple_tests.py',
|
||||
'python/samba/tests/krb5/s4u_tests.py',
|
||||
}
|
||||
|
||||
EXCLUDE_HELP = {
|
||||
|
||||
@ -393,3 +393,5 @@
|
||||
^samba.tests.ntlmdisabled.python\(ktest\).python2.ntlmdisabled.NtlmDisabledTests.test_samr_change_password\(ktest\)
|
||||
^samba.tests.ntlmdisabled.python\(ad_dc_no_ntlm\).python3.ntlmdisabled.NtlmDisabledTests.test_ntlm_connection\(ad_dc_no_ntlm\)
|
||||
^samba.tests.ntlmdisabled.python\(ad_dc_no_ntlm\).python2.ntlmdisabled.NtlmDisabledTests.test_ntlm_connection\(ad_dc_no_ntlm\)
|
||||
# Fixed upstream heimdal in PR #439
|
||||
^samba.tests.krb5.s4u_tests.samba.tests.krb5.s4u_tests.S4UKerberosTests.test_s4u2self_hmac_md5_checksum
|
||||
|
||||
@ -3,3 +3,4 @@
|
||||
.*RODC
|
||||
^samba4.ntvfs.cifs.ntlm.base.unlink
|
||||
^samba4.ntvfs.cifs.krb5.base.unlink
|
||||
^samba.tests.krb5.s4u_tests
|
||||
|
||||
@ -934,6 +934,29 @@ sub provision_raw_step2($$$)
|
||||
return undef;
|
||||
}
|
||||
|
||||
my $srv_account = "srv_account";
|
||||
$samba_tool_cmd = "";
|
||||
$samba_tool_cmd .= "RESOLV_CONF=\"$ret->{RESOLV_CONF}\" ";
|
||||
$samba_tool_cmd .= "KRB5_CONFIG=\"$ret->{KRB5_CONFIG}\" ";
|
||||
$samba_tool_cmd .= "KRB5CCNAME=\"$ret->{KRB5_CCACHE}\" ";
|
||||
$samba_tool_cmd .= Samba::bindir_path($self, "samba-tool")
|
||||
. " user create --configfile=$ctx->{smb_conf} $srv_account $ctx->{password}";
|
||||
unless (system($samba_tool_cmd) == 0) {
|
||||
warn("Unable to add $srv_account user: \n$samba_tool_cmd\n");
|
||||
return undef;
|
||||
}
|
||||
|
||||
$samba_tool_cmd = "";
|
||||
$samba_tool_cmd .= "RESOLV_CONF=\"$ret->{RESOLV_CONF}\" ";
|
||||
$samba_tool_cmd .= "KRB5_CONFIG=\"$ret->{KRB5_CONFIG}\" ";
|
||||
$samba_tool_cmd .= "KRB5CCNAME=\"$ret->{KRB5_CCACHE}\" ";
|
||||
$samba_tool_cmd .= Samba::bindir_path($self, "samba-tool")
|
||||
. " spn add HOST/$srv_account --configfile=$ctx->{smb_conf} $srv_account";
|
||||
unless (system($samba_tool_cmd) == 0) {
|
||||
warn("Unable to add spn for $srv_account: \n$samba_tool_cmd\n");
|
||||
return undef;
|
||||
}
|
||||
|
||||
my $ldbmodify = "";
|
||||
$ldbmodify .= "RESOLV_CONF=\"$ret->{RESOLV_CONF}\" ";
|
||||
$ldbmodify .= "KRB5_CONFIG=\"$ret->{KRB5_CONFIG}\" ";
|
||||
|
||||
@ -758,6 +758,10 @@ planoldpythontestsuite("ad_dc:local", "samba.tests.dckeytab", extra_args=['-U"$U
|
||||
planoldpythontestsuite("none", "samba.tests.krb5.kcrypto")
|
||||
planoldpythontestsuite("ad_dc_default", "samba.tests.krb5.simple_tests",
|
||||
environ={'SERVICE_USERNAME':'$SERVER'})
|
||||
planoldpythontestsuite("ad_dc_default:local", "samba.tests.krb5.s4u_tests",
|
||||
environ={'SERVICE_USERNAME':'srv_account',
|
||||
'SERVICE_PASSWORD':'$PASSWORD',
|
||||
'FOR_USER':'$USERNAME'})
|
||||
|
||||
for env in ["ad_dc", smbv1_disabled_testenv]:
|
||||
planoldpythontestsuite(env, "samba.tests.smb", extra_args=['-U"$USERNAME%$PASSWORD"'])
|
||||
|
||||
Loading…
Reference in New Issue
Block a user