Source Code for Module pike.test.encryption

  1  # 
  2  # Copyright (c) 2016, EMC Corporation 
  3  # All rights reserved. 
  4  # 
  5  # Redistribution and use in source and binary forms, with or without 
  6  # modification, are permitted provided that the following conditions are met: 
  7  # 
  8  # 1. Redistributions of source code must retain the above copyright notice, 
  9  # this list of conditions and the following disclaimer. 
 10  # 2. Redistributions in binary form must reproduce the above copyright notice, 
 11  # this list of conditions and the following disclaimer in the documentation 
 12  # and/or other materials provided with the distribution. 
 13  # 
 14  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
 15  # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 16  # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 17  # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 
 18  # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 19  # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 20  # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 21  # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 22  # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
 23  # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 24  # POSSIBILITY OF SUCH DAMAGE. 
 25  # 
 26  # Module Name: 
 27  # 
 28  #        encryption.py 
 29  # 
 30  # Abstract: 
 31  # 
 32  #        Test SMB3 Encryption and negotiation options 
 33  # 
 34  # Authors: Masen Furer (masen.furer@dell.com) 
 35  # 
 36   
 37  import pike.crypto as crypto 
 38  import pike.model as model 
 39  import pike.smb2 as smb2 
 40  import pike.test 
 41   
 42 -class TestEncryption(pike.test.PikeTest): 
 43 -    def test_smb_3_0_encryption(self): 
 44          client = model.Client(dialects=[smb2.DIALECT_SMB3_0]) 
 45          conn = client.connect(self.server) 
 46          conn.negotiate() 
 47          self.assertEqual(conn.negotiate_response.dialect_revision, 
 48                           smb2.DIALECT_SMB3_0) 
 49          self.assertTrue(conn.negotiate_response.capabilities & 
 50                          smb2.SMB2_GLOBAL_CAP_ENCRYPTION) 
 51          chan = conn.session_setup(self.creds) 
 52          chan.session.encrypt_data = True 
 53          self.assertIsNotNone(chan.session.encryption_context) 
 54          self.assertEqual(chan.session.encryption_context.aes_mode, 
 55                           crypto.AES.MODE_CCM) 
 56          tree = chan.tree_connect(self.share) 
 57          self.assertIsNotNone(tree.tree_connect_response.parent.parent.transform) 
 58   
 59 -    def test_smb_3_0_2_encryption(self): 
 60          client = model.Client(dialects=[smb2.DIALECT_SMB3_0_2]) 
 61          conn = client.connect(self.server) 
 62          conn.negotiate() 
 63          self.assertEqual(conn.negotiate_response.dialect_revision, 
 64                           smb2.DIALECT_SMB3_0_2) 
 65          self.assertTrue(conn.negotiate_response.capabilities & 
 66                          smb2.SMB2_GLOBAL_CAP_ENCRYPTION) 
 67          chan = conn.session_setup(self.creds) 
 68          chan.session.encrypt_data = True 
 69          self.assertIsNotNone(chan.session.encryption_context) 
 70          self.assertEqual(chan.session.encryption_context.aes_mode, 
 71                           crypto.AES.MODE_CCM) 
 72          tree = chan.tree_connect(self.share) 
 73          self.assertIsNotNone(tree.tree_connect_response.parent.parent.transform) 
 74   
 75 -    def test_smb_3_1_1_encryption_gcm(self): 
 76          client = model.Client(dialects=[smb2.DIALECT_SMB3_0, 
 77                                          smb2.DIALECT_SMB3_1_1]) 
 78          conn = client.connect(self.server) 
 79          conn.negotiate(ciphers=[crypto.SMB2_AES_128_GCM]) 
 80          self.assertEqual(conn.negotiate_response.dialect_revision, 
 81                           smb2.DIALECT_SMB3_1_1) 
 82          self.assertFalse(conn.negotiate_response.capabilities & 
 83                          smb2.SMB2_GLOBAL_CAP_ENCRYPTION) 
 84          chan = conn.session_setup(self.creds) 
 85          chan.session.encrypt_data = True 
 86          self.assertIsNotNone(chan.session.encryption_context) 
 87          self.assertEqual(chan.session.encryption_context.aes_mode, 
 88                           crypto.AES.MODE_GCM) 
 89          tree = chan.tree_connect(self.share) 
 90          self.assertIsNotNone(tree.tree_connect_response.parent.parent.transform) 
 91   
 92 -    def test_smb_3_1_1_encryption_ccm(self): 
 93          client = model.Client(dialects=[smb2.DIALECT_SMB3_0, 
 94                                          smb2.DIALECT_SMB3_1_1]) 
 95          conn = client.connect(self.server) 
 96          conn.negotiate(ciphers=[crypto.SMB2_AES_128_CCM]) 
 97          self.assertEqual(conn.negotiate_response.dialect_revision, 
 98                           smb2.DIALECT_SMB3_1_1) 
 99          self.assertFalse(conn.negotiate_response.capabilities & 
100                          smb2.SMB2_GLOBAL_CAP_ENCRYPTION) 
101          chan = conn.session_setup(self.creds) 
102          chan.session.encrypt_data = True 
103          self.assertIsNotNone(chan.session.encryption_context) 
104          self.assertEqual(chan.session.encryption_context.aes_mode, 
105                           crypto.AES.MODE_CCM) 
106          tree = chan.tree_connect(self.share) 
107          self.assertIsNotNone(tree.tree_connect_response.parent.parent.transform) 
108   
109 -    def test_smb_3_1_1_compound(self): 
110          client = model.Client(dialects=[smb2.DIALECT_SMB3_0, 
111                                          smb2.DIALECT_SMB3_1_1]) 
112          conn = client.connect(self.server) 
113          conn.negotiate(ciphers=[crypto.SMB2_AES_128_GCM]) 
114          self.assertEqual(conn.negotiate_response.dialect_revision, 
115                           smb2.DIALECT_SMB3_1_1) 
116          self.assertFalse(conn.negotiate_response.capabilities & 
117                          smb2.SMB2_GLOBAL_CAP_ENCRYPTION) 
118          chan = conn.session_setup(self.creds) 
119          chan.session.encrypt_data = True 
120          self.assertIsNotNone(chan.session.encryption_context) 
121          self.assertEqual(chan.session.encryption_context.aes_mode, 
122                           crypto.AES.MODE_GCM) 
123          chan.session.encrypt_data = True 
124          tree = chan.tree_connect(self.share) 
125          self.assertIsNotNone(tree.tree_connect_response.parent.parent.transform) 
126   
127          nb_req = chan.frame() 
128          smb_req1 = chan.request(nb_req, obj=tree) 
129          smb_req2 = chan.request(nb_req, obj=tree) 
130          create_req = smb2.CreateRequest(smb_req1) 
131          close_req = smb2.CloseRequest(smb_req2) 
132   
133          create_req.name = 'hello.txt' 
134          create_req.desired_access = pike.smb2.GENERIC_READ | pike.smb2.GENERIC_WRITE 
135          create_req.file_attributes = pike.smb2.FILE_ATTRIBUTE_NORMAL 
136          create_req.create_disposition = pike.smb2.FILE_OPEN_IF 
137   
138          max_req = pike.smb2.MaximalAccessRequest(create_req) 
139   
140          close_req.file_id = smb2.RELATED_FID 
141          smb_req2.flags |= smb2.SMB2_FLAGS_RELATED_OPERATIONS 
142          resp = chan.connection.transceive(nb_req) 
143          parent = resp[0].parent 
144          self.assertIsNotNone(parent.transform) 
145          for r in resp: 
146              self.assertEqual(r.parent, parent) 
147   
148  if __name__ == "__main__": 
149      pike.test.unittest.main() 
150