1
0
mirror of https://github.com/samba-team/samba.git synced 2025-12-23 00:23:53 +03:00

tests python krb5: PEP8 cleanups

Fix all the PEP8 warnings in samba/tests/krb5. With the exception of
rfc4120_pyasn1.py, which is generated from rfc4120.asn1.

As these tests are new, it makes sense to ensure that they conform to
PEP8. And set an aspirational goal for the rest of our python code.

Signed-off-by: Gary Lockyer <gary@catalyst.net.nz>
Reviewed-by: Andreas Schneider <asn@samba.org>

Autobuild-User(master): Gary Lockyer <gary@samba.org>
Autobuild-Date(master): Mon Dec 21 21:29:28 UTC 2020 on sn-devel-184
This commit is contained in:
Gary Lockyer
2020-12-11 11:55:01 +13:00
committed by Gary Lockyer
parent 03676a4a5c
commit c00d537526
10 changed files with 410 additions and 260 deletions

View File

@@ -31,8 +31,6 @@ import samba
from samba.auth import system_session from samba.auth import system_session
from samba.credentials import ( from samba.credentials import (
Credentials, Credentials,
CLI_CRED_NTLMv2_AUTH,
CLI_CRED_NTLM_AUTH,
DONT_USE_KERBEROS) DONT_USE_KERBEROS)
from samba.dcerpc.misc import SEC_CHAN_WKSTA from samba.dcerpc.misc import SEC_CHAN_WKSTA
from samba.dsdb import ( from samba.dsdb import (
@@ -41,7 +39,20 @@ from samba.dsdb import (
UF_NORMAL_ACCOUNT) UF_NORMAL_ACCOUNT)
from samba.samdb import SamDB from samba.samdb import SamDB
from samba.tests import delete_force, DynamicTestCase from samba.tests import delete_force, DynamicTestCase
from samba.tests.krb5.rfc4120_constants import * from samba.tests.krb5.rfc4120_constants import (
AES256_CTS_HMAC_SHA1_96,
AES128_CTS_HMAC_SHA1_96,
ARCFOUR_HMAC_MD5,
KDC_ERR_PREAUTH_REQUIRED,
KRB_AS_REP,
KU_AS_REP_ENC_PART,
KRB_ERROR,
KU_PA_ENC_TIMESTAMP,
PADATA_ENC_TIMESTAMP,
NT_ENTERPRISE_PRINCIPAL,
NT_PRINCIPAL,
NT_SRV_INST,
)
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
@@ -49,15 +60,15 @@ global_hexdump = False
@unique @unique
class TestOptions(Enum): class TestOptions(Enum):
Canonicalize = 1 Canonicalize = 1
Enterprise = 2 Enterprise = 2
UpperRealm = 4 UpperRealm = 4
UpperUserName = 8 UpperUserName = 8
NetbiosRealm = 16 NetbiosRealm = 16
UPN = 32 UPN = 32
RemoveDollar = 64 RemoveDollar = 64
AsReqSelf = 128 AsReqSelf = 128
Last = 256 Last = 256
def is_set(self, x): def is_set(self, x):
return self.value & x return self.value & x
@@ -65,7 +76,7 @@ class TestOptions(Enum):
@unique @unique
class CredentialsType(Enum): class CredentialsType(Enum):
User = 1 User = 1
Machine = 2 Machine = 2
def is_set(self, x): def is_set(self, x):
@@ -126,7 +137,8 @@ class TestData:
MACHINE_NAME = "tstkrb5cnnmch" MACHINE_NAME = "tstkrb5cnnmch"
USER_NAME = "tstkrb5cnnusr" USER_NAME = "tstkrb5cnnusr"
@DynamicTestCase @DynamicTestCase
class KerberosASCanonicalizationTests(RawKerberosTest): class KerberosASCanonicalizationTests(RawKerberosTest):
@@ -160,21 +172,21 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
cls.lp = cls.get_loadparm(cls) cls.lp = cls.get_loadparm(cls)
cls.username = os.environ["USERNAME"] cls.username = os.environ["USERNAME"]
cls.password = os.environ["PASSWORD"] cls.password = os.environ["PASSWORD"]
cls.host = os.environ["SERVER"] cls.host = os.environ["SERVER"]
c = Credentials() c = Credentials()
c.set_username(cls.username) c.set_username(cls.username)
c.set_password(cls.password) c.set_password(cls.password)
try: try:
realm = os.environ["REALM"] realm = os.environ["REALM"]
c.set_realm(realm) c.set_realm(realm)
except KeyError: except KeyError:
pass pass
try: try:
domain = os.environ["DOMAIN"] domain = os.environ["DOMAIN"]
c.set_domain(domain) c.set_domain(domain)
except KeyError: except KeyError:
pass pass
@@ -200,7 +212,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
def setUp(self): def setUp(self):
super(KerberosASCanonicalizationTests, self).setUp() super(KerberosASCanonicalizationTests, self).setUp()
self.do_asn1_print = global_asn1_print self.do_asn1_print = global_asn1_print
self.do_hexdump = global_hexdump self.do_hexdump = global_hexdump
# #
# Create a test user account # Create a test user account
@@ -340,7 +352,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
# #
# Check the protocol version, should be 5 # Check the protocol version, should be 5
self.assertEqual( self.assertEqual(
rep['pvno'], 5, "Data {0}".format(str(data))) rep['pvno'], 5, "Data {0}".format(str(data)))
self.assertEqual( self.assertEqual(
rep['msg-type'], KRB_ERROR, "Data {0}".format(str(data))) rep['msg-type'], KRB_ERROR, "Data {0}".format(str(data)))
@@ -397,7 +409,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
# #
# Check the protocol version, should be 5 # Check the protocol version, should be 5
self.assertEqual( self.assertEqual(
rep['pvno'], 5, "Data {0}".format(str(data))) rep['pvno'], 5, "Data {0}".format(str(data)))
msg_type = rep['msg-type'] msg_type = rep['msg-type']
# Should not have got an error. # Should not have got an error.

View File

@@ -25,7 +25,20 @@ os.environ["PYTHONUNBUFFERED"] = "1"
from samba.tests.krb5.raw_testcase import RawKerberosTest from samba.tests.krb5.raw_testcase import RawKerberosTest
import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1 import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
from samba.tests.krb5.rfc4120_constants import * from samba.tests.krb5.rfc4120_constants import (
AES128_CTS_HMAC_SHA1_96,
AES256_CTS_HMAC_SHA1_96,
ARCFOUR_HMAC_MD5,
KDC_ERR_PREAUTH_REQUIRED,
KRB_AS_REP,
KRB_ERROR,
KU_AS_REP_ENC_PART,
KU_PA_ENC_TIMESTAMP,
PADATA_ENC_TIMESTAMP,
PADATA_ETYPE_INFO2,
NT_PRINCIPAL,
NT_SRV_INST,
)
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
@@ -112,18 +125,17 @@ class SimpleKerberosTests(RawKerberosTest):
realm = creds.get_realm() realm = creds.get_realm()
cname = self.PrincipalName_create( cname = self.PrincipalName_create(
name_type=NT_PRINCIPAL, name_type=NT_PRINCIPAL,
names=[user]) names=[user])
sname = self.PrincipalName_create( sname = self.PrincipalName_create(
name_type=NT_SRV_INST, name_type=NT_SRV_INST,
names=["krbtgt", realm]) names=["krbtgt", realm])
till = self.get_KerberosTime(offset=36000) till = self.get_KerberosTime(offset=36000)
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None padata = None
req = self.AS_REQ_create(padata=padata, req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options), kdc_options=str(kdc_options),
cname=cname, cname=cname,

View File

@@ -64,6 +64,7 @@ from samba.credentials import Credentials
from samba import generate_random_bytes as get_random_bytes from samba import generate_random_bytes as get_random_bytes
from samba.common import get_string, get_bytes from samba.common import get_string, get_bytes
class Enctype(object): class Enctype(object):
DES_CRC = 1 DES_CRC = 1
DES_MD4 = 2 DES_MD4 = 2
@@ -112,26 +113,30 @@ def _mac_equal(mac1, mac2):
res |= x ^ y res |= x ^ y
return res == 0 return res == 0
def SIMPLE_HASH(string, algo_cls): def SIMPLE_HASH(string, algo_cls):
hash_ctx = hashes.Hash(algo_cls(), default_backend()) hash_ctx = hashes.Hash(algo_cls(), default_backend())
hash_ctx.update(string) hash_ctx.update(string)
return hash_ctx.finalize() return hash_ctx.finalize()
def HMAC_HASH(key, string, algo_cls): def HMAC_HASH(key, string, algo_cls):
hmac_ctx = hmac.HMAC(key, algo_cls(), default_backend()) hmac_ctx = hmac.HMAC(key, algo_cls(), default_backend())
hmac_ctx.update(string) hmac_ctx.update(string)
return hmac_ctx.finalize() return hmac_ctx.finalize()
def _nfold(str, nbytes): def _nfold(str, nbytes):
# Convert str to a string of length nbytes using the RFC 3961 nfold # Convert str to a string of length nbytes using the RFC 3961 nfold
# operation. # operation.
# Rotate the bytes in str to the right by nbits bits. # Rotate the bytes in str to the right by nbits bits.
def rotate_right(str, nbits): def rotate_right(str, nbits):
nbytes, remain = (nbits//8) % len(str), nbits % 8 nbytes, remain = (nbits // 8) % len(str), nbits % 8
return bytes([(str[i-nbytes] >> remain) | return bytes([
(str[i-nbytes-1] << (8-remain) & 0xff) (str[i - nbytes] >> remain)
for i in range(len(str))]) | (str[i - nbytes - 1] << (8 - remain) & 0xff)
for i in range(len(str))])
# Add equal-length strings together with end-around carry. # Add equal-length strings together with end-around carry.
def add_ones_complement(str1, str2): def add_ones_complement(str1, str2):
@@ -139,7 +144,7 @@ def _nfold(str, nbytes):
v = [a + b for a, b in zip(str1, str2)] v = [a + b for a, b in zip(str1, str2)]
# Propagate carry bits to the left until there aren't any left. # Propagate carry bits to the left until there aren't any left.
while any(x & ~0xff for x in v): while any(x & ~0xff for x in v):
v = [(v[i-n+1]>>8) + (v[i]&0xff) for i in range(n)] v = [(v[i - n + 1] >> 8) + (v[i] & 0xff) for i in range(n)]
return bytes([x for x in v]) return bytes([x for x in v])
# Concatenate copies of str to produce the least common multiple # Concatenate copies of str to produce the least common multiple
@@ -150,7 +155,7 @@ def _nfold(str, nbytes):
slen = len(str) slen = len(str)
lcm = nbytes * slen // gcd(nbytes, slen) lcm = nbytes * slen // gcd(nbytes, slen)
bigstr = b''.join((rotate_right(str, 13 * i) for i in range(lcm // slen))) bigstr = b''.join((rotate_right(str, 13 * i) for i in range(lcm // slen)))
slices = (bigstr[p:p+nbytes] for p in range(0, lcm, nbytes)) slices = (bigstr[p:p + nbytes] for p in range(0, lcm, nbytes))
return reduce(add_ones_complement, slices) return reduce(add_ones_complement, slices)
@@ -275,7 +280,7 @@ class _DES3CBC(_SimplifiedEnctype):
return b if bin(b & ~1).count('1') % 2 else b | 1 return b if bin(b & ~1).count('1') % 2 else b | 1
assert len(seed) == 7 assert len(seed) == 7
firstbytes = [parity(b & ~1) for b in seed] firstbytes = [parity(b & ~1) for b in seed]
lastbyte = parity(sum((seed[i]&1) << i+1 for i in range(7))) lastbyte = parity(sum((seed[i] & 1) << i + 1 for i in range(7)))
keybytes = bytes([b for b in firstbytes + [lastbyte]]) keybytes = bytes([b for b in firstbytes + [lastbyte]])
if _is_weak_des_key(keybytes): if _is_weak_des_key(keybytes):
keybytes[7] = bytes([keybytes[7] ^ 0xF0]) keybytes[7] = bytes([keybytes[7] ^ 0xF0])
@@ -369,7 +374,7 @@ class _AESEnctype(_SimplifiedEnctype):
if len(ciphertext) == 16: if len(ciphertext) == 16:
return aes_decrypt(ciphertext) return aes_decrypt(ciphertext)
# Split the ciphertext into blocks. The last block may be partial. # Split the ciphertext into blocks. The last block may be partial.
cblocks = [ciphertext[p:p+16] for p in range(0, len(ciphertext), 16)] cblocks = [ciphertext[p:p + 16] for p in range(0, len(ciphertext), 16)]
lastlen = len(cblocks[-1]) lastlen = len(cblocks[-1])
# CBC-decrypt all but the last two blocks. # CBC-decrypt all but the last two blocks.
prev_cblock = bytes(16) prev_cblock = bytes(16)
@@ -383,7 +388,7 @@ class _AESEnctype(_SimplifiedEnctype):
# will be the omitted bytes of ciphertext from the final # will be the omitted bytes of ciphertext from the final
# block. # block.
b = aes_decrypt(cblocks[-2]) b = aes_decrypt(cblocks[-2])
lastplaintext =_xorbytes(b[:lastlen], cblocks[-1]) lastplaintext = _xorbytes(b[:lastlen], cblocks[-1])
omitted = b[lastlen:] omitted = b[lastlen:]
# Decrypt the final cipher block plus the omitted bytes to get # Decrypt the final cipher block plus the omitted bytes to get
# the second-to-last plaintext block. # the second-to-last plaintext block.
@@ -433,7 +438,8 @@ class _RC4(_EnctypeProfile):
cksum = HMAC_HASH(ki, confounder + plaintext, hashes.MD5) cksum = HMAC_HASH(ki, confounder + plaintext, hashes.MD5)
ke = HMAC_HASH(ki, cksum, hashes.MD5) ke = HMAC_HASH(ki, cksum, hashes.MD5)
encryptor = Cipher(ciphers.ARC4(ke), None, default_backend()).encryptor() encryptor = Cipher(
ciphers.ARC4(ke), None, default_backend()).encryptor()
ctext = encryptor.update(confounder + plaintext) ctext = encryptor.update(confounder + plaintext)
return cksum + ctext return cksum + ctext
@@ -446,7 +452,8 @@ class _RC4(_EnctypeProfile):
ki = HMAC_HASH(key.contents, cls.usage_str(keyusage), hashes.MD5) ki = HMAC_HASH(key.contents, cls.usage_str(keyusage), hashes.MD5)
ke = HMAC_HASH(ki, cksum, hashes.MD5) ke = HMAC_HASH(ki, cksum, hashes.MD5)
decryptor = Cipher(ciphers.ARC4(ke), None, default_backend()).decryptor() decryptor = Cipher(
ciphers.ARC4(ke), None, default_backend()).decryptor()
basic_plaintext = decryptor.update(basic_ctext) basic_plaintext = decryptor.update(basic_ctext)
exp_cksum = HMAC_HASH(ki, basic_plaintext, hashes.MD5) exp_cksum = HMAC_HASH(ki, basic_plaintext, hashes.MD5)
@@ -636,14 +643,14 @@ def verify_checksum(cksumtype, key, keyusage, text, cksum):
c.verify(key, keyusage, text, cksum) c.verify(key, keyusage, text, cksum)
def prfplus(key, pepper, l): def prfplus(key, pepper, ln):
# Produce l bytes of output using the RFC 6113 PRF+ function. # Produce ln bytes of output using the RFC 6113 PRF+ function.
out = b'' out = b''
count = 1 count = 1
while len(out) < l: while len(out) < ln:
out += prf(key, bytes([count]) + pepper) out += prf(key, bytes([count]) + pepper)
count += 1 count += 1
return out[:l] return out[:ln]
def cf2(enctype, key1, key2, pepper1, pepper2): def cf2(enctype, key1, key2, pepper1, pepper2):
@@ -653,9 +660,11 @@ def cf2(enctype, key1, key2, pepper1, pepper2):
return e.random_to_key(_xorbytes(prfplus(key1, pepper1, e.seedsize), return e.random_to_key(_xorbytes(prfplus(key1, pepper1, e.seedsize),
prfplus(key2, pepper2, e.seedsize))) prfplus(key2, pepper2, e.seedsize)))
def h(hexstr): def h(hexstr):
return bytes.fromhex(hexstr) return bytes.fromhex(hexstr)
class KcrytoTest(TestCase): class KcrytoTest(TestCase):
"""kcrypto Test case.""" """kcrypto Test case."""
@@ -665,20 +674,21 @@ class KcrytoTest(TestCase):
conf = h('94B491F481485B9A0678CD3C4EA386AD') conf = h('94B491F481485B9A0678CD3C4EA386AD')
keyusage = 2 keyusage = 2
plain = b'9 bytesss' plain = b'9 bytesss'
ctxt = h('68FB9679601F45C78857B2BF820FD6E53ECA8D42FD4B1D7024A09205ABB7CD2E' ctxt = h('68FB9679601F45C78857B2BF820FD6E53ECA8D42FD4B1D7024A09205ABB7'
'C26C355D2F') 'CD2EC26C355D2F')
k = Key(Enctype.AES128, kb) k = Key(Enctype.AES128, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt) self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain) self.assertEqual(decrypt(k, keyusage, ctxt), plain)
def test_aes256_crypt(self): def test_aes256_crypt(self):
# AES256 encrypt and decrypt # AES256 encrypt and decrypt
kb = h('F1C795E9248A09338D82C3F8D5B567040B0110736845041347235B1404231398') kb = h('F1C795E9248A09338D82C3F8D5B567040B0110736845041347235B14042313'
'98')
conf = h('E45CA518B42E266AD98E165E706FFB60') conf = h('E45CA518B42E266AD98E165E706FFB60')
keyusage = 4 keyusage = 4
plain = b'30 bytes bytes bytes bytes byt' plain = b'30 bytes bytes bytes bytes byt'
ctxt = h('D1137A4D634CFECE924DBC3BF6790648BD5CFF7DE0E7B99460211D0DAEF3D79A' ctxt = h('D1137A4D634CFECE924DBC3BF6790648BD5CFF7DE0E7B99460211D0DAEF3'
'295C688858F3B34B9CBD6EEBAE81DAF6B734D4D498B6714F1C1D') 'D79A295C688858F3B34B9CBD6EEBAE81DAF6B734D4D498B6714F1C1D')
k = Key(Enctype.AES256, kb) k = Key(Enctype.AES256, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt) self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain) self.assertEqual(decrypt(k, keyusage, ctxt), plain)
@@ -694,7 +704,8 @@ class KcrytoTest(TestCase):
def test_aes256_checksum(self): def test_aes256_checksum(self):
# AES256 checksum # AES256 checksum
kb = h('B1AE4CD8462AFF1677053CC9279AAC30B796FB81CE21474DD3DDBCFEA4EC76D7') kb = h('B1AE4CD8462AFF1677053CC9279AAC30B796FB81CE21474DD3DDBC'
'FEA4EC76D7')
keyusage = 4 keyusage = 4
plain = b'fourteen' plain = b'fourteen'
cksum = h('E08739E3279E2903EC8E3836') cksum = h('E08739E3279E2903EC8E3836')
@@ -715,7 +726,8 @@ class KcrytoTest(TestCase):
string = b'X' * 64 string = b'X' * 64
salt = b'pass phrase equals block size' salt = b'pass phrase equals block size'
params = h('000004B0') params = h('000004B0')
kb = h('89ADEE3608DB8BC71F1BFBFE459486B05618B70CBAE22092534E56C553BA4B34') kb = h('89ADEE3608DB8BC71F1BFBFE459486B05618B70CBAE22092534E56'
'C553BA4B34')
k = string_to_key(Enctype.AES256, string, salt, params) k = string_to_key(Enctype.AES256, string, salt, params)
self.assertEqual(k.contents, kb) self.assertEqual(k.contents, kb)
@@ -741,7 +753,8 @@ class KcrytoTest(TestCase):
def test_aes256_cf2(self): def test_aes256_cf2(self):
# AES256 cf2 # AES256 cf2
kb = h('4D6CA4E629785C1F01BAF55E2E548566B9617AE3A96868C337CB93B5E72B1C7B') kb = h('4D6CA4E629785C1F01BAF55E2E548566B9617AE3A96868C337CB93B5'
'E72B1C7B')
k1 = string_to_key(Enctype.AES256, b'key1', b'key1') k1 = string_to_key(Enctype.AES256, b'key1', b'key1')
k2 = string_to_key(Enctype.AES256, b'key2', b'key2') k2 = string_to_key(Enctype.AES256, b'key2', b'key2')
k = cf2(Enctype.AES256, k1, k2, b'a', b'b') k = cf2(Enctype.AES256, k1, k2, b'a', b'b')
@@ -753,8 +766,8 @@ class KcrytoTest(TestCase):
conf = h('94690A17B2DA3C9B') conf = h('94690A17B2DA3C9B')
keyusage = 3 keyusage = 3
plain = b'13 bytes byte' plain = b'13 bytes byte'
ctxt = h('839A17081ECBAFBCDC91B88C6955DD3C4514023CF177B77BF0D0177A16F705E8' ctxt = h('839A17081ECBAFBCDC91B88C6955DD3C4514023CF177B77BF0D0177A16F7'
'49CB7781D76A316B193F8D30') '05E849CB7781D76A316B193F8D30')
k = Key(Enctype.DES3, kb) k = Key(Enctype.DES3, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt) self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), _zeropad(plain, 8)) self.assertEqual(decrypt(k, keyusage, ctxt), _zeropad(plain, 8))
@@ -790,8 +803,8 @@ class KcrytoTest(TestCase):
conf = h('37245E73A45FBF72') conf = h('37245E73A45FBF72')
keyusage = 4 keyusage = 4
plain = b'30 bytes bytes bytes bytes byt' plain = b'30 bytes bytes bytes bytes byt'
ctxt = h('95F9047C3AD75891C2E9B04B16566DC8B6EB9CE4231AFB2542EF87A7B5A0F260' ctxt = h('95F9047C3AD75891C2E9B04B16566DC8B6EB9CE4231AFB2542EF87A7B5A0'
'A99F0460508DE0CECC632D07C354124E46C5D2234EB8') 'F260A99F0460508DE0CECC632D07C354124E46C5D2234EB8')
k = Key(Enctype.RC4, kb) k = Key(Enctype.RC4, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt) self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain) self.assertEqual(decrypt(k, keyusage, ctxt), plain)

