|
16 | 16 | except ImportError: |
17 | 17 | ssl = None |
18 | 18 |
|
19 | | -from unittest import TestCase, skipUnless |
| 19 | +from unittest import mock, TestCase, skipUnless |
20 | 20 | from test import support |
21 | 21 | from test.support import requires_subprocess |
22 | 22 | from test.support import threading_helper |
@@ -1145,6 +1145,40 @@ def testTimeoutDirectAccess(self): |
1145 | 1145 | ftp.close() |
1146 | 1146 |
|
1147 | 1147 |
|
| 1148 | +class TestFtpcpSecurity(TestCase): |
| 1149 | + """ftpcp() must not trust the host a source server advertises in PASV. |
| 1150 | +
|
| 1151 | + A malicious source server can otherwise redirect the target server's |
| 1152 | + data connection to an arbitrary host:port (SSRF), so ftpcp() uses the |
| 1153 | + source server's actual peer address instead, the same as FTP.makepasv(). |
| 1154 | + """ |
| 1155 | + |
| 1156 | + def _make_pair(self, *, advertised_host, real_host, trust=False): |
| 1157 | + source = mock.Mock(spec=ftplib.FTP) |
| 1158 | + source.trust_server_pasv_ipv4_address = trust |
| 1159 | + source.sock.getpeername.return_value = (real_host, 21) |
| 1160 | + # PASV replies give the host as comma-separated octets, not dotted. |
| 1161 | + advertised = advertised_host.replace('.', ',') |
| 1162 | + source.sendcmd.side_effect = lambda cmd: ( |
| 1163 | + f'227 Entering Passive Mode ({advertised},1,2).' |
| 1164 | + if cmd == 'PASV' else '150 ok') |
| 1165 | + target = mock.Mock(spec=ftplib.FTP) |
| 1166 | + target.sendcmd.return_value = '150 ok' |
| 1167 | + return source, target |
| 1168 | + |
| 1169 | + def test_ftpcp_ignores_untrusted_pasv_host(self): |
| 1170 | + source, target = self._make_pair(advertised_host='10.0.0.5', |
| 1171 | + real_host='198.51.100.7') |
| 1172 | + ftplib.ftpcp(source, 'a', target, 'b') |
| 1173 | + target.sendport.assert_called_once_with('198.51.100.7', 258) |
| 1174 | + |
| 1175 | + def test_ftpcp_trust_server_pasv_ipv4_address(self): |
| 1176 | + source, target = self._make_pair(advertised_host='10.0.0.5', |
| 1177 | + real_host='198.51.100.7', trust=True) |
| 1178 | + ftplib.ftpcp(source, 'a', target, 'b') |
| 1179 | + target.sendport.assert_called_once_with('10.0.0.5', 258) |
| 1180 | + |
| 1181 | + |
1148 | 1182 | class MiscTestCase(TestCase): |
1149 | 1183 | def test__all__(self): |
1150 | 1184 | not_exported = { |
|
0 commit comments