sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-03-12 20:58:37 +03:00
Code

General cleanups of python code, hinted by pyflakes.

Browse Source
This commit is contained in:
Jelmer Vernooij 2010-03-01 04:46:40 +01:00
parent 1ef50d5de4
commit 31a718aa27
14 changed files with 124 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source4/scripting/python/samba/__init__.py
View File

@ -100,7 +100,7 @@ class Ldb(ldb.Ldb):
# Allow admins to force non-sync ldb for all databases
if lp is not None:
nosync_p = lp.get("nosync", "ldb")
if nosync_p is not None and nosync_p == true:
if nosync_p is not None and nosync_p == True:
flags |= FLG_NOSYNC
self.set_create_perms()

2
source4/scripting/python/samba/netcmd/newuser.py
View File

@ -20,7 +20,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import samba.getopt as options
from samba.netcmd import Command, CommandError, Option
from samba.netcmd import Command, Option
from getpass import getpass
from samba.auth import system_session

3
source4/scripting/python/samba/netcmd/pwsettings.py
View File

@ -22,10 +22,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import sys
import samba.getopt as options
import optparse
import ldb
from samba.auth import system_session

227
source4/scripting/python/samba/ntacls.py
View File

@ -19,139 +19,134 @@
#
import os
import tdb
import samba.xattr_native, samba.xattr_tdb
from samba.dcerpc import security, xattr
from samba.ndr import ndr_pack, ndr_unpack
class XattrBackendError(Exception):
"""A generic xattr backend error."""
def checkset_backend(lp,backend,eadbfile):
if backend != None:
if backend == "native":
lp.set("posix:eadb","")
elif backend == "tdb":
if eadbfile != None:
lp.set("posix:eadb",eadbfile)
else:
os.path.abspath(os.path.join(lp.get("private dir"),"eadb.tdb"))
else:
raise XattrBackendError("Unvalid xattr backend choice %s"%backend)
if backend is not None:
if backend == "native":
lp.set("posix:eadb","")
elif backend == "tdb":
if eadbfile != None:
lp.set("posix:eadb",eadbfile)
else:
os.path.abspath(os.path.join(lp.get("private dir"),"eadb.tdb"))
else:
raise XattrBackendError("Unvalid xattr backend choice %s"%backend)
def getntacl(lp,file,backend=None,eadbfile=None):
try:
checkset_backend(lp,backend,eadbfile)
except:
raise
eadbname = lp.get("posix:eadb")
if eadbname != None and eadbname != "" :
try:
attribute = samba.xattr_tdb.wrap_getxattr(eadbname,file,xattr.XATTR_NTACL_NAME)
except:
print "Fail to open %s"%eadbname
attribute = samba.xattr_native.wrap_getxattr(file,xattr.XATTR_NTACL_NAME)
else:
attribute = samba.xattr_native.wrap_getxattr(file,xattr.XATTR_NTACL_NAME)
ntacl = ndr_unpack(xattr.NTACL,attribute)
return ntacl
def getntacl(lp, file, backend=None, eadbfile=None):
checkset_backend(lp, backend, eadbfile)
eadbname = lp.get("posix:eadb")
if eadbname != None and eadbname != "" :
try:
attribute = samba.xattr_tdb.wrap_getxattr(eadbname,file,xattr.XATTR_NTACL_NAME)
except:
print "Fail to open %s" % eadbname
attribute = samba.xattr_native.wrap_getxattr(file,xattr.XATTR_NTACL_NAME)
else:
attribute = samba.xattr_native.wrap_getxattr(file,xattr.XATTR_NTACL_NAME)
ntacl = ndr_unpack(xattr.NTACL,attribute)
return ntacl
def setntacl(lp,file,sddl,domsid,backend=None,eadbfile=None):
try:
checkset_backend(lp,backend,eadbfile)
except:
raise
ntacl=xattr.NTACL()
ntacl.version = 1
sid=security.dom_sid(domsid)
sd = security.descriptor.from_sddl(sddl, sid)
ntacl.info = sd
eadbname = lp.get("posix:eadb")
if eadbname != None and eadbname != "":
try:
attribute = samba.xattr_tdb.wrap_setxattr(eadbname,file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
except:
print "Fail to open %s"%eadbname
attribute = samba.xattr_native.wrap_setxattr(file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
else:
attribute = samba.xattr_native.wrap_setxattr(file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
checkset_backend(lp,backend,eadbfile)
ntacl=xattr.NTACL()
ntacl.version = 1
sid=security.dom_sid(domsid)
sd = security.descriptor.from_sddl(sddl, sid)
ntacl.info = sd
eadbname = lp.get("posix:eadb")
if eadbname != None and eadbname != "":
try:
attribute = samba.xattr_tdb.wrap_setxattr(eadbname,file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
except:
print "Fail to open %s"%eadbname
attribute = samba.xattr_native.wrap_setxattr(file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
else:
attribute = samba.xattr_native.wrap_setxattr(file,xattr.XATTR_NTACL_NAME,ndr_pack(ntacl))
# Takes the access mask of a DS ACE and transform them in a File ACE mask
def ldapmask2filemask(ldm):
RIGHT_DS_CREATE_CHILD = 0x00000001
RIGHT_DS_DELETE_CHILD = 0x00000002
RIGHT_DS_LIST_CONTENTS = 0x00000004
ACTRL_DS_SELF = 0x00000008
RIGHT_DS_READ_PROPERTY = 0x00000010
RIGHT_DS_WRITE_PROPERTY = 0x00000020
RIGHT_DS_DELETE_TREE = 0x00000040
RIGHT_DS_LIST_OBJECT = 0x00000080
RIGHT_DS_CONTROL_ACCESS = 0x00000100
FILE_READ_DATA = 0x0001
FILE_LIST_DIRECTORY = 0x0001
FILE_WRITE_DATA = 0x0002
FILE_ADD_FILE = 0x0002
FILE_APPEND_DATA = 0x0004
FILE_ADD_SUBDIRECTORY = 0x0004
FILE_CREATE_PIPE_INSTANCE = 0x0004
FILE_READ_EA = 0x0008
FILE_WRITE_EA = 0x0010
FILE_EXECUTE = 0x0020
FILE_TRAVERSE = 0x0020
FILE_DELETE_CHILD = 0x0040
FILE_READ_ATTRIBUTES = 0x0080
FILE_WRITE_ATTRIBUTES = 0x0100
DELETE = 0x00010000
READ_CONTROL = 0x00020000
WRITE_DAC = 0x00040000
WRITE_OWNER = 0x00080000
SYNCHRONIZE = 0x00100000
STANDARD_RIGHTS_ALL = 0x001F0000
"""Takes the access mask of a DS ACE and transform them in a File ACE mask"""
RIGHT_DS_CREATE_CHILD = 0x00000001
RIGHT_DS_DELETE_CHILD = 0x00000002
RIGHT_DS_LIST_CONTENTS = 0x00000004
ACTRL_DS_SELF = 0x00000008
RIGHT_DS_READ_PROPERTY = 0x00000010
RIGHT_DS_WRITE_PROPERTY = 0x00000020
RIGHT_DS_DELETE_TREE = 0x00000040
RIGHT_DS_LIST_OBJECT = 0x00000080
RIGHT_DS_CONTROL_ACCESS = 0x00000100
FILE_READ_DATA = 0x0001
FILE_LIST_DIRECTORY = 0x0001
FILE_WRITE_DATA = 0x0002
FILE_ADD_FILE = 0x0002
FILE_APPEND_DATA = 0x0004
FILE_ADD_SUBDIRECTORY = 0x0004
FILE_CREATE_PIPE_INSTANCE = 0x0004
FILE_READ_EA = 0x0008
FILE_WRITE_EA = 0x0010
FILE_EXECUTE = 0x0020
FILE_TRAVERSE = 0x0020
FILE_DELETE_CHILD = 0x0040
FILE_READ_ATTRIBUTES = 0x0080
FILE_WRITE_ATTRIBUTES = 0x0100
DELETE = 0x00010000
READ_CONTROL = 0x00020000
WRITE_DAC = 0x00040000
WRITE_OWNER = 0x00080000
SYNCHRONIZE = 0x00100000
STANDARD_RIGHTS_ALL = 0x001F0000
filemask = ldm & STANDARD_RIGHTS_ALL
#filemask = 0
filemask = ldm & STANDARD_RIGHTS_ALL
if( (ldm & RIGHT_DS_READ_PROPERTY) and (ldm & RIGHT_DS_LIST_CONTENTS) ):
filemask = filemask | (SYNCHRONIZE | FILE_LIST_DIRECTORY |\
FILE_READ_ATTRIBUTES | FILE_READ_EA |\
FILE_READ_DATA | FILE_EXECUTE)
if (ldm & RIGHT_DS_READ_PROPERTY) and (ldm & RIGHT_DS_LIST_CONTENTS):
filemask = filemask | (SYNCHRONIZE | FILE_LIST_DIRECTORY |\
FILE_READ_ATTRIBUTES | FILE_READ_EA |\
FILE_READ_DATA | FILE_EXECUTE)
if( (ldm & RIGHT_DS_WRITE_PROPERTY) ):
filemask = filemask | (SYNCHRONIZE | FILE_WRITE_DATA |\
FILE_APPEND_DATA | FILE_WRITE_EA |\
FILE_WRITE_ATTRIBUTES | FILE_ADD_FILE |\
FILE_ADD_SUBDIRECTORY)
if ldm & RIGHT_DS_WRITE_PROPERTY:
filemask = filemask | (SYNCHRONIZE | FILE_WRITE_DATA |\
FILE_APPEND_DATA | FILE_WRITE_EA |\
FILE_WRITE_ATTRIBUTES | FILE_ADD_FILE |\
FILE_ADD_SUBDIRECTORY)
if( (ldm & RIGHT_DS_CREATE_CHILD) ):
filemask = filemask | (FILE_ADD_SUBDIRECTORY | FILE_ADD_FILE)
if ldm & RIGHT_DS_CREATE_CHILD:
filemask = filemask | (FILE_ADD_SUBDIRECTORY | FILE_ADD_FILE)
if( (ldm & RIGHT_DS_DELETE_CHILD) ):
filemask = filemask | FILE_DELETE_CHILD
if ldm & RIGHT_DS_DELETE_CHILD:
filemask = filemask | FILE_DELETE_CHILD
return filemask
return filemask
# This function takes an the SDDL representation of a DS
# ACL and return the SDDL representation of this ACL adapted
# for files. It's used for Policy object provision
def dsacl2fsacl(dssddl, domsid):
"""
This function takes an the SDDL representation of a DS
ACL and return the SDDL representation of this ACL adapted
for files. It's used for Policy object provision
"""
sid = security.dom_sid(domsid)
ref = security.descriptor.from_sddl(dssddl,sid)
fdescr = security.descriptor()
fdescr.owner_sid = ref.owner_sid
fdescr.group_sid = ref.group_sid
fdescr.type = ref.type
fdescr.revision = ref.revision
fdescr.sacl = ref.sacl
aces = ref.dacl.aces
for i in range(0,len(aces)):
ace = aces[i]
if not ace.type & security.SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT and str(ace.trustee) != security.SID_BUILTIN_PREW2K:
# if fdescr.type & security.SEC_DESC_DACL_AUTO_INHERITED:
ace.flags = ace.flags | security.SEC_ACE_FLAG_OBJECT_INHERIT | security.SEC_ACE_FLAG_CONTAINER_INHERIT
if str(ace.trustee) == security.SID_CREATOR_OWNER:
# For Creator/Owner the IO flag is set as this ACE has only a sense for child objects
ace.flags = ace.flags | security.SEC_ACE_FLAG_INHERIT_ONLY
ace.access_mask = ldapmask2filemask(ace.access_mask)
fdescr.dacl_add(ace)
def dsacl2fsacl(dssddl,domsid):
sid = security.dom_sid(domsid)
ref = security.descriptor.from_sddl(dssddl,sid)
fdescr = security.descriptor()
fdescr.owner_sid = ref.owner_sid
fdescr.group_sid = ref.group_sid
fdescr.type = ref.type
fdescr.revision = ref.revision
fdescr.sacl = ref.sacl
aces = ref.dacl.aces
for i in range(0,len(aces)):
ace = aces[i]
if not ace.type & security.SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT and str(ace.trustee) != security.SID_BUILTIN_PREW2K:
# if fdescr.type & security.SEC_DESC_DACL_AUTO_INHERITED:
ace.flags = ace.flags | security.SEC_ACE_FLAG_OBJECT_INHERIT | security.SEC_ACE_FLAG_CONTAINER_INHERIT
if str(ace.trustee) == security.SID_CREATOR_OWNER:
# For Creator/Owner the IO flag is set as this ACE has only a sense for child objects
ace.flags = ace.flags | security.SEC_ACE_FLAG_INHERIT_ONLY
ace.access_mask = ldapmask2filemask(ace.access_mask)
fdescr.dacl_add(ace)
return fdescr.as_sddl(sid)
return fdescr.as_sddl(sid)

5
source4/scripting/python/samba/provision.py
View File

@ -42,7 +42,7 @@ import ldb
from samba.auth import system_session, admin_session
from samba import glue, version, Ldb, substitute_var, valid_netbios_name
from samba import check_all_substituted, read_and_sub_file, setup_file
from samba import DS_DOMAIN_FUNCTION_2003, DS_DC_FUNCTION_2008, DS_DC_FUNCTION_2008_R2
from samba import DS_DOMAIN_FUNCTION_2003, DS_DC_FUNCTION_2008
from samba.dcerpc import security
from samba.dcerpc.misc import SEC_CHAN_BDC, SEC_CHAN_WKSTA
from samba.idmap import IDmapDB
@ -542,9 +542,6 @@ def setup_samdb_partitions(samdb_path, setup_path, message, lp, session_info,
"""
assert session_info is not None
old_partitions = None
new_partitions = None
# We use options=["modules:"] to stop the modules loading - we
# just want to wipe and re-initialise the database, not start it up

12
source4/scripting/python/samba/provisionbackend.py
View File

@ -169,8 +169,8 @@ class LDAPBackend(ProvisionBackend):
# if another instance of slapd is already running
try:
ldapi_db = Ldb(self.ldapi_uri)
search_ol_rootdse = ldapi_db.search(base="", scope=SCOPE_BASE,
expression="(objectClass=OpenLDAProotDSE)");
ldapi_db.search(base="", scope=SCOPE_BASE,
expression="(objectClass=OpenLDAProotDSE)");
try:
f = open(self.paths.slapdpid, "r")
p = f.read()
@ -181,7 +181,7 @@ class LDAPBackend(ProvisionBackend):
raise ProvisioningError("Warning: Another slapd Instance seems already running on this host, listening to " + self.ldapi_uri + ". Please shut it down before you continue. ")
except LdbError, e:
except LdbError:
pass
# Try to print helpful messages when the user has not specified the path to slapd
@ -237,11 +237,11 @@ class LDAPBackend(ProvisionBackend):
# Wait until the socket appears
try:
ldapi_db = Ldb(self.ldapi_uri, lp=self.lp, credentials=self.credentials)
search_ol_rootdse = ldapi_db.search(base="", scope=SCOPE_BASE,
ldapi_db.search(base="", scope=SCOPE_BASE,
expression="(objectClass=OpenLDAProotDSE)")
# If we have got here, then we must have a valid connection to the LDAP server!
return
except LdbError, e:
except LdbError:
time.sleep(1)
pass
@ -621,7 +621,7 @@ class FDSBackend(LDAPBackend):
lnkattr = self.schema.linked_attributes()
refint_config = data = open(self.setup_path("fedorads-refint-delete.ldif"), 'r').read()
refint_config = open(self.setup_path("fedorads-refint-delete.ldif"), 'r').read()
memberof_config = ""
index_config = ""
argnum = 3

2
source4/scripting/python/samba/schema.py
View File

@ -31,7 +31,7 @@ from samba.dcerpc import security
from samba import read_and_sub_file, substitute_var, check_all_substituted
from samba import Ldb
from samba.ndr import ndr_pack
from ldb import SCOPE_SUBTREE, SCOPE_ONELEVEL, SCOPE_BASE
from ldb import SCOPE_SUBTREE, SCOPE_ONELEVEL
import os
def get_schema_descriptor(domain_sid):

1
source4/scripting/python/samba/tests/dcerpc/sam.py
View File

@ -39,6 +39,7 @@ class SamrTests(RpcInterfaceTestCase):
def test_connect2(self):
handle = self.conn.Connect2(None, security.SEC_FLAG_MAXIMUM_ALLOWED)
self.assertTrue(handle is not None)
def test_EnumDomains(self):
handle = self.conn.Connect2(None, security.SEC_FLAG_MAXIMUM_ALLOWED)

2
source4/scripting/python/samba/tests/provision.py
View File

@ -20,8 +20,6 @@
import os
from samba.provision import setup_secretsdb, findnss
import samba.tests
from ldb import Dn
from samba import param
import unittest
lp = samba.tests.cmdline_loadparm

11
source4/scripting/python/samba/tests/samba3.py
View File

@ -200,17 +200,8 @@ class IdmapDbTestCase(unittest.TestCase):
self.idmapdb.close()
class ShareInfoTestCase(unittest.TestCase):
def setUp(self):
self.shareinfodb = ShareInfoDatabase(os.path.join(DATADIR, "share_info.tdb"))
# FIXME: needs proper data so it can be tested
def tearDown(self):
self.shareinfodb.close()
class ParamTestCase(unittest.TestCase):
def test_init(self):
file = ParamFile()
self.assertTrue(file is not None)

4
source4/scripting/python/samba/tests/samdb.py
View File

@ -17,13 +17,10 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from samba.auth import system_session
from samba.credentials import Credentials
import os
from samba.provision import setup_samdb, guess_names, make_smbconf, find_setup_dir
from samba.samdb import SamDB
from samba.tests import TestCaseInTempDir
from samba.dcerpc import security
from unittest import TestCase
import uuid
from samba import param
@ -48,7 +45,6 @@ class SamDBTestCase(TestCaseInTempDir):
domainguid = str(uuid.uuid4())
policyguid = str(uuid.uuid4())
domainsid = security.random_sid()
hostguid = str(uuid.uuid4())
path = os.path.join(self.tempdir, "samdb.ldb")
session_info = system_session()

1
source4/scripting/python/samba/tests/upgrade.py
View File

@ -17,7 +17,6 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from samba import Ldb
from samba.upgrade import import_wins
from samba.tests import LdbTestCase

1
source4/scripting/python/samba/torture/torture_samr.py
View File

@ -1,6 +1,5 @@
#!/usr/bin/python
import sys
import dcerpc, samr
def test_Connect(pipe):

3
source4/scripting/python/samba/upgradehelpers.py
View File

@ -144,7 +144,7 @@ def find_provision_key_parameters(param, credentials, session_info, paths, smbco
# dc policy guid
attrs8 = ["cn","displayName"]
res8 = samdb.search(expression="(displayName=Default Domain Controllers Policy)",base="CN=Policies,CN=System,"+basedn, \
scope=SCOPE_ONELEVEL, attrs=attrs7)
scope=SCOPE_ONELEVEL, attrs=attrs8)
if len(res8) == 1:
names.policyid_dc = str(res8[0]["cn"]).replace("{","").replace("}","")
else:
@ -168,7 +168,6 @@ def newprovision(names,setup_dir,creds,session,smbconf,provdir,messagefunc):
:param messagefunc: A function for displaying the message of the provision"""
if os.path.isdir(provdir):
shutil.rmtree(provdir)
logstd=os.path.join(provdir,"log.std")
os.chdir(os.path.join(setup_dir,".."))
os.mkdir(provdir)
messagefunc("Provision stored in %s"%provdir)