mirror of
https://github.com/samba-team/samba.git
synced 2025-01-18 06:04:06 +03:00
4f21f1ca8d
The auth-logging tests are an odd combination of server and client behaviour. On the one hand we want a IRPC connection to see the auth events being logged on the server. On the other hand, we want the auth events to appear to be happening on a client. Currently we hardcode in the use of a SOCKET_WRAPPER interface to make this happen. We can avoid this explicit socket wrapper usage by using the server smb.conf instead in the one place we actually want to act like the server (creating the IRPC connection). Then we can switch from using the 'ad_dc*:local' testenvs to use 'ad_dc*', in order to act like a client by default. The SERVERCONFFILE environment variable has already been added for the few cases where a test needs explicit access to the server's smb.conf. However, for samba.tests.auth_log, the samlogon test cases are still reliant on being run on the :local testenv, and so we can't switch them over just yet. This is because the samlogon is using the DC's machine creds underneath, which will fail on the non-local testenv. We could create separate machine creds for the client and use those, but this is a non-trivial rework of the test code. Signed-off-by: Tim Beale <timbeale@catalyst.net.nz> Reviewed-by: Andrew Bartlett <abartlet@samba.org>
144 lines
4.9 KiB
Python
144 lines
4.9 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
|
|
from samba.param import LoadParm
|
|
import time
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
msg_ctxs = []
|
|
|
|
|
|
class AuthLogTestBase(samba.tests.TestCase):
|
|
|
|
def setUp(self):
|
|
super(AuthLogTestBase, self).setUp()
|
|
# connect to the server's messaging bus (we need to explicitly load a
|
|
# different smb.conf here, because in all other respects this test
|
|
# wants to act as a separate remote client)
|
|
server_conf = os.getenv('SERVERCONFFILE')
|
|
if server_conf:
|
|
lp_ctx = LoadParm(filename_for_non_global_lp=server_conf)
|
|
else:
|
|
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):
|
|
if self.remoteAddress is None:
|
|
return True
|
|
|
|
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)
|