python/samba/tests/auth_log_netlogon.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
   3 # Copyright (C) Catalyst IT Ltd. 2017
   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 """
  20     Tests that exercise the auth logging for a successful netlogon attempt
  21
  22     NOTE: As the netlogon authentication is performed once per session,
  23           there is only one test in this routine.  If another test is added
  24           only the test executed first will generate the netlogon auth message
  25 """
  26
  27 import samba.tests
  28 import os
  29 from samba.samdb import SamDB
  30 import samba.tests.auth_log_base
  31 from samba.credentials import Credentials
  32 from samba.dcerpc import netlogon
  33 from samba.dcerpc.dcerpc import AS_SYSTEM_MAGIC_PATH_TOKEN
  34 from samba.auth import system_session
  35 from samba.tests import delete_force
  36 from samba.dsdb import UF_WORKSTATION_TRUST_ACCOUNT, UF_PASSWD_NOTREQD
  37 from samba.dcerpc.misc import SEC_CHAN_WKSTA
  38 from samba.compat import text_type
  39 from samba.dcerpc.windows_event_ids import EVT_ID_SUCCESSFUL_LOGON
  40
  41
  42 class AuthLogTestsNetLogon(samba.tests.auth_log_base.AuthLogTestBase):
  43
  44     def setUp(self):
  45         super(AuthLogTestsNetLogon, self).setUp()
  46         self.lp      = samba.tests.env_loadparm()
  47         self.creds   = Credentials()
  48
  49         self.session = system_session()
  50         self.ldb = SamDB(
  51             session_info=self.session,
  52             credentials=self.creds,
  53             lp=self.lp)
  54
  55         self.domain        = os.environ["DOMAIN"]
  56         self.netbios_name  = "NetLogonGood"
  57         self.machinepass   = "abcdefghij"
  58         self.remoteAddress = AS_SYSTEM_MAGIC_PATH_TOKEN
  59         self.base_dn       = self.ldb.domain_dn()
  60         self.dn            = ("cn=%s,cn=users,%s" %
  61                               (self.netbios_name, self.base_dn))
  62
  63         utf16pw = text_type('"' + self.machinepass + '"').encode('utf-16-le')
  64         self.ldb.add({
  65             "dn": self.dn,
  66             "objectclass": "computer",
  67             "sAMAccountName": "%s$" % self.netbios_name,
  68             "userAccountControl":
  69                 str(UF_WORKSTATION_TRUST_ACCOUNT | UF_PASSWD_NOTREQD),
  70             "unicodePwd": utf16pw})
  71
  72     def tearDown(self):
  73         super(AuthLogTestsNetLogon, self).tearDown()
  74         delete_force(self.ldb, self.dn)
  75
  76     def _test_netlogon(self, binding, checkFunction):
  77
  78         def isLastExpectedMessage(msg):
  79             return (
  80                 msg["type"] == "Authorization" and
  81                 msg["Authorization"]["serviceDescription"]  == "DCE/RPC" and
  82                 msg["Authorization"]["authType"]            == "schannel" and
  83                 msg["Authorization"]["transportProtection"] == "SEAL")
  84
  85         if binding:
  86             binding = "[schannel,%s]" % binding
  87         else:
  88             binding = "[schannel]"
  89
  90         machine_creds = Credentials()
  91         machine_creds.guess(self.get_loadparm())
  92         machine_creds.set_secure_channel_type(SEC_CHAN_WKSTA)
  93         machine_creds.set_password(self.machinepass)
  94         machine_creds.set_username(self.netbios_name + "$")
  95
  96         netlogon_conn = netlogon.netlogon("ncalrpc:%s" % binding,
  97                                           self.get_loadparm(),
  98                                           machine_creds)
  99
 100         messages = self.waitForMessages(isLastExpectedMessage, netlogon_conn)
 101         checkFunction(messages)
 102
 103     def netlogon_check(self, messages):
 104
 105         expected_messages = 5
 106         self.assertEquals(expected_messages,
 107                           len(messages),
 108                           "Did not receive the expected number of messages")
 109
 110         # Check the first message it should be an Authorization
 111         msg = messages[0]
 112         self.assertEquals("Authorization", msg["type"])
 113         self.assertEquals("DCE/RPC",
 114                           msg["Authorization"]["serviceDescription"])
 115         self.assertEquals("ncalrpc", msg["Authorization"]["authType"])
 116         self.assertEquals("NONE", msg["Authorization"]["transportProtection"])
 117         self.assertTrue(self.is_guid(msg["Authorization"]["sessionId"]))
 118
 119         # Check the fourth message it should be a NETLOGON Authentication
 120         msg = messages[3]
 121         self.assertEquals("Authentication", msg["type"])
 122         self.assertEquals("NETLOGON",
 123                           msg["Authentication"]["serviceDescription"])
 124         self.assertEquals("ServerAuthenticate",
 125                           msg["Authentication"]["authDescription"])
 126         self.assertEquals("NT_STATUS_OK",
 127                           msg["Authentication"]["status"])
 128         self.assertEquals("HMAC-SHA256",
 129                           msg["Authentication"]["passwordType"])
 130         self.assertEquals(EVT_ID_SUCCESSFUL_LOGON,
 131                           msg["Authentication"]["eventId"])
 132
 133     def test_netlogon(self):
 134         self._test_netlogon("SEAL", self.netlogon_check)