REGISTRY_DB_VERSION = 1
import os
+import struct
import tdb
+def fetch_uint32(tdb, key):
+ try:
+ data = tdb[key]
+ except KeyError:
+ return None
+ assert len(data) == 4
+ return struct.unpack("<L", data)[0]
+
+
+def fetch_int32(tdb, key):
+ try:
+ data = tdb[key]
+ except KeyError:
+ return None
+ assert len(data) == 4
+ return struct.unpack("<l", data)[0]
+
+
class TdbDatabase(object):
"""Simple Samba 3 TDB database reader."""
def __init__(self, file):
def keys(self):
"""Return list with all the keys."""
- return [k.rstrip("\x00") for k in self.tdb.keys() if not k.startswith(REGISTRY_VALUE_PREFIX)]
+ return [k.rstrip("\x00") for k in self.tdb.iterkeys() if not k.startswith(REGISTRY_VALUE_PREFIX)]
def subkeys(self, key):
"""Retrieve the subkeys for the specified key.
data = self.tdb.get("%s\x00" % key)
if data is None:
return []
- import struct
(num, ) = struct.unpack("<L", data[0:4])
keys = data[4:].split("\0")
assert keys[-1] == ""
if data is None:
return {}
ret = {}
- import struct
(num, ) = struct.unpack("<L", data[0:4])
data = data[4:]
for i in range(num):
:param file: Path to the file to open.
"""
super(PolicyDatabase, self).__init__(file)
- self.min_password_length = self.tdb.fetch_uint32("min password length\x00")
- self.password_history = self.tdb.fetch_uint32("password history\x00")
- self.user_must_logon_to_change_password = self.tdb.fetch_uint32("user must logon to change pasword\x00")
- self.maximum_password_age = self.tdb.fetch_uint32("maximum password age\x00")
- self.minimum_password_age = self.tdb.fetch_uint32("minimum password age\x00")
- self.lockout_duration = self.tdb.fetch_uint32("lockout duration\x00")
- self.reset_count_minutes = self.tdb.fetch_uint32("reset count minutes\x00")
- self.bad_lockout_minutes = self.tdb.fetch_uint32("bad lockout minutes\x00")
- self.disconnect_time = self.tdb.fetch_int32("disconnect time\x00")
- self.refuse_machine_password_change = self.tdb.fetch_uint32("refuse machine password change\x00")
+ self.min_password_length = fetch_uint32(self.tdb, "min password length\x00")
+ self.password_history = fetch_uint32(self.tdb, "password history\x00")
+ self.user_must_logon_to_change_password = fetch_uint32(self.tdb, "user must logon to change pasword\x00")
+ self.maximum_password_age = fetch_uint32(self.tdb, "maximum password age\x00")
+ self.minimum_password_age = fetch_uint32(self.tdb, "minimum password age\x00")
+ self.lockout_duration = fetch_uint32(self.tdb, "lockout duration\x00")
+ self.reset_count_minutes = fetch_uint32(self.tdb, "reset count minutes\x00")
+ self.bad_lockout_minutes = fetch_uint32(self.tdb, "bad lockout minutes\x00")
+ self.disconnect_time = fetch_int32(self.tdb, "disconnect time\x00")
+ self.refuse_machine_password_change = fetch_uint32(self.tdb, "refuse machine password change\x00")
# FIXME: Read privileges as well
class GroupMappingDatabase(TdbDatabase):
"""Samba 3 group mapping database reader."""
def _check_version(self):
- assert self.tdb.fetch_int32("INFO/version\x00") in (GROUPDB_DATABASE_VERSION_V1, GROUPDB_DATABASE_VERSION_V2)
+ assert fetch_int32(self.tdb, "INFO/version\x00") in (GROUPDB_DATABASE_VERSION_V1, GROUPDB_DATABASE_VERSION_V2)
def groupsids(self):
"""Retrieve the SIDs for the groups in this database.
:return: List with sids as strings.
"""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith(GROUP_PREFIX):
yield k[len(GROUP_PREFIX):].rstrip("\0")
data = self.tdb.get("%s%s\0" % (GROUP_PREFIX, sid))
if data is None:
return data
- import struct
(gid, sid_name_use) = struct.unpack("<lL", data[0:8])
(nt_name, comment, _) = data[8:].split("\0")
return (gid, sid_name_use, nt_name, comment)
def aliases(self):
"""Retrieve the aliases in this database."""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith(MEMBEROF_PREFIX):
yield k[len(MEMBEROF_PREFIX):].rstrip("\0")
class IdmapDatabase(TdbDatabase):
"""Samba 3 ID map database reader."""
def _check_version(self):
- assert self.tdb.fetch_int32("IDMAP_VERSION\0") == IDMAP_VERSION_V2
+ assert fetch_int32(self.tdb, "IDMAP_VERSION\0") == IDMAP_VERSION_V2
def uids(self):
"""Retrieve a list of all uids in this database."""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith(IDMAP_USER_PREFIX):
yield int(k[len(IDMAP_USER_PREFIX):].rstrip("\0"))
def gids(self):
"""Retrieve a list of all gids in this database."""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith(IDMAP_GROUP_PREFIX):
yield int(k[len(IDMAP_GROUP_PREFIX):].rstrip("\0"))
def get_user_hwm(self):
"""Obtain the user high-water mark."""
- return self.tdb.fetch_uint32(IDMAP_HWM_USER)
+ return fetch_uint32(self.tdb, IDMAP_HWM_USER)
def get_group_hwm(self):
"""Obtain the group high-water mark."""
- return self.tdb.fetch_uint32(IDMAP_HWM_GROUP)
+ return fetch_uint32(self.tdb, IDMAP_HWM_GROUP)
class SecretsDatabase(TdbDatabase):
return self.tdb.get("SECRETS/DOMGUID/%s" % host)
def ldap_dns(self):
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith("SECRETS/LDAP_BIND_PW/"):
yield k[len("SECRETS/LDAP_BIND_PW/"):].rstrip("\0")
:return: Iterator over the names of domains in this database.
"""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith("SECRETS/SID/"):
yield k[len("SECRETS/SID/"):].rstrip("\0")
return self.tdb.get("SECRETS/AFS_KEYFILE/%s" % host)
def get_machine_sec_channel_type(self, host):
- return self.tdb.fetch_uint32("SECRETS/MACHINE_SEC_CHANNEL_TYPE/%s" % host)
+ return fetch_uint32(self.tdb, "SECRETS/MACHINE_SEC_CHANNEL_TYPE/%s" % host)
def get_machine_last_change_time(self, host):
- return self.tdb.fetch_uint32("SECRETS/MACHINE_LAST_CHANGE_TIME/%s" % host)
+ return fetch_uint32(self.tdb, "SECRETS/MACHINE_LAST_CHANGE_TIME/%s" % host)
def get_machine_password(self, host):
return self.tdb.get("SECRETS/MACHINE_PASSWORD/%s" % host)
return self.tdb.get("SECRETS/$DOMTRUST.ACC/%s" % host)
def trusted_domains(self):
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith("SECRETS/$DOMTRUST.ACC/"):
yield k[len("SECRETS/$DOMTRUST.ACC/"):].rstrip("\0")
class ShareInfoDatabase(TdbDatabase):
"""Samba 3 Share Info database reader."""
def _check_version(self):
- assert self.tdb.fetch_int32("INFO/version\0") in (SHARE_DATABASE_VERSION_V1, SHARE_DATABASE_VERSION_V2)
+ assert fetch_int32(self.tdb, "INFO/version\0") in (SHARE_DATABASE_VERSION_V1, SHARE_DATABASE_VERSION_V2)
def get_secdesc(self, name):
"""Obtain the security descriptor on a particular share.
return secdesc
-class Shares:
+class Shares(object):
"""Container for share objects."""
def __init__(self, lp, shareinfo):
self.lp = lp
return ret
-class SAMUser:
+class SAMUser(object):
"""Samba 3 SAM User.
:note: Unknown or unset fields are set to None.
return False
return self.__dict__ == other.__dict__
-class SmbpasswdFile:
+
+class SmbpasswdFile(object):
"""Samba 3 smbpasswd file reader."""
def __init__(self, file):
self.users = {}
TDBSAM_USER_PREFIX = "USER_"
-class LdapSam:
+class LdapSam(object):
"""Samba 3 LDAP passdb backend reader."""
def __init__(self, url):
- self.ldap_url = ldap_url
+ self.ldap_url = url
class TdbSam(TdbDatabase):
"""Samba 3 TDB passdb backend reader."""
def _check_version(self):
- self.version = self.tdb.fetch_uint32("INFO/version\0") or 0
- assert self.version in (0, 1, 2)
+ self.version = fetch_uint32(self.tdb, "INFO/version\0") or 0
+ assert self.version in (0, 1, 2, 3)
def usernames(self):
"""Iterate over the usernames in this Tdb database."""
- for k in self.tdb.keys():
+ for k in self.tdb.iterkeys():
if k.startswith(TDBSAM_USER_PREFIX):
yield k[len(TDBSAM_USER_PREFIX):].rstrip("\0")
def __getitem__(self, name):
data = self.tdb["%s%s\0" % (TDBSAM_USER_PREFIX, name)]
user = SAMUser(name)
- import struct
def unpack_string(data):
(length, ) = struct.unpack("<L", data[:4])
for entry in hours:
for i in range(8):
user.hours.append(ord(entry) & (2 ** i) == (2 ** i))
- (user.bad_password_count, data) = unpack_uint16(data)
- (user.logon_count, data) = unpack_uint16(data)
- (user.unknown_6, data) = unpack_uint32(data)
+ # FIXME (reactivate also the tests in tests/samba3.py after fixing this)
+ #(user.bad_password_count, data) = unpack_uint16(data)
+ #(user.logon_count, data) = unpack_uint16(data)
+ #(user.unknown_6, data) = unpack_uint32(data)
assert len(data) == 0
return user
return ret
-class WinsDatabase:
+class WinsDatabase(object):
"""Samba 3 WINS database reader."""
def __init__(self, file):
self.entries = {}
def close(self): # for consistency
pass
-class Samba3:
+
+class ParamFile(object):
+ """Simple smb.conf-compatible file parser
+
+ Does not use a parameter table, unlike the "normal".
+ """
+
+ def __init__(self, sections=None):
+ self._sections = sections or {}
+
+ def _sanitize_name(self, name):
+ return name.strip().lower().replace(" ","")
+
+ def __repr__(self):
+ return "ParamFile(%r)" % self._sections
+
+ def read(self, filename):
+ """Read a file.
+
+ :param filename: Path to the file
+ """
+ section = None
+ for i, l in enumerate(open(filename, 'r').xreadlines()):
+ l = l.strip()
+ if not l or l[0] == '#' or l[0] == ';':
+ continue
+ if l[0] == "[" and l[-1] == "]":
+ section = self._sanitize_name(l[1:-1])
+ self._sections.setdefault(section, {})
+ elif "=" in l:
+ (k, v) = l.split("=", 1)
+ self._sections[section][self._sanitize_name(k)] = v
+ else:
+ raise Exception("Unable to parser line %d: %r" % (i+1,l))
+
+ def get(self, param, section=None):
+ """Return the value of a parameter.
+
+ :param param: Parameter name
+ :param section: Section name, defaults to "global"
+ :return: parameter value as string if found, None otherwise.
+ """
+ if section is None:
+ section = "global"
+ section = self._sanitize_name(section)
+ if not section in self._sections:
+ return None
+ param = self._sanitize_name(param)
+ if not param in self._sections[section]:
+ return None
+ return self._sections[section][param].strip()
+
+ def __getitem__(self, section):
+ return self._sections[section]
+
+ def get_section(self, section):
+ return self._sections.get(section)
+
+ def add_section(self, section):
+ self._sections[self._sanitize_name(section)] = {}
+
+ def set_string(self, name, value):
+ self._sections["global"][name] = value
+
+ def get_string(self, name):
+ return self._sections["global"].get(name)
+
+
+class Samba3(object):
"""Samba 3 configuration and state data reader."""
def __init__(self, libdir, smbconfpath):
"""Open the configuration and data for a Samba 3 installation.
"""
self.smbconfpath = smbconfpath
self.libdir = libdir
- import param
- self.lp = param.ParamFile()
+ self.lp = ParamFile()
self.lp.read(self.smbconfpath)
def libdir_path(self, path):
def get_sam_db(self):
lp = self.get_conf()
- backends = str(lp.get("passdb backend")).split(" ")
+ backends = (lp.get("passdb backend") or "").split(" ")
if ":" in backends[0]:
(name, location) = backends[0].split(":", 2)
else:
git.samba.org / ira / wip.git / blobdiff