Annotation of embedaddon/curl/tests/smbserver.py, revision 1.1
1.1 ! misho 1: #!/usr/bin/env python
! 2: # -*- coding: utf-8 -*-
! 3: #
! 4: # Project ___| | | | _ \| |
! 5: # / __| | | | |_) | |
! 6: # | (__| |_| | _ <| |___
! 7: # \___|\___/|_| \_\_____|
! 8: #
! 9: # Copyright (C) 2017 - 2020, Daniel Stenberg, <daniel@haxx.se>, et al.
! 10: #
! 11: # This software is licensed as described in the file COPYING, which
! 12: # you should have received as part of this distribution. The terms
! 13: # are also available at https://curl.haxx.se/docs/copyright.html.
! 14: #
! 15: # You may opt to use, copy, modify, merge, publish, distribute and/or sell
! 16: # copies of the Software, and permit persons to whom the Software is
! 17: # furnished to do so, under the terms of the COPYING file.
! 18: #
! 19: # This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
! 20: # KIND, either express or implied.
! 21: #
! 22: """Server for testing SMB"""
! 23:
! 24: from __future__ import (absolute_import, division, print_function)
! 25: # NOTE: the impacket configuration is not unicode_literals compatible!
! 26: import argparse
! 27: import os
! 28: import sys
! 29: import logging
! 30: import tempfile
! 31: if sys.version_info.major >= 3:
! 32: import configparser
! 33: else:
! 34: import ConfigParser as configparser
! 35:
! 36: # Import our curl test data helper
! 37: import curl_test_data
! 38:
! 39: # impacket needs to be installed in the Python environment
! 40: try:
! 41: import impacket
! 42: except ImportError:
! 43: sys.stderr.write('Python package impacket needs to be installed!\n')
! 44: sys.stderr.write('Use pip or your package manager to install it.\n')
! 45: sys.exit(1)
! 46: from impacket import smbserver as imp_smbserver
! 47: from impacket import smb as imp_smb
! 48: from impacket.nt_errors import (STATUS_ACCESS_DENIED, STATUS_SUCCESS,
! 49: STATUS_NO_SUCH_FILE)
! 50:
! 51: log = logging.getLogger(__name__)
! 52: SERVER_MAGIC = "SERVER_MAGIC"
! 53: TESTS_MAGIC = "TESTS_MAGIC"
! 54: VERIFIED_REQ = "verifiedserver"
! 55: VERIFIED_RSP = "WE ROOLZ: {pid}\n"
! 56:
! 57:
! 58: def smbserver(options):
! 59: """Start up a TCP SMB server that serves forever
! 60:
! 61: """
! 62: if options.pidfile:
! 63: pid = os.getpid()
! 64: # see tests/server/util.c function write_pidfile
! 65: if os.name == "nt":
! 66: pid += 65536
! 67: with open(options.pidfile, "w") as f:
! 68: f.write(str(pid))
! 69:
! 70: # Here we write a mini config for the server
! 71: smb_config = configparser.ConfigParser()
! 72: smb_config.add_section("global")
! 73: smb_config.set("global", "server_name", "SERVICE")
! 74: smb_config.set("global", "server_os", "UNIX")
! 75: smb_config.set("global", "server_domain", "WORKGROUP")
! 76: smb_config.set("global", "log_file", "")
! 77: smb_config.set("global", "credentials_file", "")
! 78:
! 79: # We need a share which allows us to test that the server is running
! 80: smb_config.add_section("SERVER")
! 81: smb_config.set("SERVER", "comment", "server function")
! 82: smb_config.set("SERVER", "read only", "yes")
! 83: smb_config.set("SERVER", "share type", "0")
! 84: smb_config.set("SERVER", "path", SERVER_MAGIC)
! 85:
! 86: # Have a share for tests. These files will be autogenerated from the
! 87: # test input.
! 88: smb_config.add_section("TESTS")
! 89: smb_config.set("TESTS", "comment", "tests")
! 90: smb_config.set("TESTS", "read only", "yes")
! 91: smb_config.set("TESTS", "share type", "0")
! 92: smb_config.set("TESTS", "path", TESTS_MAGIC)
! 93:
! 94: if not options.srcdir or not os.path.isdir(options.srcdir):
! 95: raise ScriptException("--srcdir is mandatory")
! 96:
! 97: test_data_dir = os.path.join(options.srcdir, "data")
! 98:
! 99: smb_server = TestSmbServer((options.host, options.port),
! 100: config_parser=smb_config,
! 101: test_data_directory=test_data_dir)
! 102: log.info("[SMB] setting up SMB server on port %s", options.port)
! 103: smb_server.processConfigFile()
! 104: smb_server.serve_forever()
! 105: return 0
! 106:
! 107:
! 108: class TestSmbServer(imp_smbserver.SMBSERVER):
! 109: """
! 110: Test server for SMB which subclasses the impacket SMBSERVER and provides
! 111: test functionality.
! 112: """
! 113:
! 114: def __init__(self,
! 115: address,
! 116: config_parser=None,
! 117: test_data_directory=None):
! 118: imp_smbserver.SMBSERVER.__init__(self,
! 119: address,
! 120: config_parser=config_parser)
! 121:
! 122: # Set up a test data object so we can get test data later.
! 123: self.ctd = curl_test_data.TestData(test_data_directory)
! 124:
! 125: # Override smbComNtCreateAndX so we can pretend to have files which
! 126: # don't exist.
! 127: self.hookSmbCommand(imp_smb.SMB.SMB_COM_NT_CREATE_ANDX,
! 128: self.create_and_x)
! 129:
! 130: def create_and_x(self, conn_id, smb_server, smb_command, recv_packet):
! 131: """
! 132: Our version of smbComNtCreateAndX looks for special test files and
! 133: fools the rest of the framework into opening them as if they were
! 134: normal files.
! 135: """
! 136: conn_data = smb_server.getConnectionData(conn_id)
! 137:
! 138: # Wrap processing in a try block which allows us to throw SmbException
! 139: # to control the flow.
! 140: try:
! 141: ncax_parms = imp_smb.SMBNtCreateAndX_Parameters(
! 142: smb_command["Parameters"])
! 143:
! 144: path = self.get_share_path(conn_data,
! 145: ncax_parms["RootFid"],
! 146: recv_packet["Tid"])
! 147: log.info("[SMB] Requested share path: %s", path)
! 148:
! 149: disposition = ncax_parms["Disposition"]
! 150: log.debug("[SMB] Requested disposition: %s", disposition)
! 151:
! 152: # Currently we only support reading files.
! 153: if disposition != imp_smb.FILE_OPEN:
! 154: raise SmbException(STATUS_ACCESS_DENIED,
! 155: "Only support reading files")
! 156:
! 157: # Check to see if the path we were given is actually a
! 158: # magic path which needs generating on the fly.
! 159: if path not in [SERVER_MAGIC, TESTS_MAGIC]:
! 160: # Pass the command onto the original handler.
! 161: return imp_smbserver.SMBCommands.smbComNtCreateAndX(conn_id,
! 162: smb_server,
! 163: smb_command,
! 164: recv_packet)
! 165:
! 166: flags2 = recv_packet["Flags2"]
! 167: ncax_data = imp_smb.SMBNtCreateAndX_Data(flags=flags2,
! 168: data=smb_command[
! 169: "Data"])
! 170: requested_file = imp_smbserver.decodeSMBString(
! 171: flags2,
! 172: ncax_data["FileName"])
! 173: log.debug("[SMB] User requested file '%s'", requested_file)
! 174:
! 175: if path == SERVER_MAGIC:
! 176: fid, full_path = self.get_server_path(requested_file)
! 177: else:
! 178: assert (path == TESTS_MAGIC)
! 179: fid, full_path = self.get_test_path(requested_file)
! 180:
! 181: resp_parms = imp_smb.SMBNtCreateAndXResponse_Parameters()
! 182: resp_data = ""
! 183:
! 184: # Simple way to generate a fid
! 185: if len(conn_data["OpenedFiles"]) == 0:
! 186: fakefid = 1
! 187: else:
! 188: fakefid = conn_data["OpenedFiles"].keys()[-1] + 1
! 189: resp_parms["Fid"] = fakefid
! 190: resp_parms["CreateAction"] = disposition
! 191:
! 192: if os.path.isdir(path):
! 193: resp_parms[
! 194: "FileAttributes"] = imp_smb.SMB_FILE_ATTRIBUTE_DIRECTORY
! 195: resp_parms["IsDirectory"] = 1
! 196: else:
! 197: resp_parms["IsDirectory"] = 0
! 198: resp_parms["FileAttributes"] = ncax_parms["FileAttributes"]
! 199:
! 200: # Get this file's information
! 201: resp_info, error_code = imp_smbserver.queryPathInformation(
! 202: "", full_path, level=imp_smb.SMB_QUERY_FILE_ALL_INFO)
! 203:
! 204: if error_code != STATUS_SUCCESS:
! 205: raise SmbException(error_code, "Failed to query path info")
! 206:
! 207: resp_parms["CreateTime"] = resp_info["CreationTime"]
! 208: resp_parms["LastAccessTime"] = resp_info[
! 209: "LastAccessTime"]
! 210: resp_parms["LastWriteTime"] = resp_info["LastWriteTime"]
! 211: resp_parms["LastChangeTime"] = resp_info[
! 212: "LastChangeTime"]
! 213: resp_parms["FileAttributes"] = resp_info[
! 214: "ExtFileAttributes"]
! 215: resp_parms["AllocationSize"] = resp_info[
! 216: "AllocationSize"]
! 217: resp_parms["EndOfFile"] = resp_info["EndOfFile"]
! 218:
! 219: # Let's store the fid for the connection
! 220: # smbServer.log("Create file %s, mode:0x%x" % (pathName, mode))
! 221: conn_data["OpenedFiles"][fakefid] = {}
! 222: conn_data["OpenedFiles"][fakefid]["FileHandle"] = fid
! 223: conn_data["OpenedFiles"][fakefid]["FileName"] = path
! 224: conn_data["OpenedFiles"][fakefid]["DeleteOnClose"] = False
! 225:
! 226: except SmbException as s:
! 227: log.debug("[SMB] SmbException hit: %s", s)
! 228: error_code = s.error_code
! 229: resp_parms = ""
! 230: resp_data = ""
! 231:
! 232: resp_cmd = imp_smb.SMBCommand(imp_smb.SMB.SMB_COM_NT_CREATE_ANDX)
! 233: resp_cmd["Parameters"] = resp_parms
! 234: resp_cmd["Data"] = resp_data
! 235: smb_server.setConnectionData(conn_id, conn_data)
! 236:
! 237: return [resp_cmd], None, error_code
! 238:
! 239: def get_share_path(self, conn_data, root_fid, tid):
! 240: conn_shares = conn_data["ConnectedShares"]
! 241:
! 242: if tid in conn_shares:
! 243: if root_fid > 0:
! 244: # If we have a rootFid, the path is relative to that fid
! 245: path = conn_data["OpenedFiles"][root_fid]["FileName"]
! 246: log.debug("RootFid present %s!" % path)
! 247: else:
! 248: if "path" in conn_shares[tid]:
! 249: path = conn_shares[tid]["path"]
! 250: else:
! 251: raise SmbException(STATUS_ACCESS_DENIED,
! 252: "Connection share had no path")
! 253: else:
! 254: raise SmbException(imp_smbserver.STATUS_SMB_BAD_TID,
! 255: "TID was invalid")
! 256:
! 257: return path
! 258:
! 259: def get_server_path(self, requested_filename):
! 260: log.debug("[SMB] Get server path '%s'", requested_filename)
! 261:
! 262: if requested_filename not in [VERIFIED_REQ]:
! 263: raise SmbException(STATUS_NO_SUCH_FILE, "Couldn't find the file")
! 264:
! 265: fid, filename = tempfile.mkstemp()
! 266: log.debug("[SMB] Created %s (%d) for storing '%s'",
! 267: filename, fid, requested_filename)
! 268:
! 269: contents = ""
! 270:
! 271: if requested_filename == VERIFIED_REQ:
! 272: log.debug("[SMB] Verifying server is alive")
! 273: pid = os.getpid()
! 274: # see tests/server/util.c function write_pidfile
! 275: if os.name == "nt":
! 276: pid += 65536
! 277: contents = VERIFIED_RSP.format(pid=pid).encode('utf-8')
! 278:
! 279: self.write_to_fid(fid, contents)
! 280: return fid, filename
! 281:
! 282: def write_to_fid(self, fid, contents):
! 283: # Write the contents to file descriptor
! 284: os.write(fid, contents)
! 285: os.fsync(fid)
! 286:
! 287: # Rewind the file to the beginning so a read gets us the contents
! 288: os.lseek(fid, 0, os.SEEK_SET)
! 289:
! 290: def get_test_path(self, requested_filename):
! 291: log.info("[SMB] Get reply data from 'test%s'", requested_filename)
! 292:
! 293: fid, filename = tempfile.mkstemp()
! 294: log.debug("[SMB] Created %s (%d) for storing test '%s'",
! 295: filename, fid, requested_filename)
! 296:
! 297: try:
! 298: contents = self.ctd.get_test_data(requested_filename).encode('utf-8')
! 299: self.write_to_fid(fid, contents)
! 300: return fid, filename
! 301:
! 302: except Exception:
! 303: log.exception("Failed to make test file")
! 304: raise SmbException(STATUS_NO_SUCH_FILE, "Failed to make test file")
! 305:
! 306:
! 307: class SmbException(Exception):
! 308: def __init__(self, error_code, error_message):
! 309: super(SmbException, self).__init__(error_message)
! 310: self.error_code = error_code
! 311:
! 312:
! 313: class ScriptRC(object):
! 314: """Enum for script return codes"""
! 315: SUCCESS = 0
! 316: FAILURE = 1
! 317: EXCEPTION = 2
! 318:
! 319:
! 320: class ScriptException(Exception):
! 321: pass
! 322:
! 323:
! 324: def get_options():
! 325: parser = argparse.ArgumentParser()
! 326:
! 327: parser.add_argument("--port", action="store", default=9017,
! 328: type=int, help="port to listen on")
! 329: parser.add_argument("--host", action="store", default="127.0.0.1",
! 330: help="host to listen on")
! 331: parser.add_argument("--verbose", action="store", type=int, default=0,
! 332: help="verbose output")
! 333: parser.add_argument("--pidfile", action="store",
! 334: help="file name for the PID")
! 335: parser.add_argument("--logfile", action="store",
! 336: help="file name for the log")
! 337: parser.add_argument("--srcdir", action="store", help="test directory")
! 338: parser.add_argument("--id", action="store", help="server ID")
! 339: parser.add_argument("--ipv4", action="store_true", default=0,
! 340: help="IPv4 flag")
! 341:
! 342: return parser.parse_args()
! 343:
! 344:
! 345: def setup_logging(options):
! 346: """
! 347: Set up logging from the command line options
! 348: """
! 349: root_logger = logging.getLogger()
! 350: add_stdout = False
! 351:
! 352: formatter = logging.Formatter("%(asctime)s %(levelname)-5.5s %(message)s")
! 353:
! 354: # Write out to a logfile
! 355: if options.logfile:
! 356: handler = logging.FileHandler(options.logfile, mode="w")
! 357: handler.setFormatter(formatter)
! 358: handler.setLevel(logging.DEBUG)
! 359: root_logger.addHandler(handler)
! 360: else:
! 361: # The logfile wasn't specified. Add a stdout logger.
! 362: add_stdout = True
! 363:
! 364: if options.verbose:
! 365: # Add a stdout logger as well in verbose mode
! 366: root_logger.setLevel(logging.DEBUG)
! 367: add_stdout = True
! 368: else:
! 369: root_logger.setLevel(logging.INFO)
! 370:
! 371: if add_stdout:
! 372: stdout_handler = logging.StreamHandler(sys.stdout)
! 373: stdout_handler.setFormatter(formatter)
! 374: stdout_handler.setLevel(logging.DEBUG)
! 375: root_logger.addHandler(stdout_handler)
! 376:
! 377:
! 378: if __name__ == '__main__':
! 379: # Get the options from the user.
! 380: options = get_options()
! 381:
! 382: # Setup logging using the user options
! 383: setup_logging(options)
! 384:
! 385: # Run main script.
! 386: try:
! 387: rc = smbserver(options)
! 388: except Exception as e:
! 389: log.exception(e)
! 390: rc = ScriptRC.EXCEPTION
! 391:
! 392: log.info("[SMB] Returning %d", rc)
! 393: sys.exit(rc)
FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>
![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to smbserver.py CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / curl / tests