python/samba/tests/krb5/s4u_tests.py

   1 #!/usr/bin/env python3
   2 # Unix SMB/CIFS implementation.
   3 # Copyright (C) Stefan Metzmacher 2020
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18
  19 import sys
  20 import os
  21
  22 sys.path.insert(0, "bin/python")
  23 os.environ["PYTHONUNBUFFERED"] = "1"
  24
  25 from samba.tests import env_get_var_value
  26 from samba.tests.krb5.kcrypto import Cksumtype
  27 from samba.tests.krb5.raw_testcase import RawKerberosTest
  28 from samba.tests.krb5.rfc4120_constants import (
  29     KU_PA_ENC_TIMESTAMP,
  30     KU_AS_REP_ENC_PART,
  31     KU_TGS_REP_ENC_PART_SUB_KEY,
  32 )
  33 import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
  34
  35 global_asn1_print = False
  36 global_hexdump = False
  37
  38 class S4UKerberosTests(RawKerberosTest):
  39
  40     def setUp(self):
  41         super(S4UKerberosTests, self).setUp()
  42         self.do_asn1_print = global_asn1_print
  43         self.do_hexdump = global_hexdump
  44
  45     def _test_s4u2self(self, pa_s4u2self_ctype=None):
  46         service_creds = self.get_service_creds()
  47         service = service_creds.get_username()
  48         realm = service_creds.get_realm()
  49
  50         cname = self.PrincipalName_create(name_type=1, names=[service])
  51         sname = self.PrincipalName_create(name_type=2, names=["krbtgt", realm])
  52
  53         till = self.get_KerberosTime(offset=36000)
  54
  55         kdc_options = krb5_asn1.KDCOptions('forwardable')
  56         padata = None
  57
  58         etypes=(18,17,23)
  59
  60         req = self.AS_REQ_create(padata=padata,
  61                                  kdc_options=str(kdc_options),
  62                                  cname=cname,
  63                                  realm=realm,
  64                                  sname=sname,
  65                                  from_time=None,
  66                                  till_time=till,
  67                                  renew_time=None,
  68                                  nonce=0x7fffffff,
  69                                  etypes=etypes,
  70                                  addresses=None,
  71                                  EncAuthorizationData=None,
  72                                  EncAuthorizationData_key=None,
  73                                  additional_tickets=None)
  74         rep = self.send_recv_transaction(req)
  75         self.assertIsNotNone(rep)
  76
  77         self.assertEqual(rep['msg-type'], 30)
  78         self.assertEqual(rep['error-code'], 25)
  79         rep_padata = self.der_decode(rep['e-data'], asn1Spec=krb5_asn1.METHOD_DATA())
  80
  81         for pa in rep_padata:
  82             if pa['padata-type'] == 19:
  83                 etype_info2 = pa['padata-value']
  84                 break
  85
  86         etype_info2 = self.der_decode(etype_info2, asn1Spec=krb5_asn1.ETYPE_INFO2())
  87
  88         key = self.PasswordKey_from_etype_info2(service_creds, etype_info2[0])
  89
  90         (patime, pausec) = self.get_KerberosTimeWithUsec()
  91         pa_ts = self.PA_ENC_TS_ENC_create(patime, pausec)
  92         pa_ts = self.der_encode(pa_ts, asn1Spec=krb5_asn1.PA_ENC_TS_ENC())
  93
  94         pa_ts = self.EncryptedData_create(key, KU_PA_ENC_TIMESTAMP, pa_ts)
  95         pa_ts = self.der_encode(pa_ts, asn1Spec=krb5_asn1.EncryptedData())
  96
  97         pa_ts = self.PA_DATA_create(2, pa_ts)
  98
  99         kdc_options = krb5_asn1.KDCOptions('forwardable')
 100         padata = [pa_ts]
 101
 102         req = self.AS_REQ_create(padata=padata,
 103                                  kdc_options=str(kdc_options),
 104                                  cname=cname,
 105                                  realm=realm,
 106                                  sname=sname,
 107                                  from_time=None,
 108                                  till_time=till,
 109                                  renew_time=None,
 110                                  nonce=0x7fffffff,
 111                                  etypes=etypes,
 112                                  addresses=None,
 113                                  EncAuthorizationData=None,
 114                                  EncAuthorizationData_key=None,
 115                                  additional_tickets=None)
 116         rep = self.send_recv_transaction(req)
 117         self.assertIsNotNone(rep)
 118
 119         msg_type = rep['msg-type']
 120         self.assertEqual(msg_type, 11)
 121
 122         enc_part2 = key.decrypt(KU_AS_REP_ENC_PART, rep['enc-part']['cipher'])
 123         enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncASRepPart())
 124
 125         # S4U2Self Request
 126         sname = cname
 127
 128         for_user_name = env_get_var_value('FOR_USER')
 129         uname = self.PrincipalName_create(name_type=1, names=[for_user_name])
 130
 131         kdc_options = krb5_asn1.KDCOptions('forwardable')
 132         till = self.get_KerberosTime(offset=36000)
 133         ticket = rep['ticket']
 134         ticket_session_key = self.EncryptionKey_import(enc_part2['key'])
 135         pa_s4u = self.PA_S4U2Self_create(name=uname, realm=realm,
 136                                          tgt_session_key=ticket_session_key,
 137                                          ctype=pa_s4u2self_ctype)
 138         padata = [pa_s4u]
 139
 140         subkey = self.RandomKey(ticket_session_key.etype)
 141
 142         (ctime, cusec) = self.get_KerberosTimeWithUsec()
 143
 144         req = self.TGS_REQ_create(padata=padata,
 145                                   cusec=cusec,
 146                                   ctime=ctime,
 147                                   ticket=ticket,
 148                                   kdc_options=str(kdc_options),
 149                                   cname=cname,
 150                                   realm=realm,
 151                                   sname=sname,
 152                                   from_time=None,
 153                                   till_time=till,
 154                                   renew_time=None,
 155                                   nonce=0x7ffffffe,
 156                                   etypes=etypes,
 157                                   addresses=None,
 158                                   EncAuthorizationData=None,
 159                                   EncAuthorizationData_key=None,
 160                                   additional_tickets=None,
 161                                   ticket_session_key=ticket_session_key,
 162                                   authenticator_subkey=subkey)
 163         rep = self.send_recv_transaction(req)
 164         self.assertIsNotNone(rep)
 165
 166         msg_type = rep['msg-type']
 167         if msg_type == 13:
 168             enc_part2 = subkey.decrypt(
 169                 KU_TGS_REP_ENC_PART_SUB_KEY, rep['enc-part']['cipher'])
 170             enc_part2 = self.der_decode(enc_part2, asn1Spec=krb5_asn1.EncTGSRepPart())
 171
 172         return msg_type
 173
 174     # Using the checksum type from the tgt_session_key happens to work everywhere
 175     def test_s4u2self(self):
 176         msg_type = self._test_s4u2self()
 177         self.assertEqual(msg_type, 13)
 178
 179     # Per spec, the checksum of PA-FOR-USER is HMAC_MD5, see [MS-SFU] 2.2.1
 180     def test_s4u2self_hmac_md5_checksum(self):
 181         msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.HMAC_MD5)
 182         self.assertEqual(msg_type, 13)
 183
 184     def test_s4u2self_md5_unkeyed_checksum(self):
 185         msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.MD5)
 186         self.assertEqual(msg_type, 30)
 187
 188     def test_s4u2self_sha1_unkeyed_checksum(self):
 189         msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.SHA1)
 190         self.assertEqual(msg_type, 30)
 191
 192     def test_s4u2self_crc32_unkeyed_checksum(self):
 193         msg_type = self._test_s4u2self(pa_s4u2self_ctype=Cksumtype.CRC32)
 194         self.assertEqual(msg_type, 30)
 195
 196 if __name__ == "__main__":
 197     global_asn1_print = True
 198     global_hexdump = True
 199     import unittest
 200     unittest.main()