From 4ab9df05b6b65b3ddf2155801d8ad0691d338a1e Mon Sep 17 00:00:00 2001 From: Jordan Borean Date: Mon, 26 Feb 2018 15:19:58 +1000 Subject: [PATCH] Changed Open.open to Open.create to match SMB message names --- examples/directory-management.py | 20 +- examples/file-management.py | 4 +- smbprotocol/__init__.py | 2 +- smbprotocol/open.py | 12 +- tests/test_open.py | 470 +++++++++++++++---------------- 5 files changed, 254 insertions(+), 254 deletions(-) diff --git a/examples/directory-management.py b/examples/directory-management.py index ece4f7b0..130c9f2d 100644 --- a/examples/directory-management.py +++ b/examples/directory-management.py @@ -25,7 +25,7 @@ # ensure directory is created dir_open = Open(tree, dir_name) - dir_open.open( + dir_open.create( ImpersonationLevel.Impersonation, DirectoryAccessMask.GENERIC_READ | DirectoryAccessMask.GENERIC_WRITE, FileAttributes.FILE_ATTRIBUTE_DIRECTORY, @@ -37,14 +37,14 @@ # create some files in dir and query the contents as part of a compound # request directory_file = Open(tree, r"%s\file.txt" % dir_name) - directory_file.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.GENERIC_WRITE | - FilePipePrinterAccessMask.DELETE, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_DELETE_ON_CLOSE) + directory_file.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.GENERIC_WRITE | + FilePipePrinterAccessMask.DELETE, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_DELETE_ON_CLOSE) compound_messages = [ directory_file.write("Hello World".encode('utf-8'), 0, send=False), @@ -72,7 +72,7 @@ # delete a directory (note the dir needs to be empty to delete on close) dir_open = Open(tree, dir_name) delete_msgs = [ - dir_open.open( + dir_open.create( ImpersonationLevel.Impersonation, DirectoryAccessMask.DELETE, FileAttributes.FILE_ATTRIBUTE_DIRECTORY, diff --git a/examples/file-management.py b/examples/file-management.py index b8645e4e..25b226f3 100644 --- a/examples/file-management.py +++ b/examples/file-management.py @@ -58,7 +58,7 @@ ] file_open = Open(tree, file_name) - open_info = file_open.open( + open_info = file_open.create( ImpersonationLevel.Impersonation, FilePipePrinterAccessMask.GENERIC_READ | FilePipePrinterAccessMask.GENERIC_WRITE, @@ -93,7 +93,7 @@ # read and delete a file in a single SMB packet instead of 3 file_open = Open(tree, file_name) delete_msgs = [ - file_open.open( + file_open.create( ImpersonationLevel.Impersonation, FilePipePrinterAccessMask.GENERIC_READ | FilePipePrinterAccessMask.DELETE, diff --git a/smbprotocol/__init__.py b/smbprotocol/__init__.py index 72295eb0..10ebc786 100644 --- a/smbprotocol/__init__.py +++ b/smbprotocol/__init__.py @@ -10,4 +10,4 @@ def emit(self, record): logger = logging.getLogger(__name__) logger.addHandler(NullHandler()) -__version__ = '0.0.1.dev5' +__version__ = '0.0.1.dev6' diff --git a/smbprotocol/open.py b/smbprotocol/open.py index ce87fcc9..d47709df 100644 --- a/smbprotocol/open.py +++ b/smbprotocol/open.py @@ -931,9 +931,9 @@ def __init__(self, tree, name): self.file_attributes = None self.create_disposition = None - def open(self, impersonation_level, desired_access, file_attributes, - share_access, create_disposition, create_options, - create_contexts=None, send=True): + def create(self, impersonation_level, desired_access, file_attributes, + share_access, create_disposition, create_options, + create_contexts=None, send=True): """ This will open the file based on the input parameters supplied. Any file open should also be called with Open.close() when it is finished. @@ -998,7 +998,7 @@ def open(self, impersonation_level, desired_access, file_attributes, self.create_disposition = create_disposition if not send: - return create, self._open_response + return create, self._create_response log.info("Session: %s, Tree Connect: %s - sending SMB2 Create Request " "for file %s" % (self.tree_connect.session.username, @@ -1009,9 +1009,9 @@ def open(self, impersonation_level, desired_access, file_attributes, request = self.connection.send(create, self.tree_connect.session.session_id, self.tree_connect.tree_connect_id) - return self._open_response(request) + return self._create_response(request) - def _open_response(self, request): + def _create_response(self, request): log.info("Session: %s, Tree Connect: %s - receiving SMB2 Create " "Response" % (self.tree_connect.session.username, self.tree_connect.share_name)) diff --git a/tests/test_open.py b/tests/test_open.py index 67b754d7..717828c9 100644 --- a/tests/test_open.py +++ b/tests/test_open.py @@ -970,12 +970,12 @@ def test_dialect_2_0_2(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1011,12 +1011,12 @@ def test_dialect_2_1_0(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1055,12 +1055,12 @@ def test_dialect_3_0_0(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1101,12 +1101,12 @@ def test_dialect_3_0_2(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1144,12 +1144,12 @@ def test_dialect_3_1_1(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1190,12 +1190,12 @@ def test_open_root_directory(self, smb_real): session.connect() tree.connect() - dir_open.open(ImpersonationLevel.Impersonation, - DirectoryAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_DIRECTORY, - 0, - CreateDisposition.FILE_OPEN_IF, - CreateOptions.FILE_DIRECTORY_FILE) + dir_open.create(ImpersonationLevel.Impersonation, + DirectoryAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_DIRECTORY, + 0, + CreateDisposition.FILE_OPEN_IF, + CreateOptions.FILE_DIRECTORY_FILE) dir_open.close(get_attributes=False) finally: connection.disconnect(True) @@ -1214,12 +1214,12 @@ def test_create_directory(self, smb_real): session.connect() tree.connect() - out_cont = open.open(ImpersonationLevel.Impersonation, - DirectoryAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_DIRECTORY, - 0, - CreateDisposition.FILE_OPEN_IF, - CreateOptions.FILE_DIRECTORY_FILE) + out_cont = open.create(ImpersonationLevel.Impersonation, + DirectoryAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_DIRECTORY, + 0, + CreateDisposition.FILE_OPEN_IF, + CreateOptions.FILE_DIRECTORY_FILE) assert out_cont is None assert open.allocation_size == 0 assert isinstance(open.change_time, datetime) @@ -1284,13 +1284,13 @@ def test_create_file_create_contexts(self, smb_real): query_disk, max_req ] - out_cont = open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE, - create_contexts) + out_cont = open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE, + create_contexts) assert len(out_cont) == 2 assert isinstance(out_cont[0], SMB2CreateQueryMaximalAccessResponse) or \ @@ -1311,12 +1311,12 @@ def test_create_read_write_from_file(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) actual = open.write(b"\x01\x02\x03\x04") assert actual == 4 @@ -1338,12 +1338,12 @@ def test_flush_file(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) open.flush() finally: connection.disconnect(True) @@ -1358,12 +1358,12 @@ def test_close_file_dont_get_attributes(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) old_last_write_time = open.last_write_time old_end_of_file = open.end_of_file @@ -1385,12 +1385,12 @@ def test_close_file_get_attributes(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) old_last_write_time = open.last_write_time old_end_of_file = open.end_of_file @@ -1415,12 +1415,12 @@ def test_read_file_unbuffered(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) open.write(b"\x01") actual = open.read(0, 1, unbuffered=True) @@ -1441,12 +1441,12 @@ def test_read_file_unbuffered_unsupported(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) open.write(b"\x01") with pytest.raises(SMBUnsupportedFeature) as exc: @@ -1474,13 +1474,13 @@ def test_write_file_write_through(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_WRITE_THROUGH) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_WRITE_THROUGH) actual = open.write(b"\x01", write_through=True) assert actual == 1 @@ -1499,13 +1499,13 @@ def test_write_file_write_through_unsupported(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_WRITE_THROUGH) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_WRITE_THROUGH) with pytest.raises(SMBUnsupportedFeature) as exc: open.write(b"\x01", write_through=True) @@ -1532,13 +1532,13 @@ def test_write_file_unbuffered(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_NO_INTERMEDIATE_BUFFERING) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_NO_INTERMEDIATE_BUFFERING) actual = open.write(b"\x01", unbuffered=True) assert actual == 1 @@ -1557,13 +1557,13 @@ def test_write_file_unbuffered_unsupported(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_NO_INTERMEDIATE_BUFFERING) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_NO_INTERMEDIATE_BUFFERING) with pytest.raises(SMBUnsupportedFeature) as exc: open.write(b"\x01", unbuffered=True) @@ -1588,31 +1588,31 @@ def test_query_directory(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - DirectoryAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_DIRECTORY, - ShareAccess.FILE_SHARE_READ | - ShareAccess.FILE_SHARE_WRITE | - ShareAccess.FILE_SHARE_DELETE, - CreateDisposition.FILE_OPEN_IF, - CreateOptions.FILE_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + DirectoryAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_DIRECTORY, + ShareAccess.FILE_SHARE_READ | + ShareAccess.FILE_SHARE_WRITE | + ShareAccess.FILE_SHARE_DELETE, + CreateDisposition.FILE_OPEN_IF, + CreateOptions.FILE_DIRECTORY_FILE) file1 = Open(tree, r"directory\\file1.txt") - file1.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file1.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) file1.write(b"\x01\x02\x03\x04", 0) file2 = Open(tree, r"directory\\file2.log") - file2.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file2.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) file2.write(b"\x05\x06", 0) actual = open.query_directory("*", @@ -1651,16 +1651,16 @@ def test_compounding_related_opens_encrypted(self, smb_real): tree.connect() messages = [ - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.GENERIC_READ | - FilePipePrinterAccessMask.GENERIC_WRITE | - FilePipePrinterAccessMask.DELETE, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_DELETE_ON_CLOSE, - send=False), + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.GENERIC_READ | + FilePipePrinterAccessMask.GENERIC_WRITE | + FilePipePrinterAccessMask.DELETE, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_DELETE_ON_CLOSE, + send=False), open.write(b"\x01\x02\x03\x04", send=False), open.read(0, 4, send=False), open.close(False, send=False) @@ -1695,16 +1695,16 @@ def test_compounding_related_opens_signed(self, smb_real): tree.connect() messages = [ - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.GENERIC_READ | - FilePipePrinterAccessMask.GENERIC_WRITE | - FilePipePrinterAccessMask.DELETE, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE | - CreateOptions.FILE_DELETE_ON_CLOSE, - send=False), + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.GENERIC_READ | + FilePipePrinterAccessMask.GENERIC_WRITE | + FilePipePrinterAccessMask.DELETE, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE | + CreateOptions.FILE_DELETE_ON_CLOSE, + send=False), open.write(b"\x01\x02\x03\x04", send=False), open.read(0, 4, send=False), open.close(False, send=False) @@ -1739,29 +1739,29 @@ def test_compounding_open_requests(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - DirectoryAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_DIRECTORY, - ShareAccess.FILE_SHARE_READ | - ShareAccess.FILE_SHARE_WRITE | - ShareAccess.FILE_SHARE_DELETE, - CreateDisposition.FILE_OPEN_IF, - CreateOptions.FILE_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + DirectoryAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_DIRECTORY, + ShareAccess.FILE_SHARE_READ | + ShareAccess.FILE_SHARE_WRITE | + ShareAccess.FILE_SHARE_DELETE, + CreateDisposition.FILE_OPEN_IF, + CreateOptions.FILE_DIRECTORY_FILE) file1 = Open(tree, r"directory\\file1.txt") - file1.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file1.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) file2 = Open(tree, r"directory\\file2.log") - file2.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file2.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) # create messages for each operation messages = [ @@ -1835,29 +1835,29 @@ def test_compounding_open_requests_unencrypted(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - DirectoryAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_DIRECTORY, - ShareAccess.FILE_SHARE_READ | - ShareAccess.FILE_SHARE_WRITE | - ShareAccess.FILE_SHARE_DELETE, - CreateDisposition.FILE_OPEN_IF, - CreateOptions.FILE_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + DirectoryAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_DIRECTORY, + ShareAccess.FILE_SHARE_READ | + ShareAccess.FILE_SHARE_WRITE | + ShareAccess.FILE_SHARE_DELETE, + CreateDisposition.FILE_OPEN_IF, + CreateOptions.FILE_DIRECTORY_FILE) file1 = Open(tree, r"directory\\file1.txt") - file1.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file1.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) file2 = Open(tree, r"directory\\file2.log") - file2.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - ShareAccess.FILE_SHARE_READ, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + file2.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + ShareAccess.FILE_SHARE_READ, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) # create messages for each operation messages = [ @@ -1932,12 +1932,12 @@ def test_close_file_already_closed(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) open.close() # we will just manually say it is still connected so we get the @@ -1957,12 +1957,12 @@ def test_read_greater_than_max_size(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) with pytest.raises(SMBException) as exc: open.read(0, connection.max_read_size + 1) @@ -1983,12 +1983,12 @@ def test_write_greater_than_max_size(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) with pytest.raises(SMBException) as exc: open.write(b"\x00" * (connection.max_write_size + 1), 0) @@ -2009,12 +2009,12 @@ def test_read_file_multi_credits(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) open.write(b"\x01\x02\x03\x04", 0) actual = open.read(0, 65538) assert actual == b"\x01\x02\x03\x04" @@ -2032,12 +2032,12 @@ def test_receive_message_without_request(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) read_req, unpack_func = open.write(b"\x00", 0, send=False) req = connection.send(read_req, sid=session.session_id, tid=tree.tree_connect_id) @@ -2063,12 +2063,12 @@ def test_receive_with_timeout(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) read_req, unpack_func = open.write(b"\x00", 0, send=False) req = connection.send(read_req, sid=session.session_id, tid=tree.tree_connect_id) @@ -2099,12 +2099,12 @@ def test_close_file_invalid_id(self, smb_real): session.connect() tree.connect() - open.open(ImpersonationLevel.Impersonation, - FilePipePrinterAccessMask.MAXIMUM_ALLOWED, - FileAttributes.FILE_ATTRIBUTE_NORMAL, - 0, - CreateDisposition.FILE_OVERWRITE_IF, - CreateOptions.FILE_NON_DIRECTORY_FILE) + open.create(ImpersonationLevel.Impersonation, + FilePipePrinterAccessMask.MAXIMUM_ALLOWED, + FileAttributes.FILE_ATTRIBUTE_NORMAL, + 0, + CreateDisposition.FILE_OVERWRITE_IF, + CreateOptions.FILE_NON_DIRECTORY_FILE) # create a request for a known failure and pass that into the # _close_response to ensure the exception is thrown