1# Tests for limiting processes forked on accept by the standard process model
2#
3# Copyright (C) Andrew Bartlett <abartlet@samba.org> 2018
4#
5# This program is free software; you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation; either version 3 of the License, or
8# (at your option) any later version.
9#
10# This program is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with this program.  If not, see <http://www.gnu.org/licenses/>.
17#
18
19from __future__ import print_function
20"""Tests limits on processes forked by fork on accept in the standard process
21   model.
22   NOTE: This test runs in an environment with an artificially low setting for
23         smbd max processes
24"""
25
26
27import os
28from samba.tests import TestCase
29from samba.samdb import SamDB
30from ldb import LdbError, ERR_OPERATIONS_ERROR
31
32
33class StandardModelProcessLimitTests(TestCase):
34
35    def setUp(self):
36        super(StandardModelProcessLimitTests, self).setUp()
37
38    def tearDown(self):
39        super(StandardModelProcessLimitTests, self).tearDown()
40
41    def simple_bind(self):
42        creds = self.insta_creds(template=self.get_credentials())
43        creds.set_bind_dn("%s\\%s" % (creds.get_domain(),
44                                      creds.get_username()))
45
46        return SamDB(url="ldaps://%s" % os.environ["SERVER"],
47                     lp=self.get_loadparm(),
48                     credentials=creds)
49
50    def test_process_limits(self):
51        creds = self.insta_creds(template=self.get_credentials())
52        creds.set_bind_dn("%s\\%s" % (creds.get_domain(),
53                                      creds.get_username()))
54
55        connections = []
56        try:
57            # Open a series of LDAP connections, the maximum number of
58            # active connections should be 20, so the 21st should fail.
59            # But as it is possible that there may be other processes holding
60            # connections, need to allow for earlier connection failures.
61            for _ in range(21):
62                connections.append(self.simple_bind())
63            self.fail(
64                "Processes not limited, able to make more than 20 connections")
65        except LdbError as e:
66            (errno, estr) = e.args
67            if errno != ERR_OPERATIONS_ERROR:
68                raise
69            if not (estr.endswith("NT_STATUS_CONNECTION_DISCONNECTED") or
70                    estr.endswith("NT_STATUS_CONNECTION_RESET")):
71                raise
72            pass
73        #
74        # Clean up the connections we've just opened, by deleting the
75        # connection in python. This should invoke the talloc destructor to
76        # release any resources and close the actual connection to the server.
77        for c in connections:
78            del c
79