import samba.getopt as options
import ldb
+import socket
+import samba
+from samba import sd_utils
+from samba.dcerpc import dnsserver, dnsp, security
+from samba.dnsserver import ARecord, AAAARecord
+from samba.ndr import ndr_unpack, ndr_pack, ndr_print
+from samba.remove_dc import remove_dns_references
from samba.auth import system_session
from samba.samdb import SamDB
+
from samba import (
credentials,
dsdb,
Ldb,
+ werror,
+ WERRORError
)
from samba.netcmd import (
Option,
)
+
+def _is_valid_ip(ip_string, address_families=None):
+ """Check ip string is valid address"""
+ # by default, check both ipv4 and ipv6
+ if not address_families:
+ address_families = [socket.AF_INET, socket.AF_INET6]
+
+ for address_family in address_families:
+ try:
+ socket.inet_pton(address_family, ip_string)
+ return True # if no error, return directly
+ except socket.error:
+ continue # Otherwise, check next family
+ return False
+
+
+def _is_valid_ipv4(ip_string):
+ """Check ip string is valid ipv4 address"""
+ return _is_valid_ip(ip_string, address_families=[socket.AF_INET])
+
+
+def _is_valid_ipv6(ip_string):
+ """Check ip string is valid ipv6 address"""
+ return _is_valid_ip(ip_string, address_families=[socket.AF_INET6])
+
+
+def add_dns_records(
+ samdb, name, dns_conn, change_owner_sd,
+ server, ip_address_list, logger):
+ """Add DNS A or AAAA records while creating computer. """
+ name = name.rstrip('$')
+ client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
+ select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA | dnsserver.DNS_RPC_VIEW_NO_CHILDREN
+ zone = samdb.domain_dns_name()
+ name_found = True
+ sd_helper = sd_utils.SDUtils(samdb)
+
+ try:
+ buflen, res = dns_conn.DnssrvEnumRecords2(
+ client_version,
+ 0,
+ server,
+ zone,
+ name,
+ None,
+ dnsp.DNS_TYPE_ALL,
+ select_flags,
+ None,
+ None,
+ )
+ except WERRORError as e:
+ if e.args[0] == werror.WERR_DNS_ERROR_NAME_DOES_NOT_EXIST:
+ name_found = False
+ pass
+
+ if name_found:
+ for rec in res.rec:
+ for record in rec.records:
+ if record.wType == dnsp.DNS_TYPE_A or record.wType == dnsp.DNS_TYPE_AAAA:
+ # delete record
+ del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ del_rec_buf.rec = record
+ try:
+ dns_conn.DnssrvUpdateRecord2(
+ client_version,
+ 0,
+ server,
+ zone,
+ name,
+ None,
+ del_rec_buf,
+ )
+ except WERRORError as e:
+ if e.args[0] != werror.WERR_DNS_ERROR_NAME_DOES_NOT_EXIST:
+ raise
+
+ for ip_address in ip_address_list:
+ if _is_valid_ipv6(ip_address):
+ logger.info("Adding DNS AAAA record %s.%s for IPv6 IP: %s" % (
+ name, zone, ip_address))
+ rec = AAAARecord(ip_address)
+ elif _is_valid_ipv4(ip_address):
+ logger.info("Adding DNS A record %s.%s for IPv4 IP: %s" % (
+ name, zone, ip_address))
+ rec = ARecord(ip_address)
+ else:
+ raise ValueError('Invalid IP: {}'.format(ip_address))
+
+ # Add record
+ add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
+ add_rec_buf.rec = rec
+
+ dns_conn.DnssrvUpdateRecord2(
+ client_version,
+ 0,
+ server,
+ zone,
+ name,
+ add_rec_buf,
+ None,
+ )
+
+ if (len(ip_address_list) > 0):
+ domaindns_zone_dn = ldb.Dn(
+ samdb,
+ 'DC=DomainDnsZones,%s' % samdb.get_default_basedn(),
+ )
+
+ dns_a_dn, ldap_record = samdb.dns_lookup(
+ "%s.%s" % (name, zone),
+ dns_partition=domaindns_zone_dn,
+ )
+
+ # Make the DC own the DNS record, not the administrator
+ sd_helper.modify_sd_on_dn(
+ dns_a_dn,
+ change_owner_sd,
+ controls=["sd_flags:1:%d" % (security.SECINFO_OWNER | security.SECINFO_GROUP)],
+ )
+
+
class cmd_computer_create(Command):
"""Create a new computer.
Option("--prepare-oldjoin",
help="Prepare enabled machine account for oldjoin mechanism",
action="store_true"),
+ Option("--ip-address",
+ dest='ip_address_list',
+ help=("IPv4 address for the computer's A record, or IPv6 "
+ "address for AAAA record, can be provided multiple "
+ "times"),
+ action='append'),
+ Option("--service-principal-name",
+ dest='service_principal_name_list',
+ help=("Computer's Service Principal Name, can be provided "
+ "multiple times"),
+ action='append')
]
takes_args = ["computername"]
}
def run(self, computername, credopts=None, sambaopts=None, versionopts=None,
- H=None, computerou=None, description=None, prepare_oldjoin=False):
+ H=None, computerou=None, description=None, prepare_oldjoin=False,
+ ip_address_list=None, service_principal_name_list=None):
+
+ if ip_address_list is None:
+ ip_address_list = []
+
+ if service_principal_name_list is None:
+ service_principal_name_list = []
+
+ # check each IP address if provided
+ for ip_address in ip_address_list:
+ if not _is_valid_ip(ip_address):
+ raise CommandError('Invalid IP address {}'.format(ip_address))
lp = sambaopts.get_loadparm()
creds = credopts.get_credentials(lp)
credentials=creds, lp=lp)
samdb.newcomputer(computername, computerou=computerou,
description=description,
- prepare_oldjoin=prepare_oldjoin)
+ prepare_oldjoin=prepare_oldjoin,
+ ip_address_list=ip_address_list,
+ service_principal_name_list=service_principal_name_list,
+ )
+
+ if ip_address_list:
+ # if ip_address_list provided, then we need to create DNS
+ # records for this computer.
+ filters = '(&(sAMAccountName={}$)(objectclass=computer))'.format(
+ ldb.binary_encode(computername.rstrip('$')))
+
+ recs = samdb.search(
+ base=samdb.domain_dn(),
+ scope=ldb.SCOPE_SUBTREE,
+ expression=filters,
+ attrs=['primaryGroupID', 'objectSid'])
+
+ group = recs[0]['primaryGroupID'][0]
+ owner = ndr_unpack(security.dom_sid, recs[0]["objectSid"][0])
+
+ dns_conn = dnsserver.dnsserver(
+ "ncacn_ip_tcp:{}[sign]".format(samdb.host_dns_name()),
+ lp, creds)
+
+ change_owner_sd = security.descriptor()
+ change_owner_sd.owner_sid = owner
+ change_owner_sd.group_sid = security.dom_sid(
+ "{}-{}".format(samdb.get_domain_sid(), group),
+ )
+
+ add_dns_records(
+ samdb, computername.rstrip('$'), dns_conn,
+ change_owner_sd, samdb.host_dns_name(),
+ ip_address_list, self.get_logger())
except Exception, e:
raise CommandError("Failed to create computer '%s': " %
computername, e)
self.outf.write("Computer '%s' created successfully\n" % computername)
+
class cmd_computer_delete(Command):
"""Delete a computer.
res = samdb.search(base=samdb.domain_dn(),
scope=ldb.SCOPE_SUBTREE,
expression=filter,
- attrs=["userAccountControl"])
+ attrs=["userAccountControl", "dNSHostName"])
computer_dn = res[0].dn
computer_ac = int(res[0]["userAccountControl"][0])
+ if "dNSHostName" in res[0]:
+ computer_dns_host_name = res[0]["dNSHostName"][0]
+ else:
+ computer_dns_host_name = None
except IndexError:
raise CommandError('Unable to find computer "%s"' % computername)
% computername)
try:
samdb.delete(computer_dn)
+ if computer_dns_host_name:
+ remove_dns_references(
+ samdb, self.get_logger(), computer_dns_host_name,
+ ignore_no_name=True)
except Exception, e:
raise CommandError('Failed to remove computer "%s"' %
samaccountname, e)
self.transaction_commit()
def newcomputer(self, computername, computerou=None, description=None,
- prepare_oldjoin=False):
+ prepare_oldjoin=False, ip_address_list=None,
+ service_principal_name_list=None):
"""Adds a new user with additional parameters
:param computername: Name of the new computer
:param computerou: Object container for new computer
:param description: Description of the new computer
:param prepare_oldjoin: Preset computer password for oldjoin mechanism
+ :param ip_address_list: ip address list for DNS A or AAAA record
+ :param service_principal_name_list: string list of servicePincipalName
"""
cn = re.sub(r"\$$", "", computername)
computer_dn = "CN=%s,%s" % (cn, computercontainer_dn)
- dnsdomain = ldb.Dn(self,
- self.domain_dn()).canonical_str().replace("/", "")
ldbmessage = {"dn": computer_dn,
"sAMAccountName": samaccountname,
"objectClass": "computer",
if description is not None:
ldbmessage["description"] = description
+ if service_principal_name_list:
+ ldbmessage["servicePrincipalName"] = service_principal_name_list
+
accountcontrol = str(dsdb.UF_WORKSTATION_TRUST_ACCOUNT |
dsdb.UF_ACCOUNTDISABLE)
if prepare_oldjoin:
accountcontrol = str(dsdb.UF_WORKSTATION_TRUST_ACCOUNT)
ldbmessage["userAccountControl"] = accountcontrol
+ if ip_address_list:
+ ldbmessage['dNSHostName'] = '{}.{}'.format(
+ cn, self.domain_dns_name())
+
self.transaction_start()
try:
self.add(ldbmessage)
import ldb
from samba.tests.samba_tool.base import SambaToolCmdTest
from samba import dsdb
+from samba.ndr import ndr_unpack, ndr_pack
+from samba.dcerpc import dnsp
class ComputerCmdTestCase(SambaToolCmdTest):
"""Tests for samba-tool computer subcommands"""
def setUp(self):
super(ComputerCmdTestCase, self).setUp()
- self.samdb = self.getSamDB("-H", "ldap://%s" % os.environ["DC_SERVER"],
- "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
- self.computers = []
- self.computers.append(self._randomComputer({"name": "testcomputer1"}))
- self.computers.append(self._randomComputer({"name": "testcomputer2"}))
- self.computers.append(self._randomComputer({"name": "testcomputer3$"}))
- self.computers.append(self._randomComputer({"name": "testcomputer4$"}))
+ self.creds = "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"])
+ self.samdb = self.getSamDB("-H", "ldap://%s" % os.environ["DC_SERVER"], self.creds)
+ # ips used to test --ip-address option
+ self.ipv4 = '10.10.10.10'
+ self.ipv6 = '2001:0db8:0a0b:12f0:0000:0000:0000:0001'
+ data = [
+ {
+ 'name': 'testcomputer1',
+ 'ip_address_list': [self.ipv4]
+ },
+ {
+ 'name': 'testcomputer2',
+ 'ip_address_list': [self.ipv6],
+ 'service_principal_name_list': ['SPN0']
+ },
+ {
+ 'name': 'testcomputer3$',
+ 'ip_address_list': [self.ipv4, self.ipv6],
+ 'service_principal_name_list': ['SPN0', 'SPN1']
+ },
+ {
+ 'name': 'testcomputer4$',
+ },
+ ]
+ self.computers = [self._randomComputer(base=item) for item in data]
# setup the 4 computers and ensure they are correct
for computer in self.computers:
self.assertEquals("%s" % found.get("description"),
computer["description"])
+
def tearDown(self):
super(ComputerCmdTestCase, self).tearDown()
# clean up all the left over computers, just in case
"Failed to delete computer '%s'" %
computer["name"])
+ def test_newcomputer_with_service_principal_name(self):
+ # Each computer should have correct servicePrincipalName as provided.
+ for computer in self.computers:
+ expected_names = computer.get('service_principal_name_list', [])
+ found = self._find_service_principal_name(computer['name'], expected_names)
+ self.assertTrue(found)
+
+ def test_newcomputer_with_dns_records(self):
+
+ # Each computer should have correct DNS record and ip address.
+ for computer in self.computers:
+ for ip_address in computer.get('ip_address_list', []):
+ found = self._find_dns_record(computer['name'], ip_address)
+ self.assertTrue(found)
+
+ # try to delete all the computers we just created
+ for computer in self.computers:
+ (result, out, err) = self.runsubcmd("computer", "delete",
+ "%s" % computer["name"])
+ self.assertCmdSuccess(result, out, err,
+ "Failed to delete computer '%s'" %
+ computer["name"])
+ found = self._find_computer(computer["name"])
+ self.assertIsNone(found,
+ "Deleted computer '%s' still exists" %
+ computer["name"])
+
+ # all DNS records should be gone
+ for computer in self.computers:
+ for ip_address in computer.get('ip_address_list', []):
+ found = self._find_dns_record(computer['name'], ip_address)
+ self.assertFalse(found)
def test_newcomputer(self):
"""This tests the "computer create" and "computer delete" commands"""
self.assertCmdFail(result, "Succeeded to create existing computer")
self.assertIn("already exists", err)
+
# try to delete all the computers we just created
for computer in self.computers:
(result, out, err) = self.runsubcmd("computer", "delete", "%s" %
for computer in self.computers:
(result, out, err) = self.runsubcmd(
"computer", "create", "%s" % computer["name"],
- "--description=%s" % computer["description"])
+ "--description=%s" % computer["description"])
self.assertCmdSuccess(result, out, err)
self.assertEquals(err, "", "There shouldn't be any error message")
return ou
def _create_computer(self, computer):
- return self.runsubcmd("computer", "create", "%s" % computer["name"],
- "--description=%s" % computer["description"])
+ args = '{} {} --description={}'.format(
+ computer['name'], self.creds, computer["description"])
+
+ for ip_address in computer.get('ip_address_list', []):
+ args += ' --ip-address={}'.format(ip_address)
+
+ for service_principal_name in computer.get('service_principal_name_list', []):
+ args += ' --service-principal-name={}'.format(service_principal_name)
+
+ args = args.split()
+
+ return self.runsubcmd('computer', 'create', *args)
def _create_ou(self, ou):
return self.runsubcmd("ou", "create", "OU=%s" % ou["name"],
return computerlist[0]
else:
return None
+
+ def _find_dns_record(self, name, ip_address):
+ name = name.rstrip('$') # computername
+ records = self.samdb.search(
+ base="DC=DomainDnsZones,{}".format(self.samdb.get_default_basedn()),
+ scope=ldb.SCOPE_SUBTREE,
+ expression="(&(objectClass=dnsNode)(name={}))".format(name),
+ attrs=['dnsRecord', 'dNSTombstoned'])
+
+ # unpack data and compare
+ for record in records:
+ if 'dNSTombstoned' in record and str(record['dNSTombstoned']) == 'TRUE':
+ # if a record is dNSTombstoned, ignore it.
+ continue
+ for dns_record_bin in record['dnsRecord']:
+ dns_record_obj = ndr_unpack(dnsp.DnssrvRpcRecord, dns_record_bin)
+ ip = str(dns_record_obj.data)
+
+ if str(ip) == str(ip_address):
+ return True
+
+ return False
+
+ def _find_service_principal_name(self, name, expected_service_principal_names):
+ """Find all servicePrincipalName values and compare with expected_service_principal_names"""
+ samaccountname = name.strip('$') + '$'
+ search_filter = ("(&(sAMAccountName=%s)(objectCategory=%s,%s))" %
+ (ldb.binary_encode(samaccountname),
+ "CN=Computer,CN=Schema,CN=Configuration",
+ self.samdb.domain_dn()))
+ computer_list = self.samdb.search(
+ base=self.samdb.domain_dn(),
+ scope=ldb.SCOPE_SUBTREE,
+ expression=search_filter,
+ attrs=['servicePrincipalName'])
+ names = set()
+ for computer in computer_list:
+ for name in computer.get('servicePrincipalName', []):
+ names.add(name)
+ return names == set(expected_service_principal_names)
git.samba.org / metze / samba / wip.git / commitdiff