1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
|
#!/usr/bin/env python
# Impacket - Collection of Python classes for working with network protocols.
#
# Copyright Fortra, LLC and its affiliated companies
#
# All rights reserved.
#
# This software is provided under a slightly modified version
# of the Apache Software License. See the accompanying LICENSE file
# for more information.
#
# Description:
# Base class for testing DCE/RPC Endpoints.
#
# Author:
# @martingalloar
#
from tests import RemoteTestCase
from impacket.dcerpc.v5 import transport, epm
class DCERPCTests(RemoteTestCase):
STRING_BINDING_FORMATTING = 1
STRING_BINDING_MAPPER = 2
TRANSFER_SYNTAX_NDR = ("8a885d04-1ceb-11c9-9fe8-08002b104860", "2.0")
TRANSFER_SYNTAX_NDR64 = ("71710533-BEBA-4937-8319-B5DBEF9CCC36", "1.0")
timeout = None
authn = False
authn_level = None
iface_uuid = None
protocol = None
string_binding = None
string_binding_formatting = STRING_BINDING_FORMATTING
transfer_syntax = None
machine_account = False
def connect(self, string_binding=None, iface_uuid=None):
"""Obtains a RPC Transport and a DCE interface according to the bindings and
transfer syntax specified.
:return: tuple of DCE/RPC and RPC Transport objects
:rtype: (DCERPC_v5, DCERPCTransport)
"""
string_binding = string_binding or self.string_binding
if not string_binding:
raise NotImplemented("String binding must be defined")
rpc_transport = transport.DCERPCTransportFactory(string_binding)
# Set timeout if defined
if self.timeout:
rpc_transport.set_connect_timeout(self.timeout)
# Authenticate if specified
if self.authn and hasattr(rpc_transport, 'set_credentials'):
# This method exists only for selected protocol sequences.
rpc_transport.set_credentials(self.username, self.password, self.domain, self.lmhash, self.nthash)
# Gets the DCE RPC object
dce = rpc_transport.get_dce_rpc()
# Set the authentication level
if self.authn_level:
dce.set_auth_level(self.authn_level)
# Connect
dce.connect()
# Bind if specified
iface_uuid = iface_uuid or self.iface_uuid
if iface_uuid and self.transfer_syntax:
dce.bind(iface_uuid, transfer_syntax=self.transfer_syntax)
elif iface_uuid:
dce.bind(iface_uuid)
return dce, rpc_transport
def setUp(self):
super(DCERPCTests, self).setUp()
self.set_transport_config(machine_account=self.machine_account)
if self.string_binding_formatting == self.STRING_BINDING_FORMATTING:
self.string_binding = self.string_binding.format(self)
elif self.string_binding_formatting == self.STRING_BINDING_MAPPER:
self.string_binding = epm.hept_map(self.machine, self.iface_uuid, protocol=self.protocol)
|
