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
    def test_32_06_shutdownh_download(self, env: Env, vsftpds: VsFTPD):
163
        docname = 'data-1k'
164
        curl = CurlClient(env=env)
165
        count = 1
166
        url = f'ftps://{env.ftp_domain}:{vsftpds.port}/{docname}?[0-{count-1}]'
167
        r = curl.ftp_get(urls=[url], with_stats=True, with_tcpdump=True)
168
        r.check_stats(count=count, http_status=226)
169
        # vsftp closes control connection without niceties,
170
        # look only at ports from DATA connection.
171
        data_ports = vsftpds.get_data_ports(r)
172
        assert len(data_ports), f'unable to find FTP data port connected to\n{r.dump_logs()}'
173
        assert len(r.tcpdump.get_rsts(ports=data_ports)) == 0, 'Unexpected TCP RST packets'
174

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

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

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

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

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

259
    def check_downloads(self, client, srcfile: str, count: int,
260
                        complete: bool = True):
261
        for i in range(count):
262
            dfile = client.download_file(i)
263
            assert os.path.exists(dfile)
264
            if complete and not filecmp.cmp(srcfile, dfile, shallow=False):
265
                diff = "".join(difflib.unified_diff(a=open(srcfile).readlines(),
266
                                                    b=open(dfile).readlines(),
267
                                                    fromfile=srcfile,
268
                                                    tofile=dfile,
269
                                                    n=1))
270
                assert False, f'download {dfile} differs:\n{diff}'
271

272
    def check_upload(self, env, vsftpd: VsFTPD, docname):
273
        srcfile = os.path.join(env.gen_dir, docname)
274
        dstfile = os.path.join(vsftpd.docs_dir, docname)
275
        assert os.path.exists(srcfile)
276
        assert os.path.exists(dstfile)
277
        if not filecmp.cmp(srcfile, dstfile, shallow=False):
278
            diff = "".join(difflib.unified_diff(a=open(srcfile).readlines(),
279
                                                b=open(dstfile).readlines(),
280
                                                fromfile=srcfile,
281
                                                tofile=dstfile,
282
                                                n=1))
283
            assert False, f'upload {dstfile} differs:\n{diff}'
284

285
Product

Resources

Company
