import ftplib
import argparse
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# Lock untuk thread-safe file writing
file_lock = threading.Lock()
def check_ftp_connection(ip, username, password, output_file, timeout=5, port=21):
"""
Cek koneksi FTP dengan timeout yang dapat dikonfigurasi
"""
try:
ftp = ftplib.FTP()
ftp.connect(ip, port, timeout=timeout) # Connect dengan timeout
ftp.login(username, password) # Login
# Ambil info tambahan jika berhasil
try:
welcome = ftp.getwelcome() # Welcome message
pwd = ftp.pwd() # Current directory
except:
welcome = "N/A"
pwd = "N/A"
result = f"[SUCCESS] {username}@{ip}:{port} | Pass: {password} | Dir: {pwd} | {welcome}\n"
print(result.strip())
# Thread-safe file writing
with file_lock:
output_file.write(result)
output_file.flush()
ftp.quit()
return True
except ftplib.error_perm as e:
# Error 530 = Login incorrect, 550 = Permission denied
result = f"[FAILED] {username}@{ip}:{port} - {str(e)}\n"
print(result.strip())
return False
except ftplib.error_temp as e:
# Error temporary (4xx)
result = f"[TEMP ERROR] {username}@{ip}:{port} - {str(e)}\n"
print(result.strip())
return False
except ConnectionRefusedError:
result = f"[REFUSED] {ip}:{port} - Koneksi ditolak\n"
print(result.strip())
return False
except TimeoutError:
result = f"[TIMEOUT] {ip}:{port} - Koneksi timeout\n"
print(result.strip())
return False
except OSError as e:
result = f"[NETWORK ERROR] {ip}:{port} - {str(e)}\n"
print(result.strip())
return False
except Exception as e:
result = f"[ERROR] {username}@{ip}:{port} - {str(e)}\n"
print(result.strip())
return False
def load_credentials(filename):
"""
Load dan validasi kredensial dari file
Format: ip|username|password atau ip:port|username|password
"""
credentials = []
invalid_count = 0
try:
with open(filename, 'r') as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line or line.startswith('#'): # Skip kosong & komentar
continue
parts = line.split('|')
if len(parts) != 3:
print(f"[SKIP] Line {line_num}: Format tidak valid -> {line}")
invalid_count += 1
continue
# Support format ip:port|user|pass
ip_part, username, password = parts
if ':' in ip_part:
ip, port = ip_part.split(':', 1)
try:
port = int(port)
except ValueError:
print(f"[SKIP] Line {line_num}: Port tidak valid -> {ip_part}")
invalid_count += 1
continue
else:
ip = ip_part
port = 21 # Default FTP port
# Validasi tidak kosong
if ip and username and password:
credentials.append((
ip.strip(),
username.strip(),
password.strip(),
port
))
else:
print(f"[SKIP] Line {line_num}: Ada field yang kosong")
invalid_count += 1
except FileNotFoundError:
print(f"[ERROR] File '{filename}' tidak ditemukan!")
return []
print(f"[INFO] Total kredensial valid : {len(credentials)}")
if invalid_count:
print(f"[INFO] Total kredensial invalid: {invalid_count}")
return credentials
def main():
parser = argparse.ArgumentParser(
description="Fast FTP Connection Checker",
formatter_class=argparse.RawTextHelpFormatter,
epilog="""
Format file input (live.txt):
ip|username|password
ip:port|username|password <- custom port
Contoh:
192.168.1.1|admin|admin123
192.168.1.2:2121|user|pass123
"""
)
parser.add_argument(
'-i', '--input',
default='live.txt',
help='Input file (default: live.txt)'
)
parser.add_argument(
'-o', '--output',
default='ftplives.txt',
help='Output file (default: ftplives.txt)'
)
parser.add_argument(
'-t', '--threads',
type=int,
default=50,
help='Jumlah thread concurrent (default: 50)'
)
parser.add_argument(
'--timeout',
type=int,
default=5,
help='Timeout koneksi dalam detik (default: 5)'
)
parser.add_argument(
'--port',
type=int,
default=21,
help='Default FTP port (default: 21)'
)
args = parser.parse_args()
# Load kredensial
credentials = load_credentials(args.input)
if not credentials:
print("[ERROR] Tidak ada kredensial yang valid!")
return
total = len(credentials)
success_count = 0
failed_count = 0
counter_lock = threading.Lock()
print(f"\n[INFO] Memulai pengecekan {total} target dengan {args.threads} threads...")
print(f"[INFO] Timeout : {args.timeout} detik")
print(f"[INFO] Output : {args.output}\n")
with open(args.output, 'a') as output_file:
with ThreadPoolExecutor(max_workers=args.threads) as executor:
# Submit semua task
futures = {
executor.submit(
check_ftp_connection,
ip, username, password,
output_file,
args.timeout,
port # Gunakan port dari file atau default
): (ip, username, port)
for ip, username, password, port in credentials
}
# Proses hasil
for i, future in enumerate(as_completed(futures), 1):
ip, username, port = futures[future]
try:
result = future.result()
with counter_lock:
if result:
success_count += 1
else:
failed_count += 1
except Exception as e:
print(f"[ERROR] Unexpected: {username}@{ip}:{port} -> {e}")
with counter_lock:
failed_count += 1
# Progress bar
percent = (i / total) * 100
bar = '█' * int(percent // 5) + '░' * (20 - int(percent // 5))
print(
f"[{bar}] {percent:5.1f}% | "
f"{i}/{total} | "
f"✓ {success_count} | "
f"✗ {failed_count}",
end='\r'
)
# Summary
print(f"\n\n{'='*55}")
print(f" [SELESAI] Total : {total}")
print(f" [SELESAI] Berhasil : {success_count} ✓")
print(f" [SELESAI] Gagal : {failed_count} ✗")
print(f" [SELESAI] Output : {args.output}")
print(f"{'='*55}")
if __name__ == "__main__":
main()