super(DCJoinException, self).__init__("Can't join, error: %s" % msg)
-class dc_join(object):
+class DCJoinContext(object):
"""Perform a DC join."""
def __init__(ctx, logger=None, server=None, creds=None, lp=None, site=None,
backend_store=None):
"""Join as a RODC."""
- ctx = dc_join(logger, server, creds, lp, site, netbios_name, targetdir, domain,
- machinepass, use_ntvfs, dns_backend, promote_existing,
- plaintext_secrets, backend_store=backend_store)
+ ctx = DCJoinContext(logger, server, creds, lp, site, netbios_name,
+ targetdir, domain, machinepass, use_ntvfs, dns_backend,
+ promote_existing, plaintext_secrets,
+ backend_store=backend_store)
lp.set("workgroup", ctx.domain_name)
logger.info("workgroup is %s" % ctx.domain_name)
promote_existing=False, plaintext_secrets=False,
backend_store=None):
"""Join as a DC."""
- ctx = dc_join(logger, server, creds, lp, site, netbios_name, targetdir, domain,
- machinepass, use_ntvfs, dns_backend, promote_existing,
- plaintext_secrets, backend_store=backend_store)
+ ctx = DCJoinContext(logger, server, creds, lp, site, netbios_name,
+ targetdir, domain, machinepass, use_ntvfs, dns_backend,
+ promote_existing, plaintext_secrets,
+ backend_store=backend_store)
lp.set("workgroup", ctx.domain_name)
logger.info("workgroup is %s" % ctx.domain_name)
targetdir=None, domain=None, include_secrets=False,
dns_backend="NONE"):
"""Join as a DC."""
- ctx = dc_join(logger, server, creds, lp, site=None, netbios_name=None, targetdir=targetdir, domain=domain,
- machinepass=None, use_ntvfs=False, dns_backend=dns_backend, promote_existing=False, clone_only=True)
+ ctx = DCJoinContext(logger, server, creds, lp, site=None, netbios_name=None,
+ targetdir=targetdir, domain=domain, machinepass=None,
+ use_ntvfs=False, dns_backend=dns_backend,
+ promote_existing=False, clone_only=True)
lp.set("workgroup", ctx.domain_name)
logger.info("workgroup is %s" % ctx.domain_name)
dns_backend=None, plaintext_secrets=False,
backend_store=None):
"""Join as a DC."""
- ctx = dc_join(logger, server, creds, lp, site, netbios_name, targetdir, parent_domain,
- machinepass, use_ntvfs, dns_backend, plaintext_secrets,
- backend_store=backend_store)
+ ctx = DCJoinContext(logger, server, creds, lp, site, netbios_name,
+ targetdir, parent_domain, machinepass, use_ntvfs,
+ dns_backend, plaintext_secrets,
+ backend_store=backend_store)
ctx.subdomain = True
if adminpass is None:
ctx.adminpass = samba.generate_random_password(12, 32)
import shutil
import os
from samba.tests.dns_base import DNSTKeyTest
-from samba.join import dc_join
+from samba.join import DCJoinContext
from samba.dcerpc import drsuapi, misc, dns
from samba.credentials import Credentials
self.netbios_name = "jointest1"
logger = get_logger()
- self.join_ctx = dc_join(server=self.server, creds=self.creds, lp=self.get_loadparm(),
- netbios_name=self.netbios_name,
- targetdir=self.tempdir,
- domain=None, logger=logger,
- dns_backend="SAMBA_INTERNAL")
+ self.join_ctx = DCJoinContext(server=self.server, creds=self.creds,
+ lp=self.get_loadparm(),
+ netbios_name=self.netbios_name,
+ targetdir=self.tempdir,
+ domain=None, logger=logger,
+ dns_backend="SAMBA_INTERNAL")
self.join_ctx.userAccountControl = (samba.dsdb.UF_SERVER_TRUST_ACCOUNT |
samba.dsdb.UF_TRUSTED_FOR_DELEGATION)
from samba.tests.subunitrun import SubunitOptions, TestProgram
import samba.getopt as options
-from samba.join import dc_join
+from samba.join import DCJoinContext
from ldb import (
SCOPE_BASE, SCOPE_SUBTREE, LdbError, ERR_NO_SUCH_OBJECT,
self.computerdn = "CN=%s,CN=computers,%s" % (self.computername, self.base_dn)
self.dc_dn = "CN=%s,OU=Domain Controllers,%s" % (self.dcname, self.base_dn)
self.site = "Default-First-Site-Name"
- self.rodcctx = dc_join(server=host, creds=creds, lp=lp,
- site=self.site, netbios_name=self.rodcname, targetdir=None,
- domain=None)
- self.dcctx = dc_join(server=host, creds=creds, lp=lp, site=self.site,
- netbios_name=self.dcname, targetdir=None, domain=None)
+ self.rodcctx = DCJoinContext(server=host, creds=creds, lp=lp,
+ site=self.site, netbios_name=self.rodcname,
+ targetdir=None, domain=None)
+ self.dcctx = DCJoinContext(server=host, creds=creds, lp=lp,
+ site=self.site, netbios_name=self.dcname,
+ targetdir=None, domain=None)
self.ldb_admin.newuser(self.test_user, self.user_pass)
self.ldb_user1 = self.get_ldb_connection(self.test_user, self.user_pass)
self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.test_user))
from ldb import SCOPE_BASE
from samba import WERRORError
-from samba.join import dc_join
+from samba.join import DCJoinContext
from samba.dcerpc import drsuapi, misc, drsblobs, security
from samba.drs_utils import drs_DsBind, drs_Replicate
from samba.ndr import ndr_unpack, ndr_pack
self.computer_dn = "CN=%s,OU=Domain Controllers,%s" % (self.rodc_name, self.base_dn)
- self.rodc_ctx = dc_join(server=self.ldb_dc1.host_dns_name(), creds=self.get_credentials(), lp=self.get_loadparm(),
- site=self.site, netbios_name=self.rodc_name,
- targetdir=None, domain=None, machinepass=self.rodc_pass)
+ self.rodc_ctx = DCJoinContext(server=self.ldb_dc1.host_dns_name(),
+ creds=self.get_credentials(),
+ lp=self.get_loadparm(), site=self.site,
+ netbios_name=self.rodc_name,
+ targetdir=None, domain=None,
+ machinepass=self.rodc_pass)
self._create_rodc(self.rodc_ctx)
self.rodc_ctx.create_tmp_samdb()
self.tmp_samdb = self.rodc_ctx.tmp_samdb
"""
# Create a new identical RODC with just the first letter missing
other_rodc_name = self.rodc_name[1:]
- other_rodc_ctx = dc_join(server=self.ldb_dc1.host_dns_name(), creds=self.get_credentials(), lp=self.get_loadparm(),
- site=self.site, netbios_name=other_rodc_name,
- targetdir=None, domain=None, machinepass=self.rodc_pass)
+ other_rodc_ctx = DCJoinContext(server=self.ldb_dc1.host_dns_name(),
+ creds=self.get_credentials(),
+ lp=self.get_loadparm(), site=self.site,
+ netbios_name=other_rodc_name,
+ targetdir=None, domain=None,
+ machinepass=self.rodc_pass)
self._create_rodc(other_rodc_ctx)
other_rodc_creds = Credentials()