class DCKeytabTests(tests.TestCase):
    def setUp(self):
        super(DCKeytabTests, self).setUp()
        self.lp = LoadParm()
        self.lp.load_default()
        self.creds = self.insta_creds(template=self.get_credentials())
        self.ktfile = os.path.join(self.lp.get('private dir'), 'test.keytab')
        self.principal = self.creds.get_principal()

    def tearDown(self):
        super(DCKeytabTests, self).tearDown()
        os.remove(self.ktfile)

    def test_export_keytab(self):
        net = Net(None, self.lp)
        net.export_keytab(keytab=self.ktfile, principal=self.principal)
        assert os.path.exists(self.ktfile), 'keytab was not created'
        with open_bytes(self.ktfile) as bytes_kt:
            result = ''
            for c in bytes_kt.read():
                if c in string.printable:
                    result += c
            principal_parts = self.principal.split('@')
            assert principal_parts[0] in result and \
                   principal_parts[1] in result, \
                        'Principal not found in generated keytab'

 def test_setntacl(self):
     lp = LoadParm()
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     open(self.tempf, 'w').write("empty")
     lp.set("posix:eadb",os.path.join(self.tempdir,"eadbtest.tdb"))
     setntacl(lp, self.tempf, acl, "S-1-5-21-2212615479-2695158682-2101375467")
     os.unlink(os.path.join(self.tempdir,"eadbtest.tdb"))

 def test_setntacl_forcenative(self):
     if os.getuid() == 0:
         raise SkipTest("Running test as root, test skipped")
     lp = LoadParm()
     open(self.tempf, 'w').write("empty")
     lp.set("posix:eadb", os.path.join(self.tempdir, "eadbtest.tdb"))
     self.assertRaises(Exception, setntacl, lp, self.tempf, NTACL_SDDL,
                       DOMAIN_SID, "native")

 def test_setntacl_forcenative(self):
     if os.getuid() == 0:
         raise SkipTest("Running test as root, test skipped")
     lp = LoadParm()
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     open(self.tempf, 'w').write("empty")
     lp.set("posix:eadb", os.path.join(self.tempdir,"eadbtest.tdb"))
     self.assertRaises(Exception, setntacl, lp, self.tempf ,acl,
         "S-1-5-21-2212615479-2695158682-2101375467","native")

 def test_setntacl_getntacl(self):
     lp = LoadParm()
     open(self.tempf, 'w').write("empty")
     lp.set("posix:eadb", os.path.join(self.tempdir, "eadbtest.tdb"))
     setntacl(lp, self.tempf, NTACL_SDDL, DOMAIN_SID)
     facl = getntacl(lp, self.tempf)
     anysid = security.dom_sid(security.SID_NT_SELF)
     self.assertEquals(facl.as_sddl(anysid), NTACL_SDDL)
     os.unlink(os.path.join(self.tempdir, "eadbtest.tdb"))

 def test_setntacl_getntacl(self):
     lp = LoadParm()
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     open(self.tempf, 'w').write("empty")
     lp.set("posix:eadb",os.path.join(self.tempdir,"eadbtest.tdb"))
     setntacl(lp,self.tempf,acl,"S-1-5-21-2212615479-2695158682-2101375467")
     facl = getntacl(lp,self.tempf)
     anysid = security.dom_sid(security.SID_NT_SELF)
     self.assertEquals(facl.as_sddl(anysid),acl)
     os.unlink(os.path.join(self.tempdir,"eadbtest.tdb"))

def parametros(fichero=False):
    """
    Intento ser un poco cuidadoso con la forma en que obtenemos la configuración
    """
    try:
        lp = LoadParm()
        lp.load_default() if fichero is False else lp.load(fichero)
        return lp
    except RuntimeError as e:
        log.warning(e.message)
        raise ConfiguracionException(e)

 def test_setntacl(self):
     random.seed()
     lp = LoadParm()
     path = os.environ['SELFTEST_PREFIX']
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     tempf = os.path.join(path,"pytests"+str(int(100000*random.random())))
     ntacl = xattr.NTACL()
     ntacl.version = 1
     open(tempf, 'w').write("empty")
     lp.set("posix:eadb",os.path.join(path,"eadbtest.tdb"))
     setntacl(lp, tempf, acl, "S-1-5-21-2212615479-2695158682-2101375467")
     os.unlink(tempf)

