initial

app.mjs

@@ -0,0 +1,10 @@
+import { Stream_Relay } from './stream-relay.mjs';
+import { Unix_Socket } from './unix-socket.mjs';
+import fs from 'node:fs';
+
+const source_stream = fs.createReadStream('/logsock/gitea.fifo');
+const target_socket = new Unix_Socket('/logsock/external.sock');
+
+const relay = new Stream_Relay(source_stream, target_socket);
+
+relay.serve();

autosave.mjs

@@ -0,0 +1,126 @@
+throw new Error('Not implemented');
+
+//TODO - this sketch is messy and incomplete
+
+class Autosaver {
+
+	#autosave_timer = null
+	#buffer_file_stream = null
+
+	constructor(line_buffer=[], buffer_file=null, autosave=false, autosave_interval=600) {
+		Object.assign({ line_buffer, buffer_file, autosave, autosave_interval });
+		this.apply_configuration(true);
+	}
+
+
+	perform_autosave() {
+		// TODO: flush pending lines to buffer_file
+		if (this.autosave) {
+			console.log("Autosave not implemented")
+		}
+	}
+
+	constructor(line_buffer=[], buffer_file=null, autosave=false, autosave_interval=600) {
+		Object.assign({ line_buffer, buffer_file, autosave, autosave_interval });
+		this.apply_configuration(true);
+	}
+
+	#get_buffer_writer() {
+
+		//TODO: This function should be properly vetted (ChatGPT 5.2)
+
+		if (!this.buffer_file) {
+			return null;
+		}
+
+		if (!this.#buffer_file_stream) {
+			try {
+				this.#buffer_file_stream = fs.createWriteStream(this.buffer_file, {
+					flags: 'a',
+					encoding: 'latin1'
+				});
+
+				this.#buffer_file_stream.on('error', (err) => {
+					console.log("Buffer stream error:", err);
+
+					try {
+						this.#buffer_file_stream.destroy();
+					} catch (destroy_err) {
+						console.log("Failed to destroy buffer stream due to", destroy_err);
+					}
+
+					this.#buffer_file_stream = null;
+				});
+			} catch (err) {
+				console.log("Failed to open buffer file due to", err);
+				this.#buffer_file_stream = null;
+				return null;
+			}
+		}
+
+		const stream = this.#buffer_file_stream;
+
+		const write_line = (line) => {
+
+			if (!stream.writable) {
+				this.#buffer_file_stream = null;
+				return;
+			}
+
+			stream.write(line + '\n');
+
+		};
+
+		return write_line;
+	}
+
+
+	#maybe_read_previous_buffer_file() {
+		if (this.buffer_file) {
+			let content;
+			try {
+				content = fs.readFileSync(this.buffer_file, { encoding: 'latin1' });
+			} catch (err) {
+				console.log("Failed to read buffer file due to", err);
+			}
+
+			if (content !== undefined) {
+				const lines = content.split('\n');
+				for (const line of lines) {
+					this.log_line(line);
+				}
+			}
+		}
+	}
+
+	#configure_autosave_timer() {
+		if (this.#autosave_timer) {
+			clearInterval(this.#autosave_timer);
+			this.#autosave_timer = null;
+		}
+
+		if (this.autosave && this.autosave_interval > 0) {
+			this.#autosave_timer = setInterval(() => {
+				this.perform_autosave();
+			}, this.autosave_interval * 1000);
+		}
+	}
+
+	apply_configuration(starting = false) {
+
+		if (starting === true) {
+			this.#maybe_read_previous_buffer_file();
+		}
+
+		this.#configure_autosave_timer();
+
+	}
+
+
+	buffer_line(line) {
+		const write_line = this.#get_buffer_writer();
+		write_line?.(line);
+	}
+
+
+}

probe-attempt.sh

@@ -0,0 +1 @@
+sshpass -p 'hunter1' ssh -o IdentitiesOnly=yes -o IdentityFile=none -o PubkeyAuthentication=no -o PreferredAuthentications=password -o PasswordAuthentication=yes -o KbdInteractiveAuthentication=no -o UserKnownHostsFile=/dev/null -o GlobalKnownHostsFile=/dev/null -o StrictHostKeyChecking=no evil_person@localhost -p 2222

relay.mjs

