CoCalc -- test_32_ftps

GitHub Repository: freebsd/pkg
Path: blob/main/external/curl/tests/http/test_32_ftps_vsftpd.py
²⁶⁵⁴ views
1
#!/usr/bin/env python3
2
# -*- coding: utf-8 -*-
3
#***************************************************************************
4
#                                  _   _ ____  _
5
#  Project                     ___| | | |  _ \| |
6
#                             / __| | | | |_) | |
7
#                            | (__| |_| |  _ <| |___
8
#                             \___|\___/|_| \_\_____|
9
#
10
# Copyright (C) Daniel Stenberg, <[email protected]>, et al.
11
#
12
# This software is licensed as described in the file COPYING, which
13
# you should have received as part of this distribution. The terms
14
# are also available at https://curl.se/docs/copyright.html.
15
#
16
# You may opt to use, copy, modify, merge, publish, distribute and/or sell
17
# copies of the Software, and permit persons to whom the Software is
18
# furnished to do so, under the terms of the COPYING file.
19
#
20
# This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
21
# KIND, either express or implied.
22
#
23
# SPDX-License-Identifier: curl
24
#
25
###########################################################################
26
#
27
import difflib
28
import filecmp
29
import logging
30
import os
31
import shutil
32
import pytest
33

34
from testenv import Env, CurlClient, VsFTPD
35

36

37
log = logging.getLogger(__name__)
38

39

40
@pytest.mark.skipif(condition=not Env.has_vsftpd(), reason="missing vsftpd")
41
class TestFtpsVsFTPD:
42

43
    SUPPORTS_SSL = True
44

45
    @pytest.fixture(autouse=True, scope='class')
46
    def vsftpds(self, env):
47
        if not TestFtpsVsFTPD.SUPPORTS_SSL:
48
            pytest.skip('vsftpd does not seem to support SSL')
49
        vsftpds = VsFTPD(env=env, with_ssl=True, ssl_implicit=True)
50
        if not vsftpds.initial_start():
51
            vsftpds.stop()
52
            TestFtpsVsFTPD.SUPPORTS_SSL = False
53
            pytest.skip('vsftpd does not seem to support SSL')
54
        yield vsftpds
55
        vsftpds.stop()
56

57
    def _make_docs_file(self, docs_dir: str, fname: str, fsize: int):
58
        fpath = os.path.join(docs_dir, fname)
59
        data1k = 1024*'x'
60
        flen = 0
61
        with open(fpath, 'w') as fd:
62
            while flen < fsize:
63
                fd.write(data1k)
64
                flen += len(data1k)
65
        return flen
66

67
    @pytest.fixture(autouse=True, scope='class')
68
    def _class_scope(self, env, vsftpds):
69
        if os.path.exists(vsftpds.docs_dir):
70
            shutil.rmtree(vsftpds.docs_dir)
71
        if not os.path.exists(vsftpds.docs_dir):
72
            os.makedirs(vsftpds.docs_dir)
73
        self._make_docs_file(docs_dir=vsftpds.docs_dir, fname='data-1k', fsize=1024)
74
        self._make_docs_file(docs_dir=vsftpds.docs_dir, fname='data-10k', fsize=10*1024)
75
        self._make_docs_file(docs_dir=vsftpds.docs_dir, fname='data-1m', fsize=1024*1024)
76
        self._make_docs_file(docs_dir=vsftpds.docs_dir, fname='data-10m', fsize=10*1024*1024)
77
        env.make_data_file(indir=env.gen_dir, fname="upload-1k", fsize=1024)
78
        env.make_data_file(indir=env.gen_dir, fname="upload-100k", fsize=100*1024)
79
        env.make_data_file(indir=env.gen_dir, fname="upload-1m", fsize=1024*1024)
80

81
    def test_32_01_list_dir(self, env: Env, vsftpds: VsFTPD):
82
        curl = CurlClient(env=env)
83
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/'
84
        r = curl.ftp_get(urls=[url], with_stats=True)
85
        r.check_stats(count=1, http_status=226)
86
        lines = open(os.path.join(curl.run_dir, 'download_#1.data')).readlines()
87
        assert len(lines) == 4, f'list: {lines}'
88

89
    # download 1 file, no SSL
90
    @pytest.mark.parametrize("docname", [
91
        'data-1k', 'data-1m', 'data-10m'
92
    ])
93
    def test_32_02_download_1(self, env: Env, vsftpds: VsFTPD, docname):
94
        curl = CurlClient(env=env)
95
        srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
96
        count = 1
97
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
98
        r = curl.ftp_get(urls=[url], with_stats=True)
99
        r.check_stats(count=count, http_status=226)
100
        self.check_downloads(curl, srcfile, count)
101

102
    @pytest.mark.parametrize("docname", [
103
        'data-1k', 'data-1m', 'data-10m'
104
    ])
