samba3.py

samba最新软件

        'X': ACB_PWNOEXP,   # No 'X'piry on password
        'I': ACB_DOMTRUST,  # 'I'nterdomain trust account.
        ' ': 0
        }

def decode_acb(text):
    """Decode a ACB field.

    :param text: ACB text
    :return: integer with flags set.
    """
    assert not "[" in text and not "]" in text
    ret = 0
    for x in text:
        ret |= acb_info_mapping[x]
    return ret


class SAMUser:
    """Samba 3 SAM User.
    
    :note: Unknown or unset fields are set to None.
    """
    def __init__(self, name, uid=None, lm_password=None, nt_password=None, acct_ctrl=None, 
                 last_change_time=None, nt_username=None, fullname=None, logon_time=None, logoff_time=None,
                 acct_desc=None, group_rid=None, bad_password_count=None, logon_count=None,
                 domain=None, dir_drive=None, munged_dial=None, homedir=None, logon_script=None,
                 profile_path=None, workstations=None, kickoff_time=None, bad_password_time=None,
                 pass_last_set_time=None, pass_can_change_time=None, pass_must_change_time=None,
                 user_rid=None, unknown_6=None, nt_password_history=None,
                 unknown_str=None, hours=None, logon_divs=None):
        self.username = name
        self.uid = uid
        self.lm_password = lm_password
        self.nt_password = nt_password
        self.acct_ctrl = acct_ctrl
        self.pass_last_set_time = last_change_time
        self.nt_username = nt_username
        self.fullname = fullname
        self.logon_time = logon_time
        self.logoff_time = logoff_time
        self.acct_desc = acct_desc
        self.group_rid = group_rid
        self.bad_password_count = bad_password_count
        self.logon_count = logon_count
        self.domain = domain
        self.dir_drive = dir_drive
        self.munged_dial = munged_dial
        self.homedir = homedir
        self.logon_script = logon_script
        self.profile_path = profile_path
        self.workstations = workstations
        self.kickoff_time = kickoff_time
        self.bad_password_time = bad_password_time
        self.pass_can_change_time = pass_can_change_time
        self.pass_must_change_time = pass_must_change_time
        self.user_rid = user_rid
        self.unknown_6 = unknown_6
        self.nt_password_history = nt_password_history
        self.unknown_str = unknown_str
        self.hours = hours
        self.logon_divs = logon_divs

    def __eq__(self, other): 
        if not isinstance(other, SAMUser):
            return False
        return self.__dict__ == other.__dict__

class SmbpasswdFile:
    """Samba 3 smbpasswd file reader."""
    def __init__(self, file):
        self.users = {}
        f = open(file, 'r')
        for l in f.readlines():
            if len(l) == 0 or l[0] == "#":
                continue # Skip comments and blank lines
            parts = l.split(":")
            username = parts[0]
            uid = int(parts[1])
            acct_ctrl = 0
            last_change_time = None
            if parts[2] == "NO PASSWORD":
                acct_ctrl |= ACB_PWNOTREQ
                lm_password = None
            elif parts[2][0] in ("*", "X"):
                # No password set
                lm_password = None
            else:
                lm_password = parts[2]

            if parts[3][0] in ("*", "X"):
                # No password set
                nt_password = None
            else:
                nt_password = parts[3]

            if parts[4][0] == '[':
                assert "]" in parts[4]
                acct_ctrl |= decode_acb(parts[4][1:-1])
                if parts[5].startswith("LCT-"):
                    last_change_time = int(parts[5][len("LCT-"):], 16)
            else: # old style file
                if username[-1] == "$":
                    acct_ctrl &= ~ACB_NORMAL
                    acct_ctrl |= ACB_WSTRUST

            self.users[username] = SAMUser(username, uid, lm_password, nt_password, acct_ctrl, last_change_time)

        f.close()

    def __len__(self):
        return len(self.users)

    def __getitem__(self, name):
        return self.users[name]

    def __iter__(self):
        return iter(self.users)

    def close(self): # For consistency
        pass


TDBSAM_FORMAT_STRING_V0 = "ddddddBBBBBBBBBBBBddBBwdwdBwwd"
TDBSAM_FORMAT_STRING_V1 = "dddddddBBBBBBBBBBBBddBBwdwdBwwd"
TDBSAM_FORMAT_STRING_V2 = "dddddddBBBBBBBBBBBBddBBBwwdBwwd"
TDBSAM_USER_PREFIX = "USER_"