View File

@@ -374,8 +374,8 @@ class KDCBaseTest(RawKerberosTest):
account_name = ( account_name = (
pac.info.info.info3.base.account_name) pac.info.info.info3.base.account_name)
user_sid = ( user_sid = (
str(pac.info.info.info3.base.domain_sid) + str(pac.info.info.info3.base.domain_sid)
"-" + str(pac.info.info.info3.base.rid)) + "-" + str(pac.info.info.info3.base.rid))
elif pac.type == self.PAC_LOGON_NAME: elif pac.type == self.PAC_LOGON_NAME:
logon_name = pac.info.account_name logon_name = pac.info.account_name
elif pac.type == self.PAC_UPN_DNS_INFO: elif pac.type == self.PAC_UPN_DNS_INFO:

View File

@@ -25,7 +25,20 @@ os.environ["PYTHONUNBUFFERED"] = "1"
from samba.tests.krb5.raw_testcase import RawKerberosTest from samba.tests.krb5.raw_testcase import RawKerberosTest
import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1 import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
from samba.tests.krb5.rfc4120_constants import * from samba.tests.krb5.rfc4120_constants import (
AES256_CTS_HMAC_SHA1_96,
ARCFOUR_HMAC_MD5,
KDC_ERR_PREAUTH_FAILED,
KDC_ERR_PREAUTH_REQUIRED,
KDC_ERR_SKEW,
KRB_AS_REP,
KRB_ERROR,
KU_PA_ENC_TIMESTAMP,
PADATA_ENC_TIMESTAMP,
PADATA_ETYPE_INFO2,
NT_PRINCIPAL,
NT_SRV_INST,
)
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
@@ -83,7 +96,7 @@ class KdcTests(RawKerberosTest):
break break
etype_info2 = self.der_decode( etype_info2 = self.der_decode(
etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2()) etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
key = self.PasswordKey_from_etype_info2(creds, etype_info2[0]) key = self.PasswordKey_from_etype_info2(creds, etype_info2[0])

View File

@@ -35,7 +35,10 @@ from pyasn1.codec.native.decoder import decode as pyasn1_native_decode
from pyasn1.codec.native.encoder import encode as pyasn1_native_encode from pyasn1.codec.native.encoder import encode as pyasn1_native_encode
from pyasn1.codec.ber.encoder import BitStringEncoder as BitStringEncoder from pyasn1.codec.ber.encoder import BitStringEncoder as BitStringEncoder
def BitStringEncoder_encodeValue32(self, value, asn1Spec, encodeFun, **options):
def BitStringEncoder_encodeValue32(
self, value, asn1Spec, encodeFun, **options):
# #
# BitStrings like KDCOptions or TicketFlags should at least # BitStrings like KDCOptions or TicketFlags should at least
# be 32-Bit on the wire # be 32-Bit on the wire
@@ -59,14 +62,17 @@ def BitStringEncoder_encodeValue32(self, value, asn1Spec, encodeFun, **options):
padding = 0 padding = 0
ret = b'\x00' + substrate + (b'\x00' * padding) ret = b'\x00' + substrate + (b'\x00' * padding)
return ret, False, True return ret, False, True
BitStringEncoder.encodeValue = BitStringEncoder_encodeValue32 BitStringEncoder.encodeValue = BitStringEncoder_encodeValue32
def BitString_NamedValues_prettyPrint(self, scope=0): def BitString_NamedValues_prettyPrint(self, scope=0):
ret = "%s" % self.asBinary() ret = "%s" % self.asBinary()
bits = [] bits = []
highest_bit = 32 highest_bit = 32
for byte in self.asNumbers(): for byte in self.asNumbers():
for bit in [7,6,5,4,3,2,1,0]: for bit in [7, 6, 5, 4, 3, 2, 1, 0]:
mask = 1 << bit mask = 1 << bit
if byte & mask: if byte & mask:
val = 1 val = 1
@@ -89,12 +95,21 @@ def BitString_NamedValues_prettyPrint(self, scope=0):
delim = ",\n%s " % indent delim = ",\n%s " % indent
ret += "\n%s)" % indent ret += "\n%s)" % indent
return ret return ret
krb5_asn1.TicketFlags.prettyPrintNamedValues = krb5_asn1.TicketFlagsValues.namedValues
krb5_asn1.TicketFlags.namedValues = krb5_asn1.TicketFlagsValues.namedValues
krb5_asn1.TicketFlags.prettyPrint = BitString_NamedValues_prettyPrint krb5_asn1.TicketFlags.prettyPrintNamedValues =\
krb5_asn1.KDCOptions.prettyPrintNamedValues = krb5_asn1.KDCOptionsValues.namedValues krb5_asn1.TicketFlagsValues.namedValues
krb5_asn1.KDCOptions.namedValues = krb5_asn1.KDCOptionsValues.namedValues krb5_asn1.TicketFlags.namedValues =\
krb5_asn1.KDCOptions.prettyPrint = BitString_NamedValues_prettyPrint krb5_asn1.TicketFlagsValues.namedValues
krb5_asn1.TicketFlags.prettyPrint =\
BitString_NamedValues_prettyPrint
krb5_asn1.KDCOptions.prettyPrintNamedValues =\
krb5_asn1.KDCOptionsValues.namedValues
krb5_asn1.KDCOptions.namedValues =\
krb5_asn1.KDCOptionsValues.namedValues
krb5_asn1.KDCOptions.prettyPrint =\
BitString_NamedValues_prettyPrint
def Integer_NamedValues_prettyPrint(self, scope=0): def Integer_NamedValues_prettyPrint(self, scope=0):
intval = int(self) intval = int(self)
@@ -104,16 +119,29 @@ def Integer_NamedValues_prettyPrint(self, scope=0):
name = "<__unknown__>" name = "<__unknown__>"
ret = "%d (0x%x) %s" % (intval, intval, name) ret = "%d (0x%x) %s" % (intval, intval, name)
return ret return ret
krb5_asn1.NameType.prettyPrintNamedValues = krb5_asn1.NameTypeValues.namedValues
krb5_asn1.NameType.prettyPrint = Integer_NamedValues_prettyPrint
krb5_asn1.AuthDataType.prettyPrintNamedValues = krb5_asn1.AuthDataTypeValues.namedValues krb5_asn1.NameType.prettyPrintNamedValues =\
krb5_asn1.AuthDataType.prettyPrint = Integer_NamedValues_prettyPrint krb5_asn1.NameTypeValues.namedValues
krb5_asn1.PADataType.prettyPrintNamedValues = krb5_asn1.PADataTypeValues.namedValues krb5_asn1.NameType.prettyPrint =\
krb5_asn1.PADataType.prettyPrint = Integer_NamedValues_prettyPrint Integer_NamedValues_prettyPrint
krb5_asn1.EncryptionType.prettyPrintNamedValues = krb5_asn1.EncryptionTypeValues.namedValues krb5_asn1.AuthDataType.prettyPrintNamedValues =\
krb5_asn1.EncryptionType.prettyPrint = Integer_NamedValues_prettyPrint krb5_asn1.AuthDataTypeValues.namedValues
krb5_asn1.ChecksumType.prettyPrintNamedValues = krb5_asn1.ChecksumTypeValues.namedValues krb5_asn1.AuthDataType.prettyPrint =\
krb5_asn1.ChecksumType.prettyPrint = Integer_NamedValues_prettyPrint Integer_NamedValues_prettyPrint
krb5_asn1.PADataType.prettyPrintNamedValues =\
krb5_asn1.PADataTypeValues.namedValues
krb5_asn1.PADataType.prettyPrint =\
Integer_NamedValues_prettyPrint
krb5_asn1.EncryptionType.prettyPrintNamedValues =\
krb5_asn1.EncryptionTypeValues.namedValues
krb5_asn1.EncryptionType.prettyPrint =\
Integer_NamedValues_prettyPrint
krb5_asn1.ChecksumType.prettyPrintNamedValues =\
krb5_asn1.ChecksumTypeValues.namedValues
krb5_asn1.ChecksumType.prettyPrint =\
Integer_NamedValues_prettyPrint
class Krb5EncryptionKey(object): class Krb5EncryptionKey(object):
def __init__(self, key, kvno): def __init__(self, key, kvno):
@@ -146,9 +174,10 @@ class Krb5EncryptionKey(object):
EncryptionKey_obj = { EncryptionKey_obj = {
'keytype': self.etype, 'keytype': self.etype,
'keyvalue': self.key.contents, 'keyvalue': self.key.contents,
}; }
return EncryptionKey_obj return EncryptionKey_obj
class RawKerberosTest(TestCase): class RawKerberosTest(TestCase):
"""A raw Kerberos Test case.""" """A raw Kerberos Test case."""
@@ -182,13 +211,13 @@ class RawKerberosTest(TestCase):
self.s = socket.socket(self.a[0][0], self.a[0][1], self.a[0][2]) self.s = socket.socket(self.a[0][0], self.a[0][1], self.a[0][2])
self.s.settimeout(10) self.s.settimeout(10)
self.s.connect(self.a[0][4]) self.s.connect(self.a[0][4])
except socket.error as e: except socket.error:
self.s.close() self.s.close()
raise raise
except IOError as e: except IOError:
self.s.close() self.s.close()
raise raise
except Exception as e: except Exception:
raise raise
finally: finally:
pass pass
@@ -219,8 +248,9 @@ class RawKerberosTest(TestCase):
domain = samba.tests.env_get_var_value('DOMAIN') domain = samba.tests.env_get_var_value('DOMAIN')
realm = samba.tests.env_get_var_value('REALM') realm = samba.tests.env_get_var_value('REALM')
username = samba.tests.env_get_var_value('SERVICE_USERNAME') username = samba.tests.env_get_var_value('SERVICE_USERNAME')
password = samba.tests.env_get_var_value('SERVICE_PASSWORD', password = samba.tests.env_get_var_value(
allow_missing=allow_missing_password) 'SERVICE_PASSWORD',
allow_missing=allow_missing_password)
c.set_domain(domain) c.set_domain(domain)
c.set_realm(realm) c.set_realm(realm)
c.set_username(username) c.set_username(username)
@@ -246,21 +276,34 @@ class RawKerberosTest(TestCase):
if hexdump is None: if hexdump is None:
hexdump = self.do_hexdump hexdump = self.do_hexdump
if hexdump: if hexdump:
sys.stderr.write("%s: %d\n%s" % (name, len(blob), self.hexdump(blob))) sys.stderr.write(
"%s: %d\n%s" % (name, len(blob), self.hexdump(blob)))
def der_decode(self, blob, asn1Spec=None, native_encode=True, asn1_print=None, hexdump=None): def der_decode(
self,
blob,
asn1Spec=None,
native_encode=True,
asn1_print=None,
hexdump=None):
if asn1Spec is not None: if asn1Spec is not None:
class_name = type(asn1Spec).__name__.split(':')[0] class_name = type(asn1Spec).__name__.split(':')[0]
else: else:
class_name = "<None-asn1Spec>" class_name = "<None-asn1Spec>"
self.hex_dump(class_name, blob, hexdump=hexdump) self.hex_dump(class_name, blob, hexdump=hexdump)
obj,_ = pyasn1_der_decode(blob, asn1Spec=asn1Spec) obj, _ = pyasn1_der_decode(blob, asn1Spec=asn1Spec)
self.asn1_dump(None, obj, asn1_print=asn1_print) self.asn1_dump(None, obj, asn1_print=asn1_print)
if native_encode: if native_encode:
obj = pyasn1_native_encode(obj) obj = pyasn1_native_encode(obj)
return obj return obj
def der_encode(self, obj, asn1Spec=None, native_decode=True, asn1_print=None, hexdump=None): def der_encode(
self,
obj,
asn1Spec=None,
native_decode=True,
asn1_print=None,
hexdump=None):
if native_decode: if native_decode:
obj = pyasn1_native_decode(obj, asn1Spec=asn1Spec) obj = pyasn1_native_decode(obj, asn1Spec=asn1Spec)
class_name = type(obj).__name__.split(':')[0] class_name = type(obj).__name__.split(':')[0]
@@ -273,7 +316,8 @@ class RawKerberosTest(TestCase):
def send_pdu(self, req, asn1_print=None, hexdump=None): def send_pdu(self, req, asn1_print=None, hexdump=None):
try: try:
k5_pdu = self.der_encode(req, native_decode=False, asn1_print=asn1_print, hexdump=False) k5_pdu = self.der_encode(
req, native_decode=False, asn1_print=asn1_print, hexdump=False)
header = struct.pack('>I', len(k5_pdu)) header = struct.pack('>I', len(k5_pdu))
req_pdu = header req_pdu = header
req_pdu += k5_pdu req_pdu += k5_pdu
@@ -304,7 +348,7 @@ class RawKerberosTest(TestCase):
self._disconnect("recv_raw: EOF") self._disconnect("recv_raw: EOF")
return None return None
self.hex_dump("recv_raw", rep_pdu, hexdump=hexdump) self.hex_dump("recv_raw", rep_pdu, hexdump=hexdump)
except socket.timeout as e: except socket.timeout:
self.s.settimeout(10) self.s.settimeout(10)
sys.stderr.write("recv_raw: TIMEOUT\n") sys.stderr.write("recv_raw: TIMEOUT\n")
pass pass
@@ -322,7 +366,8 @@ class RawKerberosTest(TestCase):
rep_pdu = None rep_pdu = None
rep = None rep = None
try: try:
raw_pdu = self.recv_raw(num_recv=4, hexdump=hexdump, timeout=timeout) raw_pdu = self.recv_raw(
num_recv=4, hexdump=hexdump, timeout=timeout)
if raw_pdu is None: if raw_pdu is None:
return (None, None) return (None, None)
header = struct.unpack(">I", raw_pdu[0:4]) header = struct.unpack(">I", raw_pdu[0:4])
@@ -332,22 +377,27 @@ class RawKerberosTest(TestCase):
missing = k5_len missing = k5_len
rep_pdu = b'' rep_pdu = b''
while missing > 0: while missing > 0:
raw_pdu = self.recv_raw(num_recv=missing, hexdump=hexdump, timeout=timeout) raw_pdu = self.recv_raw(
num_recv=missing, hexdump=hexdump, timeout=timeout)
self.assertGreaterEqual(len(raw_pdu), 1) self.assertGreaterEqual(len(raw_pdu), 1)
rep_pdu += raw_pdu rep_pdu += raw_pdu
missing = k5_len - len(rep_pdu) missing = k5_len - len(rep_pdu)
k5_raw = self.der_decode(rep_pdu, asn1Spec=None, native_encode=False, k5_raw = self.der_decode(
asn1_print=False, hexdump=False) rep_pdu,
pvno=k5_raw['field-0'] asn1Spec=None,
native_encode=False,
asn1_print=False,
hexdump=False)
pvno = k5_raw['field-0']
self.assertEqual(pvno, 5) self.assertEqual(pvno, 5)
msg_type=k5_raw['field-1'] msg_type = k5_raw['field-1']
self.assertIn(msg_type, [11,13,30]) self.assertIn(msg_type, [11, 13, 30])
if msg_type == 11: if msg_type == 11:
asn1Spec=krb5_asn1.AS_REP() asn1Spec = krb5_asn1.AS_REP()
elif msg_type == 13: elif msg_type == 13:
asn1Spec=krb5_asn1.TGS_REP() asn1Spec = krb5_asn1.TGS_REP()
elif msg_type == 30: elif msg_type == 30:
asn1Spec=krb5_asn1.KRB_ERROR() asn1Spec = krb5_asn1.KRB_ERROR()
rep = self.der_decode(rep_pdu, asn1Spec=asn1Spec, rep = self.der_decode(rep_pdu, asn1Spec=asn1Spec,
asn1_print=asn1_print, hexdump=False) asn1_print=asn1_print, hexdump=False)
finally: finally:
@@ -368,11 +418,17 @@ class RawKerberosTest(TestCase):
self.assertIsNone(self.s, msg="Is connected") self.assertIsNone(self.s, msg="Is connected")
return return
def send_recv_transaction(self, req, asn1_print=None, hexdump=None, timeout=None): def send_recv_transaction(
self,
req,
asn1_print=None,
hexdump=None,
timeout=None):
self.connect() self.connect()
try: try:
self.send_pdu(req, asn1_print=asn1_print, hexdump=hexdump) self.send_pdu(req, asn1_print=asn1_print, hexdump=hexdump)
rep = self.recv_pdu(asn1_print=asn1_print, hexdump=hexdump, timeout=timeout) rep = self.recv_pdu(
asn1_print=asn1_print, hexdump=hexdump, timeout=timeout)
except Exception: except Exception:
self._disconnect("transaction failed") self._disconnect("transaction failed")
raise raise
@@ -389,11 +445,15 @@ class RawKerberosTest(TestCase):
def assertPrincipalEqual(self, princ1, princ2): def assertPrincipalEqual(self, princ1, princ2):
self.assertEqual(princ1['name-type'], princ2['name-type']) self.assertEqual(princ1['name-type'], princ2['name-type'])
self.assertEqual(len(princ1['name-string']), len(princ2['name-string']), self.assertEqual(
msg="princ1=%s != princ2=%s" % (princ1, princ2)) len(princ1['name-string']),
len(princ2['name-string']),
msg="princ1=%s != princ2=%s" % (princ1, princ2))
for idx in range(len(princ1['name-string'])): for idx in range(len(princ1['name-string'])):
self.assertEqual(princ1['name-string'][idx], princ2['name-string'][idx], self.assertEqual(
msg="princ1=%s != princ2=%s" % (princ1, princ2)) princ1['name-string'][idx],
princ2['name-string'][idx],
msg="princ1=%s != princ2=%s" % (princ1, princ2))
return return
def get_KerberosTimeWithUsec(self, epoch=None, offset=None): def get_KerberosTimeWithUsec(self, epoch=None, offset=None):
@@ -421,7 +481,7 @@ class RawKerberosTest(TestCase):
salt = None salt = None
try: try:
salt = etype_info2['salt'] salt = etype_info2['salt']
except: except Exception:
pass pass
if e == kcrypto.Enctype.RC4: if e == kcrypto.Enctype.RC4:
@@ -429,7 +489,8 @@ class RawKerberosTest(TestCase):
return self.SessionKey_create(etype=e, contents=nthash, kvno=kvno) return self.SessionKey_create(etype=e, contents=nthash, kvno=kvno)
password = creds.get_password() password = creds.get_password()
return self.PasswordKey_create(etype=e, pwd=password, salt=salt, kvno=kvno) return self.PasswordKey_create(
etype=e, pwd=password, salt=salt, kvno=kvno)
def RandomKey(self, etype): def RandomKey(self, etype):
e = kcrypto._get_enctype_profile(etype) e = kcrypto._get_enctype_profile(etype)
@@ -452,14 +513,14 @@ class RawKerberosTest(TestCase):
'cipher': ciphertext 'cipher': ciphertext
} }
if key.kvno is not None: if key.kvno is not None:
EncryptedData_obj['kvno'] = key.kvno EncryptedData_obj['kvno'] = key.kvno
return EncryptedData_obj return EncryptedData_obj
def Checksum_create(self, key, usage, plaintext, ctype=None): def Checksum_create(self, key, usage, plaintext, ctype=None):
#Checksum ::= SEQUENCE { # Checksum ::= SEQUENCE {
# cksumtype [0] Int32, # cksumtype [0] Int32,
# checksum [1] OCTET STRING # checksum [1] OCTET STRING
#} # }
if ctype is None: if ctype is None:
ctype = key.ctype ctype = key.ctype
checksum = key.make_checksum(usage, plaintext, ctype=ctype) checksum = key.make_checksum(usage, plaintext, ctype=ctype)
@@ -494,10 +555,10 @@ class RawKerberosTest(TestCase):
return PA_DATA_obj return PA_DATA_obj
def PA_ENC_TS_ENC_create(self, ts, usec): def PA_ENC_TS_ENC_create(self, ts, usec):
#PA-ENC-TS-ENC ::= SEQUENCE { # PA-ENC-TS-ENC ::= SEQUENCE {
# patimestamp[0] KerberosTime, -- client's time # patimestamp[0] KerberosTime, -- client's time
# pausec[1] krb5int32 OPTIONAL # pausec[1] krb5int32 OPTIONAL
#} # }
PA_ENC_TS_ENC_obj = { PA_ENC_TS_ENC_obj = {
'patimestamp': ts, 'patimestamp': ts,
'pausec': usec, 'pausec': usec,
@@ -520,7 +581,7 @@ class RawKerberosTest(TestCase):
additional_tickets, additional_tickets,
asn1_print=None, asn1_print=None,
hexdump=None): hexdump=None):
#KDC-REQ-BODY ::= SEQUENCE { # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions, # kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL # cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --, # -- Used only in AS-REQ --,
@@ -532,20 +593,23 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime, # till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL, # rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32, # nonce [7] UInt32,
# etype [8] SEQUENCE OF Int32 -- EncryptionType # etype [8] SEQUENCE OF Int32
# -- EncryptionType
# -- in preference order --, # -- in preference order --,
# addresses [9] HostAddresses OPTIONAL, # addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL # enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --, # -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL # additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty # -- NOTE: not empty
#} # }
if EncAuthorizationData is not None: if EncAuthorizationData is not None:
enc_ad_plain = self.der_encode(EncAuthorizationData, enc_ad_plain = self.der_encode(
asn1Spec=krb5_asn1.AuthorizationData(), EncAuthorizationData,
asn1_print=asn1_print, asn1Spec=krb5_asn1.AuthorizationData(),
hexdump=hexdump) asn1_print=asn1_print,
enc_ad = self.EncryptedData_create(EncAuthorizationData_key, enc_ad_plain) hexdump=hexdump)
enc_ad = self.EncryptedData_create(
EncAuthorizationData_key, enc_ad_plain)
else: else:
enc_ad = None enc_ad = None
KDC_REQ_BODY_obj = { KDC_REQ_BODY_obj = {
@@ -590,14 +654,14 @@ class RawKerberosTest(TestCase):
asn1Spec=None, asn1Spec=None,
asn1_print=None, asn1_print=None,
hexdump=None): hexdump=None):
#KDC-REQ ::= SEQUENCE { # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0] # -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) , # pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --), # msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL # padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --, # -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY # req-body [4] KDC-REQ-BODY
#} # }
# #
KDC_REQ_BODY_obj = self.KDC_REQ_BODY_create(kdc_options, KDC_REQ_BODY_obj = self.KDC_REQ_BODY_create(kdc_options,
cname, cname,
@@ -622,39 +686,40 @@ class RawKerberosTest(TestCase):
if padata is not None: if padata is not None:
KDC_REQ_obj['padata'] = padata KDC_REQ_obj['padata'] = padata
if asn1Spec is not None: if asn1Spec is not None:
KDC_REQ_decoded = pyasn1_native_decode(KDC_REQ_obj, asn1Spec=asn1Spec) KDC_REQ_decoded = pyasn1_native_decode(
KDC_REQ_obj, asn1Spec=asn1Spec)
else: else:
KDC_REQ_decoded = None KDC_REQ_decoded = None
return KDC_REQ_obj, KDC_REQ_decoded return KDC_REQ_obj, KDC_REQ_decoded
def AS_REQ_create(self, def AS_REQ_create(self,
padata, # optional padata, # optional
kdc_options, # required kdc_options, # required
cname, # optional cname, # optional
realm, # required realm, # required
sname, # optional sname, # optional
from_time, # optional from_time, # optional
till_time, # required till_time, # required
renew_time, # optional renew_time, # optional
nonce, # required nonce, # required
etypes, # required etypes, # required
addresses, # optional addresses, # optional
EncAuthorizationData, EncAuthorizationData,
EncAuthorizationData_key, EncAuthorizationData_key,
additional_tickets, additional_tickets,
native_decoded_only=True, native_decoded_only=True,
asn1_print=None, asn1_print=None,
hexdump=None): hexdump=None):
#KDC-REQ ::= SEQUENCE { # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0] # -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) , # pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --), # msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL # padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --, # -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY # req-body [4] KDC-REQ-BODY
#} # }
# #
#KDC-REQ-BODY ::= SEQUENCE { # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions, # kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL # cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --, # -- Used only in AS-REQ --,
@@ -666,32 +731,34 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime, # till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL, # rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32, # nonce [7] UInt32,
# etype [8] SEQUENCE OF Int32 -- EncryptionType # etype [8] SEQUENCE OF Int32
# -- EncryptionType
# -- in preference order --, # -- in preference order --,
# addresses [9] HostAddresses OPTIONAL, # addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL # enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --, # -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL # additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty # -- NOTE: not empty
#} # }
obj,decoded = self.KDC_REQ_create(msg_type=10, obj, decoded = self.KDC_REQ_create(
padata=padata, msg_type=10,
kdc_options=kdc_options, padata=padata,
cname=cname, kdc_options=kdc_options,
realm=realm, cname=cname,
sname=sname, realm=realm,
from_time=from_time, sname=sname,
till_time=till_time, from_time=from_time,
renew_time=renew_time, till_time=till_time,
nonce=nonce, renew_time=renew_time,
etypes=etypes, nonce=nonce,
addresses=addresses, etypes=etypes,
EncAuthorizationData=EncAuthorizationData, addresses=addresses,
EncAuthorizationData_key=EncAuthorizationData_key, EncAuthorizationData=EncAuthorizationData,
additional_tickets=additional_tickets, EncAuthorizationData_key=EncAuthorizationData_key,
asn1Spec=krb5_asn1.AS_REQ(), additional_tickets=additional_tickets,
asn1_print=asn1_print, asn1Spec=krb5_asn1.AS_REQ(),
hexdump=hexdump) asn1_print=asn1_print,
hexdump=hexdump)
if native_decoded_only: if native_decoded_only:
return decoded return decoded
return decoded, obj return decoded, obj
@@ -703,7 +770,7 @@ class RawKerberosTest(TestCase):
# ap-options [2] APOptions, # ap-options [2] APOptions,
# ticket [3] Ticket, # ticket [3] Ticket,
# authenticator [4] EncryptedData -- Authenticator # authenticator [4] EncryptedData -- Authenticator
#} # }
AP_REQ_obj = { AP_REQ_obj = {
'pvno': 5, 'pvno': 5,
'msg-type': 14, 'msg-type': 14,
@@ -713,8 +780,9 @@ class RawKerberosTest(TestCase):
} }
return AP_REQ_obj return AP_REQ_obj
def Authenticator_create(self, crealm, cname, cksum, cusec, ctime, subkey, seq_number, def Authenticator_create(
authorization_data): self, crealm, cname, cksum, cusec, ctime, subkey, seq_number,
authorization_data):
# -- Unencrypted authenticator # -- Unencrypted authenticator
# Authenticator ::= [APPLICATION 2] SEQUENCE { # Authenticator ::= [APPLICATION 2] SEQUENCE {
# authenticator-vno [0] INTEGER (5), # authenticator-vno [0] INTEGER (5),
@@ -726,7 +794,7 @@ class RawKerberosTest(TestCase):
# subkey [6] EncryptionKey OPTIONAL, # subkey [6] EncryptionKey OPTIONAL,
# seq-number [7] UInt32 OPTIONAL, # seq-number [7] UInt32 OPTIONAL,
# authorization-data [8] AuthorizationData OPTIONAL # authorization-data [8] AuthorizationData OPTIONAL
#} # }
Authenticator_obj = { Authenticator_obj = {
'authenticator-vno': 5, 'authenticator-vno': 5,
'crealm': crealm, 'crealm': crealm,
@@ -745,20 +813,20 @@ class RawKerberosTest(TestCase):
return Authenticator_obj return Authenticator_obj
def TGS_REQ_create(self, def TGS_REQ_create(self,
padata, # optional padata, # optional
cusec, cusec,
ctime, ctime,
ticket, ticket,
kdc_options, # required kdc_options, # required
cname, # optional cname, # optional
realm, # required realm, # required
sname, # optional sname, # optional
from_time, # optional from_time, # optional
till_time, # required till_time, # required
renew_time, # optional renew_time, # optional
nonce, # required nonce, # required
etypes, # required etypes, # required
addresses, # optional addresses, # optional
EncAuthorizationData, EncAuthorizationData,
EncAuthorizationData_key, EncAuthorizationData_key,
additional_tickets, additional_tickets,
@@ -768,16 +836,16 @@ class RawKerberosTest(TestCase):
native_decoded_only=True, native_decoded_only=True,
asn1_print=None, asn1_print=None,
hexdump=None): hexdump=None):
#KDC-REQ ::= SEQUENCE { # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0] # -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) , # pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --), # msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL # padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --, # -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY # req-body [4] KDC-REQ-BODY
#} # }
# #
#KDC-REQ-BODY ::= SEQUENCE { # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions, # kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL # cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --, # -- Used only in AS-REQ --,
@@ -789,50 +857,57 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime, # till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL, # rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32, # nonce [7] UInt32,
# etype [8] SEQUENCE OF Int32 -- EncryptionType # etype [8] SEQUENCE OF Int32
# -- EncryptionType
# -- in preference order --, # -- in preference order --,
# addresses [9] HostAddresses OPTIONAL, # addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL # enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --, # -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL # additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty # -- NOTE: not empty
#} # }
req_body = self.KDC_REQ_BODY_create(kdc_options=kdc_options, req_body = self.KDC_REQ_BODY_create(
cname=None, kdc_options=kdc_options,
realm=realm, cname=None,
sname=sname, realm=realm,
from_time=from_time, sname=sname,
till_time=till_time, from_time=from_time,
renew_time=renew_time, till_time=till_time,
nonce=nonce, renew_time=renew_time,
etypes=etypes, nonce=nonce,
addresses=addresses, etypes=etypes,
EncAuthorizationData=EncAuthorizationData, addresses=addresses,
EncAuthorizationData_key=EncAuthorizationData_key, EncAuthorizationData=EncAuthorizationData,
additional_tickets=additional_tickets) EncAuthorizationData_key=EncAuthorizationData_key,
additional_tickets=additional_tickets)
req_body = self.der_encode(req_body, asn1Spec=krb5_asn1.KDC_REQ_BODY(), req_body = self.der_encode(req_body, asn1Spec=krb5_asn1.KDC_REQ_BODY(),
asn1_print=asn1_print, hexdump=hexdump) asn1_print=asn1_print, hexdump=hexdump)
req_body_checksum = self.Checksum_create(ticket_session_key, 6, req_body, req_body_checksum = self.Checksum_create(
ctype=body_checksum_type) ticket_session_key, 6, req_body, ctype=body_checksum_type)
subkey_obj = None subkey_obj = None
if authenticator_subkey is not None: if authenticator_subkey is not None:
subkey_obj = authenticator_subkey.export_obj() subkey_obj = authenticator_subkey.export_obj()
seq_number = random.randint(0, 0xfffffffe) seq_number = random.randint(0, 0xfffffffe)
authenticator = self.Authenticator_create(crealm=realm, authenticator = self.Authenticator_create(
cname=cname, crealm=realm,
cksum=req_body_checksum, cname=cname,
cusec=cusec, cksum=req_body_checksum,
ctime=ctime, cusec=cusec,
subkey=subkey_obj, ctime=ctime,
seq_number=seq_number, subkey=subkey_obj,
authorization_data=None) seq_number=seq_number,
authenticator = self.der_encode(authenticator, asn1Spec=krb5_asn1.Authenticator(), authorization_data=None)
asn1_print=asn1_print, hexdump=hexdump) authenticator = self.der_encode(
authenticator,
asn1Spec=krb5_asn1.Authenticator(),
asn1_print=asn1_print,
hexdump=hexdump)
authenticator = self.EncryptedData_create(ticket_session_key, 7, authenticator) authenticator = self.EncryptedData_create(
ticket_session_key, 7, authenticator)
ap_options = krb5_asn1.APOptions('0') ap_options = krb5_asn1.APOptions('0')
ap_req = self.AP_REQ_create(ap_options=str(ap_options), ap_req = self.AP_REQ_create(ap_options=str(ap_options),
@@ -846,24 +921,25 @@ class RawKerberosTest(TestCase):
else: else:
padata = [pa_tgs_req] padata = [pa_tgs_req]
obj,decoded = self.KDC_REQ_create(msg_type=12, obj, decoded = self.KDC_REQ_create(
padata=padata, msg_type=12,
kdc_options=kdc_options, padata=padata,
cname=None, kdc_options=kdc_options,
realm=realm, cname=None,
sname=sname, realm=realm,
from_time=from_time, sname=sname,
till_time=till_time, from_time=from_time,
renew_time=renew_time, till_time=till_time,
nonce=nonce, renew_time=renew_time,
etypes=etypes, nonce=nonce,
addresses=addresses, etypes=etypes,
EncAuthorizationData=EncAuthorizationData, addresses=addresses,
EncAuthorizationData_key=EncAuthorizationData_key, EncAuthorizationData=EncAuthorizationData,
additional_tickets=additional_tickets, EncAuthorizationData_key=EncAuthorizationData_key,
asn1Spec=krb5_asn1.TGS_REQ(), additional_tickets=additional_tickets,
asn1_print=asn1_print, asn1Spec=krb5_asn1.TGS_REQ(),
hexdump=hexdump) asn1_print=asn1_print,
hexdump=hexdump)
if native_decoded_only: if native_decoded_only:
return decoded return decoded
return decoded, obj return decoded, obj
@@ -888,5 +964,6 @@ class RawKerberosTest(TestCase):
'cksum': cksum, 'cksum': cksum,
'auth': "Kerberos", 'auth': "Kerberos",
} }
pa_s4u2self = self.der_encode(PA_S4U2Self_obj, asn1Spec=krb5_asn1.PA_S4U2Self()) pa_s4u2self = self.der_encode(
PA_S4U2Self_obj, asn1Spec=krb5_asn1.PA_S4U2Self())
return self.PA_DATA_create(129, pa_s4u2self) return self.PA_DATA_create(129, pa_s4u2self)

View File

@@ -38,31 +38,31 @@ PADATA_ETYPE_INFO2 = int(
# Error codes # Error codes
KDC_ERR_C_PRINCIPAL_UNKNOWN = 6 KDC_ERR_C_PRINCIPAL_UNKNOWN = 6
KDC_ERR_PREAUTH_FAILED = 24 KDC_ERR_PREAUTH_FAILED = 24
KDC_ERR_PREAUTH_REQUIRED = 25 KDC_ERR_PREAUTH_REQUIRED = 25
KDC_ERR_BADMATCH = 36 KDC_ERR_BADMATCH = 36
KDC_ERR_SKEW = 37 KDC_ERR_SKEW = 37
# Name types # Name types
NT_UNKNOWN = int(krb5_asn1.NameTypeValues('kRB5-NT-UNKNOWN')) NT_UNKNOWN = int(krb5_asn1.NameTypeValues('kRB5-NT-UNKNOWN'))
NT_PRINCIPAL = int(krb5_asn1.NameTypeValues('kRB5-NT-PRINCIPAL')) NT_PRINCIPAL = int(krb5_asn1.NameTypeValues('kRB5-NT-PRINCIPAL'))
NT_SRV_INST = int(krb5_asn1.NameTypeValues('kRB5-NT-SRV-INST')) NT_SRV_INST = int(krb5_asn1.NameTypeValues('kRB5-NT-SRV-INST'))
NT_ENTERPRISE_PRINCIPAL = int(krb5_asn1.NameTypeValues( NT_ENTERPRISE_PRINCIPAL = int(krb5_asn1.NameTypeValues(
'kRB5-NT-ENTERPRISE-PRINCIPAL')) 'kRB5-NT-ENTERPRISE-PRINCIPAL'))
# Authorization data ad-type values # Authorization data ad-type values
AD_IF_RELEVANT = 1 AD_IF_RELEVANT = 1
AD_INTENDED_FOR_SERVER = 2 AD_INTENDED_FOR_SERVER = 2
AD_INTENDED_FOR_APPLICATION_CLASS = 3 AD_INTENDED_FOR_APPLICATION_CLASS = 3
AD_KDC_ISSUED = 4 AD_KDC_ISSUED = 4
AD_AND_OR = 5 AD_AND_OR = 5
AD_MANDATORY_TICKET_EXTENSIONS = 6 AD_MANDATORY_TICKET_EXTENSIONS = 6
AD_IN_TICKET_EXTENSIONS = 7 AD_IN_TICKET_EXTENSIONS = 7
AD_MANDATORY_FOR_KDC = 8 AD_MANDATORY_FOR_KDC = 8
AD_INITIAL_VERIFIED_CAS = 9 AD_INITIAL_VERIFIED_CAS = 9
AD_WIN2K_PAC = 128 AD_WIN2K_PAC = 128
AD_SIGNTICKET = 512 AD_SIGNTICKET = 512
# Key usage numbers # Key usage numbers
# RFC 4120 Section 7.5.1. Key Usage Numbers # RFC 4120 Section 7.5.1. Key Usage Numbers

View File

@@ -35,6 +35,7 @@ import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
class S4UKerberosTests(RawKerberosTest): class S4UKerberosTests(RawKerberosTest):
def setUp(self): def setUp(self):
@@ -55,7 +56,7 @@ class S4UKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None padata = None
etypes=(18,17,23) etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata, req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options), kdc_options=str(kdc_options),
@@ -76,14 +77,16 @@ class S4UKerberosTests(RawKerberosTest):
self.assertEqual(rep['msg-type'], 30) self.assertEqual(rep['msg-type'], 30)
self.assertEqual(rep['error-code'], 25) self.assertEqual(rep['error-code'], 25)
rep_padata = self.der_decode(rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA()) rep_padata = self.der_decode(
rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA())
for pa in rep_padata: for pa in rep_padata:
if pa['padata-type'] == 19: if pa['padata-type'] == 19:
etype_info2 = pa['padata-value'] etype_info2 = pa['padata-value']
break break
etype_info2 = self.der_decode(etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2()) etype_info2 = self.der_decode(
etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
key = self.PasswordKey_from_etype_info2(service_creds, etype_info2[0]) key = self.PasswordKey_from_etype_info2(service_creds, etype_info2[0])
@@ -120,7 +123,8 @@ class S4UKerberosTests(RawKerberosTest):
self.assertEqual(msg_type, 11) self.assertEqual(msg_type, 11)
enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher']) enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher'])
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncASRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncASRepPart())
# S4U2Self Request # S4U2Self Request
sname = cname sname = cname
@@ -167,11 +171,13 @@ class S4UKerberosTests(RawKerberosTest):
if msg_type == 13: if msg_type == 13:
enc_part2 = subkey.decrypt( enc_part2 = subkey.decrypt(
KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher']) KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher'])
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
return msg_type return msg_type
# Using the checksum type from the tgt_session_key happens to work everywhere # Using the checksum type from the tgt_session_key happens to work
# everywhere
def test_s4u2self(self): def test_s4u2self(self):
msg_type = self._test_s4u2self() msg_type = self._test_s4u2self()
self.assertEqual(msg_type, 13) self.assertEqual(msg_type, 13)
@@ -193,6 +199,7 @@ class S4UKerberosTests(RawKerberosTest):
msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.CRC32) msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.CRC32)
self.assertEqual(msg_type, 30) self.assertEqual(msg_type, 30)
if __name__ == "__main__": if __name__ == "__main__":
global_asn1_print = True global_asn1_print = True
global_hexdump = True global_hexdump = True

