ZengQingming/JoyD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

taskkill /F /IM nginx.exe

Browse Source
This commit is contained in:
zqm
2026-03-20 16:51:31 +08:00
parent 6e0b5aa056
commit ed54296f62
11 changed files with 1190 additions and 766 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Claw/Server/gateway/Cargo.toml
View File

@@ -17,7 +17,12 @@ env_logger = "^0.10"
sha1 = "^0.10"
sha2 = "^0.10"
hex = "^0.4"
urlencoding = "^2.1"
base64 = "^0.21"
aes = "^0.8"
cbc = "^0.1"
cipher = "^0.4"
actix = "^0.13"
actix-web-actors = "^4.0"
actix-ws = "^0.2"
futures = "^0.3"
uuid = { version = "^1.0", features = ["v4"] }

528
Claw/Server/gateway/src/communication.rs
View File

@@ -1,18 +1,9 @@
use actix::prelude::*;
use actix_web_actors::ws;
use std::time::{Duration, Instant};
use std::sync::Arc;
use tokio::sync::{RwLock, mpsc};
use tokio::sync::RwLock;
use serde_json::json;
use uuid::Uuid;
use actix::ResponseFuture;
/// 测试消息发送的消息类型
#[derive(Message)]
#[rtype(result = "Result<serde_json::Value, String>")]
struct TestSendMessage;
/// WebSocket连接状态
/// 连接信息
#[derive(Debug, Clone)]
pub struct ConnectionInfo {
@@ -23,500 +14,10 @@ pub struct ConnectionInfo {
}
impl ConnectionInfo {
/// 创建新的连接信息
pub fn new(id: String) -> Self {
let now = Instant::now();
Self {
id,
connected_at: now,
last_heartbeat: now,
client_info: None,
}
}
/// 设置客户端信息
pub fn set_client_info(&mut self, info: String) {
self.client_info = Some(info);
}
/// 获取客户端信息
pub fn get_client_info(&self) -> &Option<String> {
&self.client_info
}
/// 更新心跳时间
pub fn update_heartbeat(&mut self) {
self.last_heartbeat = Instant::now();
}
}
/// WebSocket连接
#[allow(dead_code)]
pub struct WebSocketConnection {
/// 连接ID
id: String,
/// 连接信息
info: ConnectionInfo,
/// 连接管理器
manager: Arc<RwLock<ConnectionManager>>,
/// 心跳间隔
heartbeat_interval: Duration,
/// 客户端超时时间
client_timeout: Duration,
/// 心跳定时器
hb: Instant,
/// 响应通道
response_sender: Option<mpsc::Sender<serde_json::Value>>,
/// 响应接收器
response_receiver: Option<mpsc::Receiver<serde_json::Value>>,
}
impl WebSocketConnection {
pub fn new(manager: Arc<RwLock<ConnectionManager>>) -> Self {
let id = Uuid::new_v4().to_string();
let info = ConnectionInfo::new(id.clone());
let (tx, rx) = mpsc::channel(100);
println!("🔌 创建新的WebSocket连接: id={}, connected_at={:?}", info.id, info.connected_at);
Self {
id,
info,
manager,
heartbeat_interval: Duration::from_secs(30),
client_timeout: Duration::from_secs(60),
hb: Instant::now(),
response_sender: Some(tx),
response_receiver: Some(rx),
}
}
/// 获取连接信息（用于调试和监控）
pub fn get_info(&self) -> &ConnectionInfo {
println!("ℹ️ 获取连接信息: id={}, 连接时长: {:?}", self.info.id, self.info.connected_at.elapsed());
&self.info
}
/// 获取响应发送器（用于测试和调试）
pub fn get_response_sender(&self) -> &Option<mpsc::Sender<serde_json::Value>> {
println!("📤 获取响应发送器: {:?}", self.response_sender.is_some());
&self.response_sender
}
/// 获取响应接收器（用于测试和调试）
pub fn get_response_receiver(&self) -> &Option<mpsc::Receiver<serde_json::Value>> {
println!("📥 获取响应接收器: {:?}", self.response_receiver.is_some());
&self.response_receiver
}
/// 发送消息
///
/// 这个方法用于向WebSocket连接发送消息。
/// 在实际使用中，当WebSocket连接建立后，可以通过这个方法发送各种类型的消息。
///
/// # 示例
/// ```
/// let message = json!({
/// "type": "notification",
/// "data": "Hello World"
/// });
/// connection.send(message).await?;
/// ```
#[allow(dead_code)]
pub async fn send(&self, message: serde_json::Value) -> Result<(), String> {
if let Some(_sender) = &self.response_sender {
// 这里应该通过WebSocket连接发送消息
println!("📤 通过WebSocket发送消息到连接 {}: {}", self.id, message);
Ok(())
} else {
Err("发送器不可用".to_string())
}
}
/// 发送消息并等待响应
///
/// 这个方法用于向WebSocket连接发送请求消息并等待响应。
/// 适用于需要确认响应的场景，如RPC调用。
///
/// # 参数
/// * `message` - 要发送的消息
/// * `timeout_ms` - 超时时间（毫秒）
///
/// # 示例
/// ```
/// let request = json!({
/// "type": "get_status",
/// "request_id": "123"
/// });
/// let response = connection.send_and_wait(request, 5000).await?;
/// ```
#[allow(dead_code)]
pub async fn send_and_wait(&self, message: serde_json::Value, timeout_ms: u64) -> Result<serde_json::Value, String> {
let (_response_tx, mut _response_rx) = tokio::sync::mpsc::channel::<serde_json::Value>(1);
let request_id = Uuid::new_v4().to_string();
let _msg = json!({
"type": "request",
"id": request_id,
"data": message
});
println!("📤 发送WebSocket请求到连接 {}，请求ID: {}，超时: {}ms", self.id, request_id, timeout_ms);
// 首先发送消息
match self.send(message.clone()).await {
Ok(_) => {
println!("✅ 消息发送成功，等待响应...");
},
Err(e) => {
println!("❌ 消息发送失败: {}", e);
return Err(format!("发送失败: {}", e));
}
}
// 这里需要实现具体的发送逻辑
// 暂时返回模拟响应
Ok(json!({
"success": true,
"message": "WebSocket消息已发送",
"data": message,
"request_id": request_id,
"timeout": timeout_ms
}))
}
/// 内部测试方法 - 测试消息发送功能
///
/// 这个方法用于测试WebSocketConnection的消息发送功能。
/// 它会依次调用send和send_and_wait方法来验证功能是否正常。
///
/// # 返回值
/// 返回测试结果，包含send和send_and_wait的测试状态
#[allow(dead_code)]
pub async fn test_send_functionality(&self) -> Result<serde_json::Value, String> {
println!("🧪 测试WebSocket连接的消息发送功能");
let test_message = json!({
"type": "test",
"connection_id": self.id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "这是内部测试消息"
});
// 测试发送消息
match self.send(test_message.clone()).await {
Ok(_) => {
println!("✅ 测试消息发送成功");
// 测试发送并等待响应
match self.send_and_wait(test_message.clone(), 5000).await {
Ok(response) => {
println!("✅ 测试发送并等待响应成功");
Ok(json!({
"test_send": "success",
"test_send_and_wait": "success",
"response": response,
"connection_id": self.id
}))
},
Err(e) => {
println!("⚠️ 测试发送并等待响应失败: {}", e);
Ok(json!({
"test_send": "success",
"test_send_and_wait": "failed",
"error": e.to_string(),
"connection_id": self.id
}))
}
}
},
Err(e) => {
println!("❌ 测试消息发送失败: {}", e);
Err(format!("测试发送失败: {}", e))
}
}
}
/// 心跳处理
fn heartbeat(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
ctx.run_interval(self.heartbeat_interval, |act, ctx| {
// 检查客户端超时
if Instant::now().duration_since(act.hb) > act.client_timeout {
println!("❌ WebSocket连接超时: {} (最后心跳: {:?}前)", act.id, act.info.last_heartbeat.elapsed());
ctx.stop();
return;
}
// 更新心跳时间
act.info.update_heartbeat();
// 发送心跳消息
let heartbeat_msg = json!({
"type": "heartbeat",
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": act.id,
"client_info": act.info.get_client_info()
});
ctx.text(serde_json::to_string(&heartbeat_msg).unwrap());
});
}
}
impl Actor for WebSocketConnection {
type Context = ws::WebsocketContext<Self>;
fn started(&mut self, ctx: &mut Self::Context) {
println!("✅ WebSocket连接已建立: {}", self.id);
// 注册连接到管理器
let manager = self.manager.clone();
let connection_id = self.id.clone();
let mut connection_info = self.info.clone();
// 设置客户端信息
connection_info.set_client_info("SmartClaw-Service".to_string());
actix::spawn(async move {
let mut manager = manager.write().await;
manager.add_connection(connection_id, connection_info);
});
// 启动心跳机制
self.heartbeat(ctx);
// 发送欢迎消息
let welcome_msg = json!({
"type": "welcome",
"connection_id": self.id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "连接到Claw网关服务",
"client_info": "SmartClaw-Service"
});
ctx.text(serde_json::to_string(&welcome_msg).unwrap());
// 在后台测试消息发送功能
let connection_clone = ctx.address();
actix::spawn(async move {
// 延迟2秒后测试发送功能
tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
if let Ok(result) = connection_clone.send(TestSendMessage).await {
match result {
Ok(test_result) => {
println!("✅ WebSocket连接测试完成: {:?}", test_result);
},
Err(e) => {
println!("⚠️ WebSocket连接测试失败: {}", e);
}
}
}
});
}
fn stopped(&mut self, _ctx: &mut Self::Context) {
println!("🔌 WebSocket连接已断开: {} (连接时长: {:?})", self.id, self.info.connected_at.elapsed());
// 从管理器中移除连接
let manager = self.manager.clone();
let connection_id = self.id.clone();
actix::spawn(async move {
let mut manager = manager.write().await;
manager.remove_connection(&connection_id);
});
}
}
/// WebSocket消息处理
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketConnection {
fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
match msg {
Ok(ws::Message::Ping(msg)) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
Ok(ws::Message::Pong(_)) => {
self.hb = Instant::now();
}
Ok(ws::Message::Text(text)) => {
self.hb = Instant::now();
self.info.update_heartbeat(); // 更新最后心跳时间
// 解析消息
match serde_json::from_str::<serde_json::Value>(&text) {
Ok(json_msg) => {
self.handle_message(json_msg, ctx);
}
Err(e) => {
println!("❌ 解析WebSocket消息失败: {}", e);
let error_msg = json!({
"type": "error",
"message": "消息格式错误",
"error": e.to_string()
});
ctx.text(serde_json::to_string(&error_msg).unwrap());
}
}
}
Ok(ws::Message::Binary(bin)) => {
// 处理二进制消息（如果需要）
println!("📦 收到二进制消息: {} bytes", bin.len());
}
Ok(ws::Message::Close(reason)) => {
println!("🔌 WebSocket连接关闭: {:?} (连接时长: {:?})", reason, self.info.connected_at.elapsed());
ctx.stop();
}
Err(e) => {
println!("❌ WebSocket协议错误: {}", e);
ctx.stop();
}
_ => {}
}
}
}
impl WebSocketConnection {
/// 处理接收到的消息
fn handle_message(&mut self, msg: serde_json::Value, ctx: &mut ws::WebsocketContext<Self>) {
let msg_type = msg.get("type").and_then(|v| v.as_str()).unwrap_or("unknown");
match msg_type {
"heartbeat" => {
// 心跳响应
let response = json!({
"type": "heartbeat_response",
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
"task_response" => {
// 任务响应
println!("✅ 收到任务响应: {:?}", msg);
// 这里可以处理任务响应，更新状态等
// 测试使用send方法发送确认消息
let ack_message = json!({
"type": "task_ack",
"original_response": msg,
"timestamp": chrono::Utc::now().timestamp(),
"connection_id": self.id
});
// 由于send是异步方法，我们在这里模拟使用
println!("📤 模拟使用send方法发送确认: {}", ack_message);
// 测试使用send_and_wait方法（模拟）
let test_request = json!({
"type": "test_request",
"data": "测试数据",
"connection_id": self.id
});
println!("⏱️ 模拟使用send_and_wait方法发送请求: {}，超时: 3000ms", test_request);
}
"status_update" => {
// 状态更新
println!("📊 收到状态更新: {:?}", msg);
}
"test_send_direct" => {
// 直接测试send方法
println!("🧪 收到直接测试send方法的消息");
// 模拟使用send方法
let test_message = json!({
"type": "test_send_response",
"original_message": msg,
"connection_id": self.id,
"test_result": "send方法测试成功"
});
println!("📤 模拟send方法调用: {}", test_message);
// 发送响应
let response = json!({
"type": "test_send_completed",
"test_message": test_message,
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
"test_send_and_wait_direct" => {
// 直接测试send_and_wait方法
println!("🧪 收到直接测试send_and_wait方法的消息");
// 模拟使用send_and_wait方法
let test_request = json!({
"type": "test_send_and_wait_request",
"original_message": msg,
"connection_id": self.id,
"timeout": 5000
});
println!("⏱️ 模拟send_and_wait方法调用: {}", test_request);
// 模拟响应
let test_response = json!({
"success": true,
"message": "send_and_wait方法测试成功",
"data": "这是模拟的响应数据",
"connection_id": self.id
});
println!("✅ 模拟send_and_wait方法响应: {}", test_response);
// 发送响应
let response = json!({
"type": "test_send_and_wait_completed",
"request": test_request,
"response": test_response,
"connection_id": self.id
});
ctx.text(serde_json::to_string(&response).unwrap());
}
_ => {
println!("📨 收到未知类型消息: {}", msg_type);
let response = json!({
"type": "unknown_message_type",
"original_type": msg_type,
"message": "收到未知消息类型"
});
ctx.text(serde_json::to_string(&response).unwrap());
}
}
}
}
/// 处理测试消息发送的消息
impl Handler<TestSendMessage> for WebSocketConnection {
type Result = ResponseFuture<Result<serde_json::Value, String>>;
fn handle(&mut self, _msg: TestSendMessage, _ctx: &mut Self::Context) -> Self::Result {
let connection_id = self.id.clone();
Box::pin(async move {
println!("🧪 在Handler中测试WebSocket连接的消息发送功能");
let test_message = json!({
"type": "test",
"connection_id": connection_id,
"timestamp": chrono::Utc::now().timestamp(),
"message": "这是Handler中的测试消息"
});
// 模拟测试结果
Ok(json!({
"test_send": "simulated_success",
"test_send_and_wait": "simulated_success",
"connection_id": connection_id,
"test_message": test_message,
"note": "这是在Handler中模拟的测试结果"
}))
})
}
}
/// 连接管理器
@@ -531,18 +32,6 @@ impl ConnectionManager {
}
}
/// 添加连接
pub fn add_connection(&mut self, id: String, info: ConnectionInfo) {
self.connections.insert(id.clone(), info);
println!("📥 连接已注册: {} (总数: {})", id, self.connections.len());
}
/// 移除连接
pub fn remove_connection(&mut self, id: &str) {
self.connections.remove(id);
println!("📤 连接已移除: {} (剩余: {})", id, self.connections.len());
}
/// 获取连接信息
pub fn get_connection(&self, id: &str) -> Option<&ConnectionInfo> {
let conn = self.connections.get(id);
@@ -728,7 +217,7 @@ impl CommunicationConfig {
/// WebSocket客户端（用于SmartClaw服务连接网关）
pub struct WebSocketClient {
config: CommunicationConfig,
connection: Option<WebSocketConnection>,
connected: bool,
}
impl WebSocketClient {
@@ -736,7 +225,7 @@ impl WebSocketClient {
println!("🚀 创建WebSocket客户端，配置URL: {}", config.websocket_url);
Self {
config,
connection: None,
connected: false,
}
}
@@ -752,15 +241,15 @@ impl WebSocketClient {
// 这里需要实现具体的WebSocket连接逻辑
// 暂时返回模拟连接成功
println!("✅ WebSocket连接成功 (模拟)");
self.connection = Some(WebSocketConnection::new(Arc::new(RwLock::new(ConnectionManager::new()))));
self.connected = true;
Ok(())
}
/// 断开连接
pub async fn disconnect(&mut self) -> Result<(), String> {
if self.connection.is_some() {
if self.connected {
println!("🔌 断开WebSocket连接: {}", self.config.websocket_url);
self.connection = None;
self.connected = false;
} else {
println!("⚠️ 没有活动的WebSocket连接需要断开");
}
@@ -798,8 +287,7 @@ impl WebSocketClient {
/// 检查连接状态
pub fn is_connected(&self) -> bool {
let connected = self.connection.is_some();
println!("🔗 WebSocket客户端连接状态: {}", if connected { "已连接" } else { "未连接" });
connected
println!("🔗 WebSocket客户端连接状态: {}", if self.connected { "已连接" } else { "未连接" });
self.connected
}
}

738
Claw/Server/gateway/src/main.rs
View File

@@ -1,15 +1,401 @@
use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer, Responder, middleware::Logger};
use actix_web_actors::ws;
use serde::Deserialize;
use actix_web::http::Method;
use actix_ws::{Message};
use serde::{Deserialize, Serialize};
use std::env;
use std::sync::Arc;
use tokio::sync::RwLock;
use base64;
use shared::{TaskRequest, TaskResponse, HealthResponse, utils};
use sha1::{Sha1, Digest};
use futures::StreamExt;
mod communication;
use communication::{WebSocketConnection, ConnectionManager, WebSocketPool, CommunicationConfig, WebSocketClient};
use communication::{ConnectionManager, WebSocketPool, CommunicationConfig, WebSocketClient};
/// 企业微信消息发送结构体
#[derive(Serialize)]
struct WeChatMessage {
touser: String,
msgtype: String,
agentid: i32,
text: WeChatTextContent,
}
#[derive(Serialize)]
struct WeChatTextContent {
content: String,
}
#[derive(Deserialize)]
struct WeChatAccessTokenResponse {
access_token: String,
// expires_in: i32, // 未使用，暂时注释
}
#[derive(Deserialize)]
struct WeChatSendMessageResponse {
errcode: i32,
errmsg: String,
}
/// 获取企业微信访问令牌
async fn get_wechat_access_token() -> Result<String, Box<dyn std::error::Error>> {
let (_token, corp_id, _encoding_aes_key, corp_secret, _debug, _debug_config, _http) = get_wechat_config();
let url = format!("https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={}&corpsecret={}", corp_id, corp_secret);
let client = reqwest::Client::new();
let response = client.get(&url).send().await?;
let result: WeChatAccessTokenResponse = response.json().await?;
Ok(result.access_token)
}
/// 发送企业微信消息
async fn send_wechat_message(touser: &str, content: &str, debug: bool) -> Result<(), Box<dyn std::error::Error>> {
if debug {
println!("📤 开始发送企业微信消息");
}
// 获取访问令牌
let access_token = get_wechat_access_token().await?;
if debug {
println!(" 获取到访问令牌: {}", access_token);
}
// 构建消息
let message = WeChatMessage {
touser: touser.to_string(),
msgtype: "text".to_string(),
agentid: 1000002, // 企业微信应用ID
text: WeChatTextContent {
content: content.to_string(),
},
};
// 发送消息
let url = format!("https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={}", access_token);
let client = reqwest::Client::new();
let response = client.post(&url)
.json(&message)
.send()
.await?;
let result: WeChatSendMessageResponse = response.json().await?;
if result.errcode == 0 {
if debug {
println!("✅ 消息发送成功");
}
Ok(())
} else {
if debug {
println!("❌ 消息发送失败: {} - {}", result.errcode, result.errmsg);
}
Err(format!("消息发送失败: {} - {}", result.errcode, result.errmsg).into())
}
}
#[derive(Deserialize, Serialize, Debug)]
struct WeChatConfig {
wechat: WeChatSettings,
debug: DebugSettings,
}
#[derive(Deserialize, Serialize, Debug)]
struct WeChatSettings {
token: String,
corp_id: String,
encoding_aes_key: String,
corp_secret: String,
}
#[derive(Deserialize, Serialize, Debug)]
struct DebugSettings {
wechat: bool,
config: bool,
http: bool,
}
impl Default for WeChatConfig {
fn default() -> Self {
Self {
wechat: WeChatSettings {
token: "mytoken123456".to_string(),
corp_id: "wwa7bb7aec981103b4".to_string(),
encoding_aes_key: "PXP7FjoinIPc9WscGymDlf1VwMyBLh1cKJJSJFx2SO8".to_string(),
corp_secret: "your_corp_secret_here".to_string(),
},
debug: DebugSettings {
wechat: false,
config: false,
http: true,
},
}
}
}
/// 企业微信配置（从config.json文件读取，便于配置管理）
fn get_wechat_config() -> (String, String, String, String, bool, bool, bool) {
use std::fs::File;
use std::io::{Read, Write};
use std::path::Path;
// 配置文件路径
let config_path = Path::new("./config.json");
// 尝试读取配置文件
if config_path.exists() {
// 先创建默认配置，用于获取debug.config的值
let default_config = WeChatConfig::default();
let debug_config = default_config.debug.config;
if debug_config {
println!("📁 读取配置文件: {:?}", config_path);
}
match File::open(config_path) {
Ok(mut file) => {
let mut contents = String::new();
if file.read_to_string(&mut contents).is_ok() {
match serde_json::from_str::<WeChatConfig>(&contents) {
Ok(config) => {
if config.debug.config {
println!("✅ 配置文件读取成功");
}
return (config.wechat.token, config.wechat.corp_id, config.wechat.encoding_aes_key, config.wechat.corp_secret, config.debug.wechat, config.debug.config, config.debug.http);
}
Err(e) => {
if debug_config {
println!("❌ 配置文件解析失败: {}, 使用默认配置", e);
}
}
}
} else {
if debug_config {
println!("❌ 配置文件读取失败，使用默认配置");
}
}
}
Err(e) => {
let default_config = WeChatConfig::default();
if default_config.debug.config {
println!("❌ 打开配置文件失败: {}, 使用默认配置", e);
}
}
}
} else {
let default_config = WeChatConfig::default();
if default_config.debug.config {
println!("📁 配置文件不存在，生成默认配置");
}
// 生成默认配置文件
let default_config = WeChatConfig::default();
if let Ok(mut file) = File::create(config_path) {
if let Ok(contents) = serde_json::to_string_pretty(&default_config) {
if file.write_all(contents.as_bytes()).is_ok() {
if default_config.debug.config {
println!("✅ 默认配置文件生成成功: {:?}", config_path);
}
} else {
if default_config.debug.config {
println!("❌ 默认配置文件生成失败");
}
}
}
}
}
// 使用默认值
let default_config = WeChatConfig::default();
(default_config.wechat.token, default_config.wechat.corp_id, default_config.wechat.encoding_aes_key, default_config.wechat.corp_secret, default_config.debug.wechat, default_config.debug.config, default_config.debug.http)
}
/// 解析企业微信XML消息
fn parse_wechat_xml_message(xml_content: &str, debug: bool) -> (Option<String>, Option<String>, Option<String>, Option<String>) {
if debug {
println!("📄 开始解析企业微信XML消息");
}
// 检查是否为加密消息
if xml_content.contains("<Encrypt>") {
if debug {
println!("🔒 发现加密消息，开始解密");
}
// 提取Encrypt标签内容
if let Some(encrypt_content) = extract_xml_tag(xml_content, "Encrypt") {
if debug {
println!(" 提取到加密内容");
}
// 解密消息
match decrypt_wechat_message(&encrypt_content) {
Ok(decrypted_xml) => {
if debug {
println!("✅ 消息解密成功");
println!(" 解密后内容: {}", decrypted_xml);
}
// 从解密后的XML中提取发送者、内容、消息类型和事件类型
let from_user_name = extract_xml_tag(&decrypted_xml, "FromUserName");
let content = extract_xml_tag(&decrypted_xml, "Content");
let msg_type = extract_xml_tag(&decrypted_xml, "MsgType");
let event = extract_xml_tag(&decrypted_xml, "Event");
if debug {
if let Some(from_user) = &from_user_name {
println!(" 发送者: {}", from_user);
}
if let Some(msg_content) = &content {
println!(" 消息内容: {}", msg_content);
}
if let Some(msg_type_val) = &msg_type {
println!(" 消息类型: {}", msg_type_val);
}
if let Some(event_val) = &event {
println!(" 事件类型: {}", event_val);
}
}
return (from_user_name, content, msg_type, event);
}
Err(e) => {
if debug {
println!("❌ 消息解密失败: {}", e);
}
return (None, None, None, None);
}
}
} else {
if debug {
println!("❌ 无法提取Encrypt标签");
}
return (None, None, None, None);
}
} else {
// 非加密消息，直接解析
if debug {
println!("🔓 非加密消息，直接解析");
}
let from_user_name = extract_xml_tag(xml_content, "FromUserName");
let content = extract_xml_tag(xml_content, "Content");
let msg_type = extract_xml_tag(xml_content, "MsgType");
let event = extract_xml_tag(xml_content, "Event");
if debug {
if let Some(from_user) = &from_user_name {
println!(" 发送者: {}", from_user);
}
if let Some(msg_content) = &content {
println!(" 消息内容: {}", msg_content);
}
if let Some(msg_type_val) = &msg_type {
println!(" 消息类型: {}", msg_type_val);
}
if let Some(event_val) = &event {
println!(" 事件类型: {}", event_val);
}
}
return (from_user_name, content, msg_type, event);
}
}
/// 解密企业微信消息
fn decrypt_wechat_message(encrypted: &str) -> Result<String, Box<dyn std::error::Error>> {
// 获取企业微信配置
let (_, _corp_id, encoding_aes_key, _corp_secret, _debug, _debug_config, _http) = get_wechat_config();
// 企业微信官方要求：EncodingAESKey 使用URL_SAFE Base64解码，且需要补全=号
let mut aes_key_str = encoding_aes_key.to_string();
while aes_key_str.len() % 4 != 0 {
aes_key_str.push('=');
}
let key = base64::Engine::decode(&base64::engine::general_purpose::URL_SAFE, &aes_key_str)?;
if key.len() != 32 {
return Err("Invalid key length".into());
}
// 企业微信官方规定：IV 必须是 EncodingAESKey 解码后的前 16 字节
let iv = &key[0..16];
let mut iv_array = [0u8; 16];
iv_array.copy_from_slice(iv);
// 企业微信官方要求：加密消息内容（Encrypt）使用STANDARD Base64解码
let ciphertext = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, &encrypted)?;
// 使用AES-CBC-256解密
use aes::cipher::{KeyIvInit, BlockDecryptMut};
use aes::Aes256;
use cbc::Decryptor;
use cipher::block_padding::NoPadding;
let decryptor = Decryptor::<Aes256>::new_from_slices(&key, &iv_array)?;
let mut buffer = ciphertext.to_vec();
let plaintext = decryptor.decrypt_padded_mut::<NoPadding>(&mut buffer)
.map_err(|e| format!("Decryption error: {:?}", e))?;
// 企业微信官方格式：16字节随机串 + 4字节长度(网络序) + 消息内容 + CorpID
if plaintext.len() < 20 {
return Err("Decrypted data too short".into());
}
// 解析4字节长度（网络序，大端）
let msg_len = u32::from_be_bytes([plaintext[16], plaintext[17], plaintext[18], plaintext[19]]);
let msg_start = 20;
let msg_end = msg_start + msg_len as usize;
if msg_end > plaintext.len() {
return Err("Invalid message length".into());
}
// 截取消息内容
let msg = &plaintext[msg_start..msg_end];
let result = String::from_utf8_lossy(msg).to_string();
Ok(result)
}
/// 提取XML标签内容
fn extract_xml_tag(xml: &str, tag: &str) -> Option<String> {
// 尝试匹配带CDATA的格式
let start_tag_cdata = format!("<{}><![CDATA[", tag);
let cdata_end = "]]>";
let end_tag = format!("</{}", tag);
if let Some(start_idx) = xml.find(&start_tag_cdata) {
let start_idx = start_idx + start_tag_cdata.len();
// 先找到CDATA的结束
if let Some(cdata_end_idx) = xml[start_idx..].find(cdata_end) {
let content = &xml[start_idx..start_idx + cdata_end_idx];
return Some(content.to_string());
}
}
// 尝试匹配不带CDATA的格式
let start_tag = format!("<{}", tag);
if let Some(start_idx) = xml.find(&start_tag) {
// 找到标签的结束位置
if let Some(tag_end_idx) = xml[start_idx..].find(">").map(|i| start_idx + i + 1) {
if let Some(end_idx) = xml[tag_end_idx..].find(&end_tag) {
let content = &xml[tag_end_idx..tag_end_idx + end_idx];
// 去除前后空白
let trimmed_content = content.trim();
if !trimmed_content.is_empty() {
return Some(trimmed_content.to_string());
}
}
}
}
None
}
/// 任务处理服务
struct TaskService {
@@ -204,20 +590,31 @@ impl TaskService {
}
/// 验证企业微信签名
fn validate_wechat_signature(signature: &str, timestamp: &str, nonce: &str, token: &str) -> bool {
println!("🔐 验证企业微信签名:");
println!(" signature: {}", signature);
println!(" timestamp: {}", timestamp);
println!(" nonce: {}", nonce);
println!(" token: {}", token);
fn validate_wechat_signature(msg_signature: &str, timestamp: &str, nonce: &str, data: &str, debug: bool) -> bool {
if debug {
println!("🔐 验证企业微信签名:");
println!(" msg_signature: {}", msg_signature);
println!(" timestamp: {}", timestamp);
println!(" nonce: {}", nonce);
println!(" data: {}", data);
}
// 企业微信签名验证算法
// 1. 将token、timestamp、nonce三个参数进行字典序排序
let mut params = vec![token, timestamp, nonce];
// 获取企业微信配置
let (token, _corp_id, _encoding_aes_key, _corp_secret, _, _, _) = get_wechat_config();
if debug {
println!(" token: {}", token);
}
// 企业微信签名验证算法（严格按照官方要求）
// 1. 将token、timestamp、nonce、data四个参数进行字典序排序
let mut params = vec![token.as_str(), timestamp, nonce, data];
params.sort();
// 2. 将三个参数字符串拼接成一个字符串
// 2. 将排序后的参数拼接成一个字符串
let combined = params.join("");
if debug {
println!(" 排序后拼接字符串: {}", combined);
}
// 3. 进行sha1加密
let mut hasher = Sha1::new();
@@ -225,10 +622,12 @@ impl TaskService {
let result = hasher.finalize();
let computed_signature = hex::encode(result);
// 4. 与signature对比
let is_valid = computed_signature == signature;
println!(" 计算签名: {}", computed_signature);
println!(" 验证结果: {}", if is_valid { "✅ 通过" } else { "❌ 失败" });
// 4. 与msg_signature对比
let is_valid = computed_signature == msg_signature;
if debug {
println!(" 计算签名: {}", computed_signature);
println!(" 验证结果: {}", if is_valid { "✅ 通过" } else { "❌ 失败" });
}
is_valid
}
@@ -263,10 +662,15 @@ impl TaskService {
/// WebSocket连接处理器
async fn websocket_handler(
req: HttpRequest,
stream: web::Payload,
app_data: web::Data<TaskService>,
body: web::Payload,
_app_data: web::Data<TaskService>,
) -> Result<HttpResponse, actix_web::Error> {
println!("🔗 收到WebSocket连接请求: {:?}", req);
println!("🔗 收到WebSocket连接请求: ");
println!(" HttpRequest {:?} {:?}:{}", req.method(), req.version(), req.path());
println!(" headers: ");
for (name, value) in req.headers() {
println!(" {:?}: {:?}", name, value);
}
// 验证连接来源（可以添加API密钥验证）
let api_key = req.headers().get("X-API-Key")
@@ -277,30 +681,77 @@ async fn websocket_handler(
if api_key != expected_key {
println!("❌ WebSocket连接认证失败");
return Ok(HttpResponse::Unauthorized().json(serde_json::json!({
"error": "Invalid API key",
"message": "WebSocket连接认证失败"
})));
return Err(actix_web::error::ErrorUnauthorized("Invalid API key"));
}
println!("✅ WebSocket连接认证通过");
// 创建WebSocket连接
let connection = WebSocketConnection::new(app_data.connection_manager.clone());
// 获取请求路径，区分连接类型
let path = req.path();
let is_control_connection = path == "/api/v1/ws/control";
// 获取连接信息（用于调试）
let connection_info = connection.get_info();
println!("ℹ️ WebSocket连接信息: id={}, 连接时间: {:?}", connection_info.id, connection_info.connected_at.elapsed());
if is_control_connection {
println!("🎯 检测到SmartClaw服务连接 (控制通道)");
} else {
println!("📱 检测到设备连接 (任务通道)");
}
// 获取响应通道信息（用于调试）
let sender_info = connection.get_response_sender();
let receiver_info = connection.get_response_receiver();
println!("📤 响应发送器: {:?}, 📥 响应接收器: {:?}", sender_info.is_some(), receiver_info.is_some());
println!("🔗 开始WebSocket握手...");
// 使用actix-ws处理WebSocket连接
let (response, mut session, msg_stream) = match actix_ws::handle(&req, body) {
Ok(result) => {
println!("✅ WebSocket握手成功");
result
},
Err(e) => {
println!("❌ WebSocket握手失败: {}", e);
return Err(e);
}
};
let resp = ws::start(connection, &req, stream)?;
println!("🔄 启动WebSocket消息处理循环...");
// 启动WebSocket消息处理循环
actix_web::rt::spawn(async move {
println!("✅ WebSocket消息处理循环已启动");
let mut msg_stream = msg_stream;
while let Some(msg) = msg_stream.next().await {
match msg {
Ok(Message::Text(text)) => {
println!("📨 收到消息: {}", text);
// 处理消息
}
Ok(Message::Binary(bin)) => {
println!("📨 收到二进制消息: {} bytes", bin.len());
}
Ok(Message::Ping(msg)) => {
println!("📨 收到Ping");
let _ = session.pong(&msg).await;
}
Ok(Message::Pong(_)) => {
println!("📨 收到Pong");
}
Ok(Message::Close(reason)) => {
println!("📨 收到关闭消息: {:?}", reason);
break;
}
Ok(Message::Continuation(_)) => {
// 处理 continuation 消息
}
Ok(Message::Nop) => {
// 处理 nop 消息
}
Err(e) => {
println!("❌ WebSocket错误: {}", e);
break;
}
}
}
println!("🔚 WebSocket连接已关闭");
});
println!("✅ WebSocket连接已建立");
Ok(resp)
Ok(response)
}
/// 健康检查处理器
@@ -386,16 +837,26 @@ async fn test_websocket_connection_send(app_data: web::Data<TaskService>) -> imp
/// 企业微信回调处理器
async fn handle_wechat_callback(req: HttpRequest, body: web::Bytes) -> impl Responder {
println!("📱 收到企业微信回调");
// 获取企业微信配置，包括debug配置
let (_, _, _, _, debug_wechat, _debug_config, _http) = get_wechat_config();
// 获取请求方法
let method = req.method();
// 获取查询参数
let query_string = req.query_string();
println!(" 查询参数: {}", query_string);
// 根据debug配置控制日志输出
if debug_wechat {
println!("📱 收到企业微信回调");
println!(" 请求方法: {}", method);
println!(" 查询参数: {}", query_string);
}
// 解析查询参数
#[derive(Deserialize)]
struct WeChatQuery {
signature: String,
msg_signature: String,
timestamp: String,
nonce: String,
echostr: Option<String>,
@@ -404,49 +865,150 @@ async fn handle_wechat_callback(req: HttpRequest, body: web::Bytes) -> impl Resp
let query: WeChatQuery = match web::Query::<WeChatQuery>::from_query(query_string) {
Ok(q) => q.into_inner(),
Err(e) => {
println!("❌ 解析查询参数失败: {}", e);
return HttpResponse::BadRequest().json(serde_json::json!({
"error": "Invalid query parameters",
"message": e.to_string()
}));
if debug_wechat {
println!("❌ 解析查询参数失败: {}", e);
}
return HttpResponse::BadRequest().body("error");
}
};
// 获取企业微信配置
let token = env::var("WECHAT_TOKEN").unwrap_or_else(|_| "your_token_here".to_string());
// 验证签名
let is_valid = TaskService::validate_wechat_signature(
&query.signature,
&query.timestamp,
&query.nonce,
&token
);
if !is_valid {
return HttpResponse::Unauthorized().json(serde_json::json!({
"error": "Invalid signature",
"message": "签名验证失败"
}));
// 核心判断：GET 和 POST 必须分开处理
if method == &Method::GET {
// 1. GET请求 = URL 验证
if debug_wechat {
println!("🔐 开始URL验证流程");
}
// 验证签名
let is_valid = TaskService::validate_wechat_signature(
&query.msg_signature,
&query.timestamp,
&query.nonce,
&query.echostr.as_ref().unwrap_or(&String::new()),
debug_wechat
);
if !is_valid {
if debug_wechat {
println!("❌ URL验证签名失败");
}
return HttpResponse::Unauthorized().body("invalid signature");
}
// 验证通过，返回echostr
if let Some(echostr) = query.echostr {
if debug_wechat {
println!("✅ URL验证成功，返回 echostr: {}", echostr);
}
return HttpResponse::Ok().body(echostr);
} else {
if debug_wechat {
println!("❌ URL验证失败：缺少echostr参数");
}
return HttpResponse::BadRequest().body("missing echostr");
}
} else if method == &Method::POST {
// 2. POST请求 = 消息推送
if debug_wechat {
println!("📥 开始消息推送处理流程");
// 处理实际的消息回调
let body_str = String::from_utf8_lossy(&body);
println!(" 消息内容: {}", body_str);
}
// 提取Encrypt字段内容
let body_str = String::from_utf8_lossy(&body);
let encrypt_content = extract_xml_tag(&body_str, "Encrypt").unwrap_or_default();
// 验证签名（POST请求需要包含Encrypt字段）
if debug_wechat {
println!("🔐 开始验证企业微信签名");
}
let is_valid = TaskService::validate_wechat_signature(
&query.msg_signature,
&query.timestamp,
&query.nonce,
&encrypt_content,
debug_wechat
);
if !is_valid {
if debug_wechat {
println!("❌ 消息推送签名验证失败");
}
return HttpResponse::Unauthorized().body("invalid signature");
}
if debug_wechat {
println!("✅ 签名验证通过，开始处理消息");
}
// 解析XML消息
let (from_user_name, content, msg_type, event) = parse_wechat_xml_message(&body_str, debug_wechat);
// 尝试发送回复消息
if let Some(user_id) = from_user_name {
// 判断是否需要回复：只回复文本消息和非位置上报事件
let should_reply = match msg_type.as_deref() {
Some("text") => true, // 文本消息需要回复
Some("event") => {
// 事件消息中，除了位置上报事件外都回复
event.as_deref() != Some("LOCATION")
}
_ => false, // 其他类型消息不回复
};
if should_reply {
if debug_wechat {
println!("📤 准备发送回复消息给用户: {}", user_id);
}
let reply_content = if let Some(msg_content) = content {
format!("我收到你的消息啦！你说: {}", msg_content)
} else {
"我收到你的消息啦！".to_string()
};
// 异步发送消息（不阻塞主线程）
let user_id_clone = user_id.clone();
let reply_content_clone = reply_content.clone();
let debug_wechat_clone = debug_wechat;
tokio::spawn(async move {
if let Err(e) = send_wechat_message(&user_id_clone, &reply_content_clone, debug_wechat_clone).await {
if debug_wechat_clone {
println!("❌ 发送回复消息失败: {}", e);
}
}
});
if debug_wechat {
println!("✅ 已开始发送回复消息");
}
} else {
if debug_wechat {
println!("⚠️ 消息类型不需要回复，跳过回复");
}
}
} else {
if debug_wechat {
println!("⚠️ 无法获取发送者信息，跳过回复");
}
}
// 企业微信要求返回纯文本 "success"
if debug_wechat {
println!("✅ 企业微信消息处理完成，返回 success");
}
HttpResponse::Ok().body("success")
} else {
// 其他请求方法
if debug_wechat {
println!("❌ 不支持的请求方法: {}", method);
}
HttpResponse::MethodNotAllowed().body("method not allowed")
}
// 如果是验证请求（首次配置时需要）
if let Some(echostr) = query.echostr {
println!("✅ 企业微信验证请求，返回 echostr: {}", echostr);
return HttpResponse::Ok().body(echostr);
}
// 处理实际的消息回调
let body_str = String::from_utf8_lossy(&body);
println!(" 消息内容: {}", body_str);
// TODO: 解析XML消息并处理
HttpResponse::Ok().json(serde_json::json!({
"status": "success",
"message": "企业微信回调已接收",
"timestamp": utils::current_timestamp()
}))
}
/// 微信小程序回调处理器
@@ -771,8 +1333,15 @@ async fn system_info(app_data: web::Data<TaskService>) -> impl Responder {
#[actix_web::main]
async fn main() -> std::io::Result<()> {
// 初始化日志
// 获取企业微信配置，包括http日志配置
let (_token, _corp_id, _encoding_aes_key, _corp_secret, _debug, _debug_config, http_log) = get_wechat_config();
// 根据配置决定Actix Web的日志级别
let actix_web_log_level = if http_log { "info" } else { "warn" };
let log_filter = format!("info,actix_web={}", actix_web_log_level);
env_logger::init_from_env(
env_logger::Env::new().default_filter_or("info,actix_web=info")
env_logger::Env::new().default_filter_or(&log_filter)
);
// 由于nginx代理，网关服务监听在8000端口
@@ -913,6 +1482,9 @@ async fn main() -> std::io::Result<()> {
App::new()
.app_data(task_service.clone())
.wrap(Logger::default())
// 企业微信回调 - 直接匹配企业微信配置路径 /wecom
.route("/wecom", web::post().to(handle_wechat_callback))
// 其他API路由 - 通过 /api/v1 前缀
.service(
web::scope("/api/v1")
// 健康检查
@@ -923,8 +1495,7 @@ async fn main() -> std::io::Result<()> {
.route("/task", web::post().to(handle_task))
.route("/task/{task_id}", web::get().to(get_task_status))
.route("/tasks", web::get().to(list_tasks))
// 微信集成
.route("/wechat/callback", web::post().to(handle_wechat_callback))
// 微信小程序集成
.route("/wechat/miniprogram/callback", web::post().to(handle_wechat_miniprogram_callback))
// WebSocket连接（内网服务器连接）
.route("/ws/control", web::get().to(websocket_handler))
@@ -944,12 +1515,15 @@ async fn main() -> std::io::Result<()> {
println!("✅ 网关服务已启动在 {} (通过nginx代理)", bind_address);
println!("🔍 可用接口:");
println!(" 🎯 企业微信回调 - 直接匹配企业微信配置");
println!(" POST /wecom - 企业微信回调（必须直接匹配企业微信配置）");
println!("");
println!(" 📋 API接口（通过 /api/v1 前缀）:");
println!(" GET /api/v1/health - 健康检查");
println!(" GET /api/v1/system - 系统信息");
println!(" POST /api/v1/task - 处理任务");
println!(" GET /api/v1/task/{{task_id}} - 查询任务状态");
println!(" GET /api/v1/task/<task_id> - 查询任务状态");
println!(" GET /api/v1/tasks - 查询任务列表");
println!(" POST /api/v1/wechat/callback - 企业微信回调");
println!(" POST /api/v1/wechat/miniprogram/callback - 微信小程序回调");
println!(" GET /api/v1/ws/control - WebSocket控制通道");
println!(" GET /api/v1/ws/task - WebSocket任务通道");