class LdapSam:
    """Samba 3 LDAP passdb backend reader."""
    def __init__(self, url):
        self.ldap_url = ldap_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)

    def usernames(self):
        """Iterate over the usernames in this Tdb database."""
        for k in self.tdb.keys():
            if k.startswith(TDBSAM_USER_PREFIX):
                yield k[len(TDBSAM_USER_PREFIX):].rstrip("\0")

    __iter__ = usernames
    
    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])
            data = data[4:]
            if length == 0:
                return (None, data)
            return (data[:length].rstrip("\0"), data[length:])

        def unpack_int32(data):
            (value, ) = struct.unpack("<l", data[:4])
            return (value, data[4:])

        def unpack_uint32(data):
            (value, ) = struct.unpack("<L", data[:4])
            return (value, data[4:])

        def unpack_uint16(data):
            (value, ) = struct.unpack("<H", data[:2])
            return (value, data[2:])

        (logon_time, data) = unpack_int32(data)
        (logoff_time, data) = unpack_int32(data)
        (kickoff_time, data) = unpack_int32(data)

        if self.version > 0:
            (bad_password_time, data) = unpack_int32(data)
            if bad_password_time != 0:
                user.bad_password_time = bad_password_time
        (pass_last_set_time, data) = unpack_int32(data)
        (pass_can_change_time, data) = unpack_int32(data)
        (pass_must_change_time, data) = unpack_int32(data)

        if logon_time != 0:
            user.logon_time = logon_time
        user.logoff_time = logoff_time
        user.kickoff_time = kickoff_time
        if pass_last_set_time != 0:
            user.pass_last_set_time = pass_last_set_time
        user.pass_can_change_time = pass_can_change_time

        (user.username, data) = unpack_string(data)
        (user.domain, data) = unpack_string(data)
        (user.nt_username, data) = unpack_string(data)
        (user.fullname, data) = unpack_string(data)
        (user.homedir, data) = unpack_string(data)
        (user.dir_drive, data) = unpack_string(data)
        (user.logon_script, data) = unpack_string(data)
        (user.profile_path, data) = unpack_string(data)
        (user.acct_desc, data) = unpack_string(data)
        (user.workstations, data) = unpack_string(data)
        (user.unknown_str, data) = unpack_string(data)
        (user.munged_dial, data) = unpack_string(data)

        (user.user_rid, data) = unpack_int32(data)
        (user.group_rid, data) = unpack_int32(data)

        (user.lm_password, data) = unpack_string(data)
        (user.nt_password, data) = unpack_string(data)

        if self.version > 1:
            (user.nt_password_history, data) = unpack_string(data)

        (user.acct_ctrl, data) = unpack_uint16(data)
        (_, data) = unpack_uint32(data) # remove_me field
        (user.logon_divs, data) = unpack_uint16(data)
        (hours, data) = unpack_string(data)
        user.hours = []
        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)
        assert len(data) == 0
        return user


def shellsplit(text):
    """Very simple shell-like line splitting.
    
    :param text: Text to split.
    :return: List with parts of the line as strings.
    """
    ret = list()
    inquotes = False
    current = ""
    for c in text:
        if c == "\"":
            inquotes = not inquotes
        elif c in ("\t", "\n", " ") and not inquotes:
            ret.append(current)
            current = ""
        else:
            current += c
    if current != "":
        ret.append(current)
    return ret


class WinsDatabase:
    """Samba 3 WINS database reader."""
    def __init__(self, file):
        self.entries = {}
        f = open(file, 'r')
        assert f.readline().rstrip("\n") == "VERSION 1 0"
        for l in f.readlines():
            if l[0] == "#": # skip comments
                continue
            entries = shellsplit(l.rstrip("\n"))
            name = entries[0]
            ttl = int(entries[1])
            i = 2
            ips = []
            while "." in entries[i]:
                ips.append(entries[i])
                i+=1
            nb_flags = int(entries[i][:-1], 16)
            assert not name in self.entries, "Name %s exists twice" % name
            self.entries[name] = (ttl, ips, nb_flags)
        f.close()

    def __getitem__(self, name):
        return self.entries[name]

    def __len__(self):
        return len(self.entries)

    def __iter__(self):
        return iter(self.entries)

    def items(self):
        """Return the entries in this WINS database."""
        return self.entries.items()

    def close(self): # for consistency
        pass

class Samba3:
    """Samba 3 configuration and state data reader."""
    def __init__(self, libdir, smbconfpath):
        """Open the configuration and data for a Samba 3 installation.

        :param libdir: Library directory
        :param smbconfpath: Path to the smb.conf file.
        """
        self.smbconfpath = smbconfpath
        self.libdir = libdir
        import param
        self.lp = param.ParamFile()
        self.lp.read(self.smbconfpath)

    def libdir_path(self, path):
        if path[0] == "/" or path[0] == ".":
            return path
        return os.path.join(self.libdir, path)

    def get_conf(self):
        return self.lp

    def get_sam_db(self):
        lp = self.get_conf()
        backends = str(lp.get("passdb backend")).split(" ")
        if ":" in backends[0]:
            (name, location) = backends[0].split(":", 2)
        else:
            name = backends[0]
            location = None
        if name == "smbpasswd":
            return SmbpasswdFile(self.libdir_path(location or "smbpasswd"))
        elif name == "tdbsam":
            return TdbSam(self.libdir_path(location or "passdb.tdb"))
        elif name == "ldapsam":
            if location is not None:
                return LdapSam("ldap:%s" % location)
            return LdapSam(lp.get("ldap server"))
        else:
            raise NotImplementedError("unsupported passdb backend %s" % backends[0])

    def get_policy_db(self):
        return PolicyDatabase(self.libdir_path("account_policy.tdb"))
    
    def get_registry(self):
        return Registry(self.libdir_path("registry.tdb"))

    def get_secrets_db(self):
        return SecretsDatabase(self.libdir_path("secrets.tdb"))

    def get_shareinfo_db(self):
        return ShareInfoDatabase(self.libdir_path("share_info.tdb"))

    def get_idmap_db(self):
        return IdmapDatabase(self.libdir_path("winbindd_idmap.tdb"))

    def get_wins_db(self):
        return WinsDatabase(self.libdir_path("wins.dat"))

    def get_shares(self):
        return Shares(self.get_conf(), self.get_shareinfo_db())

    def get_groupmapping_db(self):
        return GroupMappingDatabase(self.libdir_path("group_mapping.tdb"))