文章总结： 本文详细介绍了如何利用Socket编程实现SOCKS5和HTTP代理服务器，将本地电脑连接到内部网络。文章首先解释了Socket的基本概念和工作原理，然后提供了完整的SOCKS5代理服务器Python代码，包括握手、请求解析、目标连接和数据转发等功能。同时，文章还提供了简化版HTTP代理服务器的实现，以及如何在Windows和Linux系统上配置使用这些代理的方法。通过这些代理服务器，用户可以将本地流量转发到内部网络，实现完整的网络接入。
综合评分： 85
文章分类： 内网渗透,网络编程,安全工具,网络安全,红队
利用socker的将本地电脑挂到某网络中
原创
Titans
骏之安科技知识库
2025年11月20日 07:00
北京
一、Socket 是什么？
Socket（套接字）是一个抽象层，它封装了底层复杂的网络协议（如 TCP/IP），为应用程序提供了一套简单的、类似文件读写的接口，用于网络通信。
可以把它想象成一个“网络插座”：
- • 应用程序就像电器
- • 网络就像电网
- • Socket 就是中间的插座，电器插上插座就能用电，无需关心电是怎么来的
二、Socket 的核心工作原理：基于 TCP/IP 模型
Socket 是应用层与传输层之间的桥梁。它的工作流程与 TCP/IP 模型紧密相关。
+-------------------+ +-------------------+
| 应用程序 A | | 应用程序 B |
| (例如：浏览器) | | (例如：Web服务器) |
+-------------------+ +-------------------+
| |
| (通过Socket API读写数据) |
| |
+-------------------+ +-------------------+
| Socket API | | Socket API | <-- 抽象接口
+-------------------+ +-------------------+
| |
+-------------------------------------------------+
| TCP/IP 协议栈 | <-- 操作系统内核
| (传输层、网络层、链路层、物理层) |
+-------------------------------------------------+
| |
物理网络 (网线、Wi-Fi、路由器等)
三. 完整的SOCKS5代理服务器
import socket
import threading
import struct
import select
from typing import Tuple
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class SOCKS5Proxy:
def __init__(self, listen_host='0.0.0.0', listen_port=1080,
upstream_proxy=None, bypass_domains=None):
self.listen_host = listen_host
self.listen_port = listen_port
self.upstream_proxy = upstream_proxy # (host, port) 上游代理
self.bypass_domains = bypass_domains or []
def handle_handshake(self, client_sock: socket.socket) -> bool:
"""处理SOCKS5握手"""
try:
# 读取版本和方法数量
data = client_sock.recv(2)
if len(data) < 2:
return False
version, n_methods = data
if version != 5: # 只支持SOCKS5
return False
# 读取方法列表
methods = client_sock.recv(n_methods)
if len(methods) < n_methods:
return False
# 响应无需认证
client_sock.sendall(b'\x05\x00')
return True
except Exception as e:
logger.error(f"握手失败: {e}")
return False
def parse_request(self, client_sock: socket.socket) -> Tuple[str, int, int]:
"""解析客户端请求"""
try:
data = client_sock.recv(4)
if len(data) < 4:
return None, None, None
version, cmd, rsv, atyp = data
# 只支持CONNECT命令
if cmd != 1:
client_sock.sendall(b'\x05\x07\x00\x01' + b'\x00' * 6)
return None, None, None
# 解析目标地址
if atyp == 1: # IPv4
ip_data = client_sock.recv(4)
if len(ip_data) < 4:
return None, None, None
target_addr = socket.inet_ntoa(ip_data)
elif atyp == 3: # 域名
domain_len = client_sock.recv(1)[0]
domain = client_sock.recv(domain_len).decode('utf-8')
target_addr = domain
elif atyp == 4: # IPv6 (不支持)
client_sock.sendall(b'\x05\x08\x00\x01' + b'\x00' * 6)
return None, None, None
else:
return None, None, None
# 解析端口
port_data = client_sock.recv(2)
if len(port_data) < 2:
return None, None, None
target_port = struct.unpack('>H', port_data)[0]
return target_addr, target_port, atyp
except Exception as e:
logger.error(f"解析请求失败: {e}")
return None, None, None
def connect_to_target(self, target_addr: str, target_port: int) -> socket.socket:
"""连接到目标服务器"""
try:
# 检查是否需要绕过代理
if self.should_bypass(target_addr):
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.settimeout(10)
remote_sock.connect((target_addr, target_port))
return remote_sock
# 如果有上游代理，通过上游代理连接
if self.upstream_proxy:
return self.connect_via_upstream(target_addr, target_port)
else:
# 直接连接目标（这里可以修改为连接到内网跳板机）
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.settimeout(10)
# 如果是内网地址，通过跳板机连接
if self.is_internal_address(target_addr):
remote_sock.connect(('jumpserver.internal.com', 8080)) # 跳板机
# 发送实际目标地址给跳板机
self.send_target_to_jump(remote_sock, target_addr, target_port)
else:
remote_sock.connect((target_addr, target_port))
return remote_sock
except Exception as e:
logger.error(f"连接目标失败 {target_addr}:{target_port}: {e}")
raise
def should_bypass(self, target_addr: str) -> bool:
"""检查是否应该绕过代理"""
for domain in self.bypass_domains:
if domain in target_addr:
return True
return False
def is_internal_address(self, addr: str) -> bool:
"""检查是否为内网地址"""
internal_prefixes = ['192.168.', '10.', '172.16.', '172.17.', '172.18.',
'172.19.', '172.20.', '172.21.', '172.22.', '172.23.',
'172.24.', '172.25.', '172.26.', '172.27.', '172.28.',
'172.29.', '172.30.', '172.31.', 'localhost']
return any(addr.startswith(prefix) for prefix in internal_prefixes)
def send_target_to_jump(self, sock: socket.socket, target_addr: str, target_port: int):
"""发送实际目标地址给跳板机（自定义协议）"""
target_info = f"{target_addr}:{target_port}".encode('utf-8')
sock.sendall(struct.pack('>I', len(target_info)) + target_info)
def connect_via_upstream(self, target_addr: str, target_port: int) -> socket.socket:
"""通过上游代理连接"""
proxy_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
proxy_sock.connect(self.upstream_proxy)
# 发送CONNECT请求到上游代理
connect_cmd = f"CONNECT {target_addr}:{target_port} HTTP/1.1\r\n\r\n"
proxy_sock.sendall(connect_cmd.encode())
# 读取响应
response = proxy_sock.recv(4096)
if b'200' not in response:
raise Exception(f"上游代理连接失败: {response.decode()}")
return proxy_sock
def send_success_response(self, client_sock: socket.socket, atyp: int):
"""发送成功响应"""
try:
# 使用客户端连接的本地地址和端口
bind_addr = client_sock.getsockname()
response = b'\x05\x00\x00' + bytes([atyp])
if atyp == 1: # IPv4
response += socket.inet_aton(bind_addr[0])
elif atyp == 3: # 域名
response += bytes([len(bind_addr[0])]) + bind_addr[0].encode()
elif atyp == 4: # IPv6
response += socket.inet_pton(socket.AF_INET6, bind_addr[0])
response += struct.pack('>H', bind_addr[1])
client_sock.sendall(response)
except Exception as e:
logger.error(f"发送响应失败: {e}")
def forward_data(self, source: socket.socket, destination: socket.socket,
direction: str, target_info: str):
"""转发数据"""
try:
while True:
# 使用select监控socket
readable, _, exceptional = select.select([source], [], [source], 60)
if exceptional:
break
for sock in readable:
data = sock.recv(4096)
if not data:
return
destination.sendall(data)
logger.debug(f"{direction} {target_info}: {len(data)} bytes")
except Exception as e:
logger.debug(f"数据转发结束: {e}")
finally:
try:
source.close()
except:
pass
try:
destination.close()
except:
pass
def handle_client(self, client_sock: socket.socket, client_addr: Tuple[str, int]):
"""处理客户端连接"""
logger.info(f"新连接来自: {client_addr}")
try:
# SOCKS5握手
if not self.handle_handshake(client_sock):
return
# 解析请求
target_addr, target_port, atyp = self.parse_request(client_sock)
if not target_addr:
return
target_info = f"{target_addr}:{target_port}"
logger.info(f"连接目标: {target_info}")
# 连接到目标
remote_sock = self.connect_to_target(target_addr, target_port)
# 发送成功响应
self.send_success_response(client_sock, atyp)
# 开始双向数据转发
client_to_remote = threading.Thread(
target=self.forward_data,
args=(client_sock, remote_sock, "→", target_info)
)
remote_to_client = threading.Thread(
target=self.forward_data,
args=(remote_sock, client_sock, "←", target_info)
)
client_to_remote.daemon = True
remote_to_client.daemon = True
client_to_remote.start()
remote_to_client.start()
# 等待线程结束
client_to_remote.join()
remote_to_client.join()
except Exception as e:
logger.error(f"处理客户端 {client_addr} 失败: {e}")
finally:
try:
client_sock.close()
except:
pass
logger.info(f"连接关闭: {client_addr} -> {target_info}")
def start(self):
"""启动代理服务器"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((self.listen_host, self.listen_port))
server.listen(100) # 较大的backlog
logger.info(f"SOCKS5代理服务器启动在 {self.listen_host}:{self.listen_port}")
if self.upstream_proxy:
logger.info(f"使用上游代理: {self.upstream_proxy}")
try:
while True:
client_sock, client_addr = server.accept()
client_thread = threading.Thread(
target=self.handle_client,
args=(client_sock, client_addr)
)
client_thread.daemon = True
client_thread.start()
except KeyboardInterrupt:
logger.info("收到停止信号，关闭服务器")
except Exception as e:
logger.error(f"服务器错误: {e}")
finally:
server.close()
# 使用示例
if __name__ == "__main__":
# 基础代理服务器
proxy = SOCKS5Proxy(
listen_host='0.0.0.0',
listen_port=1080,
bypass_domains=['localhost', '127.0.0.1']
)
# 或者使用上游代理
# proxy = SOCKS5Proxy(
# listen_port=1080,
# upstream_proxy=('internal.proxy.com', 3128) # 公司内部代理
# )
proxy.start()
四. HTTP代理服务器（简化版）
import socket
import threading
import re
class HTTPProxy:
def __init__(self, listen_port=8080, internal_gateway=None):
self.listen_port = listen_port
self.internal_gateway = internal_gateway # 内网网关地址
def parse_http_request(self, data: str):
"""解析HTTP请求"""
lines = data.split('\r\n')
if not lines:
return None
# 解析请求行
match = re.match(r'(\w+)\s+(\S+)\s+HTTP/\d\.\d', lines[0])
if not match:
return None
method, url = match.groups()
# 提取主机和路径
if url.startswith('http://'):
url = url[7:]
host_end = url.find('/')
if host_end == -1:
host = url
path = '/'
else:
host = url[:host_end]
path = url[host_end:]
# 从Host头获取主机信息
for line in lines[1:]:
if line.lower().startswith('host:'):
host = line[5:].strip()
break
return method, host, path, lines
def handle_http_connection(self, client_sock: socket.socket):
"""处理HTTP连接"""
try:
data = client_sock.recv(4096).decode('utf-8')
if not data:
return
parsed = self.parse_http_request(data)
if not parsed:
return
method, host, path, headers = parsed
# 如果是内网地址，通过网关转发
if self.is_internal_host(host) and self.internal_gateway:
target_host, target_port = self.internal_gateway
else:
# 直接连接
if ':' in host:
target_host, target_port_str = host.split(':')
target_port = int(target_port_str)
else:
target_host = host
target_port = 80
# 连接到目标服务器
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.connect((target_host, target_port))
# 重建请求
if self.internal_gateway and self.is_internal_host(host):
# 通过网关时需要修改Host头
new_headers = []
for header in headers:
if header.lower().startswith('host:'):
new_headers.append(f"Host: {host}")
else:
new_headers.append(header)
request_data = '\r\n'.join(new_headers) + '\r\n\r\n'
else:
request_data = data
remote_sock.sendall(request_data.encode())
# 转发响应
while True:
response_data = remote_sock.recv(4096)
if not response_data:
break
client_sock.sendall(response_data)
except Exception as e:
print(f"HTTP代理错误: {e}")
finally:
client_sock.close()
def is_internal_host(self, host: str) -> bool:
"""检查是否为内网主机"""
internal_patterns = ['.internal.', '.local.', '.lan.', '192.168.', '10.']
return any(pattern in host for pattern in internal_patterns)
def start(self):
"""启动HTTP代理服务器"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', self.listen_port))
server.listen(50)
print(f"HTTP代理服务器启动在端口 {self.listen_port}")
while True:
client_sock, addr = server.accept()
thread = threading.Thread(
target=self.handle_http_connection,
args=(client_sock,)
)
thread.daemon = True
thread.start()
# 启动HTTP代理
# http_proxy = HTTPProxy(
# listen_port=8080,
# internal_gateway=('gateway.internal.com', 8080)
# )
# http_proxy.start()
五. 使用说明
启动SOCKS5代理：
python socks5_proxy.py
配置系统使用代理：
Windows:
# 设置系统代理
netsh winhttp set proxy 127.0.0.1:1080
# 清除代理
netsh winhttp reset proxy
Linux/Mac:
# 临时设置环境变量
export http_proxy=socks5://127.0.0.1:1080
export https_proxy=socks5://127.0.0.1:1080
export all_proxy=socks5://127.0.0.1:1080
# 测试连接
curl --socks5 127.0.0.1:1080 http://internal-server.company.com
浏览器配置：
- • 手动配置代理：SOCKS5 127.0.0.1:1080
- • 或者使用SwitchyOmega等插件
高级功能：
- 1. 流量监控：在forward_data方法中添加流量统计
- 2. 访问控制：添加IP白名单/黑名单
- 3. 协议过滤：基于目标端口过滤流量
- 4. 负载均衡：多个内网网关的负载均衡
这个SOCKS5代理服务器可以将您的所有网络流量通过Socket连接转发到内部网络，实现完整的网络接入。
–END–
觉得不错，可以关注，点赞，转发，如果需要技术援助，可以联系我们，期待您的莅临
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：骏之安科技知识库 Titans《利用socker的将本地电脑挂到某网络中》