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