python/samba/tests/krb5/netlogon.py

   1 #!/usr/bin/env python3
   2 # Unix SMB/CIFS implementation.
   3 # Copyright (C) Stefan Metzmacher 2024
   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 import (
  26     credentials,
  27     ntstatus,
  28     NTSTATUSError,
  29     generate_random_password,
  30     generate_random_bytes,
  31 )
  32 from samba.dcerpc import netlogon, samr, misc, ntlmssp, krb5pac
  33 from samba.ndr import ndr_deepcopy, ndr_print, ndr_pack
  34 from samba.crypto import md4_hash_blob
  35 from samba.tests import DynamicTestCase, env_get_var_value
  36 from samba.tests.krb5.kdc_base_test import KDCBaseTest
  37 import samba.tests.krb5.rfc4120_pyasn1 as krb5_asn1
  38 from samba.tests.krb5.rfc4120_constants import KU_NON_KERB_CKSUM_SALT
  39
  40 global_asn1_print = False
  41 global_ndr_print = False
  42 global_hexdump = False
  43
  44 @DynamicTestCase
  45 class NetlogonSchannel(KDCBaseTest):
  46
  47     @classmethod
  48     def setUpDynamicTestCases(cls):
  49         def setup_test(name, trust, authX, flags):
  50             fnname = "test_%s" % name
  51             tname = "%s_%s_%08x" % (trust, authX, flags)
  52             targs = (trust, authX, flags)
  53             cls.generate_dynamic_test(fnname, tname, *targs)
  54             return
  55
  56         tests = [
  57             "check_passwords",
  58             "send_to_sam",
  59             "network_samlogon",
  60             "interactive_samlogon",
  61             "generic_samlogon",
  62             "ticket_samlogon",
  63         ]
  64
  65         for test in tests:
  66             for trust in ["wks", "bdc"]:
  67                 for auth3_flags in [0x603fffff, 0x613fffff, 0xe13fffff]:
  68                     setup_test(test, trust, "auth3", auth3_flags)
  69                 for auth3_flags in [0x00004004, 0x00004000, 0x01000000]:
  70                     setup_test(test, trust, "auth3", auth3_flags)
  71                 for authK_flags in [0xe13fffff, 0x80000000, 0x00000000, 0x603fbffb]:
  72                     setup_test(test, trust, "authK", authK_flags)
  73                 for authK_flags in [0x01004004, 0x01000000, 0x00004000, 0x00000004]:
  74                     setup_test(test, trust, "authK", authK_flags)
  75                 for authK_flags in [0x613fffff, 0x413fffff, 0x400001ff]:
  76                     setup_test(test, trust, "authK", authK_flags)
  77
  78     def setUp(self):
  79         super().setUp()
  80         self.do_asn1_print = global_asn1_print
  81         self.do_ndr_print = global_ndr_print
  82         self.do_hexdump = global_hexdump
  83
  84         self.empty_pwd_nt4_hash = self.get_samr_Password(md4_hash_blob(b''))
  85
  86         strong_key_support = env_get_var_value(
  87             'NETLOGON_STRONG_KEY_SUPPORT',
  88             allow_missing=True)
  89         if strong_key_support is None:
  90             strong_key_support = '1'
  91         self.strong_key_support = bool(int(strong_key_support))
  92
  93         auth_krb5_support = env_get_var_value(
  94             'NETLOGON_AUTH_KRB5_SUPPORT',
  95             allow_missing=True)
  96         if auth_krb5_support is None:
  97             auth_krb5_support = '1'
  98         self.auth_krb5_support = bool(int(auth_krb5_support))
  99
 100         self.user_creds = self.get_cached_creds(
 101                 account_type=self.AccountType.USER,
 102                 opts={'name_prefix': 'u_',
 103                       'kerberos_enabled': False})
 104
 105         samdb = self.get_samdb()
 106         self.dc_server = samdb.host_dns_name()
 107
 108     def get_wks1_creds(self):
 109         return self.get_cached_creds(
 110                 account_type=self.AccountType.COMPUTER,
 111                 use_cache=False,
 112                 opts={'name_prefix': 'w1_',
 113                       'supported_enctypes': 0x18,
 114                       'secure_channel_type': misc.SEC_CHAN_WKSTA})
 115
 116     def get_bdc1_creds(self):
 117         return self.get_cached_creds(
 118                 account_type=self.AccountType.SERVER,
 119                 use_cache=False,
 120                 opts={'name_prefix': 'b1_',
 121                       'supported_enctypes': 0x18,
 122                       'secure_channel_type': misc.SEC_CHAN_BDC})
 123
 124     def get_anon_conn(self):
 125         dc_server = self.dc_server
 126         conn = netlogon.netlogon(f'ncacn_ip_tcp:{dc_server}',
 127                                  self.get_lp())
 128         return conn
 129
 130     def get_schannel_conn(self, trust_creds, ncreds):
 131         dc_server = self.dc_server
 132         trust_creds.set_netlogon_creds(ncreds)
 133         conn = netlogon.netlogon(f'ncacn_ip_tcp:{dc_server}[schannel,seal]',
 134                                  self.get_lp(),
 135                                  trust_creds)
 136         trust_creds.set_netlogon_creds(None)
 137         return conn
 138
 139     def get_krb5_conn(self, trust_creds):
 140         dc_server = self.dc_server
 141         conn = netlogon.netlogon(f'ncacn_ip_tcp:{dc_server}[krb5,seal]',
 142                                  self.get_lp(),
 143                                  trust_creds)
 144         return conn
 145
 146     def get_samr_Password(self, nt_hash):
 147         v = samr.Password()
 148         v.hash = list(nt_hash)
 149         return v
 150
 151     def get_netr_CryptPassword(self, utf8_pwd):
 152         print("utf8_pwd len=%d" % len(utf8_pwd))
 153         pwd = netlogon.netr_CryptPassword()
 154         utf16_pwd = utf8_pwd.encode('utf-16-le')
 155         pwd_len = len(utf16_pwd)
 156         print("utf16_pwd len=%d" % pwd_len)
 157         confounder_len = len(pwd.data) - pwd_len
 158         print("confounder_len len=%d" % confounder_len)
 159         confounder = generate_random_bytes(confounder_len)
 160         pwd.length = pwd_len
 161         pwd.data = list(confounder) + list(utf16_pwd)
 162         if self.do_ndr_print:
 163             print("get_netr_CryptPassword:\n%s" % ndr_print(pwd, print_secrets=True))
 164         return pwd
 165
 166     def is_domain_trust(self, ncreds):
 167         domain_types = [
 168             misc.SEC_CHAN_DOMAIN,
 169             misc.SEC_CHAN_DNS_DOMAIN
 170         ]
 171         if ncreds.secure_channel_type in domain_types:
 172             return True
 173         return False
 174
 175     def do_Authenticate3(self, conn, trust_creds,
 176             negotiate_flags, required_flags,
 177             expect_error=None):
 178         (auth_type, auth_level) = conn.auth_info()
 179
 180         trust_account_name = trust_creds.get_username()
 181         trust_computer_name = trust_creds.get_workstation()
 182
 183         client_challenge = credentials.netlogon_creds_random_challenge()
 184
 185         (server_challenge) = \
 186             conn.netr_ServerReqChallenge(self.dc_server,
 187                                          trust_computer_name,
 188                                          client_challenge)
 189
 190         nt_hash = trust_creds.get_nt_hash()
 191         machine_password = self.get_samr_Password(nt_hash)
 192
 193         (ncreds, initial_credential) = \
 194             credentials.netlogon_creds_client_init(
 195                 client_account=trust_account_name,
 196                 client_computer_name=trust_computer_name,
 197                 secure_channel_type=trust_creds.get_secure_channel_type(),
 198                 client_challenge=client_challenge,
 199                 server_challenge=server_challenge,
 200                 machine_password=machine_password,
 201                 client_requested_flags=negotiate_flags,
 202                 negotiate_flags=negotiate_flags)
 203
 204         try:
 205             (return_credentials, negotiated_flags, client_rid) = \
 206                 conn.netr_ServerAuthenticate3(self.dc_server,
 207                                               ncreds.account_name,
 208                                               ncreds.secure_channel_type,
 209                                               ncreds.computer_name,
 210                                               initial_credential,
 211                                               ncreds.client_requested_flags)
 212         except NTSTATUSError as err:
 213             status, _ = err.args
 214             self.assertIsNotNone(expect_error,
 215                                  f'unexpectedly failed with {status:08X}')
 216             self.assertEqual(expect_error, status, 'got wrong status code')
 217             return
 218
 219         self.assertIsNone(expect_error, 'expected error')
 220
 221         credentials.netlogon_creds_client_verify(ncreds,
 222                                                  return_credentials,
 223                                                  auth_type,
 224                                                  auth_level)
 225
 226         self.assertEqual((negotiated_flags & required_flags), required_flags)
 227         self.assertEqual((negotiated_flags & negotiate_flags), negotiated_flags)
 228         credentials.netlogon_creds_client_update(ncreds,
 229                                                  negotiated_flags,
 230                                                  client_rid)
 231
 232         return ncreds
 233
 234     def do_AuthenticateKerberos(self, conn, trust_creds,
 235             negotiate_flags, required_flags,
 236             expect_error=None):
 237         (auth_type, auth_level) = conn.auth_info()
 238
 239         trust_account_name = trust_creds.get_username()
 240         trust_computer_name = trust_creds.get_workstation()
 241
 242         ncreds = \
 243             credentials.netlogon_creds_kerberos_init(
 244                 client_account=trust_account_name,
 245                 client_computer_name=trust_computer_name,
 246                 secure_channel_type=trust_creds.get_secure_channel_type(),
 247                 client_requested_flags=negotiate_flags,
 248                 negotiate_flags=negotiate_flags)
 249
 250         try:
 251             (negotiated_flags, client_rid) = \
 252                 conn.netr_ServerAuthenticateKerberos(self.dc_server,
 253                                                      ncreds.account_name,
 254                                                      ncreds.secure_channel_type,
 255                                                      ncreds.computer_name,
 256                                                      ncreds.client_requested_flags)
 257         except NTSTATUSError as err:
 258             status, _ = err.args
 259             self.assertIsNotNone(expect_error,
 260                                  f'unexpectedly failed with {status:08X}')
 261             self.assertEqual(expect_error, status, 'got wrong status code')
 262             return
 263
 264         self.assertIsNone(expect_error, 'expected error')
 265
 266         self.assertEqual((negotiated_flags & required_flags), required_flags)
 267         self.assertEqual((negotiated_flags & negotiate_flags), negotiated_flags)
 268         credentials.netlogon_creds_client_update(ncreds,
 269                                                  negotiated_flags,
 270                                                  client_rid)
 271
 272         return ncreds
 273
 274     def do_CheckCapabilities(self, ncreds, conn,
 275             expect_error1=None,
 276             expect_error2=None):
 277
 278         (auth_type, auth_level) = conn.auth_info()
 279
 280         zero_authenticator = netlogon.netr_Authenticator()
 281
 282         req_authenticator = credentials.netlogon_creds_client_authenticator(ncreds)
 283         try:
 284            (rep_authenticator, server_capabilities) = \
 285                conn.netr_LogonGetCapabilities(self.dc_server,
 286                                               ncreds.computer_name,
 287                                               req_authenticator,
 288                                               zero_authenticator,
 289                                               1)
 290         except NTSTATUSError as err:
 291             status, _ = err.args
 292             self.assertIsNotNone(expect_error1,
 293                                  f'unexpectedly failed with {status:08X}')
 294             self.assertEqual(expect_error1, status, 'got wrong status code')
 295             return
 296
 297         self.assertIsNone(expect_error1, 'expected error')
 298
 299         credentials.netlogon_creds_client_verify(ncreds,
 300                                                  rep_authenticator.cred,
 301                                                  auth_type,
 302                                                  auth_level)
 303         self.assertEqual(server_capabilities, ncreds.negotiate_flags)
 304
 305         req_authenticator = credentials.netlogon_creds_client_authenticator(ncreds)
 306         try:
 307             (rep_authenticator, requested_flags) = \
 308                 conn.netr_LogonGetCapabilities(self.dc_server,
 309                                                ncreds.computer_name,
 310                                                req_authenticator,
 311                                                zero_authenticator,
 312                                                2)
 313         except NTSTATUSError as err:
 314             status, _ = err.args
 315             self.assertIsNotNone(expect_error2,
 316                                  f'unexpectedly failed with {status:08X}')
 317             self.assertEqual(expect_error2, status, 'got wrong status code')
 318             return
 319
 320         self.assertIsNone(expect_error2, 'expected error')
 321
 322         credentials.netlogon_creds_client_verify(ncreds,
 323                                                  rep_authenticator.cred,
 324                                                  auth_type,
 325                                                  auth_level)
 326         self.assertEqual(requested_flags, ncreds.client_requested_flags)
 327
 328         return
 329
 330     def do_ServerPasswordGet(self, ncreds, conn,
 331             expect_encrypted,
 332             expect_password,
 333             expect_broken_crypto=False,
 334             expect_error=None,
 335             req_ncreds=None,
 336             rep_ncreds=None,
 337             decryption_ncreds=None):
 338
 339         if req_ncreds is None:
 340             req_ncreds = ncreds
 341         if rep_ncreds is None:
 342             rep_ncreds = ncreds
 343         if decryption_ncreds is None:
 344             decryption_ncreds = ncreds
 345
 346         (auth_type, auth_level) = conn.auth_info()
 347
 348         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 349         try:
 350             (rep_authenticator, password) = \
 351                 conn.netr_ServerPasswordGet(self.dc_server,
 352                                             ncreds.account_name,
 353                                             ncreds.secure_channel_type,
 354                                             ncreds.computer_name,
 355                                             req_authenticator)
 356         except NTSTATUSError as err:
 357             status, _ = err.args
 358             self.assertIsNotNone(expect_error,
 359                                  f'unexpectedly failed with {status:08X}')
 360             self.assertEqual(expect_error, status, 'got wrong status code')
 361             return
 362
 363         self.assertIsNone(expect_error, 'expected error')
 364
 365         credentials.netlogon_creds_client_verify(rep_ncreds,
 366                                                  rep_authenticator.cred,
 367                                                  auth_type,
 368                                                  auth_level)
 369
 370         if self.do_ndr_print:
 371             print("do_ServerPasswordGet(raw_pwd):\n%s" % ndr_print(password, print_secrets=True))
 372         if expect_encrypted and expect_password:
 373             self.assertNotEqual(ndr_print(password, print_secrets=True), ndr_print(expect_password, print_secrets=True))
 374         elif expect_password:
 375             self.assertEqual(ndr_print(password, print_secrets=True), ndr_print(expect_password, print_secrets=True))
 376         if decryption_ncreds:
 377             credentials.netlogon_creds_decrypt_samr_Password(decryption_ncreds,
 378                                                              password,
 379                                                              auth_type,
 380                                                              auth_level)
 381         if self.do_ndr_print:
 382             print("do_ServerPasswordGet(pwd):\n%s" % ndr_print(password, print_secrets=True))
 383         if expect_broken_crypto and expect_password:
 384             self.assertNotEqual(ndr_print(password, print_secrets=True), ndr_print(expect_password, print_secrets=True))
 385         elif expect_password:
 386             self.assertEqual(ndr_print(password, print_secrets=True), ndr_print(expect_password, print_secrets=True))
 387         return
 388
 389     def do_ServerTrustPasswordsGet(self, ncreds, conn,
 390             expect_encrypted,
 391             expect_new_password,
 392             expect_old_password,
 393             expect_broken_crypto=False,
 394             expect_error=None,
 395             req_ncreds=None,
 396             rep_ncreds=None,
 397             decryption_ncreds=None):
 398
 399         if req_ncreds is None:
 400             req_ncreds = ncreds
 401         if rep_ncreds is None:
 402             rep_ncreds = ncreds
 403         if decryption_ncreds is None:
 404             decryption_ncreds = ncreds
 405
 406         (auth_type, auth_level) = conn.auth_info()
 407
 408         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 409         try:
 410             (rep_authenticator, new_password, old_password) = \
 411                 conn.netr_ServerTrustPasswordsGet(self.dc_server,
 412                                                   ncreds.account_name,
 413                                                   ncreds.secure_channel_type,
 414                                                   ncreds.computer_name,
 415                                                   req_authenticator)
 416         except NTSTATUSError as err:
 417             status, _ = err.args
 418             self.assertIsNotNone(expect_error,
 419                                  f'unexpectedly failed with {status:08X}')
 420             self.assertEqual(expect_error, status, 'got wrong status code')
 421             return (None, None)
 422
 423         self.assertIsNone(expect_error, 'expected error')
 424
 425         credentials.netlogon_creds_client_verify(rep_ncreds,
 426                                                  rep_authenticator.cred,
 427                                                  auth_type,
 428                                                  auth_level)
 429
 430         if self.do_ndr_print:
 431             print("do_ServerTrustPasswordsGet(raw_new):\n%s" % ndr_print(new_password, print_secrets=True))
 432             print("do_ServerTrustPasswordsGet(raw_old):\n%s" % ndr_print(old_password, print_secrets=True))
 433         if expect_encrypted and expect_new_password:
 434             self.assertNotEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 435         elif expect_new_password:
 436             self.assertEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 437         if expect_encrypted and expect_old_password:
 438             self.assertNotEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 439         elif expect_old_password:
 440             self.assertEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 441         if decryption_ncreds:
 442             credentials.netlogon_creds_decrypt_samr_Password(decryption_ncreds,
 443                                                              new_password,
 444                                                              auth_type,
 445                                                              auth_level)
 446             credentials.netlogon_creds_decrypt_samr_Password(decryption_ncreds,
 447                                                              old_password,
 448                                                              auth_type,
 449                                                              auth_level)
 450         if self.do_ndr_print:
 451             print("do_ServerTrustPasswordsGet(new):\n%s" % ndr_print(new_password, print_secrets=True))
 452             print("do_ServerTrustPasswordsGet(old):\n%s" % ndr_print(old_password, print_secrets=True))
 453         if expect_broken_crypto and expect_new_password:
 454             self.assertNotEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 455         elif expect_new_password:
 456             self.assertEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 457         if expect_broken_crypto and expect_old_password:
 458             self.assertNotEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 459         elif expect_old_password:
 460             self.assertEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 461         return
 462
 463     def do_ServerGetTrustInfo(self, ncreds, conn,
 464             expect_encrypted,
 465             expect_new_password,
 466             expect_old_password,
 467             expect_broken_crypto=False,
 468             expect_error=None,
 469             req_ncreds=None,
 470             rep_ncreds=None,
 471             decryption_ncreds=None):
 472
 473         if req_ncreds is None:
 474             req_ncreds = ncreds
 475         if rep_ncreds is None:
 476             rep_ncreds = ncreds
 477         if decryption_ncreds is None:
 478             decryption_ncreds = ncreds
 479
 480         (auth_type, auth_level) = conn.auth_info()
 481
 482         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 483         try:
 484             (rep_authenticator, new_password, old_password, trust_info) = \
 485                 conn.netr_ServerGetTrustInfo(self.dc_server,
 486                                              ncreds.account_name,
 487                                              ncreds.secure_channel_type,
 488                                              ncreds.computer_name,
 489                                              req_authenticator)
 490         except NTSTATUSError as err:
 491             status, _ = err.args
 492             self.assertIsNotNone(expect_error,
 493                                  f'unexpectedly failed with {status:08X}')
 494             self.assertEqual(expect_error, status, 'got wrong status code')
 495             return (None, None, None)
 496
 497         self.assertIsNone(expect_error, 'expected error')
 498
 499         credentials.netlogon_creds_client_verify(rep_ncreds,
 500                                                  rep_authenticator.cred,
 501                                                  auth_type,
 502                                                  auth_level)
 503
 504         if self.do_ndr_print:
 505             print("do_ServerGetTrustInfo(raw_new):\n%s" % ndr_print(new_password, print_secrets=True))
 506             print("do_ServerGetTrustInfo(raw_old):\n%s" % ndr_print(old_password, print_secrets=True))
 507         if expect_encrypted and expect_new_password:
 508             self.assertNotEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 509         elif expect_new_password:
 510             self.assertEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 511         if expect_encrypted and expect_old_password:
 512             self.assertNotEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 513         elif expect_old_password:
 514             self.assertEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 515         if decryption_ncreds:
 516             credentials.netlogon_creds_decrypt_samr_Password(decryption_ncreds,
 517                                                              new_password,
 518                                                              auth_type,
 519                                                              auth_level)
 520             credentials.netlogon_creds_decrypt_samr_Password(decryption_ncreds,
 521                                                              old_password,
 522                                                              auth_type,
 523                                                              auth_level)
 524         if self.do_ndr_print:
 525             print("do_ServerGetTrustInfo(new):\n%s" % ndr_print(new_password, print_secrets=True))
 526             print("do_ServerGetTrustInfo(old):\n%s" % ndr_print(old_password, print_secrets=True))
 527         if expect_broken_crypto and expect_new_password:
 528             self.assertNotEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 529         elif expect_new_password:
 530             self.assertEqual(ndr_print(new_password, print_secrets=True), ndr_print(expect_new_password, print_secrets=True))
 531         if expect_broken_crypto and expect_old_password:
 532             self.assertNotEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 533         elif expect_old_password:
 534             self.assertEqual(ndr_print(old_password, print_secrets=True), ndr_print(expect_old_password, print_secrets=True))
 535         return (new_password, old_password, trust_info)
 536
 537     def do_ServerPasswordSet(self, ncreds, conn,
 538             expect_encrypted,
 539             new_password,
 540             expect_error=None,
 541             req_ncreds=None,
 542             rep_ncreds=None,
 543             encryption_ncreds=None):
 544
 545         if req_ncreds is None:
 546             req_ncreds = ncreds
 547         if rep_ncreds is None:
 548             rep_ncreds = ncreds
 549         if encryption_ncreds is None:
 550             encryption_ncreds = ncreds
 551
 552         (auth_type, auth_level) = conn.auth_info()
 553
 554         if self.do_ndr_print:
 555             print("do_ServerPasswordSet(new):\n%s" % ndr_print(new_password, print_secrets=True))
 556         send_password = ndr_deepcopy(new_password)
 557         if encryption_ncreds:
 558             credentials.netlogon_creds_encrypt_samr_Password(encryption_ncreds,
 559                                                              send_password,
 560                                                              auth_type,
 561                                                              auth_level)
 562         if self.do_ndr_print:
 563             print("do_ServerPasswordSet(send_new):\n%s" % ndr_print(send_password, print_secrets=True))
 564         if expect_encrypted:
 565             self.assertNotEqual(ndr_print(send_password, print_secrets=True), ndr_print(new_password, print_secrets=True))
 566
 567         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 568         try:
 569             rep_authenticator = \
 570                 conn.netr_ServerPasswordSet(self.dc_server,
 571                                             ncreds.account_name,
 572                                             ncreds.secure_channel_type,
 573                                             ncreds.computer_name,
 574                                             req_authenticator,
 575                                             send_password)
 576         except NTSTATUSError as err:
 577             status, _ = err.args
 578             self.assertIsNotNone(expect_error,
 579                                  f'unexpectedly failed with {status:08X}')
 580             self.assertEqual(expect_error, status, 'got wrong status code')
 581             return
 582
 583         self.assertIsNone(expect_error, 'expected error')
 584
 585         credentials.netlogon_creds_client_verify(rep_ncreds,
 586                                                  rep_authenticator.cred,
 587                                                  auth_type,
 588                                                  auth_level)
 589
 590         return
 591
 592     def do_ServerPasswordSet2(self, ncreds, conn,
 593             expect_encrypted,
 594             new_password,
 595             expect_error=None,
 596             req_ncreds=None,
 597             rep_ncreds=None,
 598             encryption_ncreds=None):
 599
 600         if req_ncreds is None:
 601             req_ncreds = ncreds
 602         if rep_ncreds is None:
 603             rep_ncreds = ncreds
 604         if encryption_ncreds is None:
 605             encryption_ncreds = ncreds
 606
 607         (auth_type, auth_level) = conn.auth_info()
 608
 609         if self.do_ndr_print:
 610             print("do_ServerPasswordSet2(new):\n%s" % ndr_print(new_password, print_secrets=True))
 611         send_password = ndr_deepcopy(new_password)
 612         if encryption_ncreds:
 613             credentials.netlogon_creds_encrypt_netr_CryptPassword(encryption_ncreds,
 614                                                                   send_password,
 615                                                                   auth_type,
 616                                                                   auth_level)
 617         if self.do_ndr_print:
 618             print("do_ServerPasswordSet2(send_new):\n%s" % ndr_print(send_password, print_secrets=True))
 619         if expect_encrypted:
 620             self.assertNotEqual(ndr_print(send_password, print_secrets=True), ndr_print(new_password, print_secrets=True))
 621
 622         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 623         try:
 624             rep_authenticator = \
 625                 conn.netr_ServerPasswordSet2(self.dc_server,
 626                                             ncreds.account_name,
 627                                             ncreds.secure_channel_type,
 628                                             ncreds.computer_name,
 629                                             req_authenticator,
 630                                             send_password)
 631         except NTSTATUSError as err:
 632             status, _ = err.args
 633             self.assertIsNotNone(expect_error,
 634                                  f'unexpectedly failed with {status:08X}')
 635             self.assertEqual(expect_error, status, 'got wrong status code')
 636             return
 637
 638         self.assertIsNone(expect_error, 'expected error')
 639
 640         credentials.netlogon_creds_client_verify(rep_ncreds,
 641                                                  rep_authenticator.cred,
 642                                                  auth_type,
 643                                                  auth_level)
 644
 645         return
 646
 647     def do_SendToSam(self, ncreds, conn,
 648             opaque_buffer,
 649             expect_send_encrypted,
 650             expect_error=None,
 651             opaque_ncreds=None,
 652             req_ncreds=None,
 653             rep_ncreds=None):
 654
 655         if opaque_ncreds is None:
 656             opaque_ncreds = ncreds
 657         if req_ncreds is None:
 658             req_ncreds = ncreds
 659         if rep_ncreds is None:
 660             rep_ncreds = ncreds
 661
 662         (auth_type, auth_level) = conn.auth_info()
 663
 664         opaque_buffer = bytes(list(opaque_buffer))
 665         if self.do_ndr_print:
 666             print("do_SendToSam(opaque_buffer):\n%r" % opaque_buffer)
 667         send_opaque_buffer = bytes(list(opaque_buffer))
 668         if opaque_ncreds:
 669             credentials.netlogon_creds_encrypt_SendToSam(opaque_ncreds,
 670                                                          send_opaque_buffer,
 671                                                          auth_type,
 672                                                          auth_level)
 673         if self.do_ndr_print:
 674             print("do_SendToSam(send_opaque_buffer):\n%r" % send_opaque_buffer)
 675         if expect_send_encrypted:
 676             self.assertNotEqual(send_opaque_buffer, opaque_buffer)
 677         else:
 678             self.assertEqual(send_opaque_buffer, opaque_buffer)
 679
 680         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 681         try:
 682             (rep_authenticator) = \
 683                 conn.netr_NetrLogonSendToSam(self.dc_server,
 684                                              ncreds.computer_name,
 685                                              req_authenticator,
 686                                              list(send_opaque_buffer))
 687         except NTSTATUSError as err:
 688             status, _ = err.args
 689             self.assertIsNotNone(expect_error,
 690                                  f'unexpectedly failed with {status:08X}')
 691             self.assertEqual(expect_error, status, 'got wrong status code')
 692             return
 693
 694         self.assertIsNone(expect_error, 'expected error')
 695
 696         credentials.netlogon_creds_client_verify(rep_ncreds,
 697                                                  rep_authenticator.cred,
 698                                                  auth_type,
 699                                                  auth_level)
 700         return
 701
 702     def _prepare_samlogon(self, ncreds, conn, logon_type, user_creds,
 703                           generic_package=None, generic_data=None,
 704                           trust_creds=None,
 705                           encrypt=False):
 706
 707         username, domain = user_creds.get_ntlm_username_domain()
 708         workstation = ncreds.computer_name
 709
 710         identity_info = netlogon.netr_IdentityInfo()
 711         identity_info.domain_name.string = domain
 712         identity_info.account_name.string = username
 713         identity_info.parameter_control = (
 714             netlogon.MSV1_0_ALLOW_SERVER_TRUST_ACCOUNT) | (
 715                 netlogon.MSV1_0_ALLOW_WORKSTATION_TRUST_ACCOUNT)
 716         identity_info.workstation.string = workstation
 717
 718         (auth_type, auth_level) = conn.auth_info()
 719
 720         if logon_type == netlogon.NetlogonInteractiveInformation:
 721             logon = netlogon.netr_PasswordInfo()
 722
 723             lm_pass = samr.Password()
 724             lm_pass.hash = [0] * 16
 725
 726             nt_pass = samr.Password()
 727             nt_pass.hash = list(user_creds.get_nt_hash())
 728
 729             logon.lmpassword = lm_pass
 730             logon.ntpassword = nt_pass
 731
 732         elif logon_type == netlogon.NetlogonNetworkInformation:
 733             computername = ntlmssp.AV_PAIR()
 734             computername.AvId = ntlmssp.MsvAvNbComputerName
 735             computername.Value = workstation
 736
 737             domainname = ntlmssp.AV_PAIR()
 738             domainname.AvId = ntlmssp.MsvAvNbDomainName
 739             domainname.Value = domain
 740
 741             eol = ntlmssp.AV_PAIR()
 742             eol.AvId = ntlmssp.MsvAvEOL
 743
 744             target_info = ntlmssp.AV_PAIR_LIST()
 745             target_info.count = 3
 746             target_info.pair = [domainname, computername, eol]
 747
 748             target_info_blob = ndr_pack(target_info)
 749
 750             challenge = b'fixedval'
 751             response = user_creds.get_ntlm_response(flags=0,
 752                                                     challenge=challenge,
 753                                                     target_info=target_info_blob)
 754
 755             logon = netlogon.netr_NetworkInfo()
 756
 757             logon.challenge = list(challenge)
 758             logon.nt = netlogon.netr_ChallengeResponse()
 759             logon.nt.length = len(response['nt_response'])
 760             logon.nt.data = list(response['nt_response'])
 761
 762         elif logon_type == netlogon.NetlogonGenericInformation:
 763             self.assertIsNotNone(generic_package)
 764             self.assertIsNotNone(generic_data)
 765
 766             logon = netlogon.netr_GenericInfo()
 767             logon.package_name.string = generic_package
 768             logon.length = len(generic_data)
 769             logon.data = list(generic_data)
 770
 771             identity_info = netlogon.netr_IdentityInfo()
 772             identity_info.domain_name.string = domain
 773
 774         elif logon_type == netlogon.NetlogonTicketLogonInformation:
 775             user_tgt = self.get_tgt(user_creds)
 776
 777             service_ticket = self.get_service_ticket(user_tgt, trust_creds)
 778             service_ticket_data = self.der_encode(service_ticket.ticket,
 779                                                   asn1Spec=krb5_asn1.Ticket())
 780
 781             logon = netlogon.netr_TicketLogonInfo()
 782             logon.request_options = 0
 783             logon.service_ticket_length = len(service_ticket_data)
 784             logon.service_ticket = list(service_ticket_data)
 785             logon.additional_ticket_length = 0
 786             logon.additional_ticket = None
 787
 788             identity_info = netlogon.netr_IdentityInfo()
 789             identity_info.domain_name.string = domain
 790         else:
 791             self.fail(f'unknown logon type {logon_type}')
 792
 793         logon.identity_info = identity_info
 794
 795         if encrypt:
 796             credentials.netlogon_creds_encrypt_netr_LogonLevel(ncreds,
 797                                                                logon_type,
 798                                                                logon,
 799                                                                auth_type,
 800                                                                auth_level)
 801
 802         return logon
 803
 804     def do_LogonWithFlags(self, ncreds, conn,
 805             logon_type, logon_info, validation_level,
 806             expect_send_encrypted,
 807             expect_recv_encrypted,
 808             expect_error=None,
 809             logon_ncreds=None,
 810             req_ncreds=None,
 811             rep_ncreds=None,
 812             validation_ncreds=None):
 813
 814         if logon_ncreds is None:
 815             logon_ncreds = ncreds
 816         if req_ncreds is None:
 817             req_ncreds = ncreds
 818         if rep_ncreds is None:
 819             rep_ncreds = ncreds
 820         if validation_ncreds is None:
 821             validation_ncreds = ncreds
 822
 823         (auth_type, auth_level) = conn.auth_info()
 824
 825         logon_info = ndr_deepcopy(logon_info)
 826         if self.do_ndr_print:
 827             print("do_LogonWithFlags(logon):\n%s" % ndr_print(logon_info, print_secrets=True))
 828         send_logon_info = ndr_deepcopy(logon_info)
 829         if logon_ncreds:
 830             credentials.netlogon_creds_encrypt_netr_LogonLevel(logon_ncreds,
 831                                                                logon_type,
 832                                                                send_logon_info,
 833                                                                auth_type,
 834                                                                auth_level)
 835         if self.do_ndr_print:
 836             print("do_LogonWithFlags(send_logon):\n%s" % ndr_print(send_logon_info, print_secrets=True))
 837         if expect_send_encrypted:
 838             self.assertNotEqual(ndr_print(send_logon_info, print_secrets=True), ndr_print(logon_info, print_secrets=True))
 839         else:
 840             self.assertEqual(ndr_print(send_logon_info, print_secrets=True), ndr_print(logon_info, print_secrets=True))
 841
 842         zero_authenticator = netlogon.netr_Authenticator()
 843
 844         req_flags = 0
 845         req_authenticator = credentials.netlogon_creds_client_authenticator(req_ncreds)
 846         try:
 847             (rep_authenticator, recv_validation, authoritative, rep_flags) = \
 848                 conn.netr_LogonSamLogonWithFlags(self.dc_server,
 849                                                  ncreds.computer_name,
 850                                                  req_authenticator,
 851                                                  zero_authenticator,
 852                                                  logon_type,
 853                                                  send_logon_info,
 854                                                  validation_level,
 855                                                  req_flags)
 856         except NTSTATUSError as err:
 857             status, _ = err.args
 858             self.assertIsNotNone(expect_error,
 859                                  f'unexpectedly failed with {status:08X}')
 860             self.assertEqual(expect_error, status, 'got wrong status code')
 861             return None
 862
 863         self.assertIsNone(expect_error, 'expected error')
 864
 865         self.assertEqual(1, authoritative)
 866         self.assertEqual(0, rep_flags)
 867
 868         credentials.netlogon_creds_client_verify(rep_ncreds,
 869                                                  rep_authenticator.cred,
 870                                                  auth_type,
 871                                                  auth_level)
 872
 873         recv_validation = ndr_deepcopy(recv_validation)
 874         if self.do_ndr_print:
 875             print("do_LogonWithFlags(%d,recv):\n%s" % (logon_type, ndr_print(recv_validation, print_secrets=True)))
 876         validation = ndr_deepcopy(recv_validation)
 877         if validation_ncreds:
 878             credentials.netlogon_creds_decrypt_netr_Validation(validation_ncreds,
 879                                                                validation_level,
 880                                                                validation,
 881                                                                auth_type,
 882                                                                auth_level)
 883         if self.do_ndr_print:
 884             print("do_LogonWithFlags(%d):\n%s" % (logon_type, ndr_print(validation, print_secrets=True)))
 885         if expect_recv_encrypted:
 886             self.assertNotEqual(ndr_print(validation, print_secrets=True), ndr_print(recv_validation, print_secrets=True))
 887         else:
 888             self.assertEqual(ndr_print(validation, print_secrets=True), ndr_print(recv_validation, print_secrets=True))
 889         return validation
 890
 891     def do_LogonEx(self, ncreds, conn,
 892             logon_type, logon_info, validation_level,
 893             expect_send_encrypted,
 894             expect_recv_encrypted,
 895             expect_error=None,
 896             logon_ncreds=None,
 897             validation_ncreds=None):
 898
 899         if logon_ncreds is None:
 900             logon_ncreds = ncreds
 901         if validation_ncreds is None:
 902             validation_ncreds = ncreds
 903
 904         (auth_type, auth_level) = conn.auth_info()
 905
 906         logon_info = ndr_deepcopy(logon_info)
 907         if self.do_ndr_print:
 908             print("do_LogonEx(logon):\n%s" % ndr_print(logon_info, print_secrets=True))
 909         send_logon_info = ndr_deepcopy(logon_info)
 910         if logon_ncreds:
 911             credentials.netlogon_creds_encrypt_netr_LogonLevel(logon_ncreds,
 912                                                                logon_type,
 913                                                                send_logon_info,
 914                                                                auth_type,
 915                                                                auth_level)
 916         if self.do_ndr_print:
 917             print("do_LogonEx(send_logon):\n%s" % ndr_print(send_logon_info, print_secrets=True))
 918         if expect_send_encrypted:
 919             self.assertNotEqual(ndr_print(send_logon_info, print_secrets=True), ndr_print(logon_info, print_secrets=True))
 920         else:
 921             self.assertEqual(ndr_print(send_logon_info, print_secrets=True), ndr_print(logon_info, print_secrets=True))
 922
 923         req_flags = 0
 924         try:
 925             (recv_validation, authoritative, rep_flags) = \
 926                 conn.netr_LogonSamLogonEx(self.dc_server,
 927                                           ncreds.computer_name,
 928                                           logon_type,
 929                                           send_logon_info,
 930                                           validation_level,
 931                                           req_flags)
 932         except NTSTATUSError as err:
 933             status, _ = err.args
 934             self.assertIsNotNone(expect_error,
 935                                  f'unexpectedly failed with {status:08X}')
 936             self.assertEqual(expect_error, status, 'got wrong status code')
 937             return None
 938
 939         self.assertIsNone(expect_error, 'expected error')
 940
 941         self.assertEqual(1, authoritative)
 942         self.assertEqual(0, rep_flags)
 943
 944         recv_validation = ndr_deepcopy(recv_validation)
 945         if self.do_ndr_print:
 946             print("do_LogonEx(%d,recv):\n%s" % (logon_type, ndr_print(recv_validation, print_secrets=True)))
 947         validation = ndr_deepcopy(recv_validation)
 948         if validation_ncreds:
 949             credentials.netlogon_creds_decrypt_netr_Validation(validation_ncreds,
 950                                                                validation_level,
 951                                                                validation,
 952                                                                auth_type,
 953                                                                auth_level)
 954         if self.do_ndr_print:
 955             print("do_LogonEx(%d):\n%s" % (logon_type, ndr_print(validation, print_secrets=True)))
 956         if expect_recv_encrypted:
 957             self.assertNotEqual(ndr_print(validation, print_secrets=True), ndr_print(recv_validation, print_secrets=True))
 958         else:
 959             self.assertEqual(ndr_print(validation, print_secrets=True), ndr_print(recv_validation, print_secrets=True))
 960         return validation
 961
 962     def _prepare_ncreds_conn_with_args(self, trust, authX, flags):
 963         if trust == "wks":
 964             creds = self.get_wks1_creds()
 965         elif trust == "bdc":
 966             creds = self.get_bdc1_creds()
 967         self.assertIsNotNone(creds)
 968
 969         proposed_flags = flags
 970         required_flags = flags
 971
 972         if authX == "auth3":
 973             anon_conn = self.get_anon_conn()
 974             expect_error = None
 975             if flags & 0x80000000 and not self.auth_krb5_support:
 976                 required_flags &= ~0x80000000
 977             if not (flags & 0x01000000) and not self.strong_key_support:
 978                 expect_error = ntstatus.NT_STATUS_DOWNGRADE_DETECTED
 979             ncreds = self.do_Authenticate3(anon_conn, creds,
 980                                            proposed_flags,
 981                                            required_flags,
 982                                            expect_error=expect_error)
 983             if expect_error is not None:
 984                 self.skipTest('Requires NETLOGON_STRONG_KEY_SUPPORT')
 985                 return (None, None, None, None)
 986             if proposed_flags != required_flags:
 987                 self.assertEqual(ncreds.negotiate_flags, required_flags)
 988                 self.skipTest('Requires NETLOGON_AUTH_KRB5_SUPPORT')
 989                 return (None, None, None, None)
 990             conn = self.get_schannel_conn(creds, ncreds)
 991         elif authX == "authK":
 992             conn = self.get_krb5_conn(creds)
 993             expect_error = None
 994             if not self.auth_krb5_support:
 995                 expect_error = ntstatus.NT_STATUS_RPC_PROCNUM_OUT_OF_RANGE
 996             ncreds = self.do_AuthenticateKerberos(conn, creds,
 997                                                   proposed_flags,
 998                                                   required_flags,
 999                                                   expect_error=expect_error)
