File: test_epm.py

package info (click to toggle)
impacket 0.13.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 9,276 kB
  • sloc: python: 137,329; makefile: 10; sh: 3
file content (126 lines) | stat: -rw-r--r-- 4,565 bytes parent folder | download | duplicates (2)
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
# 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.
#
# Tested so far:
#   (h)ept_lookup
#   (h)ept_map
#
from __future__ import division
from __future__ import print_function
import socket
import pytest
import unittest
from tests.dcerpc import DCERPCTests

from impacket.dcerpc.v5 import epm
from impacket.dcerpc.v5.ndr import NULL
from impacket.uuid import string_to_bin, uuidtup_to_bin


class EPMTests(DCERPCTests):
    iface_uuid = epm.MSRPC_UUID_PORTMAP
    string_binding = r"ncacn_np:{0.machine}[\pipe\epmapper]"
    authn = True

    def test_lookup(self):
        dce, rpctransport = self.connect()
        request = epm.ept_lookup()
        request['inquiry_type'] = epm.RPC_C_EP_ALL_ELTS
        request['object'] = NULL
        request['Ifid'] = NULL
        request['vers_option'] = epm.RPC_C_VERS_ALL
        request['max_ents'] = 499
      
        resp = dce.request(request)
        for entry in resp['entries']:
            tower = entry['tower']['tower_octet_string']
            epm.EPMTower(b''.join(tower))

    def test_hlookup(self):
        epm.hept_lookup(self.machine)
        MSRPC_UUID_SAMR = uuidtup_to_bin(('12345778-1234-ABCD-EF00-0123456789AC', '1.0'))
        epm.hept_lookup(self.machine, inquiry_type=epm.RPC_C_EP_MATCH_BY_IF, ifId=MSRPC_UUID_SAMR)
        MSRPC_UUID_ATSVC = uuidtup_to_bin(('1FF70682-0A51-30E8-076D-740BE8CEE98B', '1.0'))
        epm.hept_lookup(self.machine, inquiry_type=epm.RPC_C_EP_MATCH_BY_IF, ifId=MSRPC_UUID_ATSVC)
        MSRPC_UUID_SCMR = uuidtup_to_bin(('367ABB81-9844-35F1-AD32-98F038001003', '2.0'))
        epm.hept_lookup(self.machine, inquiry_type=epm.RPC_C_EP_MATCH_BY_IF, ifId=MSRPC_UUID_SCMR)

    def test_map(self):
        dce, rpctransport = self.connect()
        tower = epm.EPMTower()
        interface = epm.EPMRPCInterface()
        interface['InterfaceUUID'] = string_to_bin('12345778-1234-ABCD-EF00-0123456789AC')
        interface['MajorVersion'] = 1
        interface['MinorVersion'] = 0

        dataRep = epm.EPMRPCDataRepresentation()
        dataRep['DataRepUuid'] = string_to_bin('8a885d04-1ceb-11c9-9fe8-08002b104860')
        dataRep['MajorVersion'] = 2
        dataRep['MinorVersion'] = 0

        protId = epm.EPMProtocolIdentifier()
        protId['ProtIdentifier'] = 0xb

        pipeName = epm.EPMPipeName()
        pipeName['PipeName'] = b'\x00'

        portAddr = epm.EPMPortAddr()
        portAddr['IpPort'] = 0

        hostAddr = epm.EPMHostAddr()
        hostAddr['Ip4addr'] = socket.inet_aton('0.0.0.0')

        hostName = epm.EPMHostName()
        hostName['HostName'] = b'\x00'

        tower['NumberOfFloors'] = 5
        tower['Floors'] = interface.getData() + dataRep.getData() + protId.getData() + portAddr.getData() + hostAddr.getData()
        request = epm.ept_map()
        request['max_towers'] = 4
        request['map_tower']['tower_length'] = len(tower)
        request['map_tower']['tower_octet_string'] = tower.getData()
        resp = dce.request(request)
        resp.dump()

    def test_hept_map(self):
        MSRPC_UUID_SAMR = uuidtup_to_bin(('12345778-1234-ABCD-EF00-0123456789AC', '1.0'))
        epm.hept_map(self.machine, MSRPC_UUID_SAMR)
        epm.hept_map(self.machine, MSRPC_UUID_SAMR, protocol='ncacn_ip_tcp')
        MSRPC_UUID_ATSVC = uuidtup_to_bin(('1FF70682-0A51-30E8-076D-740BE8CEE98B', '1.0'))
        epm.hept_map(self.machine, MSRPC_UUID_ATSVC)
        MSRPC_UUID_SCMR = uuidtup_to_bin(('367ABB81-9844-35F1-AD32-98F038001003', '2.0'))
        epm.hept_map(self.machine, MSRPC_UUID_SCMR, protocol='ncacn_ip_tcp')


@pytest.mark.remote
class EPMTestsSMBTransport(EPMTests, unittest.TestCase):
    transfer_syntax = DCERPCTests.TRANSFER_SYNTAX_NDR


@pytest.mark.remote
class EPMTestsSMBTransport64(EPMTests, unittest.TestCase):
    transfer_syntax = DCERPCTests.TRANSFER_SYNTAX_NDR64


@pytest.mark.remote
class EPMTestsTCPTransport(EPMTests, unittest.TestCase):
    string_binding = r"ncacn_ip_tcp:{0.machine}[135]"
    transfer_syntax = DCERPCTests.TRANSFER_SYNTAX_NDR


@pytest.mark.remote
class EPMTestsTCPTransport64(EPMTests, unittest.TestCase):
    string_binding = r"ncacn_ip_tcp:{0.machine}[135]"
    transfer_syntax = DCERPCTests.TRANSFER_SYNTAX_NDR64


# Process command-line arguments.
if __name__ == "__main__":
    unittest.main(verbosity=1)