r26590: Parsing routines for the smbpasswd file and idmap database.

(This used to be commit 0c14e16a2bc965d3319ca990089566bff98a47fe)

2 files changed, 140 insertions, 9 deletions

diff --git a/source4/scripting/python/samba/samba3.py b/source4/scripting/python/samba/samba3.py
index 509ee29c1d..b75b24ba34 100644
--- a/source4/scripting/python/samba/samba3.py
+++ b/source4/scripting/python/samba/samba3.py
@@ -108,16 +108,47 @@ class GroupMappingDatabase:
 
 
 # High water mark keys
-HWM_GROUP = "GROUP HWM"
-HWM_USER = "USER HWM"
+IDMAP_HWM_GROUP = "GROUP HWM\0"
+IDMAP_HWM_USER = "USER HWM\0"
+
+IDMAP_GROUP_PREFIX = "GID "
+IDMAP_USER_PREFIX = "UID "
 
 # idmap version determines auto-conversion
-IDMAP_VERSION = 2
+IDMAP_VERSION_V2 = 2
 
 class IdmapDatabase:
     def __init__(self, file):
         self.tdb = tdb.Tdb(file, flags=os.O_RDONLY)
-        assert self.tdb.fetch_int32("IDMAP_VERSION") == IDMAP_VERSION
+        assert self.tdb.fetch_int32("IDMAP_VERSION\0") == IDMAP_VERSION_V2
+
+    def uids(self):
+        for k in self.tdb.keys():
+            if k.startswith(IDMAP_USER_PREFIX):
+                yield int(k[len(IDMAP_USER_PREFIX):].rstrip("\0"))
+
+    def gids(self):
+        for k in self.tdb.keys():
+            if k.startswith(IDMAP_GROUP_PREFIX):
+                yield int(k[len(IDMAP_GROUP_PREFIX):].rstrip("\0"))
+
+    def get_user_sid(self, uid):
+        data = self.tdb.get("%s%d\0" % (IDMAP_USER_PREFIX, uid))
+        if data is None:
+            return data
+        return data.rstrip("\0")
+
+    def get_group_sid(self, gid):
+        data = self.tdb.get("%s%d\0" % (IDMAP_GROUP_PREFIX, gid))
+        if data is None:
+            return data
+        return data.rstrip("\0")
+
+    def get_user_hwm(self):
+        return self.tdb.fetch_uint32(IDMAP_HWM_USER)
+
+    def get_group_hwm(self):
+        return self.tdb.fetch_uint32(IDMAP_HWM_GROUP)
 
     def close(self):
         self.tdb.close()
@@ -181,6 +212,7 @@ class ShareInfoDatabase:
     def get_secdesc(self, name):
         secdesc = self.tdb.get("SECDESC/%s" % name)
         # FIXME: Run ndr_pull_security_descriptor
+        return secdesc
 
     def close(self):
         self.tdb.close()
@@ -220,11 +252,65 @@ acb_info_mapping = {
         ' ': 0
         }
 
+def decode_acb(text):
+    assert not "[" in text and not "]" in text
+    ret = 0
+    for x in text:
+        ret |= acb_info_mapping[x]
+    return ret
 
-class Smbpasswd:
+
+class SmbpasswdFile:
     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] = (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 close(self): # For consistency
         pass
 
+
 TDBSAM_FORMAT_STRING_V0 = "ddddddBBBBBBBBBBBBddBBwdwdBwwd"
 TDBSAM_FORMAT_STRING_V1 = "dddddddBBBBBBBBBBBBddBBwdwdBwwd"
 TDBSAM_FORMAT_STRING_V2 = "dddddddBBBBBBBBBBBBddBBBwwdBwwd"
diff --git a/source4/scripting/python/samba/tests/samba3.py b/source4/scripting/python/samba/tests/samba3.py
index 92d11d11d4..580bcfa3c2 100644
--- a/source4/scripting/python/samba/tests/samba3.py
+++ b/source4/scripting/python/samba/tests/samba3.py
@@ -19,7 +19,7 @@
 
 import unittest
 from samba.samba3 import (GroupMappingDatabase, Registry, PolicyDatabase, SecretsDatabase, TdbSam,
-                          WinsDatabase)
+                          WinsDatabase, SmbpasswdFile, ACB_NORMAL, IdmapDatabase)
 import os
 
 DATADIR=os.path.join(os.path.dirname(__file__), "../../../../../testdata/samba3")
@@ -109,6 +109,51 @@ class WinsDatabaseTestCase(unittest.TestCase):
     def tearDown(self):
         self.winsdb.close()
 
-# FIXME: smbpasswd
-# FIXME: idmapdb
-# FIXME: Shares
+class SmbpasswdTestCase(unittest.TestCase):
+    def setUp(self):
+        self.samdb = SmbpasswdFile(os.path.join(DATADIR, "smbpasswd"))
+
+    def test_length(self):
+        self.assertEquals(3, len(self.samdb))
+
+    def test_get_user(self):
+        self.assertEquals((0, "552902031BEDE9EFAAD3B435B51404EE", "878D8014606CDA29677A44EFA1353FC7", ACB_NORMAL, int(1125418267)), self.samdb["rootpw"])
+
+    def tearDown(self):
+        self.samdb.close()
+
+
+class IdmapDbTestCase(unittest.TestCase):
+    def setUp(self):
+        self.idmapdb = IdmapDatabase(os.path.join(DATADIR, "winbindd_idmap.tdb"))
+
+    def test_user_hwm(self):
+        self.assertEquals(10000, self.idmapdb.get_user_hwm())
+
+    def test_group_hwm(self):
+        self.assertEquals(10002, self.idmapdb.get_group_hwm())
+
+    def test_uids(self):
+        self.assertEquals(1, len(list(self.idmapdb.uids())))
+
+    def test_gids(self):
+        self.assertEquals(3, len(list(self.idmapdb.gids())))
+
+    def test_get_user_sid(self):
+        self.assertEquals("S-1-5-21-58189338-3053988021-627566699-501", self.idmapdb.get_user_sid(65534))
+
+    def test_get_group_sid(self):
+        self.assertEquals("S-1-5-21-2447931902-1787058256-3961074038-3007", self.idmapdb.get_group_sid(10001))
+
+    def tearDown(self):
+        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()