View File

@@ -33,6 +33,7 @@ import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
class SimpleKerberosTests(RawKerberosTest): class SimpleKerberosTests(RawKerberosTest):
def setUp(self): def setUp(self):
@@ -53,7 +54,7 @@ class SimpleKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None padata = None
etypes=(18,17,23) etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata, req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options), kdc_options=str(kdc_options),
@@ -74,14 +75,16 @@ class SimpleKerberosTests(RawKerberosTest):
self.assertEqual(rep['msg-type'], 30) self.assertEqual(rep['msg-type'], 30)
self.assertEqual(rep['error-code'], 25) self.assertEqual(rep['error-code'], 25)
rep_padata = self.der_decode(rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA()) rep_padata = self.der_decode(
rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA())
for pa in rep_padata: for pa in rep_padata:
if pa['padata-type'] == 19: if pa['padata-type'] == 19:
etype_info2 = pa['padata-value'] etype_info2 = pa['padata-value']
break break
etype_info2 = self.der_decode(etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2()) etype_info2 = self.der_decode(
etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
key = self.PasswordKey_from_etype_info2(user_creds, etype_info2[0]) key = self.PasswordKey_from_etype_info2(user_creds, etype_info2[0])
@@ -119,17 +122,21 @@ class SimpleKerberosTests(RawKerberosTest):
enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher']) enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher'])
# MIT KDC encodes both EncASRepPart and EncTGSRepPart with application tag 26 # MIT KDC encodes both EncASRepPart and EncTGSRepPart with
# application tag 26
try: try:
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncASRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncASRepPart())
except Exception: except Exception:
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
# TGS Request # TGS Request
service_creds = self.get_service_creds(allow_missing_password=True) service_creds = self.get_service_creds(allow_missing_password=True)
service_name = service_creds.get_username() service_name = service_creds.get_username()
sname = self.PrincipalName_create(name_type=2, names=["host", service_name]) sname = self.PrincipalName_create(
name_type=2, names=["host", service_name])
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
till = self.get_KerberosTime(offset=36000) till = self.get_KerberosTime(offset=36000)
ticket = rep['ticket'] ticket = rep['ticket']
@@ -167,7 +174,8 @@ class SimpleKerberosTests(RawKerberosTest):
enc_part2 = subkey.decrypt( enc_part2 = subkey.decrypt(
KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher']) KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher'])
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
return return

