| Server IP : 134.29.175.74 / Your IP : 216.73.216.160 Web Server : nginx/1.10.2 System : Windows NT CST-WEBSERVER 10.0 build 19045 (Windows 10) i586 User : Administrator ( 0) PHP Version : 7.1.0 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : OFF | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : C:/Program Files (x86)/Certbot/pkgs/win32comext/adsi/demos/ |
Upload File : |
"""A re-implementation of the MS DirectoryService samples related to services.
* Adds and removes an ActiveDirectory "Service Connection Point",
including managing the security on the object.
* Creates and registers Service Principal Names.
* Changes the username for a domain user.
Some of these functions are likely to become move to a module - but there
is also a little command-line-interface to try these functions out.
For example:
scp.py --account-name=domain\\user --service-class=PythonScpTest \\
--keyword=foo --keyword=bar --binding-string=bind_info \\
ScpCreate SpnCreate SpnRegister
would:
* Attempt to delete a Service Connection Point for the service class
'PythonScpTest'
* Attempt to create a Service Connection Point for that class, with 2
keywords and a binding string of 'bind_info'
* Create a Service Principal Name for the service and register it
to undo those changes, you could execute:
scp.py --account-name=domain\\user --service-class=PythonScpTest \\
SpnCreate SpnUnregister ScpDelete
which will:
* Create a SPN
* Unregister that SPN from the Active Directory.
* Delete the Service Connection Point
Executing with --test will create and remove one of everything.
"""
from win32com.adsi.adsicon import *
from win32com.adsi import adsi
import win32api, win32con, winerror
from win32com.client import Dispatch
import ntsecuritycon as dscon
import win32security
import optparse, textwrap
import traceback
verbose = 1
g_createdSCP = None
g_createdSPNs = []
g_createdSPNLast = None
import logging
logger = logging # use logging module global methods for now.
# still a bit confused about log(n, ...) vs logger.info/debug()
# Returns distinguished name of SCP.
def ScpCreate(
service_binding_info,
service_class_name, # Service class string to store in SCP.
account_name = None, # Logon account that needs access to SCP.
container_name = None,
keywords = None,
object_class = "serviceConnectionPoint",
dns_name_type = "A",
dn = None,
dns_name = None,
):
container_name = container_name or service_class_name
if not dns_name:
# Get the DNS name of the local computer
dns_name = win32api.GetComputerNameEx(win32con.ComputerNameDnsFullyQualified)
# Get the distinguished name of the computer object for the local computer
if dn is None:
dn = win32api.GetComputerObjectName(win32con.NameFullyQualifiedDN)
# Compose the ADSpath and bind to the computer object for the local computer
comp = adsi.ADsGetObject("LDAP://" + dn, adsi.IID_IDirectoryObject)
# Publish the SCP as a child of the computer object
keywords = keywords or []
# Fill in the attribute values to be stored in the SCP.
attrs = [
("cn", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (container_name,)),
("objectClass", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (object_class,)),
("keywords", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, keywords),
("serviceDnsName", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (dns_name,)),
("serviceDnsNameType", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (dns_name_type,)),
("serviceClassName", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (service_class_name,)),
("serviceBindingInformation", ADS_ATTR_UPDATE, ADSTYPE_CASE_IGNORE_STRING, (service_binding_info,)),
]
new = comp.CreateDSObject("cn=" + container_name, attrs)
logger.info("New connection point is at %s", container_name)
# Wrap in a usable IDispatch object.
new = Dispatch(new)
# And allow access to the SCP for the specified account name
AllowAccessToScpProperties(account_name, new)
return new
def ScpDelete(container_name, dn = None):
if dn is None:
dn = win32api.GetComputerObjectName(win32con.NameFullyQualifiedDN)
logger.debug("Removing connection point '%s' from %s", container_name, dn)
# Compose the ADSpath and bind to the computer object for the local computer
comp = adsi.ADsGetObject("LDAP://" + dn, adsi.IID_IDirectoryObject)
comp.DeleteDSObject("cn=" + container_name)
logger.info("Deleted service connection point '%s'", container_name)
# This function is described in detail in the MSDN article titled
# "Enabling Service Account to Access SCP Properties"
# From that article:
# The following sample code sets a pair of ACEs on a service connection point
# (SCP) object. The ACEs grant read/write access to the user or computer account
# under which the service instance will be running. Your service installation
# program calls this code to ensure that the service will be allowed to update
# its properties at run time. If you don't set ACEs like these, your service
# will get access-denied errors if it tries to modify the SCP's properties.
#
# The code uses the IADsSecurityDescriptor, IADsAccessControlList, and
# IADsAccessControlEntry interfaces to do the following:
# * Get the SCP object's security descriptor.
# * Set ACEs in the DACL of the security descriptor.
# * Set the security descriptor back on the SCP object.
def AllowAccessToScpProperties(
accountSAM, #Service account to allow access.
scpObject, # The IADs SCP object.
schemaIDGUIDs = # Attributes to allow write-access to.
("{28630eb8-41d5-11d1-a9c1-0000f80367c1}", # serviceDNSName
"{b7b1311c-b82e-11d0-afee-0000f80367c1}", # serviceBindingInformation
)
):
# If no service account is specified, service runs under LocalSystem.
# So allow access to the computer account of the service's host.
if accountSAM:
trustee = accountSAM
else:
# Get the SAM account name of the computer object for the server.
trustee = win32api.GetComputerObjectName(win32con.NameSamCompatible)
# Get the nTSecurityDescriptor attribute
attribute = "nTSecurityDescriptor"
sd = getattr(scpObject, attribute)
acl = sd.DiscretionaryAcl
for sguid in schemaIDGUIDs:
ace = Dispatch(adsi.CLSID_AccessControlEntry)
# Set the properties of the ACE.
# Allow read and write access to the property.
ace.AccessMask = ADS_RIGHT_DS_READ_PROP | ADS_RIGHT_DS_WRITE_PROP
# Set the trustee, which is either the service account or the
# host computer account.
ace.Trustee = trustee
# Set the ACE type.
ace.AceType = ADS_ACETYPE_ACCESS_ALLOWED_OBJECT
# Set AceFlags to zero because ACE is not inheritable.
ace.AceFlags = 0
# Set Flags to indicate an ACE that protects a specified object.
ace.Flags = ADS_FLAG_OBJECT_TYPE_PRESENT
# Set ObjectType to the schemaIDGUID of the attribute.
ace.ObjectType = sguid
# Add the ACEs to the DACL.
acl.AddAce(ace)
# Write the modified DACL back to the security descriptor.
sd.DiscretionaryAcl = acl
# Write the ntSecurityDescriptor property to the property cache.
setattr(scpObject, attribute, sd)
# SetInfo updates the SCP object in the directory.
scpObject.SetInfo()
logger.info("Set security on object for account '%s'" % (trustee,))
# Service Principal Names functions from the same sample.
# The example calls the DsWriteAccountSpn function, which stores the SPNs in
# Microsoft Active Directory under the servicePrincipalName attribute of the
# account object specified by the serviceAcctDN parameter. The account object
# corresponds to the logon account specified in the CreateService call for this
# service instance. If the logon account is a domain user account,
# serviceAcctDN must be the distinguished name of the account object in
# Active Directory for that user account. If the service's logon account is the
# LocalSystem account, serviceAcctDN must be the distinguished name of the
# computer account object for the host computer on which the service is
# installed. win32api.TranslateNames and win32security.DsCrackNames can
# be used to convert a domain\account format name to a distinguished name.
def SpnRegister(
serviceAcctDN, # DN of the service's logon account
spns, # List of SPNs to register
operation, # Add, replace, or delete SPNs
):
assert type(spns) not in [str, str] and hasattr(spns, "__iter__"), \
"spns must be a sequence of strings (got %r)" % spns
# Bind to a domain controller.
# Get the domain for the current user.
samName = win32api.GetUserNameEx(win32api.NameSamCompatible)
samName = samName.split('\\', 1)[0]
if not serviceAcctDN:
# Get the SAM account name of the computer object for the server.
serviceAcctDN = win32api.GetComputerObjectName(win32con.NameFullyQualifiedDN)
logger.debug("SpnRegister using DN '%s'", serviceAcctDN)
# Get the name of a domain controller in that domain.
info = win32security.DsGetDcName(
domainName=samName,
flags=dscon.DS_IS_FLAT_NAME |
dscon.DS_RETURN_DNS_NAME |
dscon.DS_DIRECTORY_SERVICE_REQUIRED)
# Bind to the domain controller.
handle = win32security.DsBind( info['DomainControllerName'] )
# Write the SPNs to the service account or computer account.
logger.debug("DsWriteAccountSpn with spns %s")
win32security.DsWriteAccountSpn(
handle, # handle to the directory
operation, # Add or remove SPN from account's existing SPNs
serviceAcctDN, # DN of service account or computer account
spns) # names
# Unbind the DS in any case (but Python would do it anyway)
handle.Close()
def UserChangePassword(username_dn, new_password):
# set the password on the account.
# Use the distinguished name to bind to the account object.
accountPath = "LDAP://" + username_dn
user = adsi.ADsGetObject(accountPath, adsi.IID_IADsUser)
# Set the password on the account.
user.SetPassword(new_password)
# functions related to the command-line interface
def log(level, msg, *args):
if verbose >= level:
print(msg % args)
class _NoDefault: pass
def _get_option(po, opt_name, default = _NoDefault):
parser, options = po
ret = getattr(options, opt_name, default)
if not ret and default is _NoDefault:
parser.error("The '%s' option must be specified for this operation" % opt_name)
if not ret:
ret = default
return ret
def _option_error(po, why):
parser = po[0]
parser.error(why)
def do_ScpCreate(po):
"""Create a Service Connection Point"""
global g_createdSCP
scp = ScpCreate(_get_option(po, "binding_string"),
_get_option(po, "service_class"),
_get_option(po, "account_name_sam", None),
keywords=_get_option(po, "keywords", None))
g_createdSCP = scp
return scp.distinguishedName
def do_ScpDelete(po):
"""Delete a Service Connection Point"""
sc = _get_option(po, "service_class")
try:
ScpDelete(sc)
except adsi.error as details:
if details[0] != winerror.ERROR_DS_OBJ_NOT_FOUND:
raise
log(2, "ScpDelete ignoring ERROR_DS_OBJ_NOT_FOUND for service-class '%s'",
sc)
return sc
def do_SpnCreate(po):
"""Create a Service Principal Name"""
# The 'service name' is the dn of our scp.
if g_createdSCP is None:
# Could accept an arg to avoid this?
_option_error(po, "ScpCreate must have been specified before SpnCreate")
# Create a Service Principal Name"
spns = win32security.DsGetSpn(dscon.DS_SPN_SERVICE,
_get_option(po, "service_class"),
g_createdSCP.distinguishedName,
_get_option(po, "port", 0),
None, None)
spn = spns[0]
log(2, "Created SPN: %s", spn)
global g_createdSPNLast
g_createdSPNLast = spn
g_createdSPNs.append(spn)
return spn
def do_SpnRegister(po):
"""Register a previously created Service Principal Name"""
if not g_createdSPNLast:
_option_error(po, "SpnCreate must appear before SpnRegister")
SpnRegister(_get_option(po, "account_name_dn", None),
(g_createdSPNLast,),
dscon.DS_SPN_ADD_SPN_OP)
return g_createdSPNLast
def do_SpnUnregister(po):
"""Unregister a previously created Service Principal Name"""
if not g_createdSPNLast:
_option_error(po, "SpnCreate must appear before SpnUnregister")
SpnRegister(_get_option(po, "account_name_dn", None),
(g_createdSPNLast,),
dscon.DS_SPN_DELETE_SPN_OP)
return g_createdSPNLast
def do_UserChangePassword(po):
"""Change the password for a specified user"""
UserChangePassword(_get_option(po, "account_name_dn"),
_get_option(po, "password"))
return "Password changed OK"
handlers = (
('ScpCreate', do_ScpCreate),
('ScpDelete', do_ScpDelete),
('SpnCreate', do_SpnCreate),
('SpnRegister', do_SpnRegister),
('SpnUnregister', do_SpnUnregister),
('UserChangePassword', do_UserChangePassword),
)
class HelpFormatter(optparse.IndentedHelpFormatter):
def format_description(self, description):
return description
def main():
global verbose
_handlers_dict = {}
arg_descs = []
for arg, func in handlers:
this_desc = "\n".join(textwrap.wrap(func.__doc__,
subsequent_indent = " " * 8))
arg_descs.append(" %s: %s" % (arg, this_desc))
_handlers_dict[arg.lower()] = func
description = __doc__ + "\ncommands:\n" + "\n".join(arg_descs) + "\n"
parser = optparse.OptionParser(usage = "%prog [options] command ...",
description=description,
formatter=HelpFormatter())
parser.add_option("-v", action="count",
dest="verbose", default=1,
help="increase the verbosity of status messages")
parser.add_option("-q", "--quiet", action="store_true",
help="Don't print any status messages")
parser.add_option("-t", "--test", action="store_true",
help="Execute a mini-test suite, providing defaults for most options and args"),
parser.add_option("", "--show-tracebacks", action="store_true",
help="Show the tracebacks for any exceptions")
parser.add_option("", "--service-class",
help="The service class name to use")
parser.add_option("", "--port", default=0,
help="The port number to associate with the SPN")
parser.add_option("", "--binding-string",
help="The binding string to use for SCP creation")
parser.add_option("", "--account-name",
help="The account name to use (default is LocalSystem)")
parser.add_option("", "--password",
help="The password to set.")
parser.add_option("", "--keyword", action="append", dest="keywords",
help="""A keyword to add to the SCP. May be specified
multiple times""")
parser.add_option("", "--log-level",
help="""The log-level to use - may be a number or a logging
module constant""", default=str(logging.WARNING))
options, args = parser.parse_args()
po = (parser, options)
# fixup misc
try:
options.port = int(options.port)
except (TypeError, ValueError):
parser.error("--port must be numeric")
# fixup log-level
try:
log_level = int(options.log_level)
except (TypeError, ValueError):
try:
log_level = int(getattr(logging, options.log_level.upper()))
except (ValueError, TypeError, AttributeError):
parser.error("Invalid --log-level value")
try:
sl = logger.setLevel
# logger is a real logger
except AttributeError:
# logger is logging module
sl = logging.getLogger().setLevel
sl(log_level)
# Check -q/-v
if options.quiet and options.verbose:
parser.error("Can't specify --quiet and --verbose")
if options.quiet:
options.verbose -= 1
verbose = options.verbose
# --test
if options.test:
if args:
parser.error("Can't specify args with --test")
args = "ScpDelete ScpCreate SpnCreate SpnRegister SpnUnregister ScpDelete"
log(1, "--test - pretending args are:\n %s", args)
args = args.split()
if not options.service_class:
options.service_class = "PythonScpTest"
log(2, "--test: --service-class=%s", options.service_class)
if not options.keywords:
options.keywords = "Python Powered".split()
log(2, "--test: --keyword=%s", options.keywords)
if not options.binding_string:
options.binding_string = "test binding string"
log(2, "--test: --binding-string=%s", options.binding_string)
# check args
if not args:
parser.error("No command specified (use --help for valid commands)")
for arg in args:
if arg.lower() not in _handlers_dict:
parser.error("Invalid command '%s' (use --help for valid commands)" % arg)
# Patch up account-name.
if options.account_name:
log(2, "Translating account name '%s'", options.account_name)
options.account_name_sam = win32security.TranslateName(options.account_name,
win32api.NameUnknown,
win32api.NameSamCompatible)
log(2, "NameSamCompatible is '%s'",options.account_name_sam)
options.account_name_dn = win32security.TranslateName(options.account_name,
win32api.NameUnknown,
win32api.NameFullyQualifiedDN)
log(2, "NameFullyQualifiedDNis '%s'",options.account_name_dn)
# do it.
for arg in args:
handler = _handlers_dict[arg.lower()] # already been validated
if handler is None:
parser.error("Invalid command '%s'" % arg)
err_msg = None
try:
try:
log(2, "Executing '%s'...", arg)
result = handler(po)
log(1, "%s: %s", arg, result)
except:
if options.show_tracebacks:
print("--show-tracebacks specified - dumping exception")
traceback.print_exc()
raise
except adsi.error as xxx_todo_changeme:
(hr, desc, exc, argerr) = xxx_todo_changeme.args
if exc:
extra_desc = exc[2]
else:
extra_desc = ""
err_msg = desc
if extra_desc:
err_msg += "\n\t" + extra_desc
except win32api.error as xxx_todo_changeme1:
(hr, func, msg) = xxx_todo_changeme1.args
err_msg = msg
if err_msg:
log(1, "Command '%s' failed: %s", arg, err_msg)
if __name__=='__main__':
try:
main()
except KeyboardInterrupt:
print("*** Interrupted")