105
    def test_32_03_download_10_serial(self, env: Env, vsftpds: VsFTPD, docname):
106
        curl = CurlClient(env=env)
107
        srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
108
        count = 10
109
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
110
        r = curl.ftp_get(urls=[url], with_stats=True)
111
        r.check_stats(count=count, http_status=226)
112
        self.check_downloads(curl, srcfile, count)
113
        assert r.total_connects == count + 1, 'should reuse the control conn'
114

115
    # 2 serial transfers, first with 'ftps://' and second with 'ftp://'
116
    # we want connection reuse in this case
117
    def test_32_03b_ftp_compat_ftps(self, env: Env, vsftpds: VsFTPD):
118
        curl = CurlClient(env=env)
119
        docname = 'data-1k'
120
        count = 2
121
        url1= f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}'
122
        url2 = f'ftp://{env.ftp_domain}:{vsftpds.port}/{docname}'
123
        r = curl.ftp_get(urls=[url1, url2], with_stats=True)
124
        r.check_stats(count=count, http_status=226)
125
        assert r.total_connects == count + 1, 'should reuse the control conn'
126

127
    @pytest.mark.parametrize("docname", [
128
        'data-1k', 'data-1m', 'data-10m'
129
    ])
130
    def test_32_04_download_10_parallel(self, env: Env, vsftpds: VsFTPD, docname):
131
        curl = CurlClient(env=env)
132
        srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
133
        count = 10
134
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
135
        r = curl.ftp_get(urls=[url], with_stats=True, extra_args=[
136
            '--parallel'
137
        ])
138
        r.check_stats(count=count, http_status=226)
139
        self.check_downloads(curl, srcfile, count)
140
        assert r.total_connects > count + 1, 'should have used several control conns'
141

142
    @pytest.mark.parametrize("docname", [
143
        'upload-1k', 'upload-100k', 'upload-1m'
144
    ])
145
    def test_32_05_upload_1(self, env: Env, vsftpds: VsFTPD, docname):
146
        curl = CurlClient(env=env)
147
        srcfile = os.path.join(env.gen_dir, docname)
148
        dstfile = os.path.join(vsftpds.docs_dir, docname)
149
        self._rmf(dstfile)
150
        count = 1
151
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/'
152
        r = curl.ftp_upload(urls=[url], fupload=f'{srcfile}', with_stats=True)
153
        r.check_stats(count=count, http_status=226)
154
        self.check_upload(env, vsftpds, docname=docname)
155

156
    def _rmf(self, path):
157
        if os.path.exists(path):
158
            return os.remove(path)
159

160
    # check with `tcpdump` if curl causes any TCP RST packets
161
    @pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
162
    @pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
163
    @pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
164
    def test_32_06_shutdownh_download(self, env: Env, vsftpds: VsFTPD):
165
        docname = 'data-1k'
166
        curl = CurlClient(env=env)
167
        count = 1
168
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
169
        r = curl.ftp_get(urls=[url], with_stats=True, with_tcpdump=True)
170
        r.check_stats(count=count, http_status=226)
171
        # vsftp closes control connection without niceties,
172
        # look only at ports from DATA connection.
173
        data_ports = vsftpds.get_data_ports(r)
174
        assert len(data_ports), f'unable to find FTP data port connected to\n{r.dump_logs()}'
175
        assert len(r.tcpdump.get_rsts(ports=data_ports)) == 0, 'Unexpected TCP RST packets'
176

177
    # check with `tcpdump` if curl causes any TCP RST packets
178
    @pytest.mark.skipif(condition=not Env.tcpdump(), reason="tcpdump not available")
179
    @pytest.mark.skipif(condition=not Env.curl_is_debug(), reason="needs curl debug")
180
    @pytest.mark.skipif(condition=not Env.curl_is_verbose(), reason="needs curl verbose strings")
181
    def test_32_07_shutdownh_upload(self, env: Env, vsftpds: VsFTPD):
182
        docname = 'upload-1k'
183
        curl = CurlClient(env=env)
184
        srcfile = os.path.join(env.gen_dir, docname)
185
        dstfile = os.path.join(vsftpds.docs_dir, docname)
186
        self._rmf(dstfile)
187
        count = 1
188
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/'
189
        r = curl.ftp_upload(urls=[url], fupload=f'{srcfile}', with_stats=True, with_tcpdump=True)
190
        r.check_stats(count=count, http_status=226)
191
        # vsftp closes control connection without niceties,
192
        # look only at ports from DATA connection.
193
        data_ports = vsftpds.get_data_ports(r)
194
        assert len(data_ports), f'unable to find FTP data port connected to\n{r.dump_logs()}'
195
        assert len(r.tcpdump.get_rsts(ports=data_ports)) == 0, 'Unexpected TCP RST packets'
196

197
    def test_32_08_upload_ascii(self, env: Env, vsftpds: VsFTPD):
198
        docname = 'upload-ascii'