View File

@@ -34,6 +34,7 @@ import samba.tests
global_asn1_print = False global_asn1_print = False
global_hexdump = False global_hexdump = False
class XrealmKerberosTests(RawKerberosTest): class XrealmKerberosTests(RawKerberosTest):
def setUp(self): def setUp(self):
@@ -54,7 +55,7 @@ class XrealmKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None padata = None
etypes=(18,17,23) etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata, req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options), kdc_options=str(kdc_options),
@@ -75,14 +76,16 @@ class XrealmKerberosTests(RawKerberosTest):
self.assertEqual(rep['msg-type'], 30) self.assertEqual(rep['msg-type'], 30)
self.assertEqual(rep['error-code'], 25) self.assertEqual(rep['error-code'], 25)
rep_padata = self.der_decode(rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA()) rep_padata = self.der_decode(
rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA())
for pa in rep_padata: for pa in rep_padata:
if pa['padata-type'] == 19: if pa['padata-type'] == 19:
etype_info2 = pa['padata-value'] etype_info2 = pa['padata-value']
break break
etype_info2 = self.der_decode(etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2()) etype_info2 = self.der_decode(
etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
key = self.PasswordKey_from_etype_info2(user_creds, etype_info2[0]) key = self.PasswordKey_from_etype_info2(user_creds, etype_info2[0])
@@ -120,15 +123,19 @@ class XrealmKerberosTests(RawKerberosTest):
enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher']) enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher'])
# MIT KDC encodes both EncASRepPart and EncTGSRepPart with application tag 26 # MIT KDC encodes both EncASRepPart and EncTGSRepPart with
# application tag 26
try: try:
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncASRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncASRepPart())
except Exception: except Exception:
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
# TGS Request (for cross-realm TGT) # TGS Request (for cross-realm TGT)
trust_realm = samba.tests.env_get_var_value('TRUST_REALM') trust_realm = samba.tests.env_get_var_value('TRUST_REALM')
sname = self.PrincipalName_create(name_type=2, names=["krbtgt", trust_realm]) sname = self.PrincipalName_create(
name_type=2, names=["krbtgt", trust_realm])
kdc_options = krb5_asn1.KDCOptions('forwardable') kdc_options = krb5_asn1.KDCOptions('forwardable')
till = self.get_KerberosTime(offset=36000) till = self.get_KerberosTime(offset=36000)
@@ -167,10 +174,11 @@ class XrealmKerberosTests(RawKerberosTest):
enc_part2 = subkey.decrypt( enc_part2 = subkey.decrypt(
KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher']) KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher'])
enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart()) enc_part2 = self.der_decode(
enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
# Check the forwardable flag # Check the forwardable flag
fwd_pos = len(tuple(krb5_asn1.TicketFlags('forwardable'))) -1 fwd_pos = len(tuple(krb5_asn1.TicketFlags('forwardable'))) - 1
assert(krb5_asn1.TicketFlags(enc_part2['flags'])[fwd_pos]) assert(krb5_asn1.TicketFlags(enc_part2['flags'])[fwd_pos])
return return