python/samba/tests/libsmb-basic.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright Volker Lendecke <vl@samba.org> 2012
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 """Tests for samba.samba3.libsmb."""
  19
  20 from samba.samba3 import libsmb_samba_internal as libsmb
  21 from samba.dcerpc import security
  22 from samba import NTSTATUSError,ntstatus
  23 from samba.ntstatus import NT_STATUS_DELETE_PENDING
  24 from samba.credentials import SMB_ENCRYPTION_REQUIRED
  25 import samba.tests.libsmb
  26 import threading
  27 import sys
  28 import random
  29
  30
  31 class LibsmbTestCase(samba.tests.libsmb.LibsmbTests):
  32
  33     class OpenClose(threading.Thread):
  34
  35         def __init__(self, conn, filename, num_ops):
  36             threading.Thread.__init__(self)
  37             self.conn = conn
  38             self.filename = filename
  39             self.num_ops = num_ops
  40             self.exc = False
  41
  42         def run(self):
  43             c = self.conn
  44             try:
  45                 for i in range(self.num_ops):
  46                     f = c.create(self.filename, CreateDisposition=3,
  47                                  DesiredAccess=security.SEC_STD_DELETE)
  48                     c.delete_on_close(f, True)
  49                     c.close(f)
  50             except Exception:
  51                 self.exc = sys.exc_info()
  52
  53     def test_OpenClose(self):
  54
  55         c = libsmb.Conn(
  56             self.server_ip,
  57             "tmp",
  58             self.lp,
  59             self.creds,
  60             multi_threaded=True,
  61             force_smb1=True)
  62
  63         mythreads = []
  64
  65         for i in range(3):
  66             t = LibsmbTestCase.OpenClose(c, "test" + str(i), 10)
  67             mythreads.append(t)
  68
  69         for t in mythreads:
  70             t.start()
  71
  72         for t in mythreads:
  73             t.join()
  74             if t.exc:
  75                 raise t.exc[0](t.exc[1])
  76
  77     def test_SMB3EncryptionRequired(self):
  78         test_dir = 'testing_%d' % random.randint(0, 0xFFFF)
  79
  80         self.creds.set_smb_encryption(SMB_ENCRYPTION_REQUIRED)
  81
  82         c = libsmb.Conn(self.server_ip, "tmp", self.lp, self.creds)
  83
  84         c.mkdir(test_dir)
  85         c.rmdir(test_dir)
  86
  87     def test_SMB1EncryptionRequired(self):
  88         test_dir = 'testing_%d' % random.randint(0, 0xFFFF)
  89
  90         self.creds.set_smb_encryption(SMB_ENCRYPTION_REQUIRED)
  91
  92         c = libsmb.Conn(
  93             self.server_ip,
  94             "tmp",
  95             self.lp,
  96             self.creds,
  97             force_smb1=True)
  98
  99         c.mkdir(test_dir)
 100         c.rmdir(test_dir)
 101
 102     def test_RenameDstDelOnClose(self):
 103
 104         dstdir = "\\dst-subdir"
 105
 106         c1 = libsmb.Conn(self.server_ip, "tmp", self.lp, self.creds)
 107         c2 = libsmb.Conn(self.server_ip, "tmp", self.lp, self.creds)
 108
 109         try:
 110             c1.deltree(dstdir)
 111         except:
 112             pass
 113
 114         c1.mkdir(dstdir)
 115         dnum = c1.create(dstdir, DesiredAccess=security.SEC_STD_DELETE)
 116         c1.delete_on_close(dnum,1)
 117         c2.savefile("\\src.txt", b"Content")
 118
 119         with self.assertRaises(NTSTATUSError) as cm:
 120             c2.rename("\\src.txt", dstdir + "\\dst.txt")
 121         if (cm.exception.args[0] != NT_STATUS_DELETE_PENDING):
 122             raise AssertionError("Rename must fail with DELETE_PENDING")
 123
 124         c1.delete_on_close(dnum,0)
 125         c1.close(dnum)
 126
 127         try:
 128             c1.deltree(dstdir)
 129             c1.unlink("\\src.txt")
 130         except:
 131             pass
 132
 133     def test_libsmb_CreateContexts(self):
 134         c = libsmb.Conn(self.server_ip, "tmp", self.lp, self.creds)
 135         cc_in = [(libsmb.SMB2_CREATE_TAG_MXAC, b'')]
 136         fnum,cr,cc = c.create_ex("",CreateContexts=cc_in)
 137         self.assertEqual(
 138             cr['file_attributes'] & libsmb.FILE_ATTRIBUTE_DIRECTORY,
 139             libsmb.FILE_ATTRIBUTE_DIRECTORY)
 140         self.assertEqual(cc[0][0],libsmb.SMB2_CREATE_TAG_MXAC)
 141         self.assertEqual(len(cc[0][1]),8)
 142         c.close(fnum)
 143
 144     def test_libsmb_TortureCaseSensitivity(self):
 145         testdir = "test_libsmb_torture_case_sensitivity"
 146         filename = "file"
 147         filepath = testdir + "/" + filename
 148
 149         c = libsmb.Conn(self.server_ip, "tmp", self.lp, self.creds)
 150
 151         try:
 152             c.deltree(testdir)
 153         except:
 154             pass
 155
 156         c.mkdir(testdir)
 157
 158         try:
 159             # Now check for all possible upper-/lowercase combinations:
 160             # - testdir/file
 161             # - TESTDIR/file
 162             # - testdir/FILE
 163             # - TESTDIR/FILE
 164
 165             dircases = [testdir, testdir, testdir.upper(), testdir.upper()]
 166             filecases = [filename, filename.upper(), filename, filename.upper()]
 167             tcases = [{'dir':dir, 'file':file} for dir,file in zip(dircases,filecases)]
 168
 169             for tcase in tcases:
 170                 testpath = tcase['dir'] + "/" + tcase['file']
 171
 172                 # Create the testfile
 173                 h = c.create(filepath,
 174                              DesiredAccess=security.SEC_FILE_ALL,
 175                              CreateDisposition=libsmb.FILE_OPEN_IF)
 176                 c.close(h)
 177
 178                 # Open
 179                 c.loadfile(testpath)
 180
 181                 # Search
 182                 ls = [f['name'] for f in c.list(tcase['dir'], mask=tcase['file'])]
 183                 self.assertIn(filename, ls, msg='When searching for "%s" not found in "%s"' % (tcase['file'], tcase['dir']))
 184
 185                 # Rename
 186                 c.rename(testpath, tcase['dir'] + "/tmp")
 187                 c.rename(tcase['dir'] + "/TMP", filepath)
 188                 c.loadfile(testpath)
 189
 190                 # Delete
 191                 c.unlink(testpath)
 192
 193         finally:
 194             c.deltree(testdir)
 195
 196     def test_libsmb_TortureDirCaseSensitive(self):
 197         c = libsmb.Conn(self.server_ip, "lowercase", self.lp, self.creds)
 198         c.mkdir("subdir")
 199         c.mkdir("subdir/b")
 200         ret = c.chkpath("SubDir/b")
 201         c.rmdir("subdir/b")
 202         c.rmdir("subdir")
 203         self.assertTrue(ret)
 204
 205     def test_libsmb_shadow_depot(self):
 206         c = libsmb.Conn(self.server_ip, "shadow_depot", self.lp, self.creds)
 207         try:
 208             fnum=c.create("x:y",CreateDisposition=libsmb.FILE_CREATE)
 209             c.close(fnum)
 210         except:
 211             self.fail()
 212         finally:
 213             # "c" might have crashed, get a new connection
 214             c1 = libsmb.Conn(self.server_ip, "shadow_depot", self.lp, self.creds)
 215             c1.unlink("x")
 216             c1 = None
 217
 218 if __name__ == "__main__":
 219     import unittest
 220     unittest.main()