mirror of
https://github.com/samba-team/samba.git
synced 2025-01-18 06:04:06 +03:00
81043cdf0e
Some tests (especially samba.tests.auth_log_netlogon_bad_creds) are failing due to not receiving expected messages. There seems to be some timing issue or race around the messaging bus being set up and getting the expected events resulting from the failed netlogon. Specifically the the order of destruction of the messaging.Messaging() c-py objects is different under python2. Under python2 all of the messaging.Messaging() objects are destructed *after* all the tests are run. Note: each instance of the TestCase has it's own Messaging() instance which is created by TestCaseXYZ.setUp, so it appears the unittest destroys the test instances when all the tests have run whereas in python3 we see each messaging.Messaging() instance destroyed after each test runs. Ok, what difference does that make ? well it seems in python3 because each Messaging() instance is destructed after a test runs that the associated messaging_dgm_destroy() also runs, this destroys the global_dgm_context context which means when the next test runs the whole messaging infrastructure needs to be built again when the next Messaging() object is created. On the server-side this seems to result in attempts to send messages to the listener failing first with get_event_server: Failed to find 'auth_event' registered on the message bus to send JSON audit events to: NT_STATUS_CONNECTION_REFUSED and subsequently with get_event_server: Failed to find 'auth_event' registered on the message bus to send JSON audit events to: NT_STATUS_UNSUCCESSFUL client doesn't get any more messages, test fails :-( So, what's the difference in python2, well because the destructors for the (4 in the case of netlogon_bad_creds) instances of Messagaging() don't run till the end of the tests this doesn't happen and the global_dgm_context never gets destroyed untill all the tests complete. There is some race condition at play here, a simple sleep at the start of a failing test fixes the problem. But... ok that isn't a possible solution here, instead I have adjusted the base auth tests to store the Messaging() objects in a global list forcing them to remain in scope until the tests are complete. This ensure the behaviour is consistent across python2 & python3. Signed-off-by: Noel Power <noel.power@suse.com> Reviewed-by: Andrew Bartlett <abartlet@samba.org>
132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
# Unix SMB/CIFS implementation.
|
|
# Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
|
|
from __future__ import print_function
|
|
"""Tests for the Auth and AuthZ logging.
|
|
"""
|
|
|
|
import samba.tests
|
|
from samba.messaging import Messaging
|
|
from samba.dcerpc.messaging import MSG_AUTH_LOG, AUTH_EVENT_NAME
|
|
import time
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
msg_ctxs = []
|
|
|
|
class AuthLogTestBase(samba.tests.TestCase):
|
|
|
|
def setUp(self):
|
|
super(AuthLogTestBase, self).setUp()
|
|
lp_ctx = self.get_loadparm()
|
|
self.msg_ctx = Messaging((1,), lp_ctx=lp_ctx)
|
|
global msg_ctxs
|
|
msg_ctxs.append(self.msg_ctx)
|
|
self.msg_ctx.irpc_add_name(AUTH_EVENT_NAME)
|
|
|
|
def messageHandler(context, msgType, src, message):
|
|
# This does not look like sub unit output and it
|
|
# makes these tests much easier to debug.
|
|
print(message)
|
|
jsonMsg = json.loads(message)
|
|
context["messages"].append(jsonMsg)
|
|
|
|
self.context = {"messages": []}
|
|
self.msg_handler_and_context = (messageHandler, self.context)
|
|
self.msg_ctx.register(self.msg_handler_and_context,
|
|
msg_type=MSG_AUTH_LOG)
|
|
|
|
self.discardMessages()
|
|
|
|
self.remoteAddress = None
|
|
self.server = os.environ["SERVER"]
|
|
self.connection = None
|
|
|
|
def tearDown(self):
|
|
if self.msg_handler_and_context:
|
|
self.msg_ctx.deregister(self.msg_handler_and_context,
|
|
msg_type=MSG_AUTH_LOG)
|
|
self.msg_ctx.irpc_remove_name(AUTH_EVENT_NAME)
|
|
|
|
def waitForMessages(self, isLastExpectedMessage, connection=None):
|
|
"""Wait for all the expected messages to arrive
|
|
The connection is passed through to keep the connection alive
|
|
until all the logging messages have been received.
|
|
"""
|
|
|
|
def completed(messages):
|
|
for message in messages:
|
|
if isRemote(message) and isLastExpectedMessage(message):
|
|
return True
|
|
return False
|
|
|
|
def isRemote(message):
|
|
remote = None
|
|
if message["type"] == "Authorization":
|
|
remote = message["Authorization"]["remoteAddress"]
|
|
elif message["type"] == "Authentication":
|
|
remote = message["Authentication"]["remoteAddress"]
|
|
else:
|
|
return False
|
|
|
|
try:
|
|
addr = remote.split(":")
|
|
return addr[1] == self.remoteAddress
|
|
except IndexError:
|
|
return False
|
|
|
|
self.connection = connection
|
|
|
|
start_time = time.time()
|
|
while not completed(self.context["messages"]):
|
|
self.msg_ctx.loop_once(0.1)
|
|
if time.time() - start_time > 1:
|
|
self.connection = None
|
|
return []
|
|
|
|
self.connection = None
|
|
return list(filter(isRemote, self.context["messages"]))
|
|
|
|
# Discard any previously queued messages.
|
|
def discardMessages(self):
|
|
self.msg_ctx.loop_once(0.001)
|
|
while len(self.context["messages"]):
|
|
self.msg_ctx.loop_once(0.001)
|
|
self.context["messages"] = []
|
|
|
|
# Remove any NETLOGON authentication messages
|
|
# NETLOGON is only performed once per session, so to avoid ordering
|
|
# dependencies within the tests it's best to strip out NETLOGON messages.
|
|
#
|
|
def remove_netlogon_messages(self, messages):
|
|
def is_not_netlogon(msg):
|
|
if "Authentication" not in msg:
|
|
return True
|
|
sd = msg["Authentication"]["serviceDescription"]
|
|
return sd != "NETLOGON"
|
|
|
|
return list(filter(is_not_netlogon, messages))
|
|
|
|
GUID_RE = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
|
|
|
|
#
|
|
# Is the supplied GUID string correctly formatted
|
|
#
|
|
def is_guid(self, guid):
|
|
return re.match(self.GUID_RE, guid)
|