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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
# Unix SMB/CIFS implementation.
#
# Copyright (C) Catalyst.Net Ltd. 2017
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
"""Test whether various python calls segfault when given unexpected input.
"""
import samba.tests
import os
import sys
from samba.net import Net, LIBNET_JOIN_AUTOMATIC
from samba.credentials import DONT_USE_KERBEROS
from samba import NTSTATUSError, ntstatus
from samba.dcerpc import misc, drsuapi, samr, unixinfo, dnsserver
from samba import auth, gensec
from samba.samdb import SamDB
from samba import netbios
from samba import registry
from samba import ldb
from samba import messaging
import traceback
def segfault_detector(f):
def wrapper(*args, **kwargs):
pid = os.fork()
if pid == 0:
try:
f(*args, **kwargs)
except Exception as e:
traceback.print_exc()
sys.stderr.flush()
sys.stdout.flush()
os._exit(0)
pid2, status = os.waitpid(pid, 0)
signal = status & 255
if os.WIFSIGNALED(status):
signal = os.WTERMSIG(status)
raise AssertionError("Failed with signal %d" % signal)
return wrapper
class SegfaultTests(samba.tests.TestCase):
def get_lp_et_al(self):
server = os.environ["SERVER"]
lp = self.get_loadparm()
creds = self.insta_creds(template=self.get_credentials(),
kerberos_state=DONT_USE_KERBEROS)
return lp, creds, server
def get_samdb(self):
lp, creds, server = self.get_lp_et_al()
url = 'ldap://' + server
ldb = SamDB(url, credentials=creds, lp=lp)
return ldb
@segfault_detector
def test_net_replicate_init__1(self):
lp, creds, server = self.get_lp_et_al()
net = Net(creds, lp, server=server)
net.replicate_init(42, lp, None, misc.GUID())
@segfault_detector
def test_net_replicate_init__3(self):
# third argument is also unchecked
samdb = self.get_samdb()
lp, creds, server = self.get_lp_et_al()
net = Net(creds, lp, server=server)
net.replicate_init(samdb, lp, 42, misc.GUID())
@segfault_detector
def test_net_replicate_chunk_1(self):
lp, creds, server = self.get_lp_et_al()
ctr = drsuapi.DsGetNCChangesCtr6()
net = Net(creds, lp, server=server)
net.replicate_chunk(42, 1, ctr)
@segfault_detector
def test_auth_context_gensec_start_server(self):
a = auth.AuthContext(ldb=42, methods=['sam'])
# there is no failure yet because the ldb is not actually
# dereferenced.
g = gensec.Security.start_server(auth_context=a)
# and still the ldb is not dereferenced...
@segfault_detector
def test_auth_user_session(self):
s = auth.user_session(ldb=42, principal='foo')
@segfault_detector
def test_gensec_start_server(self):
gensec.Security.start_server(auth_context=42)
@segfault_detector
def test_netbios_query_name(self):
n = netbios.Node()
t = n.query_name((42, 'foo'), 'localhost')
@segfault_detector
def test_encrypt_netr_crypt_password(self):
lp, creds, server = self.get_lp_et_al()
creds.encrypt_netr_crypt_password(42)
@segfault_detector
def test_hive_open_ldb(self):
# we don't need to provide a valid path because we segfault first
try:
registry.open_ldb('', credentials=42)
except ldb.LdbError as e:
print("failed with %s" % e)
@segfault_detector
def test_hive_open_hive(self):
# we don't need to provide a valid path because we segfault first
try:
registry.open_hive('s', 's', 's', 's')
except ldb.LdbError as e:
print("failed with %s" % e)
@segfault_detector
def test_ldb_add_nameless_element(self):
m = ldb.Message()
e = ldb.MessageElement('q')
try:
m.add(e)
except ldb.LdbError:
pass
str(m)
@segfault_detector
def test_ldb_register_module(self):
ldb.register_module('')
@segfault_detector
def test_messaging_deregister(self):
messaging.deregister('s', 's', 's', False)
@segfault_detector
def test_rpcecho(self):
from dcerpc import echo
echo.rpcecho("")
@segfault_detector
def test_dcerpc_idl_ref_elements(self):
"""There are many pidl generated functions that crashed on this
pattern, where a NULL pointer was created rather than an empty
structure."""
samr.Connect5().out_info_out = 1
@segfault_detector
def test_dcerpc_idl_unixinfo_elements(self):
"""Dereferencing is sufficient to crash"""
unixinfo.GetPWUid().out_infos
@segfault_detector
def test_dcerpc_idl_inline_arrays(self):
"""Inline arrays were incorrectly handled."""
dnsserver.DNS_RPC_SERVER_INFO_DOTNET().pExtensions
@segfault_detector
def test_ldb_dn_explode_crash(self):
for i in range(106, 550, 5):
dn = ldb.Dn(ldb.Ldb(), "a=b%s,c= " % (' ' * i))
dn.validate()
|
