admin/FACULTATE-LICENTA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

network chunk v11

Browse Source
This commit is contained in:
andrei-mihnea-cerbu
2024-11-13 13:00:05 +02:00
parent 4568bc12f4
commit 664b987269
10 changed files with 340 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
UC/src/network/socket_communicator/socket_communicator_base.ts
+129 -2
View File
@@ -1,24 +1,114 @@
import { ParsedMessage } from '../message_handler';
import { OperationHandler } from '../operations_base/operation_handler';
import ping from "ping";
import {
constants,
createCipheriv,
createDecipheriv,
generateKeyPairSync,
privateEncrypt,
publicDecrypt,
randomBytes
} from "crypto";
export abstract class SocketCommunicatorBase {
protected readonly ip: string;
protected readonly port: number;
protected readonly operationHandler: OperationHandler;
protected handlerResult: ParsedMessage | null;
protected networkSpeed: number | null = null; // Estimated network speed in bytes/ms
protected networkSpeed: number | null = null;
protected chunkBuffers: { [messageId: string]: string[] };
protected readonly EOP = '<EOP>';
protected privateKey: string | null;
protected publicKey: string | null;
protected aesKey: Buffer | null;
protected aesIv: Buffer | null;
protected constructor(ip: string, port: number, operationHandler: OperationHandler) {
this.ip = ip;
this.port = port;
this.operationHandler = operationHandler
this.handlerResult = null;
this.chunkBuffers = {};
this.privateKey = null;
this.publicKey = null;
this.aesKey = null;
this.aesIv = null;
}
// Getter for the handler result
getHandlerResult(): ParsedMessage | null {
return this.handlerResult;
const result = this.handlerResult;
this.handlerResult = null;
return result;
}
protected generateKeyPair(): void {
const { privateKey, publicKey } = generateKeyPairSync('rsa', {
modulusLength: 2048,
publicKeyEncoding: { type: 'spki', format: 'pem' },
privateKeyEncoding: { type: 'pkcs8', format: 'pem' },
});
this.privateKey = privateKey;
this.publicKey = publicKey;
}
protected generateAesKey(): void {
this.aesKey = randomBytes(32);
this.aesIv = randomBytes(16);
}
protected encryptWithAes(message: string): string {
if (!this.aesKey || !this.aesIv) {
throw new Error('AES key or IV is not set.');
}
const cipher = createCipheriv('aes-256-cbc', this.aesKey, this.aesIv);
let encrypted = cipher.update(message, 'utf-8');
encrypted = Buffer.concat([encrypted, cipher.final()]);
return encrypted.toString('base64');
}
protected decryptWithAes(encryptedMessage: string): string {
if (!this.aesKey || !this.aesIv) {
throw new Error('AES key or IV is not set.');
}
const decipher = createDecipheriv('aes-256-cbc', this.aesKey, this.aesIv);
let decrypted = decipher.update(Buffer.from(encryptedMessage, 'base64'));
decrypted = Buffer.concat([decrypted, decipher.final()]);
return decrypted.toString('utf-8');
}
protected decryptWithRsa(message: string): string {
if (!this.publicKey) {
throw new Error('Server public key not set.');
}
try {
const encryptedMessage = Buffer.from(message, 'base64');
const decrypted = publicDecrypt(
{
key: this.publicKey,
padding: constants.RSA_PKCS1_PADDING,
},
encryptedMessage
);
return decrypted.toString('utf-8');
} catch (error) {
throw new Error('Failed to decrypt RSA message.');
}
}
protected encryptWithRsa(message: string): string {
if (!this.privateKey) throw new Error('Server private key not set.');
return privateEncrypt(
{
key: this.privateKey,
padding: constants.RSA_PKCS1_PADDING,
},
Buffer.from(message)
).toString('base64');
}
protected async scanNetworkLatency(): Promise<number> {
@@ -54,4 +144,41 @@ export abstract class SocketCommunicatorBase {
return chunkSize || 1024; // Fallback to 1024 if alignment adjustment results in 0
}
async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingData = data.toString().trim();
// Split the incoming data by <EOP> to handle multiple chunks concatenated by TCP
const messages = incomingData.split(this.EOP).filter(Boolean); // Filter out any empty strings from split
for (const incomingMessage of messages) {
const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson);
// Initialize an array for chunks if it's the first chunk received for this messageId
if (!this.chunkBuffers[header.messageId]) {
this.chunkBuffers[header.messageId] = [];
}
// Directly set the chunk at the correct index, adjusting for 1-based indexing
this.chunkBuffers[header.messageId][header.sequenceNumber - 1] = chunkContent;
console.log(`Received chunk: ${incomingMessage}`);
console.log(`Chunks received so far: ${this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length}`);
// Check if all chunks have been received by confirming the length
if (this.chunkBuffers[header.messageId].filter(chunk => chunk !== undefined).length === header.totalChunks) {
const chunks = this.chunkBuffers[header.messageId];
const fullMessage = chunks.join('');
await this.handleIncomingMessage(fullMessage);
delete this.chunkBuffers[header.messageId];
}
}
}
abstract handleIncomingMessage(incomingMessage: string): Promise<void>;
abstract sendMessage(message: string): Promise<void>;
}
UC/src/network/socket_communicator/tcp_server_communicator.ts
+18 -112
View File
@@ -1,118 +1,48 @@
import { Socket } from 'net';
import { privateEncrypt, generateKeyPairSync, createCipheriv, createDecipheriv, randomBytes, constants } from 'crypto';
import { MessageHandler } from '../message_handler';
import { SocketCommunicatorBase } from './socket_communicator_base';
import { OperationHandler } from '../operations_base/operation_handler';
export class TcpServerCommunicator extends SocketCommunicatorBase {
private readonly socket: Socket;
private privateKey: string | null;
private publicKey: string | null;
private aesKey: Buffer | null;
private aesIv: Buffer | null;
private chunkBuffers: { [messageId: string]: string[] };
constructor(socket: Socket, ip: string, port: number, operationHandler: OperationHandler) {
super(ip, port, operationHandler);
this.socket = socket;
this.privateKey = null;
this.publicKey = null; // Client public key will be set later
this.aesKey = null;
this.aesIv = null;
this.chunkBuffers = {};
this.generateKeyPair();
}
generateKeyPair(): void {
const { privateKey, publicKey } = generateKeyPairSync('rsa', {
modulusLength: 2048,
publicKeyEncoding: { type: 'spki', format: 'pem' },
privateKeyEncoding: { type: 'pkcs8', format: 'pem' },
});
this.privateKey = privateKey;
this.publicKey = publicKey;
}
async sendPublicKey(): Promise<void> {
if (!this.publicKey) {
throw new Error('Public key is not available. Please generate RSA key pair.');
this.generateKeyPair();
if (!this.publicKey || !this.privateKey) {
throw new Error('RSA key pair is not available. Please generate RSA key pair.');
}
await this.sendChunkedMessage('SET_PUBLIC_KEY', { publicKey: this.publicKey });
await this.sendMessage('SET_PUBLIC_KEY', { publicKey: this.publicKey });
}
async sendAesKey(): Promise<void> {
this.aesKey = randomBytes(32);
this.aesIv = randomBytes(16);
this.generateAesKey();
if (!this.aesKey || !this.aesIv) {
throw new Error('AES key or IV is not available. Please generate AES key.');
}
const aesKeyBase64 = this.aesKey.toString('base64');
const aesIvBase64 = this.aesIv.toString('base64');
await this.sendChunkedMessage('SET_AES_KEY', { aesKey: aesKeyBase64, aesIv: aesIvBase64 });
await this.sendMessage('SET_AES_KEY', { aesKey: aesKeyBase64, aesIv: aesIvBase64 });
}
private encryptWithRsa(message: string): string {
if (!this.privateKey) throw new Error('Server private key not set.');
return privateEncrypt(
{
key: this.privateKey,
padding: constants.RSA_PKCS1_PADDING,
},
Buffer.from(message)
).toString('base64');
async handleIncomingMessage(incomingMessage: string): Promise<void> {
const messageToProcess = this.decryptWithAes(incomingMessage);
this.handlerResult = await this.operationHandler.handleOperation(messageToProcess);
}
private decryptWithAes(encryptedMessage: string): string {
if (!this.aesKey || !this.aesIv) {
throw new Error('AES key not set.');
}
const decipher = createDecipheriv('aes-256-cbc', this.aesKey, this.aesIv);
let decrypted = decipher.update(Buffer.from(encryptedMessage, 'base64'));
decrypted = Buffer.concat([decrypted, decipher.final()]);
return decrypted.toString('utf-8');
}
private encryptWithAes(message: string): string {
if (!this.aesKey || !this.aesIv) {
throw new Error('AES key or IV is not set.');
}
const cipher = createCipheriv('aes-256-cbc', this.aesKey, this.aesIv);
let encrypted = cipher.update(message, 'utf-8');
encrypted = Buffer.concat([encrypted, cipher.final()]);
return encrypted.toString('base64');
}
async handleIncomingChunk(data: Buffer): Promise<void> {
const incomingMessage = data.toString().trim();
const [headerJson, chunkContent] = incomingMessage.split('|');
const header = JSON.parse(headerJson);
if (!this.chunkBuffers[header.messageId]) {
this.chunkBuffers[header.messageId] = new Array(header.totalChunks).fill(undefined);
}
this.chunkBuffers[header.messageId][header.sequenceNumber - 1] = chunkContent;
if (this.chunkBuffers[header.messageId].every((chunk) => chunk !== undefined)) {
await this.processCompleteMessage(header.messageId);
}
}
private async processCompleteMessage(messageId: string): Promise<void> {
const chunks = this.chunkBuffers[messageId];
if (chunks && chunks.every((chunk) => chunk !== undefined)) {
const fullMessage = chunks.join('');
await this.handleIncomingMessage(fullMessage);
delete this.chunkBuffers[messageId];
}
}
async sendChunkedMessage(
operationCode: string,
metaInfo?: { [key: string]: any },
fileContent?: Buffer
): Promise<void> {
// Check if latency-based chunk size needs calculation
async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
if (!this.networkSpeed) {
this.networkSpeed = await this.scanNetworkLatency();
}
// Format the message for sending
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
let outgoingMessage: string;
switch (operationCode) {
case 'SET_PUBLIC_KEY':
@@ -138,32 +68,8 @@ export class TcpServerCommunicator extends SocketCommunicatorBase {
sequenceNumber: i + 1,
totalChunks,
});
const chunkWithHeader = `${chunkHeader}|${chunk}`;
await this.writeToSocket(chunkWithHeader);
// Set delay based on latency for smoother transmission
const delay = Math.max(100, Math.min(300, this.networkSpeed * 10));
await new Promise((resolve) => setTimeout(resolve, delay));
const chunkWithHeader = `${chunkHeader}|${chunk}${this.EOP}`;
if(!this.socket.write(chunkWithHeader)) this.socket.end();
}
}
async handleIncomingMessage(incomingMessage: string): Promise<void> {
let messageToProcess = incomingMessage;
if (this.aesKey && this.aesIv) {
messageToProcess = this.decryptWithAes(incomingMessage);
}
this.handlerResult = await this.operationHandler.handleOperation(messageToProcess);
}
protected writeToSocket(message: string): Promise<void> {
return new Promise((resolve, reject) => {
this.socket.write(message, (err: any) => {
if (err) {
console.error('Error sending message over TCP:', err);
return reject(err);
}
resolve();
});
});
}
}
UC/src/network/socket_communicator/udp_socket_communicator.ts
+6 -12
View File
@@ -11,20 +11,9 @@ export class UdpSocketCommunicator extends SocketCommunicatorBase {
this.socket = socket;
}
// Handle incoming message (no decryption needed for UDP)
handleIncomingMessage(incomingMessage: string): void {
this.handlerResult = this.operationHandler.handleOperation(incomingMessage);
}
// Send plain message over UDP (metaInfo as JSON and fileContent as Buffer)
async sendMessage(operationCode: string, metaInfo?: any, fileContent?: Buffer): Promise<void> {
async sendMessage(operationCode: string, metaInfo?: { [key: string]: any }, fileContent?: Buffer): Promise<void> {
const message = MessageHandler.formatMessage(operationCode, metaInfo, fileContent);
await this.sendUdpMessage(message);
}
// Helper method to wrap socket.send in a Promise for async/await support
private sendUdpMessage(message: string): Promise<void> {
return new Promise((resolve, reject) => {
this.socket.send(message, this.port, this.ip, (err: any) => {
if (err) {
@@ -36,4 +25,9 @@ export class UdpSocketCommunicator extends SocketCommunicatorBase {
});
});
}
// Handle incoming message (no decryption needed for UDP)
async handleIncomingMessage(incomingMessage: string): Promise<void> {
this.handlerResult = await this.operationHandler.handleOperation(incomingMessage);
}
}
UC/src/tcp_server.ts
+1 -2
View File
@@ -58,7 +58,6 @@ export class TcpServer {
const tcpCommunicator = new TcpServerCommunicator(socket, ip, port, this.operationHandler);
this.connectionManager.addConnection(ip, port, tcpCommunicator);
tcpCommunicator.generateKeyPair();
tcpCommunicator.sendPublicKey()
.then(() => tcpCommunicator.sendAesKey())
.then(() => this.log('Public key and AES key sent successfully.'))
@@ -106,7 +105,7 @@ export class TcpServer {
const handlerResult = communicator.getHandlerResult();
if (handlerResult) {
try {
await communicator.sendChunkedMessage(
await communicator.sendMessage(
handlerResult.operationCode,
handlerResult.metaInfo,
handlerResult.fileContent