199
        line_length = 21
200
        srcfile = os.path.join(env.gen_dir, docname)
201
        dstfile = os.path.join(vsftpds.docs_dir, docname)
202
        env.make_data_file(indir=env.gen_dir, fname=docname, fsize=100*1024,
203
                           line_length=line_length)
204
        srcsize = os.path.getsize(srcfile)
205
        self._rmf(dstfile)
206
        count = 1
207
        curl = CurlClient(env=env)
208
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/'
209
        r = curl.ftp_upload(urls=[url], fupload=f'{srcfile}', with_stats=True,
210
                            extra_args=['--use-ascii'])
211
        r.check_stats(count=count, http_status=226)
212
        # expect the uploaded file to be number of converted newlines larger
213
        dstsize = os.path.getsize(dstfile)
214
        newlines = len(open(srcfile).readlines())
215
        assert (srcsize + newlines) == dstsize, \
216
            f'expected source with {newlines} lines to be that much larger,'\
217
            f'instead srcsize={srcsize}, upload size={dstsize}, diff={dstsize-srcsize}'
218

219
    def test_32_08_active_download(self, env: Env, vsftpds: VsFTPD):
220
        docname = 'data-10k'
221
        curl = CurlClient(env=env)
222
        srcfile = os.path.join(vsftpds.docs_dir, f'{docname}')
223
        count = 1
224
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
225
        r = curl.ftp_get(urls=[url], with_stats=True, extra_args=[
226
            '--ftp-port', '127.0.0.1'
227
        ])
228
        r.check_stats(count=count, http_status=226)
229
        self.check_downloads(curl, srcfile, count)
230

231
    def test_32_09_active_upload(self, env: Env, vsftpds: VsFTPD):
232
        docname = 'upload-1k'
233
        curl = CurlClient(env=env)
234
        srcfile = os.path.join(env.gen_dir, docname)
235
        dstfile = os.path.join(vsftpds.docs_dir, docname)
236
        self._rmf(dstfile)
237
        count = 1
238
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/'
239
        r = curl.ftp_upload(urls=[url], fupload=f'{srcfile}', with_stats=True, extra_args=[
240
            '--ftp-port', '127.0.0.1'
241
        ])
242
        r.check_stats(count=count, http_status=226)
243
        self.check_upload(env, vsftpds, docname=docname)
244

245
    @pytest.mark.parametrize("indata", [
246
        pytest.param('1234567890', id='10-bytes'),
247
        pytest.param('', id='0-bytes'),
248
    ])
249
    def test_32_10_upload_stdin(self, env: Env, vsftpds: VsFTPD, indata):
250
        curl = CurlClient(env=env)
251
        docname = "upload_31_10"
252
        dstfile = os.path.join(vsftpds.docs_dir, docname)
253
        self._rmf(dstfile)
254
        count = 1
255
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}'
256
        r = curl.ftp_upload(urls=[url], updata=indata, with_stats=True)
257
        r.check_stats(count=count, http_status=226)
258
        assert os.path.exists(dstfile)
259
        destdata = open(dstfile).readlines()
260
        expdata = [indata] if len(indata) else []
261
        assert expdata == destdata, f'expected: {expdata}, got: {destdata}'
262

263
    def test_32_11_download_non_existing(self, env: Env, vsftpds: VsFTPD):
264
        curl = CurlClient(env=env)
265
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/does-not-exist'
266
        r = curl.ftp_get(urls=[url], with_stats=True)
267
        r.check_exit_code(78)
268
        r.check_stats(count=1, exitcode=78)
269

270
    def check_downloads(self, client, srcfile: str, count: int,
271
                        complete: bool = True):
272
        for i in range(count):
273
            dfile = client.download_file(i)
274
            assert os.path.exists(dfile)
275
            if complete and not filecmp.cmp(srcfile, dfile, shallow=False):
276
                diff = "".join(difflib.unified_diff(a=open(srcfile).readlines(),
277
                                                    b=open(dfile).readlines(),
278
                                                    fromfile=srcfile,
279
                                                    tofile=dfile,
280
                                                    n=1))
281
                assert False, f'download {dfile} differs:\n{diff}'
282

283
    def check_upload(self, env, vsftpd: VsFTPD, docname):
284
        srcfile = os.path.join(env.gen_dir, docname)
285
        dstfile = os.path.join(vsftpd.docs_dir, docname)
286
        assert os.path.exists(srcfile)
287
        assert os.path.exists(dstfile)
288
        if not filecmp.cmp(srcfile, dstfile, shallow=False):
289
            diff = "".join(difflib.unified_diff(a=open(srcfile).readlines(),
290
                                                b=open(dstfile).readlines(),
291
                                                fromfile=srcfile,
292
                                                tofile=dstfile,
293
                                                n=1))
294
            assert False, f'upload {dstfile} differs:\n{diff}'
295

296
Product

Resources

Company
