summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
authorGary Lockyer <gary@catalyst.net.nz>2020-12-11 11:55:01 +1300
committerGary Lockyer <gary@samba.org>2020-12-21 21:29:28 +0000
commitc00d537526ca881c540ff66e703ad9c96dd1face (patch)
tree67ff35567a1303560bab1c20b29960ba0c2cb61c /python
parent03676a4a5c55ab5f4958a86cbd4d7be0f0a8a294 (diff)
downloadsamba-c00d537526ca881c540ff66e703ad9c96dd1face.tar.gz
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
Diffstat (limited to 'python')
-rwxr-xr-xpython/samba/tests/krb5/as_canonicalization_tests.py54
-rwxr-xr-xpython/samba/tests/krb5/compatability_tests.py24
-rwxr-xr-xpython/samba/tests/krb5/kcrypto.py67
-rw-r--r--python/samba/tests/krb5/kdc_base_test.py4
-rwxr-xr-xpython/samba/tests/krb5/kdc_tests.py17
-rw-r--r--python/samba/tests/krb5/raw_testcase.py409
-rw-r--r--python/samba/tests/krb5/rfc4120_constants.py32
-rwxr-xr-xpython/samba/tests/krb5/s4u_tests.py19
-rwxr-xr-xpython/samba/tests/krb5/simple_tests.py24
-rwxr-xr-xpython/samba/tests/krb5/xrealm_tests.py26
10 files changed, 413 insertions, 263 deletions
diff --git a/python/samba/tests/krb5/as_canonicalization_tests.py b/python/samba/tests/krb5/as_canonicalization_tests.py
index e89b40eab8f..43f532dc483 100755
--- a/python/samba/tests/krb5/as_canonicalization_tests.py
+++ b/python/samba/tests/krb5/as_canonicalization_tests.py
@@ -31,8 +31,6 @@ import samba
from samba.auth import system_session
from samba.credentials import (
Credentials,
- CLI_CRED_NTLMv2_AUTH,
- CLI_CRED_NTLM_AUTH,
DONT_USE_KERBEROS)
from samba.dcerpc.misc import SEC_CHAN_WKSTA
from samba.dsdb import (
@@ -41,7 +39,20 @@ from samba.dsdb import (
UF_NORMAL_ACCOUNT)
from samba.samdb import SamDB
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_hexdump = False
@@ -49,15 +60,15 @@ global_hexdump = False
@unique
class TestOptions(Enum):
- Canonicalize = 1
- Enterprise = 2
- UpperRealm = 4
- UpperUserName = 8
- NetbiosRealm = 16
- UPN = 32
- RemoveDollar = 64
- AsReqSelf = 128
- Last = 256
+ Canonicalize = 1
+ Enterprise = 2
+ UpperRealm = 4
+ UpperUserName = 8
+ NetbiosRealm = 16
+ UPN = 32
+ RemoveDollar = 64
+ AsReqSelf = 128
+ Last = 256
def is_set(self, x):
return self.value & x
@@ -65,7 +76,7 @@ class TestOptions(Enum):
@unique
class CredentialsType(Enum):
- User = 1
+ User = 1
Machine = 2
def is_set(self, x):
@@ -126,7 +137,8 @@ class TestData:
MACHINE_NAME = "tstkrb5cnnmch"
-USER_NAME = "tstkrb5cnnusr"
+USER_NAME = "tstkrb5cnnusr"
+
@DynamicTestCase
class KerberosASCanonicalizationTests(RawKerberosTest):
@@ -160,21 +172,21 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
@classmethod
def setUpClass(cls):
- cls.lp = cls.get_loadparm(cls)
+ cls.lp = cls.get_loadparm(cls)
cls.username = os.environ["USERNAME"]
cls.password = os.environ["PASSWORD"]
- cls.host = os.environ["SERVER"]
+ cls.host = os.environ["SERVER"]
c = Credentials()
c.set_username(cls.username)
c.set_password(cls.password)
try:
- realm = os.environ["REALM"]
+ realm = os.environ["REALM"]
c.set_realm(realm)
except KeyError:
pass
try:
- domain = os.environ["DOMAIN"]
+ domain = os.environ["DOMAIN"]
c.set_domain(domain)
except KeyError:
pass
@@ -200,7 +212,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
def setUp(self):
super(KerberosASCanonicalizationTests, self).setUp()
self.do_asn1_print = global_asn1_print
- self.do_hexdump = global_hexdump
+ self.do_hexdump = global_hexdump
#
# Create a test user account
@@ -340,7 +352,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
#
# Check the protocol version, should be 5
self.assertEqual(
- rep['pvno'], 5, "Data {0}".format(str(data)))
+ rep['pvno'], 5, "Data {0}".format(str(data)))
self.assertEqual(
rep['msg-type'], KRB_ERROR, "Data {0}".format(str(data)))
@@ -397,7 +409,7 @@ class KerberosASCanonicalizationTests(RawKerberosTest):
#
# Check the protocol version, should be 5
self.assertEqual(
- rep['pvno'], 5, "Data {0}".format(str(data)))
+ rep['pvno'], 5, "Data {0}".format(str(data)))
msg_type = rep['msg-type']
# Should not have got an error.
diff --git a/python/samba/tests/krb5/compatability_tests.py b/python/samba/tests/krb5/compatability_tests.py
index 0b3701cd60d..5a1ef02ef80 100755
--- a/python/samba/tests/krb5/compatability_tests.py
+++ b/python/samba/tests/krb5/compatability_tests.py
@@ -25,7 +25,20 @@ os.environ["PYTHONUNBUFFERED"] = "1"
from samba.tests.krb5.raw_testcase import RawKerberosTest
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_hexdump = False
@@ -112,18 +125,17 @@ class SimpleKerberosTests(RawKerberosTest):
realm = creds.get_realm()
cname = self.PrincipalName_create(
- name_type=NT_PRINCIPAL,
- names=[user])
+ name_type=NT_PRINCIPAL,
+ names=[user])
sname = self.PrincipalName_create(
- name_type=NT_SRV_INST,
- names=["krbtgt", realm])
+ name_type=NT_SRV_INST,
+ names=["krbtgt", realm])
till = self.get_KerberosTime(offset=36000)
kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None
-
req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options),
cname=cname,
diff --git a/python/samba/tests/krb5/kcrypto.py b/python/samba/tests/krb5/kcrypto.py
index 64bdbecd8b2..c8fef4c876d 100755
--- a/python/samba/tests/krb5/kcrypto.py
+++ b/python/samba/tests/krb5/kcrypto.py
@@ -64,6 +64,7 @@ from samba.credentials import Credentials
from samba import generate_random_bytes as get_random_bytes
from samba.common import get_string, get_bytes
+
class Enctype(object):
DES_CRC = 1
DES_MD4 = 2
@@ -112,26 +113,30 @@ def _mac_equal(mac1, mac2):
res |= x ^ y
return res == 0
+
def SIMPLE_HASH(string, algo_cls):
hash_ctx = hashes.Hash(algo_cls(), default_backend())
hash_ctx.update(string)
return hash_ctx.finalize()
+
def HMAC_HASH(key, string, algo_cls):
hmac_ctx = hmac.HMAC(key, algo_cls(), default_backend())
hmac_ctx.update(string)
return hmac_ctx.finalize()
+
def _nfold(str, nbytes):
# Convert str to a string of length nbytes using the RFC 3961 nfold
# operation.
# Rotate the bytes in str to the right by nbits bits.
def rotate_right(str, nbits):
- nbytes, remain = (nbits//8) % len(str), nbits % 8
- return bytes([(str[i-nbytes] >> remain) |
- (str[i-nbytes-1] << (8-remain) & 0xff)
- for i in range(len(str))])
+ nbytes, remain = (nbits // 8) % len(str), nbits % 8
+ return bytes([
+ (str[i - nbytes] >> remain)
+ | (str[i - nbytes - 1] << (8 - remain) & 0xff)
+ for i in range(len(str))])
# Add equal-length strings together with end-around carry.
def add_ones_complement(str1, str2):
@@ -139,7 +144,7 @@ def _nfold(str, nbytes):
v = [a + b for a, b in zip(str1, str2)]
# Propagate carry bits to the left until there aren't any left.
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])
# Concatenate copies of str to produce the least common multiple
@@ -150,7 +155,7 @@ def _nfold(str, nbytes):
slen = len(str)
lcm = nbytes * slen // gcd(nbytes, 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)
@@ -275,7 +280,7 @@ class _DES3CBC(_SimplifiedEnctype):
return b if bin(b & ~1).count('1') % 2 else b | 1
assert len(seed) == 7
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]])
if _is_weak_des_key(keybytes):
keybytes[7] = bytes([keybytes[7] ^ 0xF0])
@@ -369,7 +374,7 @@ class _AESEnctype(_SimplifiedEnctype):
if len(ciphertext) == 16:
return aes_decrypt(ciphertext)
# 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])
# CBC-decrypt all but the last two blocks.
prev_cblock = bytes(16)
@@ -383,7 +388,7 @@ class _AESEnctype(_SimplifiedEnctype):
# will be the omitted bytes of ciphertext from the final
# block.
b = aes_decrypt(cblocks[-2])
- lastplaintext =_xorbytes(b[:lastlen], cblocks[-1])
+ lastplaintext = _xorbytes(b[:lastlen], cblocks[-1])
omitted = b[lastlen:]
# Decrypt the final cipher block plus the omitted bytes to get
# the second-to-last plaintext block.
@@ -433,7 +438,8 @@ class _RC4(_EnctypeProfile):
cksum = HMAC_HASH(ki, confounder + plaintext, 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)
return cksum + ctext
@@ -446,7 +452,8 @@ class _RC4(_EnctypeProfile):
ki = HMAC_HASH(key.contents, cls.usage_str(keyusage), 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)
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)
-def prfplus(key, pepper, l):
- # Produce l bytes of output using the RFC 6113 PRF+ function.
+def prfplus(key, pepper, ln):
+ # Produce ln bytes of output using the RFC 6113 PRF+ function.
out = b''
count = 1
- while len(out) < l:
+ while len(out) < ln:
out += prf(key, bytes([count]) + pepper)
count += 1
- return out[:l]
+ return out[:ln]
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),
prfplus(key2, pepper2, e.seedsize)))
+
def h(hexstr):
return bytes.fromhex(hexstr)
+
class KcrytoTest(TestCase):
"""kcrypto Test case."""
@@ -665,20 +674,21 @@ class KcrytoTest(TestCase):
conf = h('94B491F481485B9A0678CD3C4EA386AD')
keyusage = 2
plain = b'9 bytesss'
- ctxt = h('68FB9679601F45C78857B2BF820FD6E53ECA8D42FD4B1D7024A09205ABB7CD2E'
- 'C26C355D2F')
+ ctxt = h('68FB9679601F45C78857B2BF820FD6E53ECA8D42FD4B1D7024A09205ABB7'
+ 'CD2EC26C355D2F')
k = Key(Enctype.AES128, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain)
def test_aes256_crypt(self):
# AES256 encrypt and decrypt
- kb = h('F1C795E9248A09338D82C3F8D5B567040B0110736845041347235B1404231398')
+ kb = h('F1C795E9248A09338D82C3F8D5B567040B0110736845041347235B14042313'
+ '98')
conf = h('E45CA518B42E266AD98E165E706FFB60')
keyusage = 4
plain = b'30 bytes bytes bytes bytes byt'
- ctxt = h('D1137A4D634CFECE924DBC3BF6790648BD5CFF7DE0E7B99460211D0DAEF3D79A'
- '295C688858F3B34B9CBD6EEBAE81DAF6B734D4D498B6714F1C1D')
+ ctxt = h('D1137A4D634CFECE924DBC3BF6790648BD5CFF7DE0E7B99460211D0DAEF3'
+ 'D79A295C688858F3B34B9CBD6EEBAE81DAF6B734D4D498B6714F1C1D')
k = Key(Enctype.AES256, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain)
@@ -694,7 +704,8 @@ class KcrytoTest(TestCase):
def test_aes256_checksum(self):
# AES256 checksum
- kb = h('B1AE4CD8462AFF1677053CC9279AAC30B796FB81CE21474DD3DDBCFEA4EC76D7')
+ kb = h('B1AE4CD8462AFF1677053CC9279AAC30B796FB81CE21474DD3DDBC'
+ 'FEA4EC76D7')
keyusage = 4
plain = b'fourteen'
cksum = h('E08739E3279E2903EC8E3836')
@@ -715,7 +726,8 @@ class KcrytoTest(TestCase):
string = b'X' * 64
salt = b'pass phrase equals block size'
params = h('000004B0')
- kb = h('89ADEE3608DB8BC71F1BFBFE459486B05618B70CBAE22092534E56C553BA4B34')
+ kb = h('89ADEE3608DB8BC71F1BFBFE459486B05618B70CBAE22092534E56'
+ 'C553BA4B34')
k = string_to_key(Enctype.AES256, string, salt, params)
self.assertEqual(k.contents, kb)
@@ -741,7 +753,8 @@ class KcrytoTest(TestCase):
def test_aes256_cf2(self):
# AES256 cf2
- kb = h('4D6CA4E629785C1F01BAF55E2E548566B9617AE3A96868C337CB93B5E72B1C7B')
+ kb = h('4D6CA4E629785C1F01BAF55E2E548566B9617AE3A96868C337CB93B5'
+ 'E72B1C7B')
k1 = string_to_key(Enctype.AES256, b'key1', b'key1')
k2 = string_to_key(Enctype.AES256, b'key2', b'key2')
k = cf2(Enctype.AES256, k1, k2, b'a', b'b')
@@ -753,8 +766,8 @@ class KcrytoTest(TestCase):
conf = h('94690A17B2DA3C9B')
keyusage = 3
plain = b'13 bytes byte'
- ctxt = h('839A17081ECBAFBCDC91B88C6955DD3C4514023CF177B77BF0D0177A16F705E8'
- '49CB7781D76A316B193F8D30')
+ ctxt = h('839A17081ECBAFBCDC91B88C6955DD3C4514023CF177B77BF0D0177A16F7'
+ '05E849CB7781D76A316B193F8D30')
k = Key(Enctype.DES3, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), _zeropad(plain, 8))
@@ -790,8 +803,8 @@ class KcrytoTest(TestCase):
conf = h('37245E73A45FBF72')
keyusage = 4
plain = b'30 bytes bytes bytes bytes byt'
- ctxt = h('95F9047C3AD75891C2E9B04B16566DC8B6EB9CE4231AFB2542EF87A7B5A0F260'
- 'A99F0460508DE0CECC632D07C354124E46C5D2234EB8')
+ ctxt = h('95F9047C3AD75891C2E9B04B16566DC8B6EB9CE4231AFB2542EF87A7B5A0'
+ 'F260A99F0460508DE0CECC632D07C354124E46C5D2234EB8')
k = Key(Enctype.RC4, kb)
self.assertEqual(encrypt(k, keyusage, plain, conf), ctxt)
self.assertEqual(decrypt(k, keyusage, ctxt), plain)
diff --git a/python/samba/tests/krb5/kdc_base_test.py b/python/samba/tests/krb5/kdc_base_test.py
index e835d389f1c..bef5458c881 100644
--- a/python/samba/tests/krb5/kdc_base_test.py
+++ b/python/samba/tests/krb5/kdc_base_test.py
@@ -374,8 +374,8 @@ class KDCBaseTest(RawKerberosTest):
account_name = (
pac.info.info.info3.base.account_name)
user_sid = (
- str(pac.info.info.info3.base.domain_sid) +
- "-" + str(pac.info.info.info3.base.rid))
+ str(pac.info.info.info3.base.domain_sid)
+ + "-" + str(pac.info.info.info3.base.rid))
elif pac.type == self.PAC_LOGON_NAME:
logon_name = pac.info.account_name
elif pac.type == self.PAC_UPN_DNS_INFO:
diff --git a/python/samba/tests/krb5/kdc_tests.py b/python/samba/tests/krb5/kdc_tests.py
index 17b9d154bd9..c7c53953a86 100755
--- a/python/samba/tests/krb5/kdc_tests.py
+++ b/python/samba/tests/krb5/kdc_tests.py
@@ -25,7 +25,20 @@ os.environ["PYTHONUNBUFFERED"] = "1"
from samba.tests.krb5.raw_testcase import RawKerberosTest
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_hexdump = False
@@ -83,7 +96,7 @@ class KdcTests(RawKerberosTest):
break
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])
diff --git a/python/samba/tests/krb5/raw_testcase.py b/python/samba/tests/krb5/raw_testcase.py
index e67f5464e59..82e68ee7019 100644
--- a/python/samba/tests/krb5/raw_testcase.py
+++ b/python/samba/tests/krb5/raw_testcase.py
@@ -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.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
# be 32-Bit on the wire
@@ -59,14 +62,17 @@ def BitStringEncoder_encodeValue32(self, value, asn1Spec, encodeFun, **options):
padding = 0
ret = b'\x00' + substrate + (b'\x00' * padding)
return ret, False, True
+
+
BitStringEncoder.encodeValue = BitStringEncoder_encodeValue32
+
def BitString_NamedValues_prettyPrint(self, scope=0):
ret = "%s" % self.asBinary()
bits = []
highest_bit = 32
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
if byte & mask:
val = 1
@@ -89,12 +95,21 @@ def BitString_NamedValues_prettyPrint(self, scope=0):
delim = ",\n%s " % indent
ret += "\n%s)" % indent
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.KDCOptions.prettyPrintNamedValues = krb5_asn1.KDCOptionsValues.namedValues
-krb5_asn1.KDCOptions.namedValues = krb5_asn1.KDCOptionsValues.namedValues
-krb5_asn1.KDCOptions.prettyPrint = BitString_NamedValues_prettyPrint
+
+
+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.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):
intval = int(self)
@@ -104,16 +119,29 @@ def Integer_NamedValues_prettyPrint(self, scope=0):
name = "<__unknown__>"
ret = "%d (0x%x) %s" % (intval, intval, name)
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.AuthDataType.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
+
+
+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.AuthDataType.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):
def __init__(self, key, kvno):
@@ -146,9 +174,10 @@ class Krb5EncryptionKey(object):
EncryptionKey_obj = {
'keytype': self.etype,
'keyvalue': self.key.contents,
- };
+ }
return EncryptionKey_obj
+
class RawKerberosTest(TestCase):
"""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.settimeout(10)
self.s.connect(self.a[0][4])
- except socket.error as e:
+ except socket.error:
self.s.close()
raise
- except IOError as e:
+ except IOError:
self.s.close()
raise
- except Exception as e:
+ except Exception:
raise
finally:
pass
@@ -219,8 +248,9 @@ class RawKerberosTest(TestCase):
domain = samba.tests.env_get_var_value('DOMAIN')
realm = samba.tests.env_get_var_value('REALM')
username = samba.tests.env_get_var_value('SERVICE_USERNAME')
- password = samba.tests.env_get_var_value('SERVICE_PASSWORD',
- allow_missing=allow_missing_password)
+ password = samba.tests.env_get_var_value(
+ 'SERVICE_PASSWORD',
+ allow_missing=allow_missing_password)
c.set_domain(domain)
c.set_realm(realm)
c.set_username(username)
@@ -246,21 +276,34 @@ class RawKerberosTest(TestCase):
if hexdump is None:
hexdump = self.do_hexdump
if hexdump:
- 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):
+ 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):
if asn1Spec is not None:
class_name = type(asn1Spec).__name__.split(':')[0]
else:
class_name = "<None-asn1Spec>"
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)
if native_encode:
obj = pyasn1_native_encode(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:
obj = pyasn1_native_decode(obj, asn1Spec=asn1Spec)
class_name = type(obj).__name__.split(':')[0]
@@ -273,7 +316,8 @@ class RawKerberosTest(TestCase):
def send_pdu(self, req, asn1_print=None, hexdump=None):
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))
req_pdu = header
req_pdu += k5_pdu
@@ -304,7 +348,7 @@ class RawKerberosTest(TestCase):
self._disconnect("recv_raw: EOF")
return None
self.hex_dump("recv_raw", rep_pdu, hexdump=hexdump)
- except socket.timeout as e:
+ except socket.timeout:
self.s.settimeout(10)
sys.stderr.write("recv_raw: TIMEOUT\n")
pass
@@ -322,7 +366,8 @@ class RawKerberosTest(TestCase):
rep_pdu = None
rep = None
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:
return (None, None)
header = struct.unpack(">I", raw_pdu[0:4])
@@ -332,22 +377,27 @@ class RawKerberosTest(TestCase):
missing = k5_len
rep_pdu = b''
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)
rep_pdu += raw_pdu
missing = k5_len - len(rep_pdu)
- k5_raw = self.der_decode(rep_pdu, asn1Spec=None, native_encode=False,
- asn1_print=False, hexdump=False)
- pvno=k5_raw['field-0']
+ k5_raw = self.der_decode(
+ rep_pdu,
+ asn1Spec=None,
+ native_encode=False,
+ asn1_print=False,
+ hexdump=False)
+ pvno = k5_raw['field-0']
self.assertEqual(pvno, 5)
- msg_type=k5_raw['field-1']
- self.assertIn(msg_type, [11,13,30])
+ msg_type = k5_raw['field-1']
+ self.assertIn(msg_type, [11, 13, 30])
if msg_type == 11:
- asn1Spec=krb5_asn1.AS_REP()
+ asn1Spec = krb5_asn1.AS_REP()
elif msg_type == 13:
- asn1Spec=krb5_asn1.TGS_REP()
+ asn1Spec = krb5_asn1.TGS_REP()
elif msg_type == 30:
- asn1Spec=krb5_asn1.KRB_ERROR()
+ asn1Spec = krb5_asn1.KRB_ERROR()
rep = self.der_decode(rep_pdu, asn1Spec=asn1Spec,
asn1_print=asn1_print, hexdump=False)
finally:
@@ -368,11 +418,17 @@ class RawKerberosTest(TestCase):
self.assertIsNone(self.s, msg="Is connected")
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()
try:
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:
self._disconnect("transaction failed")
raise
@@ -389,11 +445,15 @@ class RawKerberosTest(TestCase):
def assertPrincipalEqual(self, princ1, princ2):
self.assertEqual(princ1['name-type'], princ2['name-type'])
- self.assertEqual(len(princ1['name-string']), len(princ2['name-string']),
- msg="princ1=%s != princ2=%s" % (princ1, princ2))
+ self.assertEqual(
+ len(princ1['name-string']),
+ len(princ2['name-string']),
+ msg="princ1=%s != princ2=%s" % (princ1, princ2))
for idx in range(len(princ1['name-string'])):
- self.assertEqual(princ1['name-string'][idx], princ2['name-string'][idx],
- msg="princ1=%s != princ2=%s" % (princ1, princ2))
+ self.assertEqual(
+ princ1['name-string'][idx],
+ princ2['name-string'][idx],
+ msg="princ1=%s != princ2=%s" % (princ1, princ2))
return
def get_KerberosTimeWithUsec(self, epoch=None, offset=None):
@@ -421,7 +481,7 @@ class RawKerberosTest(TestCase):
salt = None
try:
salt = etype_info2['salt']
- except:
+ except Exception:
pass
if e == kcrypto.Enctype.RC4:
@@ -429,7 +489,8 @@ class RawKerberosTest(TestCase):
return self.SessionKey_create(etype=e, contents=nthash, kvno=kvno)
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):
e = kcrypto._get_enctype_profile(etype)
@@ -452,14 +513,14 @@ class RawKerberosTest(TestCase):
'cipher': ciphertext
}
if key.kvno is not None:
- EncryptedData_obj['kvno'] = key.kvno
+ EncryptedData_obj['kvno'] = key.kvno
return EncryptedData_obj
def Checksum_create(self, key, usage, plaintext, ctype=None):
- #Checksum ::= SEQUENCE {
+ # Checksum ::= SEQUENCE {
# cksumtype [0] Int32,
# checksum [1] OCTET STRING
- #}
+ # }
if ctype is None:
ctype = key.ctype
checksum = key.make_checksum(usage, plaintext, ctype=ctype)
@@ -494,10 +555,10 @@ class RawKerberosTest(TestCase):
return PA_DATA_obj
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
# pausec[1] krb5int32 OPTIONAL
- #}
+ # }
PA_ENC_TS_ENC_obj = {
'patimestamp': ts,
'pausec': usec,
@@ -520,7 +581,7 @@ class RawKerberosTest(TestCase):
additional_tickets,
asn1_print=None,
hexdump=None):
- #KDC-REQ-BODY ::= SEQUENCE {
+ # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --,
@@ -532,20 +593,23 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32,
- # etype [8] SEQUENCE OF Int32 -- EncryptionType
+ # etype [8] SEQUENCE OF Int32
+ # -- EncryptionType
# -- in preference order --,
# addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty
- #}
+ # }
if EncAuthorizationData is not None:
- enc_ad_plain = self.der_encode(EncAuthorizationData,
- asn1Spec=krb5_asn1.AuthorizationData(),
- asn1_print=asn1_print,
- hexdump=hexdump)
- enc_ad = self.EncryptedData_create(EncAuthorizationData_key, enc_ad_plain)
+ enc_ad_plain = self.der_encode(
+ EncAuthorizationData,
+ asn1Spec=krb5_asn1.AuthorizationData(),
+ asn1_print=asn1_print,
+ hexdump=hexdump)
+ enc_ad = self.EncryptedData_create(
+ EncAuthorizationData_key, enc_ad_plain)
else:
enc_ad = None
KDC_REQ_BODY_obj = {
@@ -590,14 +654,14 @@ class RawKerberosTest(TestCase):
asn1Spec=None,
asn1_print=None,
hexdump=None):
- #KDC-REQ ::= SEQUENCE {
+ # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY
- #}
+ # }
#
KDC_REQ_BODY_obj = self.KDC_REQ_BODY_create(kdc_options,
cname,
@@ -622,39 +686,40 @@ class RawKerberosTest(TestCase):
if padata is not None:
KDC_REQ_obj['padata'] = padata
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:
KDC_REQ_decoded = None
return KDC_REQ_obj, KDC_REQ_decoded
def AS_REQ_create(self,
- padata, # optional
- kdc_options, # required
- cname, # optional
- realm, # required
- sname, # optional
- from_time, # optional
- till_time, # required
- renew_time, # optional
- nonce, # required
- etypes, # required
- addresses, # optional
+ padata, # optional
+ kdc_options, # required
+ cname, # optional
+ realm, # required
+ sname, # optional
+ from_time, # optional
+ till_time, # required
+ renew_time, # optional
+ nonce, # required
+ etypes, # required
+ addresses, # optional
EncAuthorizationData,
EncAuthorizationData_key,
additional_tickets,
native_decoded_only=True,
asn1_print=None,
hexdump=None):
- #KDC-REQ ::= SEQUENCE {
+ # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY
- #}
+ # }
#
- #KDC-REQ-BODY ::= SEQUENCE {
+ # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --,
@@ -666,32 +731,34 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32,
- # etype [8] SEQUENCE OF Int32 -- EncryptionType
+ # etype [8] SEQUENCE OF Int32
+ # -- EncryptionType
# -- in preference order --,
# addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty
- #}
- obj,decoded = self.KDC_REQ_create(msg_type=10,
- padata=padata,
- kdc_options=kdc_options,
- cname=cname,
- realm=realm,
- sname=sname,
- from_time=from_time,
- till_time=till_time,
- renew_time=renew_time,
- nonce=nonce,
- etypes=etypes,
- addresses=addresses,
- EncAuthorizationData=EncAuthorizationData,
- EncAuthorizationData_key=EncAuthorizationData_key,
- additional_tickets=additional_tickets,
- asn1Spec=krb5_asn1.AS_REQ(),
- asn1_print=asn1_print,
- hexdump=hexdump)
+ # }
+ obj, decoded = self.KDC_REQ_create(
+ msg_type=10,
+ padata=padata,
+ kdc_options=kdc_options,
+ cname=cname,
+ realm=realm,
+ sname=sname,
+ from_time=from_time,
+ till_time=till_time,
+ renew_time=renew_time,
+ nonce=nonce,
+ etypes=etypes,
+ addresses=addresses,
+ EncAuthorizationData=EncAuthorizationData,
+ EncAuthorizationData_key=EncAuthorizationData_key,
+ additional_tickets=additional_tickets,
+ asn1Spec=krb5_asn1.AS_REQ(),
+ asn1_print=asn1_print,
+ hexdump=hexdump)
if native_decoded_only:
return decoded
return decoded, obj
@@ -703,7 +770,7 @@ class RawKerberosTest(TestCase):
# ap-options [2] APOptions,
# ticket [3] Ticket,
# authenticator [4] EncryptedData -- Authenticator
- #}
+ # }
AP_REQ_obj = {
'pvno': 5,
'msg-type': 14,
@@ -713,8 +780,9 @@ class RawKerberosTest(TestCase):
}
return AP_REQ_obj
- def Authenticator_create(self, crealm, cname, cksum, cusec, ctime, subkey, seq_number,
- authorization_data):
+ def Authenticator_create(
+ self, crealm, cname, cksum, cusec, ctime, subkey, seq_number,
+ authorization_data):
# -- Unencrypted authenticator
# Authenticator ::= [APPLICATION 2] SEQUENCE {
# authenticator-vno [0] INTEGER (5),
@@ -726,7 +794,7 @@ class RawKerberosTest(TestCase):
# subkey [6] EncryptionKey OPTIONAL,
# seq-number [7] UInt32 OPTIONAL,
# authorization-data [8] AuthorizationData OPTIONAL
- #}
+ # }
Authenticator_obj = {
'authenticator-vno': 5,
'crealm': crealm,
@@ -745,20 +813,20 @@ class RawKerberosTest(TestCase):
return Authenticator_obj
def TGS_REQ_create(self,
- padata, # optional
+ padata, # optional
cusec,
ctime,
ticket,
- kdc_options, # required
- cname, # optional
- realm, # required
- sname, # optional
- from_time, # optional
- till_time, # required
- renew_time, # optional
- nonce, # required
- etypes, # required
- addresses, # optional
+ kdc_options, # required
+ cname, # optional
+ realm, # required
+ sname, # optional
+ from_time, # optional
+ till_time, # required
+ renew_time, # optional
+ nonce, # required
+ etypes, # required
+ addresses, # optional
EncAuthorizationData,
EncAuthorizationData_key,
additional_tickets,
@@ -768,16 +836,16 @@ class RawKerberosTest(TestCase):
native_decoded_only=True,
asn1_print=None,
hexdump=None):
- #KDC-REQ ::= SEQUENCE {
+ # KDC-REQ ::= SEQUENCE {
# -- NOTE: first tag is [1], not [0]
# pvno [1] INTEGER (5) ,
# msg-type [2] INTEGER (10 -- AS -- | 12 -- TGS --),
# padata [3] SEQUENCE OF PA-DATA OPTIONAL
# -- NOTE: not empty --,
# req-body [4] KDC-REQ-BODY
- #}
+ # }
#
- #KDC-REQ-BODY ::= SEQUENCE {
+ # KDC-REQ-BODY ::= SEQUENCE {
# kdc-options [0] KDCOptions,
# cname [1] PrincipalName OPTIONAL
# -- Used only in AS-REQ --,
@@ -789,50 +857,57 @@ class RawKerberosTest(TestCase):
# till [5] KerberosTime,
# rtime [6] KerberosTime OPTIONAL,
# nonce [7] UInt32,
- # etype [8] SEQUENCE OF Int32 -- EncryptionType
+ # etype [8] SEQUENCE OF Int32
+ # -- EncryptionType
# -- in preference order --,
# addresses [9] HostAddresses OPTIONAL,
# enc-authorization-data [10] EncryptedData OPTIONAL
# -- AuthorizationData --,
# additional-tickets [11] SEQUENCE OF Ticket OPTIONAL
# -- NOTE: not empty
- #}
-
- req_body = self.KDC_REQ_BODY_create(kdc_options=kdc_options,
- cname=None,
- realm=realm,
- sname=sname,
- from_time=from_time,
- till_time=till_time,
- renew_time=renew_time,
- nonce=nonce,
- etypes=etypes,
- addresses=addresses,
- EncAuthorizationData=EncAuthorizationData,
- EncAuthorizationData_key=EncAuthorizationData_key,
- additional_tickets=additional_tickets)
+ # }
+
+ req_body = self.KDC_REQ_BODY_create(
+ kdc_options=kdc_options,
+ cname=None,
+ realm=realm,
+ sname=sname,
+ from_time=from_time,
+ till_time=till_time,
+ renew_time=renew_time,
+ nonce=nonce,
+ etypes=etypes,
+ addresses=addresses,
+ EncAuthorizationData=EncAuthorizationData,
+ EncAuthorizationData_key=EncAuthorizationData_key,
+ additional_tickets=additional_tickets)
req_body = self.der_encode(req_body, asn1Spec=krb5_asn1.KDC_REQ_BODY(),
asn1_print=asn1_print, hexdump=hexdump)
- req_body_checksum = self.Checksum_create(ticket_session_key, 6, req_body,
- ctype=body_checksum_type)
+ req_body_checksum = self.Checksum_create(
+ ticket_session_key, 6, req_body, ctype=body_checksum_type)
subkey_obj = None
if authenticator_subkey is not None:
subkey_obj = authenticator_subkey.export_obj()
seq_number = random.randint(0, 0xfffffffe)
- authenticator = self.Authenticator_create(crealm=realm,
- cname=cname,
- cksum=req_body_checksum,
- cusec=cusec,
- ctime=ctime,
- subkey=subkey_obj,
- seq_number=seq_number,
- authorization_data=None)
- 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.Authenticator_create(
+ crealm=realm,
+ cname=cname,
+ cksum=req_body_checksum,
+ cusec=cusec,
+ ctime=ctime,
+ subkey=subkey_obj,
+ seq_number=seq_number,
+ authorization_data=None)
+ authenticator = self.der_encode(
+ authenticator,
+ asn1Spec=krb5_asn1.Authenticator(),
+ asn1_print=asn1_print,
+ hexdump=hexdump)
+
+ authenticator = self.EncryptedData_create(
+ ticket_session_key, 7, authenticator)
ap_options = krb5_asn1.APOptions('0')
ap_req = self.AP_REQ_create(ap_options=str(ap_options),
@@ -846,24 +921,25 @@ class RawKerberosTest(TestCase):
else:
padata = [pa_tgs_req]
- obj,decoded = self.KDC_REQ_create(msg_type=12,
- padata=padata,
- kdc_options=kdc_options,
- cname=None,
- realm=realm,
- sname=sname,
- from_time=from_time,
- till_time=till_time,
- renew_time=renew_time,
- nonce=nonce,
- etypes=etypes,
- addresses=addresses,
- EncAuthorizationData=EncAuthorizationData,
- EncAuthorizationData_key=EncAuthorizationData_key,
- additional_tickets=additional_tickets,
- asn1Spec=krb5_asn1.TGS_REQ(),
- asn1_print=asn1_print,
- hexdump=hexdump)
+ obj, decoded = self.KDC_REQ_create(
+ msg_type=12,
+ padata=padata,
+ kdc_options=kdc_options,
+ cname=None,
+ realm=realm,
+ sname=sname,
+ from_time=from_time,
+ till_time=till_time,
+ renew_time=renew_time,
+ nonce=nonce,
+ etypes=etypes,
+ addresses=addresses,
+ EncAuthorizationData=EncAuthorizationData,
+ EncAuthorizationData_key=EncAuthorizationData_key,
+ additional_tickets=additional_tickets,
+ asn1Spec=krb5_asn1.TGS_REQ(),
+ asn1_print=asn1_print,
+ hexdump=hexdump)
if native_decoded_only:
return decoded
return decoded, obj
@@ -888,5 +964,6 @@ class RawKerberosTest(TestCase):
'cksum': cksum,
'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)
diff --git a/python/samba/tests/krb5/rfc4120_constants.py b/python/samba/tests/krb5/rfc4120_constants.py
index 9de56578c99..5bbf1229d09 100644
--- a/python/samba/tests/krb5/rfc4120_constants.py
+++ b/python/samba/tests/krb5/rfc4120_constants.py
@@ -38,31 +38,31 @@ PADATA_ETYPE_INFO2 = int(
# Error codes
KDC_ERR_C_PRINCIPAL_UNKNOWN = 6
-KDC_ERR_PREAUTH_FAILED = 24
-KDC_ERR_PREAUTH_REQUIRED = 25
-KDC_ERR_BADMATCH = 36
-KDC_ERR_SKEW = 37
+KDC_ERR_PREAUTH_FAILED = 24
+KDC_ERR_PREAUTH_REQUIRED = 25
+KDC_ERR_BADMATCH = 36
+KDC_ERR_SKEW = 37
# 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_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(
'kRB5-NT-ENTERPRISE-PRINCIPAL'))
# Authorization data ad-type values
-AD_IF_RELEVANT = 1
-AD_INTENDED_FOR_SERVER = 2
+AD_IF_RELEVANT = 1
+AD_INTENDED_FOR_SERVER = 2
AD_INTENDED_FOR_APPLICATION_CLASS = 3
-AD_KDC_ISSUED = 4
-AD_AND_OR = 5
-AD_MANDATORY_TICKET_EXTENSIONS = 6
-AD_IN_TICKET_EXTENSIONS = 7
-AD_MANDATORY_FOR_KDC = 8
-AD_INITIAL_VERIFIED_CAS = 9
-AD_WIN2K_PAC = 128
-AD_SIGNTICKET = 512
+AD_KDC_ISSUED = 4
+AD_AND_OR = 5
+AD_MANDATORY_TICKET_EXTENSIONS = 6
+AD_IN_TICKET_EXTENSIONS = 7
+AD_MANDATORY_FOR_KDC = 8
+AD_INITIAL_VERIFIED_CAS = 9
+AD_WIN2K_PAC = 128
+AD_SIGNTICKET = 512
# Key usage numbers
# RFC 4120 Section 7.5.1. Key Usage Numbers
diff --git a/python/samba/tests/krb5/s4u_tests.py b/python/samba/tests/krb5/s4u_tests.py
index 2e1bd3fbe1f..30a58d6345a 100755
--- a/python/samba/tests/krb5/s4u_tests.py
+++ b/python/samba/tests/krb5/s4u_tests.py
@@ -35,6 +35,7 @@ import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
global_asn1_print = False
global_hexdump = False
+
class S4UKerberosTests(RawKerberosTest):
def setUp(self):
@@ -55,7 +56,7 @@ class S4UKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None
- etypes=(18,17,23)
+ etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options),
@@ -76,14 +77,16 @@ class S4UKerberosTests(RawKerberosTest):
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())
+ 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())
+ etype_info2 = self.der_decode(
+ etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
key = self.PasswordKey_from_etype_info2(service_creds, etype_info2[0])
@@ -120,7 +123,8 @@ class S4UKerberosTests(RawKerberosTest):
self.assertEqual(msg_type, 11)
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
sname = cname
@@ -167,11 +171,13 @@ class S4UKerberosTests(RawKerberosTest):
if msg_type == 13:
enc_part2 = subkey.decrypt(
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
- # 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):
msg_type = self._test_s4u2self()
self.assertEqual(msg_type, 13)
@@ -193,6 +199,7 @@ class S4UKerberosTests(RawKerberosTest):
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
diff --git a/python/samba/tests/krb5/simple_tests.py b/python/samba/tests/krb5/simple_tests.py
index 6c090af3d46..889b91a9bf0 100755
--- a/python/samba/tests/krb5/simple_tests.py
+++ b/python/samba/tests/krb5/simple_tests.py
@@ -33,6 +33,7 @@ import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
global_asn1_print = False
global_hexdump = False
+
class SimpleKerberosTests(RawKerberosTest):
def setUp(self):
@@ -53,7 +54,7 @@ class SimpleKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None
- etypes=(18,17,23)
+ etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options),
@@ -74,14 +75,16 @@ class SimpleKerberosTests(RawKerberosTest):
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())
+ 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())
+ etype_info2 = self.der_decode(
+ etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
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'])
- # MIT KDC encodes both EncASRepPart and EncTGSRepPart with application tag 26
+ # MIT KDC encodes both EncASRepPart and EncTGSRepPart with
+ # application tag 26
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:
- enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
+ enc_part2 = self.der_decode(
+ enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
# TGS Request
service_creds = self.get_service_creds(allow_missing_password=True)
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')
till = self.get_KerberosTime(offset=36000)
ticket = rep['ticket']
@@ -167,7 +174,8 @@ class SimpleKerberosTests(RawKerberosTest):
enc_part2 = subkey.decrypt(
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
diff --git a/python/samba/tests/krb5/xrealm_tests.py b/python/samba/tests/krb5/xrealm_tests.py
index b4a02bff33a..efb953bdf7e 100755
--- a/python/samba/tests/krb5/xrealm_tests.py
+++ b/python/samba/tests/krb5/xrealm_tests.py
@@ -34,6 +34,7 @@ import samba.tests
global_asn1_print = False
global_hexdump = False
+
class XrealmKerberosTests(RawKerberosTest):
def setUp(self):
@@ -54,7 +55,7 @@ class XrealmKerberosTests(RawKerberosTest):
kdc_options = krb5_asn1.KDCOptions('forwardable')
padata = None
- etypes=(18,17,23)
+ etypes = (18, 17, 23)
req = self.AS_REQ_create(padata=padata,
kdc_options=str(kdc_options),
@@ -75,14 +76,16 @@ class XrealmKerberosTests(RawKerberosTest):
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())
+ 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())
+ etype_info2 = self.der_decode(
+ etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
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'])
- # MIT KDC encodes both EncASRepPart and EncTGSRepPart with application tag 26
+ # MIT KDC encodes both EncASRepPart and EncTGSRepPart with
+ # application tag 26
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:
- 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)
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')
till = self.get_KerberosTime(offset=36000)
@@ -167,10 +174,11 @@ class XrealmKerberosTests(RawKerberosTest):
enc_part2 = subkey.decrypt(
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
- 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])
return