class GPOTests(tests.TestCase):
    def setUp(self):
        super(GPOTests, self).setUp()
        self.server = os.environ["SERVER"]
        self.lp = LoadParm()
        self.lp.load_default()
        self.creds = self.insta_creds(template=self.get_credentials())

    def tearDown(self):
        super(GPOTests, self).tearDown()

    def test_gpo_list(self):
        global poldir, dspath
        ads = gpo.ADS_STRUCT(self.server, self.lp, self.creds)
        if ads.connect():
            gpos = ads.get_gpo_list(self.creds.get_username())
        guid = '{31B2F340-016D-11D2-945F-00C04FB984F9}'
        names = ['Local Policy', guid]
        file_sys_paths = [None, '%s\\%s' % (poldir, guid)]
        ds_paths = [None, 'CN=%s,%s' % (guid, dspath)]
        for i in range(0, len(gpos)):
            assert gpos[i].name == names[i], \
              'The gpo name did not match expected name %s' % gpos[i].name
            assert gpos[i].file_sys_path == file_sys_paths[i], \
              'file_sys_path did not match expected %s' % gpos[i].file_sys_path
            assert gpos[i].ds_path == ds_paths[i], \
              'ds_path did not match expected %s' % gpos[i].ds_path


    def test_gpo_ads_does_not_segfault(self):
        try:
            ads = gpo.ADS_STRUCT(self.server, 42, self.creds)
        except:
            pass

    def test_gpt_version(self):
        global gpt_data
        local_path = self.lp.get("path", "sysvol")
        policies = 'addom.samba.example.com/Policies'
        guid = '{31B2F340-016D-11D2-945F-00C04FB984F9}'
        gpo_path = os.path.join(local_path, policies, guid)
        old_vers = gpo.gpo_get_sysvol_gpt_version(gpo_path)[1]

        with open(os.path.join(gpo_path, 'GPT.INI'), 'w') as gpt:
            gpt.write(gpt_data % 42)
        assert gpo.gpo_get_sysvol_gpt_version(gpo_path)[1] == 42, \
          'gpo_get_sysvol_gpt_version() did not return the expected version'

        with open(os.path.join(gpo_path, 'GPT.INI'), 'w') as gpt:
            gpt.write(gpt_data % old_vers)
        assert gpo.gpo_get_sysvol_gpt_version(gpo_path)[1] == old_vers, \
          'gpo_get_sysvol_gpt_version() did not return the expected version'

class Options:
	def __init__(self):
		self.function_level = None
		self.use_xattrs = "auto" 
		self.lp = LoadParm();
		self.lp.load_default();
		self.realm = self.lp.get('realm') #default
		self.domain = self.lp.get('workgroup') #default
		self.adminpass = ''
		self.smbconf = self.lp.configfile
		self.server_role = self.lp.get("server role") #default
		self.samdb_fill = FILL_FULL #default
		self.blank = False
		self.partitions_only = False

