Created
October 2, 2023 13:40
-
-
Save joshfinley/a4bb127e115f8a4f26c9812bc60da75e to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import paramiko | |
import socket | |
def http_connect_proxy(proxy_host, proxy_port, target_host, target_port): | |
""" | |
Establish a socket connection through an HTTP proxy. | |
""" | |
proxy = (proxy_host, proxy_port) | |
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
s.connect(proxy) | |
# Send HTTP CONNECT request to the proxy | |
connect_request = f"CONNECT {target_host}:{target_port} HTTP/1.1\r\nHost: {target_host}:{target_port}\r\n\r\n" | |
s.sendall(connect_request.encode()) | |
# Receive the response from the proxy | |
buffer = b"" | |
while not buffer.endswith(b"\r\n\r\n"): | |
buffer += s.recv(4096) | |
# Check the response | |
response = buffer.decode().split("\r\n")[0] | |
if not response.startswith("HTTP/1.1 200 "): | |
raise Exception(f"Failed to connect through proxy: {response}") | |
return s | |
def main(): | |
# Proxy settings | |
proxy_host = "proxy_server_address" | |
proxy_port = 8080 | |
# Target SSH server settings | |
target_host = "ssh_server_address" | |
target_port = 22 | |
username = "your_username" | |
password = "your_password" | |
# Establish connection through HTTP proxy | |
try: | |
proxy_sock = http_connect_proxy(proxy_host, proxy_port, target_host, target_port) | |
except Exception as e: | |
print(f"Failed to establish proxy connection: {e}") | |
return | |
# Establish SSH connection | |
ssh_client = paramiko.SSHClient() | |
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) | |
try: | |
ssh_client.connect(target_host, port=target_port, username=username, password=password, sock=proxy_sock) | |
except Exception as e: | |
print(f"Failed to establish SSH connection: {e}") | |
return | |
# Executing a command to test the SSH connection | |
stdin, stdout, stderr = ssh_client.exec_command("ls") | |
print(stdout.read().decode()) | |
ssh_client.close() | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment