SMSS 安全通讯框架

bool Certification::Verify(const char *username, const char *password)
{
#ifdef DEBUG
    return true;
#endif
    struct spwd *sp;
    if ((sp = getspnam(username)) == NULL)
    {
        LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "系统未注册该用户或应用无法访问系统");
        return false;
    }
    char salt[512] = {0};
    int i, j;
    for (i = 0, j = 0; sp->sp_pwdp[i] && j != 3; i++)
    {
        if (sp->sp_pwdp[i] == '$')
        {
            j++;
        }
    }
    strncpy(salt, sp->sp_pwdp, i - 1);
    if (strcmp(sp->sp_pwdp, crypt(password, salt)) == 0)
    {
        return true;
    }
    return false;
}

bool WorkGroup::Init()
{
    // 直接初始化指定的工作线程
    for (int i = 0; i < num_; i++)
    {
        int id = group_.size() + 1;
        WorkThread *work = new WorkThread(this, id, net_bus_);
        if (!work->Init())
        {
            return false;
        }
        work->Start(); // thread start...
        group_.push_back(work);
        // 将当前初始化完成的工作线程注册进消息总线
        net_bus_->Regist(work); // regist thread to netbus
    }
    return true;
}

void WorkGroup::CreateConnection(int sock)
{
    int min = -1;
    WorkThread *work = nullptr;
    // 遍历寻找负载最轻的线程
    for (auto it = group_.begin(); it != group_.end(); it++)
    {
        if (min == -1)
        {
            min = (*it)->connect_num();
            work = (*it);
        }
        else if ((*it)->connect_num() < min)
        {
            min = (*it)->connect_num();
            work = (*it);
        }
    }
    // 添加一条socket fd进队列并通过管道激活
    work->AddSocket(sock);
    work->Notify("c");
}

void NetBus::Emit()
{
    for (auto it = work_thread_list_.begin(); it != work_thread_list_.end(); it++)
    {
        (*it)->Notify("e");
    }
}
void NetBus::Add(const char *header, int header_size, const char *data, int data_size)
{
    mtx_.lock();
    Metadata *meta = new Metadata(header, header_size, data, data_size);
    meta->set_ttl(work_thread_list_.size());
    metadata_list_.push_back(meta);
    if(metadata_list_.size() == 1)
    {
        this->Emit();
    }
    mtx_.unlock();
}
void NetBus::ReadFinish()
{
    mtx_.lock();
    Metadata *m = metadata_list_.front();
    m->Decrement();
    if (m->get_ttl() == 0)
    {
        delete m;
        metadata_list_.pop_front();
        if(metadata_list_.size() > 0)
        {
            this->Emit();
        }
    }
    
    mtx_.unlock();
}

if (msg_header.token() != this->token_)
{
        LOG4CPLUS_ERROR(SimpleLogger::Get()->LoggerRef(), "用户通信令牌(token)验证错误！");
        return;
}
LOG4CPLUS_DEBUG(SimpleLogger::Get()->LoggerRef(), "MsgType::MSG_SEND_REQ");
work_thread_->SendToNetBus(msg_header.SerializeAsString().c_str(), msg_header.ByteSizeLong(), msg_buff, msg_header.msg_size());

void SocketManager::ReadFromNetBus(Metadata *meta)
{
    // 解析消息头
    MsgHeader header;
    header.ParseFromArray(meta->GetDataHeader(), meta->GetDataHeaderSize());
    // 只接受发送给自己的消息或广播消息
    if (header.to() == this->user_id_ || (header.to() == 0
#ifndef DEBUG
                                          && header.from() != this->user_id_
#endif
                                          ))
    {
        header.set_to(this->user_id_);
        this->SendMessage(header.SerializeAsString().c_str(), header.ByteSizeLong(), meta->GetData(), meta->GetDataSize());
        // 如果是消息发送请求的类型则在向客户端推送完成后向发送端返回一条消息确认作为回执
        // 确认回执会被重新发送到总线上由其他连接负责读取
        if (header.msg_type() == MsgType::MSG_SEND_REQ)
        {
            // reply sender
            MsgHeader resp_header;
            resp_header.set_to(header.from());
            resp_header.set_from(0);
            resp_header.set_msg_type(MsgType::MSG_RECV_RESP);
            resp_header.set_msg_size(0);
            resp_header.set_msg_id(header.msg_id());
            work_thread_->SendToNetBus(resp_header.SerializeAsString().c_str(), resp_header.ByteSizeLong(), NULL, 0);
        }
    }
}