class AuthSMB4(object):
	def __init__(self,user,password):
		self.user = user
		self.password = password
		_isLastErrorAvailable=False
		self.lp = LoadParm()
		self.lp.load_default()
                self.ip = '127.0.0.1'
                self.WorkGroup = str(self.lp.get("workgroup"))
                self.creds = credentials.Credentials()
                self.creds.set_username(self.user)
                self.creds.set_password(self.password)
                self.creds.set_domain(self.WorkGroup)
                self.creds.set_workstation("")


		self.logger = logging.getLogger(__name__)
		self.logger.addHandler(logging.StreamHandler(sys.stdout))
		self.logger.setLevel(logging.INFO)
		
        def Autenticate(self):
                try:
                        session_info_flags = ( AUTH_SESSION_INFO_DEFAULT_GROUPS | AUTH_SESSION_INFO_AUTHENTICATED )

                        LdapConn = samba.Ldb("ldap://%s" % self.ip,lp=self.lp,credentials=self.creds)
                        DomainDN = LdapConn.get_default_basedn()
                        search_filter="sAMAccountName=%s" % self.user
                        res = LdapConn.search(base=DomainDN, scope=SCOPE_SUBTREE,expression=search_filter, attrs=["dn"])
                        if len(res) == 0:
                                return False

                        user_dn = res[0].dn
                        session = samba.auth.user_session(LdapConn, lp_ctx=self.lp, dn=user_dn,session_info_flags=session_info_flags)
                        token = session.security_token


                        if (token.has_builtin_administrators()):
                                return True

                        if(token.is_system()):
                                return True

                except Exception,e:
                        if(len(e.args)>1):
                                self.logger.info("%s %s" % (e.args[1],e.args[0]))
                                self.SetError(e.args[1],e.args[0])
                        else:
                                self.logger.info("%s " % (e.args[0]))
                                self.SetError(e.args,0)
                return False

 def test_setntacl_forcenative(self):
     if os.getuid() == 0:
         raise TestSkipped("Running test as root, test skipped")
     random.seed()
     lp = LoadParm()
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     path = os.environ['SELFTEST_PREFIX']
     tempf = os.path.join(path,"pytests"+str(int(100000*random.random())))
     ntacl = xattr.NTACL()
     ntacl.version = 1
     open(tempf, 'w').write("empty")
     lp.set("posix:eadb", os.path.join(path,"eadbtest.tdb"))
     self.assertRaises(Exception, setntacl, lp, tempf ,acl,
         "S-1-5-21-2212615479-2695158682-2101375467","native")
     os.unlink(tempf)

 def __init__(self):
     self.samba_lp = LoadParm()
     self.samba_lp.set("debug level", "0")
     self.samba_lp.load_default()
     url = self.samba_lp.get("dcerpc_mapiproxy:samdb_url") or self.samba_lp.private_path("sam.ldb")
     self.samdb = SamDB(url=url, lp=self.samba_lp, session_info=system_session())
     self.conn = self._open_mysql_connection()

 def test_setntacl_getntacl(self):
     random.seed()
     lp = LoadParm()
     path = None
     path = os.environ['SELFTEST_PREFIX']
     acl = "O:S-1-5-21-2212615479-2695158682-2101375467-512G:S-1-5-21-2212615479-2695158682-2101375467-513D:(A;OICI;0x001f01ff;;;S-1-5-21-2212615479-2695158682-2101375467-512)"
     tempf = os.path.join(path,"pytests"+str(int(100000*random.random())))
     ntacl = xattr.NTACL()
     ntacl.version = 1
     open(tempf, 'w').write("empty")
     lp.set("posix:eadb",os.path.join(path,"eadbtest.tdb"))
     setntacl(lp,tempf,acl,"S-1-5-21-2212615479-2695158682-2101375467")
     facl = getntacl(lp,tempf)
     anysid = security.dom_sid(security.SID_NT_SELF)
     self.assertEquals(facl.info.as_sddl(anysid),acl)
     os.unlink(tempf)

 def __init__(self, username=None, no_dry_run=False, mysql_string=None):
     self.lp = LoadParm()
     self.lp.set('debug level', '0')
     self.lp.load_default()
     self.username = username
     self.no_dry_run = no_dry_run
     self.conn = self._open_mysql_conn(mysql_string)

 def setUp(self):
     super(DCKeytabTests, self).setUp()
     self.lp = LoadParm()
     self.lp.load_default()
     self.creds = self.insta_creds(template=self.get_credentials())
     self.ktfile = os.path.join(self.lp.get('private dir'), 'test.keytab')
     self.principal = self.creds.get_principal()

