Python中如何检测ss心跳包或检测ss节点连通性?

最近用 django 写了个 sspanel,

项目在这: https://github.com/Ehco1996/django-sspanel

Demo: http://www.ehcozone.club

现在希望能够判断一下 节点的运行状态,并返回检测值,

现在网上好多资料都没有了,查了好久,貌似需要检测心跳包?

不太明白相关的知识

谁能给个思路么?


Python中如何检测ss心跳包或检测ss节点连通性?

21 回复

都写 sspanel 了这个搞不定不科学呀……怎么和 ssserver 通信的…


import socket
import struct
import time

def check_ss_heartbeat(host, port, timeout=5):
    """
    检测SS节点连通性(模拟心跳包检测)
    原理:建立TCP连接后发送协议头,验证响应
    """
    try:
        # 创建socket连接
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        
        # 尝试连接
        start_time = time.time()
        sock.connect((host, port))
        connect_time = time.time() - start_time
        
        # 构造简单的协议头(模拟SS协议)
        # 实际SS协议需要更完整的实现
        protocol_header = b'\x05\x01\x00'
        sock.send(protocol_header)
        
        # 接收响应(至少1字节)
        response = sock.recv(1)
        
        sock.close()
        
        if response:
            return {
                'status': 'online',
                'connect_time': round(connect_time * 1000, 2),  # 毫秒
                'response': response.hex()
            }
        else:
            return {'status': 'no_response', 'connect_time': connect_time}
            
    except socket.timeout:
        return {'status': 'timeout'}
    except ConnectionRefusedError:
        return {'status': 'refused'}
    except Exception as e:
        return {'status': f'error: {str(e)}'}

def continuous_heartbeat_check(host, port, interval=10, count=5):
    """持续心跳检测"""
    results = []
    for i in range(count):
        result = check_ss_heartbeat(host, port)
        result['check_time'] = time.strftime('%H:%M:%S')
        results.append(result)
        print(f"Check {i+1}: {result}")
        
        if i < count - 1:
            time.sleep(interval)
    
    # 统计成功率
    success_rate = sum(1 for r in results if r['status'] == 'online') / count * 100
    return {
        'host': f"{host}:{port}",
        'results': results,
        'success_rate': round(success_rate, 1)
    }

# 使用示例
if __name__ == "__main__":
    # 单次检测
    result = check_ss_heartbeat('example.com', 8388)
    print(f"单次检测结果: {result}")
    
    # 持续监测(示例)
    # stats = continuous_heartbeat_check('your_ss_host', 8388, interval=5, count=3)
    # print(f"\n监测统计: {stats}")

核心要点:

  1. TCP连接检测:通过socket建立连接测试基本连通性
  2. 协议验证:发送SS协议头验证服务响应(这里是最简示例)
  3. 实际SS检测需要完整的协议实现,包括认证、加密协商等
  4. 建议使用现成库shadowsocksaiohttp进行更完整的检测

一句话建议: 用socket测试连接+协议验证最直接,但完整检测建议用现成库。

如果是检测进程是否在运行:pgrep ssserver
如果是检测 ss 能不能用,在监控机上用 sslocal 连你的 ss,然后:proxychains4 curl ifconfig.co

两者都是不需要深度知识的方法

前端和 server 是通过数据库通讯的,关于 ss 是如何运行的我还真不太明白…



谢谢回复,
我用的还是 manyuser python 版,
运行是通过 run.sh 运行人,
用 ps 一下,检测不到进程…

我刚去试了 用 pgrep ssserver 也检测不到进程

请问怎么办?

感谢!

歪个楼,怎么检测流量?

不会,先给顶一个!


run.sh 里面,在运行之前它会先找到正在运行的 process,然后 kill,再重新启动。(我看到的这个版本是这样的)
你改改就变成检测是否运行的脚本了,大致是 pgrep python 这样,run.sh 里面做得更详细。

user 表里的 u+d 就是流量了


知道怎么做了, 脚本运行的话 进程是 server.py

用这条命令就行了就能检测到了
ps -ef | grep server.py| grep -v grep | wc -l

再次感谢

我是说你是怎么实现流量统计的

ps -ef | grep server.py| grep -v grep | wc -l 并不准确,我我有一个 server.py 的脚本启动也会统计进去

1. 找一下 ss server 启动时候没有存储 pid 的参数
2. 比较好的方法是利用 supervisor 启动,然后监控 supervisor 的进程
3. 如果不想用 supervisor 的话,你可自己写一个 daemon 启动然后 fork 子进程出来启动 ss 后台,然后记下 pid

或者用 lsof 判断 ss 有没有监听端口,有的话表明程序还在运行

通过数据库通信……莫非你后端用的不是官方 server ……

不需要用多用户版,直接用官方的 Python 实现就可以,可以参考: https://github.com/ZhuFaner/shadowsocks-manage-system



谢谢,一会我去试试看


我用的后端还是破娃的最后一次更新的版本,
接口用的是 sspanelv2

也没找到开发的文档,不知道怎么拓展,

你写得管理系统很棒哎! 不过不会 php,没找到你是怎么检测的

是直接跟 ssserver 进程进行 udp 数据交换,每次发数据都会收到返回,判断返回就知道是否存活了。

可以开个固定端口。然后监控那个端口在不在 netstat -ntpl |grep 443 |wc -l

你读一下 fsgmhoward 的 mu 的代码,这个版本会向服务器主动推送心跳包表达节点所在服务器状态的。

我有点懵逼啊,ss 不是预留了一个 manager 接口吗? Python 版和 libev 版都支持的,一般默认采用 unix socket 进行通讯,发送 ping 命令就能得到统计数据。这难道不就是个现成的非常好用的心跳检查吗?直接判断进程名,不怕同名(虽然不太可能)或者服务端卡死了吗?虽然检查进程名不是不可以,不过总感觉有点不专业啊。




那个 manager 接口很好用,但是如果想要用这个 api 必须再 manager mode 下才能进行操作,

而我的后端是运行在 mysql manyuser 下的, 接口用的 sspanelv2

看起来貌似也只能先用判断进程名的方法了。

再次感谢!

回到顶部