char buf[1024 * 10] = {0};
while (1)
{
    int len = bufferevent_read(bev_, buf, sizeof(buf));
    if (len > 0)
    {
        fwrite(buf, 1, len, pf_);
        remain_size_ += len;
    }
    if (len == 0)
    {
        break;
    }
}

char buf[1024] = {0};
while (1)
{
	int len = fread(buf, 1, sizeof(buf), pf_);
	if (len > 0)
	{
		bufferevent_write(bev_, buf, len);
		send_size += len;
	}
	if (len == 0)
	{
		is_stream_ = false;
		send_size = 0;
		fclose(pf_);
		pf_ = nullptr;
		break;
	}
}

unsigned int Utils::StringHashCode(const char *src)
{
    unsigned int seed = 1313;
    unsigned int hash = 0;
    while (*src)
    {
        hash = (*src++) + seed * hash;
    }
    return (hash & 0x7FFFFFFF);
}

// 读取服务器通讯公钥
fs.readFile("./data/.shadow/server.pem", (err, data) => {
	if (err) {
		reject(err);
	}
	let cipherBuffer = crypto.publicEncrypt(
		{
			key: data.toString(), // 秘钥
			padding: crypto.constants.RSA_PKCS1_PADDING // 填充方式
		},
		loginReq.serializeBinary()
	);
	// 构造消息头
	let msgHeader = new MsgHeader();
	msgHeader.setMsgSize(cipherBuffer.length);
	msgHeader.setMsgId(0);
	msgHeader.setMsgType(MsgType.CLIENT_LOGIN_REQ);
	msgHeader.setFrom(userID);
	msgHeader.setTo(0); // 发送给服务器
	const headerBuffer = msgHeader.serializeBinary();
	let packageHeader = Buffer.alloc(8);
	packageHeader.write("AB47");
	packageHeader.writeInt32LE(headerBuffer.length, 4);
	const packageBuffer = Buffer.concat([
		packageHeader,
		headerBuffer,
		cipherBuffer
	]);
	socket.write(...);

/**
 * 文件上传流程
 * 
 * @param {*} selfId 
 * @param {*} userToken 
 * @param {*} otherId 
 */
function UploadEvent(selfId, userToken, otherId) {
    dialog.showOpenDialog(require("electron").remote.getCurrentWindow(), {
        title: "选择上传文件",
        properties: ["openFile"]
    }).then(result => {
        return new Promise((resolve, reject) => {
            if (result.canceled) {
                reject(result);
            }
            resolve(result.filePaths[0]);
        });
    }).then(filePath => { // 上传文件的本地路径
        return new Promise((resolve, reject) => {
            (...)
        })
    }).then(obj => { // 文件属性
        return new Promise((resolve, reject) => {
            (...)
        });
    }).then(obj => { // 用户私钥
        return new Promise((resolve, reject) => {
            (...)
        });
    }).then(obj => {
        (...)
        return new Promise((resolve, reject) => {
            (...)
        });
    }).then(obj => { // 执行上传获取应答
		(...)
        return new Promise((resolve, reject) => {
            (...)
        });
    }).then(obj => {
        (...)
    });
}

const rs = fs.createReadStream(filePath);
rs.pipe(netSocket);
rs.on("end", () => {
	obj.netSocket.once("end", () => {
		console.log("上传连接关闭");
	})
});