class SambaOCHelper(object):
    def __init__(self):
        self.samba_lp = LoadParm()
        self.samba_lp.set('debug level', '0')
        self.samba_lp.load_default()
        url = self.samba_lp.get('dcerpc_mapiproxy:samdb_url') or \
            self.samba_lp.private_path("sam.ldb")
        self.samdb = SamDB(url=url,
                           lp=self.samba_lp,
                           session_info=system_session())
        self.conn = self._open_mysql_connection()

    def _open_mysql_connection(self):
        connection_string = self.samba_lp.get('mapiproxy:openchangedb')
        if not connection_string:
            raise Exception("Not found mapiproxy:openchangedb on samba configuration")
        # mysql://openchange:[email protected]/openchange
        m = re.search(r'(?P<scheme>.+)://(?P<user>.+):(?P<pass>.+)@(?P<host>.+)/(?P<db>.+)',
                      connection_string)
        if not m:
            raise Exception("Unable to parse mapiproxy:openchangedb: %s" %
                            connection_string)
        group_dict = m.groupdict()
        if group_dict['scheme'] != 'mysql':
            raise Exception("mapiproxy:openchangedb should start with mysql:// (we got %s)",
                            group_dict['scheme'])

        conn = MySQLdb.connect(host=group_dict['host'], user=group_dict['user'],
                               passwd=group_dict['pass'], db=group_dict['db'])
        conn.autocommit(True)
        return conn

    def invalid_user(self, username):
        ret = self.samdb.search(base=self.samdb.domain_dn(),
                                scope=ldb.SCOPE_SUBTREE,
                                expression="(sAMAccountName=%s)" % ldb.binary_encode(username))
        return len(ret) != 1

    def find_email_of(self, username):
        ret = self.samdb.search(base=self.samdb.domain_dn(),
                                scope=ldb.SCOPE_SUBTREE, attrs=["mail"],
                                expression="(sAMAccountName=%s)" % ldb.binary_encode(username))
        return ret[0]["mail"][0]

    def active_openchange_users(self):
        c = self.conn.cursor()
        c.execute("SELECT name FROM mailboxes")
        return sorted([row[0] for row in c.fetchall()])

    def get_indexing_cache(self):
        memcached_server = self.samba_lp.get('mapistore:indexing_cache')
        if not memcached_server:
            return "127.0.0.1:11211"
        # This should has a format like: --SERVER=11.22.33.44:11211
        return memcached_server.split('=')[1]

	def __init__(self,user,password):
		self.user = user
		self.password = password
		_isLastErrorAvailable=False
		self.lp = LoadParm()
		self.lp.load_default()

		self.logger = logging.getLogger(__name__)
		self.logger.addHandler(logging.StreamHandler(sys.stdout))
		self.logger.setLevel(logging.INFO)

 def __init__(self):
     self.samba_lp = LoadParm()
     self.samba_lp.set('debug level', '0')
     self.samba_lp.load_default()
     self.next_fmid = None
     url = self.samba_lp.get('dcerpc_mapiproxy:samdb_url') or \
         self.samba_lp.private_path("sam.ldb")
     self.samdb = SamDB(url=url,
                        lp=self.samba_lp,
                        session_info=system_session())
     self.conn = self._open_mysql_connection()

    def apply(cls, cur, **kwargs):
        # Mimetise what mapistore_interface.c (mapistore_init) does
        # to get the mapping path
        if 'lp' in kwargs:
            mapping_path = kwargs['lp'].private_path("mapistore")
        else:
            lp = LoadParm()
            lp.load_default()
            mapping_path = lp.private_path("mapistore")

        if mapping_path is None:
            return

        cur.execute("START TRANSACTION")
        try:
            # Get all mailboxes
            cur.execute("SELECT name FROM mailboxes")
            for row in cur.fetchall():
                username = row[0]
                path = "{0}/{1}/replica_mapping.tdb".format(mapping_path, username)
                try:
                    tdb_file = tdb.Tdb(path, 0, tdb.DEFAULT, os.O_RDONLY)
                    for k in tdb_file.iterkeys():
                        # Check if the key is an integer
                        try:
                            repl_id = int(k, base=16)
                            cls._insert_map(cur, username, repl_id, tdb_file[k])
                        except ValueError:
                            # Cannot convert to int, so no repl_id
                            continue
                except IOError:
                    # Cannot read any replica mapping
                    continue

            cur.execute("COMMIT")
        except Exception as e:
            print("Error migrating TDB files into the database {}, rollback".format(e), file=sys.stderr)
            cur.execute("ROLLBACK")
            raise