import { ConfigService } from '@nestjs/config'; import Big from 'big.js'; import * as bitcoinjs from 'bitcoinjs-lib'; import { plainToInstance } from 'class-transformer'; import { validate, ValidatorOptions } from 'class-validator'; import * as crypto from 'crypto'; import { Socket } from 'net'; import { firstValueFrom, Subscription } from 'rxjs'; import { clearInterval } from 'timers'; import { AddressSettingsService } from '../ORM/address-settings/address-settings.service'; import { BlocksService } from '../ORM/blocks/blocks.service'; import { ClientStatisticsService } from '../ORM/client-statistics/client-statistics.service'; import { ClientEntity } from '../ORM/client/client.entity'; import { ClientService } from '../ORM/client/client.service'; import { BitcoinRpcService } from '../services/bitcoin-rpc.service'; import { NotificationService } from '../services/notification.service'; import { IJobTemplate, StratumV1JobsService } from '../services/stratum-v1-jobs.service'; import { eRequestMethod } from './enums/eRequestMethod'; import { eResponseMethod } from './enums/eResponseMethod'; import { eStratumErrorCode } from './enums/eStratumErrorCode'; import { MiningJob } from './MiningJob'; import { AuthorizationMessage } from './stratum-messages/AuthorizationMessage'; import { ConfigurationMessage } from './stratum-messages/ConfigurationMessage'; import { MiningSubmitMessage } from './stratum-messages/MiningSubmitMessage'; import { StratumErrorMessage } from './stratum-messages/StratumErrorMessage'; import { SubscriptionMessage } from './stratum-messages/SubscriptionMessage'; import { SuggestDifficulty } from './stratum-messages/SuggestDifficultyMessage'; import { StratumV1ClientStatistics } from './StratumV1ClientStatistics'; export class StratumV1Client { public clientSubscription: SubscriptionMessage; private clientConfiguration: ConfigurationMessage; private clientAuthorization: AuthorizationMessage; private clientSuggestedDifficulty: SuggestDifficulty; private stratumSubscription: Subscription; private backgroundWork: NodeJS.Timer[] = []; private statistics: StratumV1ClientStatistics; private stratumInitialized = false; private usedSuggestedDifficulty = false; private sessionDifficulty: number = 100000; private clientEntity: ClientEntity; private creatingEntity: Promise; public extraNonceAndSessionId: string; public sessionStart: Date; //public noFee: boolean; //public hashRate: number = 0; private buffer: string = ''; private miningSubmissionHashes = new Set() constructor( public readonly socket: Socket, private readonly stratumV1JobsService: StratumV1JobsService, private readonly bitcoinRpcService: BitcoinRpcService, private readonly clientService: ClientService, private readonly clientStatisticsService: ClientStatisticsService, private readonly notificationService: NotificationService, private readonly blocksService: BlocksService, private readonly configService: ConfigService, private readonly addressSettingsService: AddressSettingsService ) { this.socket.on('data', (data: Buffer) => { this.buffer += data.toString(); let lines = this.buffer.split('\n'); this.buffer = lines.pop() || ''; // Save the last part of the data (incomplete line) to the buffer lines .filter(m => m.length > 0) .forEach(async (m) => { try { await this.handleMessage(m); } catch (e) { await this.socket.end(); console.error(e); } }); }); } public async destroy() { if (this.clientEntity?.id) { await this.clientService.delete(this.clientEntity.id); } if (this.stratumSubscription != null) { this.stratumSubscription.unsubscribe(); } this.backgroundWork.forEach(work => { clearInterval(work); }); } private getRandomHexString() { const randomBytes = crypto.randomBytes(4); // 4 bytes = 32 bits const randomNumber = randomBytes.readUInt32BE(0); // Convert bytes to a 32-bit unsigned integer const hexString = randomNumber.toString(16).padStart(8, '0'); // Convert to hex and pad with zeros return hexString; } private async handleMessage(message: string) { //console.log(`Received from ${this.extraNonceAndSessionId}`, message); // Parse the message and check if it's the initial subscription message let parsedMessage = null; try { parsedMessage = JSON.parse(message); } catch (e) { //console.log("Invalid JSON"); await this.socket.end(); return; } switch (parsedMessage.method) { case eRequestMethod.SUBSCRIBE: { const subscriptionMessage = plainToInstance( SubscriptionMessage, parsedMessage, ); const validatorOptions: ValidatorOptions = { whitelist: true, //forbidNonWhitelisted: true, }; const errors = await validate(subscriptionMessage, validatorOptions); if (errors.length === 0) { if (this.sessionStart == null) { this.sessionStart = new Date(); this.statistics = new StratumV1ClientStatistics(this.clientStatisticsService); this.extraNonceAndSessionId = this.getRandomHexString(); //console.log(`New client ID: : ${this.extraNonceAndSessionId}, ${this.socket.remoteAddress}:${this.socket.remotePort}`); } this.clientSubscription = subscriptionMessage; const success = await this.write(JSON.stringify(this.clientSubscription.response(this.extraNonceAndSessionId)) + '\n'); if (!success) { return; } } else { console.error('Subscription validation error'); const err = new StratumErrorMessage( subscriptionMessage.id, eStratumErrorCode.OtherUnknown, 'Subscription validation error', errors).response(); console.error(err); const success = await this.write(err); if (!success) { return; } } break; } case eRequestMethod.CONFIGURE: { const configurationMessage = plainToInstance( ConfigurationMessage, parsedMessage, ); const validatorOptions: ValidatorOptions = { whitelist: true, //forbidNonWhitelisted: true, }; const errors = await validate(configurationMessage, validatorOptions); if (errors.length === 0) { this.clientConfiguration = configurationMessage; //const response = this.buildSubscriptionResponse(configurationMessage.id); const success = await this.write(JSON.stringify(this.clientConfiguration.response()) + '\n'); if (!success) { return; } } else { console.log('Configuration validation error'); const err = new StratumErrorMessage( configurationMessage.id, eStratumErrorCode.OtherUnknown, 'Configuration validation error', errors).response(); console.error(err); const success = await this.write(err); if (!success) { return; } } break; } case eRequestMethod.AUTHORIZE: { const authorizationMessage = plainToInstance( AuthorizationMessage, parsedMessage, ); const validatorOptions: ValidatorOptions = { whitelist: true, //forbidNonWhitelisted: true, }; const errors = await validate(authorizationMessage, validatorOptions); if (errors.length === 0) { this.clientAuthorization = authorizationMessage; if (this.clientSuggestedDifficulty == null && this.clientAuthorization.startingDiff != null && this.clientAuthorization.startingDiff > this.sessionDifficulty) { this.sessionDifficulty = this.clientAuthorization.startingDiff; } const success = await this.write(JSON.stringify(this.clientAuthorization.response()) + '\n'); if (!success) { return; } } else { const err = new StratumErrorMessage( authorizationMessage.id, eStratumErrorCode.OtherUnknown, 'Authorization validation error', errors).response(); //console.log(err); const success = await this.write(err); if (!success) { return; } } break; } case eRequestMethod.SUGGEST_DIFFICULTY: { if (this.usedSuggestedDifficulty == true) { return; } const suggestDifficultyMessage = plainToInstance( SuggestDifficulty, parsedMessage ); const validatorOptions: ValidatorOptions = { whitelist: true, //forbidNonWhitelisted: true, }; const errors = await validate(suggestDifficultyMessage, validatorOptions); if (errors.length === 0) { this.clientSuggestedDifficulty = suggestDifficultyMessage; this.sessionDifficulty = suggestDifficultyMessage.suggestedDifficulty; const success = await this.write(JSON.stringify(this.clientSuggestedDifficulty.response(this.sessionDifficulty)) + '\n'); if (!success) { return; } this.usedSuggestedDifficulty = true; } else { console.error('Suggest difficulty validation error'); const err = new StratumErrorMessage( suggestDifficultyMessage.id, eStratumErrorCode.OtherUnknown, 'Suggest difficulty validation error', errors).response(); console.error(err); const success = await this.write(err); if (!success) { return; } } break; } case eRequestMethod.SUBMIT: { if (this.stratumInitialized == false) { //console.log('Submit before initalized'); await this.socket.end(); return; } const miningSubmitMessage = plainToInstance( MiningSubmitMessage, parsedMessage, ); const validatorOptions: ValidatorOptions = { whitelist: true, //forbidNonWhitelisted: true, }; const errors = await validate(miningSubmitMessage, validatorOptions); if (errors.length === 0 && this.stratumInitialized == true) { const result = await this.handleMiningSubmission(miningSubmitMessage); if (result == true) { const success = await this.write(JSON.stringify(miningSubmitMessage.response()) + '\n'); if (!success) { return; } } } else { console.log('Mining Submit validation error'); const err = new StratumErrorMessage( miningSubmitMessage.id, eStratumErrorCode.OtherUnknown, 'Mining Submit validation error', errors).response(); console.error(err); const success = await this.write(err); if (!success) { return; } } break; } // default: { // console.log("Invalid message"); // console.log(parsedMessage); // await this.socket.end(); // return; // } } if (this.clientSubscription != null && this.clientAuthorization != null && this.stratumInitialized == false) { this.initStratum(); } } private async initStratum() { this.stratumInitialized = true; if (this.validateHeaderCompliance(this.clientSubscription.userAgent)) { console.log(`Non compliant connection from userAgent: ${this.clientSubscription.userAgent}`); await this.socket.end(); return; } switch (this.clientSubscription.userAgent) { case 'cpuminer': { this.sessionDifficulty = 0.1; } } if (this.clientSuggestedDifficulty == null) { //console.log(`Setting difficulty to ${this.sessionDifficulty}`) const setDifficulty = JSON.stringify(new SuggestDifficulty().response(this.sessionDifficulty)); const success = await this.write(setDifficulty + '\n'); if (!success) { return; } } this.stratumSubscription = this.stratumV1JobsService.newMiningJob$.subscribe(async (jobTemplate) => { try { if(jobTemplate.blockData.clearJobs){ this.miningSubmissionHashes.clear(); } await this.sendNewMiningJob(jobTemplate); } catch (e) { await this.socket.end(); console.error(e); } }); this.backgroundWork.push( setInterval(async () => { await this.checkDifficulty(); }, 60 * 1000) ); // this.backgroundWork.push( // setInterval(async () => { // await this.statistics.saveShares(this.clientEntity); // }, 60 * 1000) // ); } private async sendNewMiningJob(jobTemplate: IJobTemplate) { let payoutInformation= [ { address: this.clientAuthorization.address, percent: 100 } ]; // const devFeeAddress = this.configService.get('DEV_FEE_ADDRESS'); // //50Th/s // this.noFee = false; // if (this.clientEntity) { // this.hashRate = await this.clientStatisticsService.getHashRateForSession(this.clientEntity.id); // // 250Gh/s // if(this.hashRate < 250000000000){ // this.statistics.targetSubmitShareEveryNSeconds = 10; // } // this.noFee = this.hashRate != 0 && this.hashRate < 50000000000000; // } // if (this.noFee || devFeeAddress == null || devFeeAddress.length < 1) { // payoutInformation = [ // { address: this.clientAuthorization.address, percent: 100 } // ]; // } else { // payoutInformation = [ // { address: devFeeAddress, percent: 1.5 }, // { address: this.clientAuthorization.address, percent: 98.5 } // ]; // } const networkConfig = this.configService.get('NETWORK'); let network; if (networkConfig === 'mainnet') { network = bitcoinjs.networks.bitcoin; } else if (networkConfig === 'testnet') { network = bitcoinjs.networks.testnet; } else if (networkConfig === 'regtest') { network = bitcoinjs.networks.regtest; } else { throw new Error('Invalid network configuration'); } const job = new MiningJob( network, this.stratumV1JobsService.getNextId(), payoutInformation, jobTemplate ); this.stratumV1JobsService.addJob(job); const success = await this.write(job.response(jobTemplate)); if (!success) { return; } //console.log(`Sent new job to ${this.clientAuthorization.worker}.${this.extraNonceAndSessionId}. (clearJobs: ${jobTemplate.blockData.clearJobs}, fee?: ${!this.noFee})`) } private async handleMiningSubmission(submission: MiningSubmitMessage) { if (this.clientEntity == null) { if (this.creatingEntity == null) { this.creatingEntity = new Promise(async (resolve, reject) => { try { this.clientEntity = await this.clientService.insert({ sessionId: this.extraNonceAndSessionId, address: this.clientAuthorization.address, clientName: this.clientAuthorization.worker, userAgent: this.clientSubscription.userAgent, startTime: new Date(), bestDifficulty: 0 }); } catch (e) { reject(e); } resolve(); }); await this.creatingEntity; } else { await this.creatingEntity; } } const submissionHash = submission.hash(); if(this.miningSubmissionHashes.has(submissionHash)){ const err = new StratumErrorMessage( submission.id, eStratumErrorCode.DuplicateShare, 'Duplicate share').response(); const success = await this.write(err); if (!success) { return false; } return false; }else{ this.miningSubmissionHashes.add(submissionHash); } const job = this.stratumV1JobsService.getJobById(submission.jobId); // a miner may submit a job that doesn't exist anymore if it was removed by a new block notification (or expired, 5 min) if (job == null) { const err = new StratumErrorMessage( submission.id, eStratumErrorCode.JobNotFound, 'Job not found').response(); //console.log(err); const success = await this.write(err); if (!success) { return false; } return false; } const jobTemplate = this.stratumV1JobsService.getJobTemplateById(job.jobTemplateId); if (jobTemplate == null) { const err = new StratumErrorMessage( submission.id, eStratumErrorCode.JobNotFound, 'Job Template not found').response(); //console.log(err); const success = await this.write(err); if (!success) { return false; } return false; } const updatedJobBlock = job.copyAndUpdateBlock( jobTemplate, parseInt(submission.versionMask, 16), parseInt(submission.nonce, 16), this.extraNonceAndSessionId, submission.extraNonce2, parseInt(submission.ntime, 16) ); const header = updatedJobBlock.toBuffer(true); const { submissionDifficulty } = this.calculateDifficulty(header); //console.log(`DIFF: ${submissionDifficulty} of ${this.sessionDifficulty} from ${this.clientAuthorization.worker + '.' + this.extraNonceAndSessionId}`); if (submissionDifficulty >= this.sessionDifficulty) { if (submissionDifficulty >= jobTemplate.blockData.networkDifficulty) { console.log('!!! BLOCK FOUND !!!'); const blockHex = updatedJobBlock.toHex(false); const result = await this.bitcoinRpcService.SUBMIT_BLOCK(blockHex); await this.blocksService.save({ height: jobTemplate.blockData.height, minerAddress: this.clientAuthorization.address, worker: this.clientAuthorization.worker, sessionId: this.extraNonceAndSessionId, blockData: blockHex }); await this.notificationService.notifySubscribersBlockFound(this.clientAuthorization.address, jobTemplate.blockData.height, updatedJobBlock, result); //success if (result == null) { await this.addressSettingsService.resetBestDifficultyAndShares(); } } try { await this.statistics.addShares(this.clientEntity, this.sessionDifficulty); const now = new Date(); // only update every minute //if (this.clientEntity.updatedAt == null || now.getTime() - this.clientEntity.updatedAt.getTime() > 1000 * 60) { await this.clientService.heartbeatBulkAsync(this.clientEntity.id, this.statistics.hashRate, now); this.clientEntity.updatedAt = now; //} } catch (e) { console.log(e); } if (submissionDifficulty > this.clientEntity.bestDifficulty) { await this.clientService.updateBestDifficulty(this.clientEntity.id, submissionDifficulty); this.clientEntity.bestDifficulty = submissionDifficulty; if (submissionDifficulty > (await this.addressSettingsService.getSettings(this.clientAuthorization.address, true)).bestDifficulty) { await this.addressSettingsService.updateBestDifficulty(this.clientAuthorization.address, submissionDifficulty, this.clientEntity.userAgent); } } } else { const err = new StratumErrorMessage( submission.id, eStratumErrorCode.LowDifficultyShare, 'Difficulty too low').response(); const success = await this.write(err); if (!success) { return false; } return false; } //await this.checkDifficulty(); return true; } private async checkDifficulty() { const targetDiff = this.statistics.getSuggestedDifficulty(this.sessionDifficulty); if (targetDiff == null) { return; } if (targetDiff != this.sessionDifficulty) { //console.log(`Adjusting ${this.extraNonceAndSessionId} difficulty from ${this.sessionDifficulty} to ${targetDiff}`); this.sessionDifficulty = targetDiff; const data = JSON.stringify({ id: null, method: eResponseMethod.SET_DIFFICULTY, params: [targetDiff] }) + '\n'; await this.socket.write(data); const jobTemplate = await firstValueFrom(this.stratumV1JobsService.newMiningJob$); // we need to clear the jobs so that the difficulty set takes effect. Otherwise the different miner implementations can cause issues jobTemplate.blockData.clearJobs = true; await this.sendNewMiningJob(jobTemplate); } } private calculateDifficulty(header: Buffer): { submissionDifficulty: number, submissionHash: string } { const hashResult = bitcoinjs.crypto.hash256(header); let s64 = this.le256todouble(hashResult); const truediffone = Big('26959535291011309493156476344723991336010898738574164086137773096960'); const difficulty = truediffone.div(s64.toString()); return { submissionDifficulty: difficulty.toNumber(), submissionHash: hashResult.toString('hex') }; } private le256todouble(target: Buffer): bigint { const number = target.reduceRight((acc, byte) => { // Shift the number 8 bits to the left and OR with the current byte return (acc << BigInt(8)) | BigInt(byte); }, BigInt(0)); return number; } private validateHeaderCompliance(userAgent: string): boolean { const headerCompliance = this.configService.get('COMPLIANT_HEADERS'); if (!headerCompliance || headerCompliance.trim() === '') { return false; } const complianceList = headerCompliance.split(',').map(ua => ua.trim().toLowerCase()); const userAgentLower = userAgent.toLowerCase(); return complianceList.some(compliant => compliant.length > 0 && userAgentLower.includes(compliant)); } private async write(message: string): Promise { try { if (!this.socket.destroyed && !this.socket.writableEnded) { await new Promise((resolve, reject) => { this.socket.write(message, (error) => { if (error) { reject(error); } else { resolve(true); } }); }); return true; } else { //console.error(`Error: Cannot write to closed or ended socket. ${this.extraNonceAndSessionId} ${message}`); this.destroy(); if (!this.socket.destroyed) { this.socket.destroy(); } return false; } } catch (error) { this.destroy(); if (!this.socket.writableEnded) { await this.socket.end(); } else if (!this.socket.destroyed) { this.socket.destroy(); } //console.error(`Error occurred while writing to socket: ${this.extraNonceAndSessionId}`, error); return false; } } }