@@ -0,0 +1,116 @@
+import readline from 'node:readline';
+import net from 'node:net';
+import fs from 'node:fs';
+
+
+export class Relay {
+
+	#target_stream = null;
+	#socket_server = null;
+	#line_buffer_tally = 0;
+
+	constructor(listening_socket, target_socket, line_buffer=[], line_buffer_max_size=1024**2) {
+		Object.assign(this, { listening_socket, target_socket, line_buffer, line_buffer_max_size });
+	}
+	serve() {
+
+		const server = net.createServer((socket) => {
+
+			console.log("socket connected");
+
+			const rli = readline.createInterface({
+				input: socket,
+				crlfDelay: Infinity
+			});
+
+			rli.on('line', (line) => this.log_line(line));
+
+			socket.on('close', () => {
+				console.log("socket closed");
+			});
+
+			socket.on('error', (err) => {
+				console.log("socket error", { err });
+			});
+
+		});
+
+
+		server.on('error', (err) => {
+			console.log("server error", { err });
+		});
+
+		this.listening_socket.listen(server, () => {
+			this.#socket_server = server;
+		});
+
+	}
+
+	buffer_line(line) {
+
+		this.#line_buffer_tally += line.length + 1;
+		this.line_buffer.push(line);	// NOTE: We may of course briefly overshoot here but we are not expecting or caring about huge oneliners for now
+
+		while (this.line_buffer.length && (this.#line_buffer_tally > this.line_buffer_max_size)) {
+			const discarded = this.line_buffer.shift();
+			this.#line_buffer_tally -= discarded.length + 1;
+			console.log({ discarded });
+		}
+
+	}
+
+	log_line(line) {
+		this.buffer_line(line);
+
+		if (this.#target_stream) {
+			this.#flush();
+		} else {
+			this.#connect_target();
+		}
+
+	}
+
+	#connect_target() {
+
+		if (this.#target_stream) {
+			return;
+		}
+
+		const socket = net.createConnection(this.target_socket.connect_options);
+
+		socket.on('connect', () => {
+			this.#target_stream = socket;
+			this.#flush();
+		});
+
+		socket.on('error', (err) => {
+			console.log("target error", { err });
+			socket.destroy();
+			this.#target_stream = null;
+		});
+
+		socket.on('close', () => {
+			this.#target_stream = null;
+		});
+
+	}
+
+
+	#flush() {
+
+		while (this.line_buffer.length && this.#target_stream) {
+
+			const line = this.line_buffer[0];
+			const ok = this.#target_stream.write(line + '\n');
+
+			if (ok === false) {
+				return;
+			}
+
+			this.line_buffer.shift();
+		}
+	}
+
+
+
+}

stream-relay.mjs

@@ -0,0 +1,95 @@
+import readline from 'node:readline';
+import fs from 'node:fs';
+import net from 'node:net';
+
+export class Stream_Relay {
+
+	#target_stream = null;
+	#line_buffer_tally = 0;
+	#connecting = false;
+
+	constructor(source_stream, target_socket, line_buffer=[], line_buffer_max_size=1024**2) {
+		Object.assign(this, { source_stream, target_socket, line_buffer, line_buffer_max_size });
+	}
+	serve() {
+
+		const rli = readline.createInterface({
+			input: this.source_stream,
+			crlfDelay: Infinity
+		});
+
+		rli.on('line', (line) => this.log_line(line));
+	}
+
+	buffer_line(line) {
+
+		this.#line_buffer_tally += line.length + 1;
+		this.line_buffer.push(line);	// NOTE: We may of course briefly overshoot here but we are not expecting or caring about huge oneliners for now
+
+		while (this.line_buffer.length && (this.#line_buffer_tally > this.line_buffer_max_size)) {
+			const discarded = this.line_buffer.shift();
+			this.#line_buffer_tally -= discarded.length + 1;
+			console.log({ discarded });
+		}
+
+	}
+
+	log_line(line) {
+		this.buffer_line(line);
+
+		if (this.#target_stream) {
+			this.#flush();
+		} else {
+			this.#connect_target();
+		}
+
+	}
+
+
+
+	#connect_target() {
+
+		if (this.#target_stream || this.#connecting) {
+			return;
+		}
+
+		this.#connecting = true;
+
+		const socket = net.createConnection(this.target_socket.connect_options);
+
+		socket.on('connect', () => {
+			this.#connecting = false;
+			this.#target_stream = socket;
+			this.#flush();
+		});
+
+		socket.on('error', (err) => {
+			console.log("target error", { err });
+			socket.destroy();
+			this.#connecting = false;
+		});
+
+		socket.on('close', () => {
+			this.#target_stream = null;
+		});
+	}
+
+
+	#flush() {
+
+		while (this.line_buffer.length && this.#target_stream) {
+
+			const line = this.line_buffer[0];
+			const ok = this.#target_stream.write(line + '\n');
+
+			if (ok === false) {
+				return;
+			}
+
+			this.line_buffer.shift();
+		}
+	}
+
+
+
+}

t2.mjs

@@ -0,0 +1,36 @@
+import { Unix_Socket } from './unix-socket.mjs';
+import readline from 'node:readline';
+import net from 'node:net';
+import fs from 'node:fs';
+
+const listening_socket = new Unix_Socket('./logrelay.sock');
+
+const server = net.createServer((socket) => {
+
+	console.log("socket connected");
+
+	const rli = readline.createInterface({
+		input: socket,
+		crlfDelay: Infinity
+	});
+
+	rli.on('line', (line) => {
+		console.log({ line });
+	});
+
+	socket.on('close', () => {
+		console.log("socket closed");
+	});
+
+	socket.on('error', (err) => {
+		console.log("socket error", { err });
+	});
+
+});
+
+
+server.on('error', (err) => {
+	console.log("server error", { err });
+});
+
+listening_socket.listen(server);

t3.mjs

@@ -0,0 +1,11 @@
+import { Stream_Relay } from './stream-relay.mjs';
+import { Unix_Socket } from './unix-socket.mjs';
+import fs from 'node:fs';
+
+const source_stream = fs.createReadStream('/logsock/gitea.fifo');
+
+const target_socket = new Unix_Socket('./target.sock');
+
+const relay = new Relay(listening_socket, target_socket);
+
+relay.serve();

unix-socket.mjs

@@ -0,0 +1,27 @@
+import fs from 'node:fs';
+
+export class Unix_Socket {
+	constructor(path) {
+		Object.assign(this, { path });
+	}
+
+	get connect_options() {
+		const { path } = this;
+		return { path };
+	}
+
+	listen(server, on_listening) {
+		const { path } = this;
+		if (fs.existsSync(path)) {
+			fs.unlinkSync(path);
+		}
+
+		server.listen(path, () => {
+			fs.chmodSync(path, 0o666);
+			on_listening?.();
+		});
+
+	}
+
+
+}