Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if service['service_name'].lower().startswith("paexec"):
paexec_services.append(service['service_name'])
smb_tree = TreeConnect(client.session,
r"\\%s\ADMIN$" % client.connection.server_name)
smb_tree.connect()
share = Open(smb_tree, "")
share.create(ImpersonationLevel.Impersonation,
DirectoryAccessMask.FILE_READ_ATTRIBUTES |
DirectoryAccessMask.SYNCHRONIZE |
DirectoryAccessMask.FILE_LIST_DIRECTORY,
FileAttributes.FILE_ATTRIBUTE_DIRECTORY,
ShareAccess.FILE_SHARE_READ |
ShareAccess.FILE_SHARE_WRITE |
ShareAccess.FILE_SHARE_DELETE,
CreateDisposition.FILE_OPEN,
CreateOptions.FILE_DIRECTORY_FILE)
try:
paexec_files = share.query_directory("PAExec-*.exe",
FileInformationClass.
FILE_NAMES_INFORMATION)
except SMBResponseException as exc:
if exc.status != NtStatus.STATUS_NO_SUCH_FILE:
raise exc
paexec_files = []
return client, paexec_services, paexec_files
paexec_file.create(ImpersonationLevel.Impersonation,
FilePipePrinterAccessMask.FILE_WRITE_DATA,
FileAttributes.FILE_ATTRIBUTE_NORMAL,
ShareAccess.FILE_SHARE_READ,
CreateDisposition.FILE_OVERWRITE_IF,
CreateOptions.FILE_NON_DIRECTORY_FILE)
paexec_file.write(pkgutil.get_data('pypsexec', 'paexec.exe'), 0)
paexec_file.close(get_attributes=False)
yield session
finally:
paexec_file.create(ImpersonationLevel.Impersonation,
FilePipePrinterAccessMask.DELETE,
FileAttributes.FILE_ATTRIBUTE_NORMAL,
ShareAccess.FILE_SHARE_DELETE,
CreateDisposition.FILE_OVERWRITE_IF,
CreateOptions.FILE_DELETE_ON_CLOSE)
paexec_file.close(get_attributes=False)
connection.disconnect(True)
else:
pytest.skip("PYPSEXEC_SERVER, PYPSEXEC_USERNAME, PYPSEXEC_PASSWORD"
" environment variables were not set. Integration "
def create_pipe(tree, name, access_mask):
pipe = Open(tree, name)
pipe.open(ImpersonationLevel.Impersonation,
access_mask,
FileAttributes.FILE_ATTRIBUTE_NORMAL,
ShareAccess.FILE_SHARE_READ |
ShareAccess.FILE_SHARE_WRITE |
ShareAccess.FILE_SHARE_DELETE,
CreateDisposition.FILE_OPEN,
CreateOptions.FILE_NON_DIRECTORY_FILE |
CreateOptions.FILE_SYNCHRONOUS_IO_NONALERT)
return pipe
def open(self):
log.debug("Connecting to SMB Tree %s for SCMR" % self.tree.share_name)
self.tree.connect()
log.debug("Opening handle to svcctl pipe")
self.handle.create(ImpersonationLevel.Impersonation,
FilePipePrinterAccessMask.GENERIC_READ |
FilePipePrinterAccessMask.GENERIC_WRITE,
0,
ShareAccess.FILE_SHARE_READ |
ShareAccess.FILE_SHARE_WRITE |
ShareAccess.FILE_SHARE_DELETE,
CreateDisposition.FILE_OPEN,
CreateOptions.FILE_NON_DIRECTORY_FILE)
# we need to bind svcctl to SCManagerW over DCE/RPC
bind = BindPDU()
bind['pfx_flags'].set_flag(PFlags.PFC_FIRST_FRAG)
bind['pfx_flags'].set_flag(PFlags.PFC_LAST_FRAG)
bind['packed_drep'] = DataRepresentationFormat()
bind['call_id'] = self.call_id
self.call_id += 1
context_ndr = ContextElement()
context_ndr['context_id'] = 0
context_ndr['abstract_syntax'] = SyntaxIdElement()
context_ndr['abstract_syntax']['uuid'] = \