1000             if expect_error is not None:
1001                 self.skipTest('Requires NETLOGON_AUTH_KRB5_SUPPORT')
1002                 return (None, None, None, None)
1003         self.assertIsNotNone(ncreds)
1004         self.assertIsNotNone(conn)
1005
1006         if ncreds.authenticate_kerberos:
1007             expect_encrypted = False
1008         elif flags & 0x80000000:
1009             expect_encrypted = False
1010         else:
1011             expect_encrypted = True
1012
1013         return (creds, ncreds, conn, expect_encrypted)
1014
1015     def _test_check_passwords(self, trust_creds, ncreds, conn,
1016                               expect_encrypted):
1017         self.do_CheckCapabilities(ncreds, conn)
1018
1019         nt_hash = trust_creds.get_nt_hash()
1020         expect_new_password = self.get_samr_Password(nt_hash)
1021         old_nt_hash = None
1022         if self.is_domain_trust(ncreds):
1023             old_nt_hash = trust_creds.get_old_hash()
1024         if old_nt_hash:
1025             expect_old_password = self.get_samr_Password(old_nt_hash)
1026         else:
1027             expect_old_password = self.empty_pwd_nt4_hash
1028
1029         expect_broken_crypto = False
1030         expect_broken_set2_crypto = False
1031         if ncreds.authenticate_kerberos:
1032             self.assertEqual(expect_encrypted, False)
1033             if ncreds.negotiate_flags & 0x80000000:
1034                 # This is the expected case of a sane client
1035                 pass
1036             elif ncreds.negotiate_flags & 0x01000004:
1037                 # This fails as there is aes or arfour
1038                 # encryption with a random key
1039                 expect_broken_crypto = True
1040                 expect_broken_set2_crypto = True
1041             else:
1042                 # This fails as there is des
1043                 # encryption with a random key
1044                 # This applies to things using
1045                 # the NT-HASH (samr_Password)
1046                 #
1047                 # But for {samr,netr}_CryptPassword
1048                 # there's no encryption for
1049                 # ServerPasswordSet2
1050                 expect_broken_crypto = True
1051
1052         if expect_broken_crypto:
1053             expect_encrypted = True
1054             expect_set_encrypted = False
1055             encryption_set_ncreds = False
1056             expect_set2_encrypted = False
1057             encryption_set2_ncreds = False
1058         else:
1059             expect_set_encrypted = expect_encrypted
1060             encryption_set_ncreds = ncreds
1061             expect_set2_encrypted = expect_encrypted
1062             encryption_set2_ncreds = ncreds
1063
1064         if not (ncreds.negotiate_flags & 0x01000004):
1065             # Without aes or arcfour this uses no encryption
1066             expect_set2_encrypted = False
1067
1068         if ncreds.secure_channel_type == misc.SEC_CHAN_WKSTA:
1069             expect_get_error = ntstatus.NT_STATUS_ACCESS_DENIED
1070         else:
1071             expect_get_error = None
1072         self.do_ServerPasswordGet(ncreds, conn,
1073                                   expect_encrypted,
1074                                   expect_new_password,
1075                                   expect_broken_crypto=expect_broken_crypto,
1076                                   expect_error=expect_get_error)
1077         self.do_ServerTrustPasswordsGet(ncreds, conn,
1078                                         expect_encrypted,
1079                                         expect_new_password,
1080                                         expect_old_password,
1081                                         expect_broken_crypto=expect_broken_crypto)
1082         self.do_ServerGetTrustInfo(ncreds, conn,
1083                                    expect_encrypted,
1084                                    expect_new_password,
1085                                    expect_old_password,
1086                                    expect_broken_crypto=expect_broken_crypto)
1087
1088         if self.is_domain_trust(ncreds):
1089             self.do_CheckCapabilities(ncreds, conn)
1090             return
1091
1092         if expect_encrypted:
1093             old_utf8 = trust_creds.get_password()
1094             new_utf8 = generate_random_password(120, 120)
1095             tmp_creds = credentials.Credentials()
1096             tmp_creds.set_password(new_utf8)
1097             tmp_nt_hash = tmp_creds.get_nt_hash()
1098             new_set_password = self.get_samr_Password(tmp_nt_hash)
1099             expect_set_error = None
1100         else:
1101             old_utf8 = None
1102             new_utf8 = None
1103             tmp_nt_hash = generate_random_bytes(16)
1104             new_set_password = self.get_samr_Password(tmp_nt_hash)
1105             expect_set_error = ntstatus.NT_STATUS_NOT_SUPPORTED
1106         self.do_ServerPasswordSet(ncreds, conn,
1107                                   expect_set_encrypted,
1108                                   new_set_password,
1109                                   encryption_ncreds=encryption_set_ncreds,
1110                                   expect_error=expect_set_error)
1111         if expect_broken_crypto and expect_set_error is None:
1112             #
1113             # This is tricky!
1114             #
1115             # As the encryption works with a random key,
1116             # ServerPasswordSet and ServerGetTrustInfo
1117             # both use the same key to decrypt
1118             # -> store -> retrieve -> encrypt
1119             # As a result we get back the same value we passed
1120             # to ServerPasswordSet.
1121             #
1122             #
1123             self.do_ServerGetTrustInfo(ncreds, conn,
1124                                        False, # expect_encrypted
1125                                        new_set_password, # expect_new_password
1126                                        None,  # expect_old_password
1127                                        decryption_ncreds=False)
1128             # For the old password we're not able to
1129             # decrypt it...
1130             self.do_ServerGetTrustInfo(ncreds, conn,
1131                                        expect_encrypted,
1132                                        None, #expect_new_password,
1133                                        expect_old_password,
1134                                        expect_broken_crypto=expect_broken_crypto)
1135             self.do_CheckCapabilities(ncreds, conn)
1136             # We re-negotiate the flags with
1137             # NETLOGON_NEG_SUPPORTS_KERBEROS_AUTH, so that
1138             # we can get the value the server stored.
1139             orig_flags = ncreds.negotiate_flags
1140             krb5_flags = orig_flags | 0x80000000
1141             ncreds = self.do_AuthenticateKerberos(conn, trust_creds, krb5_flags, krb5_flags)
1142             self.do_CheckCapabilities(ncreds, conn)
1143             # The value store should not be the one
1144             # we passed to ServerPasswordSet...
1145             self.do_ServerGetTrustInfo(ncreds, conn,
1146                                        True, # expect_encrypted
1147                                        new_set_password, # expect_new_password
1148                                        None,  # expect_old_password
1149                                        expect_broken_crypto=expect_broken_crypto)
1150             # We get the old one fixed self.empty_pwd_nt4_hash now
1151             self.do_ServerGetTrustInfo(ncreds, conn,
1152                                        False, # expect_encrypted
1153                                        None, # expect_new_password,
1154                                        expect_old_password)
1155             # Now we reset the password using ServerPasswordSet2
1156             # in order to do useful testing below...
1157             fix_set2_password = self.get_netr_CryptPassword(new_utf8)
1158             self.do_ServerPasswordSet2(ncreds, conn,
1159                                        False, # expect__encrypted
1160                                        fix_set2_password)
1161             self.do_ServerGetTrustInfo(ncreds, conn,
1162                                        False, # expect_encrypted
1163                                        new_set_password, # expect_new_password
1164                                        expect_old_password)
1165             self.do_CheckCapabilities(ncreds, conn)
1166             #
1167             # Now we test with the original flags again
1168             krb5_flags = orig_flags
1169             ncreds = self.do_AuthenticateKerberos(conn, trust_creds, krb5_flags, krb5_flags)
1170             self.do_CheckCapabilities(ncreds, conn)
1171
1172         if old_utf8:
1173             trust_creds.set_old_password(old_utf8)
1174         if new_utf8:
1175             trust_creds.set_password(new_utf8)
1176             tmp_nt_hash = trust_creds.get_nt_hash()
1177             expect_new_password = self.get_samr_Password(tmp_nt_hash)
1178
1179         self.do_ServerPasswordGet(ncreds, conn,
1180                                   expect_encrypted,
1181                                   expect_new_password,
1182                                   expect_broken_crypto=expect_broken_crypto,
1183                                   expect_error=expect_get_error)
1184         self.do_ServerTrustPasswordsGet(ncreds, conn,
1185                                         expect_encrypted,
1186                                         expect_new_password,
1187                                         expect_old_password,
1188                                         expect_broken_crypto=expect_broken_crypto)
1189         self.do_ServerGetTrustInfo(ncreds, conn,
1190                                    expect_encrypted,
1191                                    expect_new_password,
1192                                    expect_old_password,
1193                                    expect_broken_crypto=expect_broken_crypto)
1194
1195         if expect_broken_set2_crypto:
1196             old_utf8 = None
1197             new_utf8 = None
1198             tmp_utf8 = generate_random_password(120, 120)
1199             new_set2_password = self.get_netr_CryptPassword(tmp_utf8)
1200             expect_set2_error = ntstatus.NT_STATUS_ACCESS_DENIED
1201         else:
1202             old_utf8 = trust_creds.get_password()
1203             new_utf8 = generate_random_password(120, 120)
1204             new_set2_password = self.get_netr_CryptPassword(new_utf8)
1205             expect_set2_error = None
1206         self.do_ServerPasswordSet2(ncreds, conn,
1207                                    expect_set2_encrypted,
1208                                    new_set2_password,
1209                                    encryption_ncreds=encryption_set2_ncreds,
1210                                    expect_error=expect_set2_error)
1211         if old_utf8:
1212             trust_creds.set_old_password(old_utf8)
1213         if new_utf8:
1214             trust_creds.set_password(new_utf8)
1215             tmp_nt_hash = trust_creds.get_nt_hash()
1216             expect_new_password = self.get_samr_Password(tmp_nt_hash)
1217
1218         self.do_ServerPasswordGet(ncreds, conn,
1219                                   expect_encrypted,
1220                                   expect_new_password,
1221                                   expect_broken_crypto=expect_broken_crypto,
1222                                   expect_error=expect_get_error)
1223         self.do_ServerTrustPasswordsGet(ncreds, conn,
1224                                         expect_encrypted,
1225                                         expect_new_password,
1226                                         expect_old_password,
1227                                         expect_broken_crypto=expect_broken_crypto)
1228         self.do_ServerGetTrustInfo(ncreds, conn,
1229                                    expect_encrypted,
1230                                    expect_new_password,
1231                                    expect_old_password,
1232                                    expect_broken_crypto=expect_broken_crypto)
1233
1234         if expect_broken_crypto and not expect_broken_set2_crypto:
1235             #
1236             # This is tricky!
1237             #
1238             # ServerPasswordSet2 isn't affected by
1239             # broken crypto, so we can get
1240             # back the nthashes related to
1241             # the unencrypted plaintext password
1242             # set passed to ServerPasswordSet2
1243             orig_flags = ncreds.negotiate_flags
1244             krb5_flags = orig_flags | 0x80000000
1245             ncreds = self.do_AuthenticateKerberos(conn, trust_creds, krb5_flags, krb5_flags)
1246             self.do_CheckCapabilities(ncreds, conn)
1247             self.do_ServerPasswordGet(ncreds, conn,
1248                                       False, # expect_encrypted
1249                                       expect_new_password,
1250                                       expect_error=expect_get_error)
1251             self.do_ServerTrustPasswordsGet(ncreds, conn,
1252                                             False, # expect_encrypted
1253                                             expect_new_password,
1254                                             expect_old_password)
1255             self.do_ServerGetTrustInfo(ncreds, conn,
1256                                        False, # expect_encrypted
1257                                        expect_new_password,
1258                                        expect_old_password)
1259
1260         self.do_CheckCapabilities(ncreds, conn)
1261         return
1262
1263     def _test_check_passwords_with_args(self, trust, authX, flags):
1264         (creds, ncreds, conn, expect_encrypted) = \
1265             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1266
1267         if conn is None:
1268             return
1269
1270         return self._test_check_passwords(creds,
1271                                           ncreds,
1272                                           conn,
1273                                           expect_encrypted)
1274
1275     def _test_send_to_sam(self, trust_creds, ncreds, conn,
1276                           expect_encrypted):
1277         self.do_CheckCapabilities(ncreds, conn)
1278
1279         expect_broken_crypto = False
1280         if ncreds.authenticate_kerberos:
1281             self.assertEqual(expect_encrypted, False)
1282             if ncreds.negotiate_flags & 0x80000000:
1283                 # This is the expected case of a sane client
1284                 pass
1285             elif ncreds.negotiate_flags & 0x01000004:
1286                 # This fails as there is aes or arfour
1287                 # encryption with a random key
1288                 expect_broken_crypto = True
1289                 expect_encrypted = True
1290             else:
1291                 # There's no encryption with des
1292                 pass
1293
1294         if not (ncreds.negotiate_flags & 0x01000004):
1295             # Without aes or arcfour this uses no encryption
1296             expect_encrypted = False
1297
1298         opaque_buffer = b'invalid_opaque_buffer'
1299         if ncreds.secure_channel_type == misc.SEC_CHAN_WKSTA:
1300             expect_invalid_error = ntstatus.NT_STATUS_ACCESS_DENIED
1301         else:
1302             expect_invalid_error = ntstatus.NT_STATUS_INVALID_PARAMETER
1303         self.do_SendToSam(ncreds, conn, opaque_buffer,
1304                           expect_encrypted,
1305                           expect_error=expect_invalid_error)
1306
1307         rbpc = netlogon.netr_SendToSamResetBadPasswordCount()
1308         rbpc.guid = misc.GUID("00000001-0001-0001-0001-000000000001")
1309         bmsg = netlogon.netr_SendToSamBase()
1310         bmsg.message_type = netlogon.SendToSamResetBadPasswordCount
1311         bmsg.message_size = 16
1312         bmsg.message = rbpc
1313         opaque_buffer = ndr_pack(bmsg)
1314         if ncreds.secure_channel_type == misc.SEC_CHAN_WKSTA:
1315             expect_not_found_error = ntstatus.NT_STATUS_ACCESS_DENIED
1316         elif expect_broken_crypto:
1317             expect_not_found_error = ntstatus.NT_STATUS_INVALID_PARAMETER
1318         else:
1319             expect_not_found_error = ntstatus.NT_STATUS_OBJECT_NAME_NOT_FOUND
1320         self.do_SendToSam(ncreds, conn, opaque_buffer,
1321                           expect_encrypted,
1322                           expect_error=expect_not_found_error)
1323
1324         rbpc = netlogon.netr_SendToSamResetBadPasswordCount()
1325         rbpc.guid = misc.GUID(self.user_creds.get_guid())
1326         bmsg = netlogon.netr_SendToSamBase()
1327         bmsg.message_type = netlogon.SendToSamResetBadPasswordCount
1328         bmsg.message_size = 16
1329         bmsg.message = rbpc
1330         opaque_buffer = ndr_pack(bmsg)
1331         if ncreds.secure_channel_type == misc.SEC_CHAN_WKSTA:
1332             expect_no_error = ntstatus.NT_STATUS_ACCESS_DENIED
1333         elif expect_broken_crypto:
1334             expect_no_error = ntstatus.NT_STATUS_INVALID_PARAMETER
1335         else:
1336             expect_no_error = None
1337         self.do_SendToSam(ncreds, conn, opaque_buffer,
1338                           expect_encrypted,
1339                           expect_error=expect_no_error)
1340
1341         self.do_CheckCapabilities(ncreds, conn)
1342         return
1343
1344     def _test_send_to_sam_with_args(self, trust, authX, flags):
1345         (creds, ncreds, conn, expect_encrypted) = \
1346             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1347
1348         if conn is None:
1349             return
1350
1351         return self._test_send_to_sam(creds,
1352                                       ncreds,
1353                                       conn,
1354                                       expect_encrypted)
1355
1356     def _test_network_samlogon(self, trust_creds, ncreds, conn,
1357                                expect_encrypted):
1358         self.do_CheckCapabilities(ncreds, conn)
1359
1360         expect_broken_nt_crypto = False
1361         expect_broken_lm_crypto = False
1362         if ncreds.authenticate_kerberos:
1363             self.assertEqual(expect_encrypted, False)
1364             if ncreds.negotiate_flags & 0x80000000:
1365                 # This is the expected case of a sane client
1366                 pass
1367             elif ncreds.negotiate_flags & 0x01000004:
1368                 # This fails as there is aes or arfour
1369                 # encryption with a random key
1370                 expect_broken_nt_crypto = True
1371                 expect_broken_lm_crypto = True
1372                 expect_encrypted = True
1373             else:
1374                 # This fails as there is des
1375                 # encryption with a random key
1376                 # but it only encrypts the LMSessKey
1377                 expect_encrypted = True
1378                 expect_broken_lm_crypto = True
1379
1380         validation_level6 = netlogon.NetlogonValidationSamInfo4
1381         validation_level3 = netlogon.NetlogonValidationSamInfo2
1382         validation_level2 = netlogon.NetlogonValidationSamInfo
1383         logon_type_n = netlogon.NetlogonNetworkInformation
1384         logon_info_n = self._prepare_samlogon(ncreds,
1385                                               conn,
1386                                               logon_type_n,
1387                                               self.user_creds)
1388
1389         expect_send_encrypted = False
1390         expect_recv_encrypted = False
1391         validationRef_n6 = self.do_LogonEx(ncreds, conn,
1392                                            logon_type_n, logon_info_n,
1393                                            validation_level6,
1394                                            expect_send_encrypted,
1395                                            expect_recv_encrypted)
1396         self.assertNotEqual(validationRef_n6.base.rid, 0)
1397         self.assertNotEqual(validationRef_n6.base.key.key, list(b'\x00' *16))
1398         self.assertEqual(validationRef_n6.base.LMSessKey.key, list(b'\x00' *8))
1399
1400         expect_send_encrypted = False
1401         expect_recv_encrypted = expect_encrypted
1402         validationWF_n2 = self.do_LogonWithFlags(ncreds, conn,
1403                                                  logon_type_n, logon_info_n,
1404                                                  validation_level2,
1405                                                  expect_send_encrypted,
1406                                                  expect_recv_encrypted)
1407         self.assertEqual(validationWF_n2.base.rid, validationRef_n6.base.rid)
1408         if expect_broken_nt_crypto:
1409             self.assertNotEqual(validationWF_n2.base.key.key, list(b'\x00' *16))
1410             self.assertNotEqual(validationWF_n2.base.key.key, validationRef_n6.base.key.key)
1411         else:
1412             self.assertEqual(validationWF_n2.base.key.key, validationRef_n6.base.key.key)
1413         if expect_broken_lm_crypto:
1414             self.assertNotEqual(validationWF_n2.base.LMSessKey.key, list(b'\x00' *8))
1415             self.assertNotEqual(validationWF_n2.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1416         else:
1417             self.assertEqual(validationWF_n2.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1418         validationEx_n2 = self.do_LogonEx(ncreds, conn,
1419                                           logon_type_n, logon_info_n,
1420                                           validation_level2,
1421                                           expect_send_encrypted,
1422                                           expect_recv_encrypted)
1423         self.assertEqual(validationEx_n2.base.rid, validationRef_n6.base.rid)
1424         if expect_broken_nt_crypto:
1425             self.assertNotEqual(validationEx_n2.base.key.key, list(b'\x00' *16))
1426             self.assertNotEqual(validationEx_n2.base.key.key, validationRef_n6.base.key.key)
1427         else:
1428             self.assertEqual(validationEx_n2.base.key.key, validationRef_n6.base.key.key)
1429         if expect_broken_lm_crypto:
1430             self.assertNotEqual(validationEx_n2.base.LMSessKey.key, list(b'\x00' *8))
1431             self.assertNotEqual(validationEx_n2.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1432         else:
1433             self.assertEqual(validationEx_n2.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1434
1435         expect_send_encrypted = False
1436         expect_recv_encrypted = expect_encrypted
1437         validationWF_n3 = self.do_LogonWithFlags(ncreds, conn,
1438                                                  logon_type_n, logon_info_n,
1439                                                  validation_level3,
1440                                                  expect_send_encrypted,
1441                                                  expect_recv_encrypted)
1442         self.assertEqual(validationWF_n3.base.rid, validationRef_n6.base.rid)
1443         if expect_broken_nt_crypto:
1444             self.assertNotEqual(validationWF_n3.base.key.key, list(b'\x00' *16))
1445             self.assertNotEqual(validationWF_n3.base.key.key, validationRef_n6.base.key.key)
1446         else:
1447             self.assertEqual(validationWF_n3.base.key.key, validationRef_n6.base.key.key)
1448         if expect_broken_lm_crypto:
1449             self.assertNotEqual(validationWF_n3.base.LMSessKey.key, list(b'\x00' *8))
1450             self.assertNotEqual(validationWF_n3.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1451         else:
1452             self.assertEqual(validationWF_n3.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1453         validationEx_n3 = self.do_LogonEx(ncreds, conn,
1454                                           logon_type_n, logon_info_n,
1455                                           validation_level3,
1456                                           expect_send_encrypted,
1457                                           expect_recv_encrypted)
1458         self.assertEqual(validationEx_n3.base.rid, validationRef_n6.base.rid)
1459         if expect_broken_nt_crypto:
1460             self.assertNotEqual(validationEx_n3.base.key.key, list(b'\x00' *16))
1461             self.assertNotEqual(validationEx_n3.base.key.key, validationRef_n6.base.key.key)
1462         else:
1463             self.assertEqual(validationEx_n3.base.key.key, validationRef_n6.base.key.key)
1464         if expect_broken_lm_crypto:
1465             self.assertNotEqual(validationEx_n3.base.LMSessKey.key, list(b'\x00' *8))
1466             self.assertNotEqual(validationEx_n3.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1467         else:
1468             self.assertEqual(validationEx_n3.base.LMSessKey.key, validationRef_n6.base.LMSessKey.key)
1469
1470         expect_send_encrypted = False
1471         expect_recv_encrypted = False
1472         validationWF_n6 = self.do_LogonWithFlags(ncreds, conn,
1473                                                  logon_type_n, logon_info_n,
1474                                                  validation_level6,
1475                                                  expect_send_encrypted,
1476                                                  expect_recv_encrypted)
1477         self.assertEqual(validationWF_n6.base.rid, validationRef_n6.base.rid)
1478         self.assertEqual(validationWF_n6.base.key.key, validationRef_n6.base.key.key)
1479         validationEx_n6 = self.do_LogonEx(ncreds, conn,
1480                                           logon_type_n, logon_info_n,
1481                                           validation_level6,
1482                                           expect_send_encrypted,
1483                                           expect_recv_encrypted)
1484         self.assertEqual(validationEx_n6.base.rid, validationRef_n6.base.rid)
1485         self.assertEqual(validationEx_n6.base.key.key, validationRef_n6.base.key.key)
1486
1487         self.do_CheckCapabilities(ncreds, conn)
1488         return
1489
1490     def _test_network_samlogon_with_args(self, trust, authX, flags):
1491         (creds, ncreds, conn, expect_encrypted) = \
1492             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1493
1494         if conn is None:
1495             return
1496
1497         return self._test_network_samlogon(creds,
1498                                            ncreds,
1499                                            conn,
1500                                            expect_encrypted)
1501
1502     def _test_interactive_samlogon(self, trust_creds, ncreds, conn,
1503                                    expect_encrypted):
1504         self.do_CheckCapabilities(ncreds, conn)
1505
1506         validation_level6 = netlogon.NetlogonValidationSamInfo4
1507         validation_level3 = netlogon.NetlogonValidationSamInfo2
1508         validation_level2 = netlogon.NetlogonValidationSamInfo
1509         logon_type_i = netlogon.NetlogonInteractiveInformation
1510         logon_info_i = self._prepare_samlogon(ncreds,
1511                                               conn,
1512                                               logon_type_i,
1513                                               self.user_creds)
1514
1515         expect_broken_crypto = False
1516         if ncreds.authenticate_kerberos:
1517             self.assertEqual(expect_encrypted, False)
1518             if ncreds.negotiate_flags & 0x80000000:
1519                 # This is the expected case of a sane client
1520                 pass
1521             else:
1522                 # This fails as there is aes, arcfour or des
1523                 # encryption with a random key
1524                 expect_broken_crypto = True
1525                 expect_encrypted = True
1526
1527         if expect_broken_crypto:
1528             expect_error = ntstatus.NT_STATUS_WRONG_PASSWORD
1529         else:
1530             expect_error = None
1531
1532         expect_send_encrypted = expect_encrypted
1533         expect_recv_encrypted = False
1534         validationRef_i6 = self.do_LogonEx(ncreds, conn,
1535                                            logon_type_i, logon_info_i,
1536                                            validation_level6,
1537                                            expect_send_encrypted,
1538                                            expect_recv_encrypted,
1539                                            expect_error=expect_error)
1540         if expect_error is not None:
1541             self.do_CheckCapabilities(ncreds, conn)
1542             return
1543         self.assertNotEqual(validationRef_i6.base.rid, 0)
1544         self.assertEqual(validationRef_i6.base.key.key, list(b'\x00' *16))
1545         self.assertEqual(validationRef_i6.base.LMSessKey.key, list(b'\x00' *8))
1546
1547         expect_send_encrypted = expect_encrypted
1548         expect_recv_encrypted = False
1549         validationWF_i2 = self.do_LogonWithFlags(ncreds, conn,
1550                                                  logon_type_i, logon_info_i,
1551                                                  validation_level2,
1552                                                  expect_send_encrypted,
1553                                                  expect_recv_encrypted)
1554         self.assertEqual(validationWF_i2.base.rid, validationRef_i6.base.rid)
1555         self.assertEqual(validationWF_i2.base.key.key, validationRef_i6.base.key.key)
1556         self.assertEqual(validationWF_i2.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1557         validationEx_i2 = self.do_LogonEx(ncreds, conn,
1558                                           logon_type_i, logon_info_i,
1559                                           validation_level2,
1560                                           expect_send_encrypted,
1561                                           expect_recv_encrypted)
1562         self.assertEqual(validationEx_i2.base.rid, validationRef_i6.base.rid)
1563         self.assertEqual(validationEx_i2.base.key.key, validationRef_i6.base.key.key)
1564         self.assertEqual(validationEx_i2.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1565
1566         expect_send_encrypted = expect_encrypted
1567         expect_recv_encrypted = False
1568         validationWF_i3 = self.do_LogonWithFlags(ncreds, conn,
1569                                                  logon_type_i, logon_info_i,
1570                                                  validation_level3,
1571                                                  expect_send_encrypted,
1572                                                  expect_recv_encrypted)
1573         self.assertEqual(validationWF_i3.base.rid, validationRef_i6.base.rid)
1574         self.assertEqual(validationWF_i3.base.key.key, validationRef_i6.base.key.key)
1575         self.assertEqual(validationWF_i3.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1576         validationEx_i3 = self.do_LogonEx(ncreds, conn,
1577                                           logon_type_i, logon_info_i,
1578                                           validation_level3,
1579                                           expect_send_encrypted,
1580                                           expect_recv_encrypted)
1581         self.assertEqual(validationEx_i3.base.rid, validationRef_i6.base.rid)
1582         self.assertEqual(validationEx_i3.base.key.key, validationRef_i6.base.key.key)
1583         self.assertEqual(validationEx_i3.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1584
1585         expect_send_encrypted = expect_encrypted
1586         expect_recv_encrypted = False
1587         validationWF_i6 = self.do_LogonWithFlags(ncreds, conn,
1588                                                  logon_type_i, logon_info_i,
1589                                                  validation_level6,
1590                                                  expect_send_encrypted,
1591                                                  expect_recv_encrypted)
1592         self.assertEqual(validationWF_i6.base.rid, validationRef_i6.base.rid)
1593         self.assertEqual(validationWF_i6.base.key.key, validationRef_i6.base.key.key)
1594         self.assertEqual(validationWF_i6.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1595         validationEx_i6 = self.do_LogonEx(ncreds, conn,
1596                                           logon_type_i, logon_info_i,
1597                                           validation_level6,
1598                                           expect_send_encrypted,
1599                                           expect_recv_encrypted)
1600         self.assertEqual(validationEx_i6.base.rid, validationRef_i6.base.rid)
1601         self.assertEqual(validationEx_i6.base.key.key, validationRef_i6.base.key.key)
1602         self.assertEqual(validationEx_i6.base.LMSessKey.key, validationRef_i6.base.LMSessKey.key)
1603
1604         self.do_CheckCapabilities(ncreds, conn)
1605         return
1606
1607     def _test_interactive_samlogon_with_args(self, trust, authX, flags):
1608         (creds, ncreds, conn, expect_encrypted) = \
1609             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1610
1611         if conn is None:
1612             return
1613
1614         return self._test_interactive_samlogon(creds,
1615                                                ncreds,
1616                                                conn,
1617                                                expect_encrypted)
1618
1619     def _test_generic_samlogon(self, trust_creds, ncreds, conn,
1620                                expect_encrypted):
1621         self.do_CheckCapabilities(ncreds, conn)
1622
1623         expect_broken_crypto = False
1624         if ncreds.authenticate_kerberos:
1625             self.assertEqual(expect_encrypted, False)
1626             if ncreds.negotiate_flags & 0x80000000:
1627                 # This is the expected case of a sane client
1628                 pass
1629             elif ncreds.negotiate_flags & 0x01000004:
1630                 # This fails as there is aes or arfour
1631                 # encryption with a random key
1632                 expect_broken_crypto = True
1633                 expect_encrypted = True
1634             else:
1635                 # There's no aes nor arcfour, so no encryption
1636                 pass
1637
1638         if expect_broken_crypto:
1639             expect_error = ntstatus.NT_STATUS_INVALID_PARAMETER
1640         else:
1641             expect_error = None
1642
1643         if not (ncreds.negotiate_flags & 0x01000004):
1644             # Without aes or arcfour this uses no encryption
1645             expect_encrypted = False
1646
1647         krbtgt_creds = self.get_krbtgt_creds()
1648         krbtgt_key = self.TicketDecryptionKey_from_creds(krbtgt_creds)
1649
1650         cks = krbtgt_key.make_checksum(KU_NON_KERB_CKSUM_SALT, b'')
1651         sig = krbtgt_key.make_checksum(KU_NON_KERB_CKSUM_SALT, cks)
1652
1653         pv = krb5pac.PAC_Validate()
1654         pv.ChecksumLength = len(cks)
1655         pv.SignatureType = krbtgt_key.ctype
1656         pv.SignatureLength = len(sig)
1657         pv.ChecksumAndSignature = cks + sig
1658         pv_blob = ndr_pack(pv)
1659
1660         validation_level = netlogon.NetlogonValidationGenericInfo2
1661         logon_type = netlogon.NetlogonGenericInformation
1662         logon_info_pv = self._prepare_samlogon(ncreds,
1663                                                conn,
1664                                                logon_type,
1665                                                self.user_creds,
1666                                                generic_package="Kerberos",
1667                                                generic_data=pv_blob)
1668
1669         expect_send_encrypted = expect_encrypted
1670         expect_recv_encrypted = False
1671         validationEx = self.do_LogonEx(ncreds, conn,
1672                                        logon_type, logon_info_pv,
1673                                        validation_level,
1674                                        expect_send_encrypted,
1675                                        expect_recv_encrypted,
1676                                        expect_error=expect_error)
1677         if expect_error is None:
1678             self.assertEqual(validationEx.length, 0)
1679
1680         expect_send_encrypted = expect_encrypted
1681         expect_recv_encrypted = False
1682         validationWF = self.do_LogonWithFlags(ncreds, conn,
1683                                               logon_type, logon_info_pv,
1684                                               validation_level,
1685                                               expect_send_encrypted,
1686                                               expect_recv_encrypted,
1687                                               expect_error=expect_error)
1688         if expect_error is None:
1689             self.assertEqual(validationWF.length, 0)
1690
1691         self.do_CheckCapabilities(ncreds, conn)
1692         return
1693
1694     def _test_generic_samlogon_with_args(self, trust, authX, flags):
1695         (creds, ncreds, conn, expect_encrypted) = \
1696             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1697
1698         if conn is None:
1699             return
1700
1701         return self._test_generic_samlogon(creds,
1702                                            ncreds,
1703                                            conn,
1704                                            expect_encrypted)
1705
1706     def _test_ticket_samlogon(self, trust_creds, ncreds, conn,
1707                               expect_encrypted):
1708         self.do_CheckCapabilities(ncreds, conn)
1709
1710         validation_level = netlogon.NetlogonValidationTicketLogon
1711         logon_type = netlogon.NetlogonTicketLogonInformation
1712         logon_info = self._prepare_samlogon(ncreds,
1713                                             conn,
1714                                             logon_type,
1715                                             self.user_creds,
1716                                             trust_creds=trust_creds)
1717
1718         expect_send_encrypted = False
1719         expect_recv_encrypted = False
1720         validationEx = self.do_LogonEx(ncreds, conn,
1721                                        logon_type, logon_info,
1722                                        validation_level,
1723                                        expect_send_encrypted,
1724                                        expect_recv_encrypted)
1725         self.assertEqual(validationEx.results,
1726                          netlogon.NETLOGON_TICKET_LOGON_FULL_SIGNATURE_PRESENT)
1727         self.assertEqual(validationEx.kerberos_status[0], ntstatus.NT_STATUS_OK)
1728         self.assertEqual(validationEx.netlogon_status[0], ntstatus.NT_STATUS_OK)
1729         self.assertIsNone(validationEx.source_of_status.string)
1730         self.assertIsNotNone(validationEx.user_information)
1731         self.assertNotEqual(validationEx.user_information.base.rid, 0)
1732         self.assertEqual(validationEx.user_information.base.key.key, list(b'\x00' *16))
1733         self.assertIsNone(validationEx.device_information)
1734
1735         expect_send_encrypted = False
1736         expect_recv_encrypted = False
1737         validationWF = self.do_LogonWithFlags(ncreds, conn,
1738                                               logon_type, logon_info,
1739                                               validation_level,
1740                                               expect_send_encrypted,
1741                                               expect_recv_encrypted)
1742         self.assertEqual(validationWF.results,
1743                          netlogon.NETLOGON_TICKET_LOGON_FULL_SIGNATURE_PRESENT)
1744         self.assertEqual(validationWF.kerberos_status[0], ntstatus.NT_STATUS_OK)
1745         self.assertEqual(validationWF.netlogon_status[0], ntstatus.NT_STATUS_OK)
1746         self.assertIsNone(validationWF.source_of_status.string)
1747         self.assertIsNotNone(validationWF.user_information)
1748         self.assertEqual(validationWF.user_information.base.rid,
1749                          validationEx.user_information.base.rid)
1750         self.assertEqual(validationWF.user_information.base.key.key, list(b'\x00' *16))
1751         self.assertIsNone(validationWF.device_information)
1752
1753         self.do_CheckCapabilities(ncreds, conn)
1754         return
1755
1756     def _test_ticket_samlogon_with_args(self, trust, authX, flags):
1757         (creds, ncreds, conn, expect_encrypted) = \
1758             self._prepare_ncreds_conn_with_args(trust, authX, flags)
1759
1760         if conn is None:
1761             return
1762
1763         return self._test_ticket_samlogon(creds,
1764                                           ncreds,
1765                                           conn,
1766                                           expect_encrypted)
1767
1768     def test_wks1_authenticate_flags(self):
1769
1770         wks1_creds = self.get_wks1_creds()
1771
1772         anon_conn = self.get_anon_conn()
1773
1774         des_flags = 0
1775         self.do_Authenticate3(anon_conn, wks1_creds,
1776                               des_flags, des_flags,
1777                               expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1778
1779         strong_flags = 0x00004000
1780         if self.strong_key_support:
1781             strong_ncreds = self.do_Authenticate3(anon_conn, wks1_creds,
1782                                                   strong_flags, strong_flags)
1783             strong_conn = self.get_schannel_conn(wks1_creds, strong_ncreds)
1784             tmp_ncreds = ndr_deepcopy(strong_ncreds)
1785             self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1786                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1787             self.do_CheckCapabilities(strong_ncreds, strong_conn)
1788         else:
1789             self.do_Authenticate3(anon_conn, wks1_creds,
1790                                   strong_flags, strong_flags,
1791                                   expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1792             strong_conn = None
1793
1794         aes_flags = 0x01000000
1795         aes_ncreds = self.do_Authenticate3(anon_conn, wks1_creds,
1796                                            aes_flags, aes_flags)
1797         aes_conn = self.get_schannel_conn(wks1_creds, aes_ncreds)
1798         tmp_ncreds = ndr_deepcopy(aes_ncreds)
1799         self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1800                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1801         self.do_CheckCapabilities(aes_ncreds, aes_conn)
1802         if strong_conn:
1803             self.do_CheckCapabilities(aes_ncreds, strong_conn)
1804
1805         krb5_flags = 0x80000000
1806         krb5_ncreds = self.do_Authenticate3(anon_conn, wks1_creds,
1807                                             krb5_flags, krb5_flags,
1808                                             expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1809
1810         if strong_conn:
1811             aes_ncreds = self.do_Authenticate3(strong_conn, wks1_creds,
1812                                                aes_flags, aes_flags)
1813             tmp_ncreds = ndr_deepcopy(aes_ncreds)
1814             self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1815                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1816             self.do_CheckCapabilities(aes_ncreds, aes_conn)
1817             self.do_CheckCapabilities(strong_ncreds, anon_conn,
1818                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1819             self.do_CheckCapabilities(aes_ncreds, strong_conn)
1820
1821         aes_ncreds = self.do_Authenticate3(aes_conn, wks1_creds,
1822                                            aes_flags, aes_flags)
1823         tmp_ncreds = ndr_deepcopy(aes_ncreds)
1824         self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1825                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1826         self.do_CheckCapabilities(aes_ncreds, aes_conn)
1827         if strong_conn:
1828             self.do_CheckCapabilities(aes_ncreds, strong_conn)
1829
1830         krb5_conn = self.get_krb5_conn(wks1_creds)
1831
1832         des_flags = 0
1833         self.do_Authenticate3(krb5_conn, wks1_creds,
1834                               des_flags, des_flags,
1835                               expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1836         strong_flags = 0x00004000
1837         if self.strong_key_support:
1838             strong_ncreds = self.do_Authenticate3(krb5_conn, wks1_creds,
1839                                                   strong_flags, strong_flags)
1840             tmp_ncreds = ndr_deepcopy(strong_ncreds)
1841             self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1842                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1843             if self.auth_krb5_support:
1844                 self.do_CheckCapabilities(strong_ncreds, krb5_conn)
1845             else:
1846                 tmp_ncreds = ndr_deepcopy(strong_ncreds)
1847                 self.do_CheckCapabilities(tmp_ncreds, krb5_conn,
1848                                           expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1849             self.do_CheckCapabilities(strong_ncreds, strong_conn)
1850             self.do_CheckCapabilities(strong_ncreds, aes_conn)
1851             self.do_CheckCapabilities(aes_ncreds, strong_conn,
1852                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1853             self.do_CheckCapabilities(strong_ncreds, aes_conn,
1854                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1855         else:
1856             self.do_Authenticate3(krb5_conn, wks1_creds,
1857                                   strong_flags, strong_flags,
1858                                   expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1859         aes_flags = 0x01000000
1860         aes_ncreds = self.do_Authenticate3(krb5_conn, wks1_creds,
1861                                            aes_flags, aes_flags)
1862         tmp_ncreds = ndr_deepcopy(aes_ncreds)
1863         self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1864                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1865         if self.auth_krb5_support:
1866             self.do_CheckCapabilities(aes_ncreds, krb5_conn)
1867         else:
1868             tmp_ncreds = ndr_deepcopy(aes_ncreds)
1869             self.do_CheckCapabilities(tmp_ncreds, krb5_conn,
1870                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1871         if strong_conn:
1872             self.do_CheckCapabilities(aes_ncreds, strong_conn)
1873         self.do_CheckCapabilities(aes_ncreds, aes_conn)
1874         krb5_flags = 0x80000000
1875         self.do_Authenticate3(krb5_conn, wks1_creds,
1876                               krb5_flags, krb5_flags,
1877                               expect_error=ntstatus.NT_STATUS_DOWNGRADE_DETECTED)
1878
1879         tmp_ncreds = ndr_deepcopy(aes_ncreds)
1880         self.do_CheckCapabilities(tmp_ncreds, anon_conn,
1881                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1882         if self.auth_krb5_support:
1883             self.do_CheckCapabilities(aes_ncreds, krb5_conn)
1884         else:
1885             tmp_ncreds = ndr_deepcopy(aes_ncreds)
1886             self.do_CheckCapabilities(tmp_ncreds, krb5_conn,
1887                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1888         if strong_conn:
1889             self.do_CheckCapabilities(aes_ncreds, strong_conn)
1890         self.do_CheckCapabilities(aes_ncreds, aes_conn)
1891         self.do_CheckCapabilities(tmp_ncreds, aes_conn,
1892                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1893         self.do_CheckCapabilities(aes_ncreds, aes_conn,
1894                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1895
1896         krb5_flags = 0
1897         if not self.auth_krb5_support:
1898             self.do_AuthenticateKerberos(krb5_conn, wks1_creds,
1899                                          krb5_flags, krb5_flags,
1900                                          expect_error=ntstatus.NT_STATUS_RPC_PROCNUM_OUT_OF_RANGE)
1901             return
1902         krb5_ncreds = self.do_AuthenticateKerberos(krb5_conn, wks1_creds,
1903                                                    krb5_flags, krb5_flags)
1904         self.do_CheckCapabilities(krb5_ncreds, anon_conn,
1905                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1906         self.do_CheckCapabilities(krb5_ncreds, krb5_conn)
1907         if strong_conn:
1908             self.do_CheckCapabilities(krb5_ncreds, strong_conn,
1909                                       expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1910         self.do_CheckCapabilities(krb5_ncreds, aes_conn,
1911                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1912         self.do_CheckCapabilities(krb5_ncreds, krb5_conn)
1913         return
1914
1915     def test_wks1_vs_bdc1_authK(self):
1916
1917         if not self.auth_krb5_support:
1918             self.skipTest('Required NETLOGON_AUTH_KRB5_SUPPORT')
1919
1920         wks1_creds = self.get_wks1_creds()
1921         bdc1_creds = self.get_bdc1_creds()
1922
1923         wks1_conn = self.get_krb5_conn(wks1_creds)
1924         bdc1_conn = self.get_krb5_conn(bdc1_creds)
1925
1926         krb5_flags = 0xe13fffff
1927         wks1_ncreds = self.do_AuthenticateKerberos(wks1_conn, wks1_creds,
1928                                                    krb5_flags, krb5_flags)
1929         self.do_CheckCapabilities(wks1_ncreds, wks1_conn)
1930         bdc1_ncreds = self.do_AuthenticateKerberos(bdc1_conn, bdc1_creds,
1931                                                    krb5_flags, krb5_flags)
1932         self.do_CheckCapabilities(bdc1_ncreds, bdc1_conn)
1933
1934         self.do_CheckCapabilities(wks1_ncreds, bdc1_conn,
1935                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1936         self.do_CheckCapabilities(bdc1_ncreds, wks1_conn,
1937                                   expect_error1=ntstatus.NT_STATUS_ACCESS_DENIED)
1938
1939         self.do_CheckCapabilities(wks1_ncreds, wks1_conn)
1940         self.do_CheckCapabilities(bdc1_ncreds, bdc1_conn)
1941
1942         self.do_AuthenticateKerberos(wks1_conn, bdc1_creds,
1943                                      krb5_flags, krb5_flags,
1944                                      expect_error=ntstatus.NT_STATUS_ACCESS_DENIED)
1945         self.do_AuthenticateKerberos(bdc1_conn, wks1_creds,
1946                                      krb5_flags, krb5_flags,
1947                                      expect_error=ntstatus.NT_STATUS_ACCESS_DENIED)
1948
1949         self.do_CheckCapabilities(wks1_ncreds, wks1_conn)
1950         self.do_CheckCapabilities(bdc1_ncreds, bdc1_conn)
1951         return
1952
1953
1954 if __name__ == "__main__":
1955     global_asn1_print = True
1956     global_ndr_print = True
1957     global_hexdump = True
1958     import unittest
1959     unittest.main()