326 lines
10 KiB
Python
326 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
FTP 上传+下载测试脚本
|
||
用法：python ftp_up_dl_test.py --host 192.168.1.100
|
||
"""
|
||
|
||
import socket
|
||
import sys
|
||
import time
|
||
import re
|
||
import argparse
|
||
import os
|
||
|
||
HOST = "192.168.1.100"
|
||
PORT = 21
|
||
TIMEOUT = 15.0
|
||
|
||
TEST_DATA = b"Hello from Python FTP test!\r\nThis is line 2.\r\nLine 3: " + b"ABCDEFGH" * 50 + b"\r\n"
|
||
|
||
def recv_line(sock, timeout=TIMEOUT):
|
||
data = b""
|
||
sock.settimeout(timeout)
|
||
while True:
|
||
try:
|
||
ch = sock.recv(1)
|
||
except socket.timeout:
|
||
print(f" [TIMEOUT] partial: {data[:80]!r}")
|
||
return None
|
||
if not ch:
|
||
return data if data else None
|
||
data += ch
|
||
if data.endswith(b"\r\n"):
|
||
return data
|
||
|
||
def send_cmd(sock, cmd, expect_code=None):
|
||
print(f" > {cmd}")
|
||
sock.sendall((cmd + "\r\n").encode())
|
||
time.sleep(0.15)
|
||
response = recv_line(sock)
|
||
if response:
|
||
text = response.decode(errors='replace').rstrip()
|
||
print(f" < {text}")
|
||
if expect_code and not text.startswith(str(expect_code)):
|
||
print(f" *** Expected {expect_code}, got: {text}")
|
||
return None
|
||
return response
|
||
print(" < [NO RESPONSE]")
|
||
return None
|
||
|
||
def parse_pasv(resp):
|
||
m = re.search(r"(\d+),(\d+),(\d+),(\d+),(\d+),(\d+)", resp.decode())
|
||
if m:
|
||
a, b, c, d, e, f = [int(x) for x in m.groups()]
|
||
return f"{a}.{b}.{c}.{d}", e * 256 + f
|
||
return None, None
|
||
|
||
def ftp_transfer_data(host, pasv_ip, pasv_port, cmd, data_to_send=None, read_data=False):
|
||
"""通用数据通道传输：发送命令 + 连数据端口 + 收发数据"""
|
||
print(f"\n [{cmd}] Opening data channel {pasv_ip}:{pasv_port}...")
|
||
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
data_sock.settimeout(10.0)
|
||
try:
|
||
data_sock.connect((pasv_ip, pasv_port))
|
||
print(f" Data channel connected.")
|
||
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(TIMEOUT)
|
||
sock.connect((host, PORT))
|
||
recv_line(sock) # 220
|
||
send_cmd(sock, "USER anonymous")
|
||
send_cmd(sock, "PASS test@test.com")
|
||
|
||
# PASV
|
||
resp = send_cmd(sock, "PASV")
|
||
_, data_port = parse_pasv(resp)
|
||
print(f" PASV port: {data_port}")
|
||
|
||
# Connect data FIRST
|
||
data_sock2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
data_sock2.settimeout(10.0)
|
||
data_sock2.connect((pasv_ip, data_port))
|
||
|
||
# Send command
|
||
time.sleep(0.2)
|
||
send_cmd(sock, cmd)
|
||
|
||
result = None
|
||
if data_to_send:
|
||
print(f" Sending {len(data_to_send)} bytes to data channel...")
|
||
total = 0
|
||
while total < len(data_to_send):
|
||
n = data_sock2.send(data_to_send[total:])
|
||
if n <= 0:
|
||
break
|
||
total += n
|
||
print(f" Sent {total} bytes.")
|
||
data_sock2.shutdown(socket.SHUT_WR)
|
||
|
||
if read_data:
|
||
result = b""
|
||
while True:
|
||
try:
|
||
chunk = data_sock2.recv(4096)
|
||
if not chunk:
|
||
break
|
||
result += chunk
|
||
except socket.timeout:
|
||
break
|
||
print(f" Received {len(result)} bytes from data channel.")
|
||
|
||
data_sock2.close()
|
||
resp = recv_line(sock)
|
||
if resp:
|
||
print(f" Control response: {resp.decode().rstrip()}")
|
||
|
||
send_cmd(sock, "QUIT")
|
||
sock.close()
|
||
data_sock.close()
|
||
return result
|
||
except Exception as e:
|
||
print(f" Data transfer error: {e}")
|
||
try:
|
||
data_sock.close()
|
||
except:
|
||
pass
|
||
return None
|
||
|
||
def test_upload_download(host, port):
|
||
print(f"=== FTP Upload/Download Test: {host}:{port} ===\n")
|
||
checks = []
|
||
|
||
def check(label, ok, detail=""):
|
||
checks.append((label, ok, detail))
|
||
mark = "PASS" if ok else "FAIL"
|
||
print(f" [{mark}] {label}{(' - ' + detail) if detail else ''}")
|
||
|
||
def finish(sock):
|
||
passed = all(ok for _, ok, _ in checks)
|
||
print("\n" + "=" * 40)
|
||
print("FTP Upload/Download Summary:")
|
||
for label, ok, _ in checks:
|
||
print(f" [{'PASS' if ok else 'FAIL'}] {label}")
|
||
print("=" * 40)
|
||
if passed:
|
||
print(">>> TEST PASSED <<<")
|
||
else:
|
||
print(">>> TEST FAILED <<<")
|
||
print("=== Test Complete ===")
|
||
if sock:
|
||
try:
|
||
sock.close()
|
||
except Exception:
|
||
pass
|
||
return passed
|
||
|
||
# 1. Connect + Login
|
||
print("[1] Connect + Login")
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(TIMEOUT)
|
||
try:
|
||
sock.connect((host, port))
|
||
except Exception as e:
|
||
print(f" CONNECT FAILED: {e}")
|
||
check("Connect", False, str(e))
|
||
return finish(sock)
|
||
recv_line(sock) # 220
|
||
|
||
if not send_cmd(sock, "USER anonymous", 230):
|
||
check("Login USER 230", False)
|
||
return finish(sock)
|
||
if not send_cmd(sock, "PASS test@test.com", 230):
|
||
check("Login PASS 230", False)
|
||
return finish(sock)
|
||
check("Login (USER/PASS 230)", True)
|
||
|
||
# 2. TYPE I (binary mode)
|
||
print("\n[2] Set binary mode")
|
||
resp = send_cmd(sock, "TYPE I")
|
||
type_ok = bool(resp) and resp.startswith(b"200")
|
||
check("TYPE I 200", type_ok)
|
||
|
||
# 3. PASV + STOR (upload)
|
||
print("\n[3] Upload: STOR up_test.dat")
|
||
resp = send_cmd(sock, "PASV")
|
||
if not resp:
|
||
check("PASV", False, "no response")
|
||
return finish(sock)
|
||
pasv_ip, pasv_port = parse_pasv(resp)
|
||
print(f" PASV: {pasv_ip}:{pasv_port}")
|
||
|
||
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
data_sock.settimeout(10.0)
|
||
try:
|
||
data_sock.connect((pasv_ip, pasv_port))
|
||
print(f" Data connected.")
|
||
except Exception as e:
|
||
print(f" Data connect FAILED: {e}")
|
||
check("PASV data connect", False, str(e))
|
||
return finish(sock)
|
||
check("PASV data connect", True, f"{pasv_ip}:{pasv_port}")
|
||
|
||
send_cmd(sock, "STOR up_test.dat")
|
||
total = 0
|
||
while total < len(TEST_DATA):
|
||
try:
|
||
n = data_sock.send(TEST_DATA[total:])
|
||
if n <= 0:
|
||
break
|
||
total += n
|
||
except Exception as e:
|
||
print(f" Send error: {e}")
|
||
break
|
||
print(f" Uploaded {total}/{len(TEST_DATA)} bytes.")
|
||
data_sock.shutdown(socket.SHUT_WR)
|
||
data_sock.close()
|
||
|
||
resp = recv_line(sock)
|
||
stor_ok = bool(resp) and resp.startswith(b"226") and total == len(TEST_DATA)
|
||
check("STOR 226 + bytes", stor_ok,
|
||
f"uploaded={total}/{len(TEST_DATA)} resp={resp.decode().rstrip() if resp else 'none'}")
|
||
|
||
# 4. PASV + LIST (verify)
|
||
print("\n[4] Verify: LIST")
|
||
resp = send_cmd(sock, "PASV")
|
||
if not resp:
|
||
check("LIST PASV", False, "no response")
|
||
return finish(sock)
|
||
pasv_ip, pasv_port = parse_pasv(resp)
|
||
|
||
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
data_sock.settimeout(10.0)
|
||
data_sock.connect((pasv_ip, pasv_port))
|
||
|
||
send_cmd(sock, "LIST")
|
||
data = b""
|
||
while True:
|
||
try:
|
||
chunk = data_sock.recv(4096)
|
||
if not chunk:
|
||
break
|
||
data += chunk
|
||
except socket.timeout:
|
||
break
|
||
print(f" LIST: {data.decode(errors='replace').rstrip() or '(empty directory)'}")
|
||
data_sock.close()
|
||
|
||
resp = recv_line(sock)
|
||
list_ok = bool(data) and b"UP_TEST.DAT" in data.upper()
|
||
check("LIST shows UP_TEST.DAT", list_ok,
|
||
resp.decode().rstrip() if resp else "no resp")
|
||
|
||
# 5. PASV + RETR (download)
|
||
time.sleep(1.0)
|
||
print("\n[5] Download: RETR up_test.dat")
|
||
resp = send_cmd(sock, "PASV")
|
||
if not resp:
|
||
check("RETR PASV", False, "no response")
|
||
return finish(sock)
|
||
pasv_ip, pasv_port = parse_pasv(resp)
|
||
|
||
data_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
data_sock.settimeout(10.0)
|
||
try:
|
||
data_sock.connect((pasv_ip, pasv_port))
|
||
except Exception as e:
|
||
print(f" Data connect FAILED: {e}")
|
||
check("RETR data connect", False, str(e))
|
||
return finish(sock)
|
||
check("RETR data connect", True, f"{pasv_ip}:{pasv_port}")
|
||
|
||
send_cmd(sock, "RETR up_test.dat")
|
||
dl_data = b""
|
||
while True:
|
||
try:
|
||
chunk = data_sock.recv(4096)
|
||
if not chunk:
|
||
break
|
||
dl_data += chunk
|
||
except socket.timeout:
|
||
break
|
||
data_sock.close()
|
||
resp = recv_line(sock)
|
||
retr_ok = bool(resp) and resp.startswith(b"226")
|
||
check("RETR 226", retr_ok, resp.decode().rstrip() if resp else "no resp")
|
||
|
||
# Compare
|
||
print(f"\n[6] Compare")
|
||
print(f" Uploaded: {len(TEST_DATA)} bytes")
|
||
print(f" Downloaded: {len(dl_data)} bytes")
|
||
match = (dl_data == TEST_DATA)
|
||
if match:
|
||
print(f" *** MATCH! Upload/Download success. ***")
|
||
else:
|
||
print(f" *** MISMATCH! ***")
|
||
if dl_data:
|
||
print(f" First 80 bytes uploaded: {TEST_DATA[:80]!r}")
|
||
print(f" First 80 bytes downloaded: {dl_data[:80]!r}")
|
||
check("Upload/Download MATCH", match, f"up={len(TEST_DATA)} dl={len(dl_data)}")
|
||
|
||
# 7. DELE (delete test file)
|
||
print("\n[7] Cleanup: DELE up_test.dat")
|
||
resp = send_cmd(sock, "DELE up_test.dat")
|
||
dele_ok = bool(resp) and resp.startswith(b"250")
|
||
check("DELE 250", dele_ok, resp.decode().rstrip() if resp else "no resp")
|
||
|
||
# 8. QUIT
|
||
print("\n[8] QUIT")
|
||
resp = send_cmd(sock, "QUIT")
|
||
quit_ok = bool(resp) and resp.startswith(b"221")
|
||
check("QUIT 221", quit_ok, resp.decode().rstrip() if resp else "no resp")
|
||
|
||
return finish(sock)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="FTP Upload/Download Test")
|
||
parser.add_argument("--host", default=HOST)
|
||
parser.add_argument("--port", type=int, default=PORT)
|
||
args = parser.parse_args()
|
||
success = test_upload_download(args.host, args.port)
|
||
sys.exit(0 if success else 1)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|