GarlicGrater: call for testing

General Discussion
1

Abstract

GarlicGrater its simple proxy for i2p SAM bridge what provides isolation between pseudonyms and real identity, works like onion-grater
For more information see my guide about i2p in qubes link

How to works

Gralicgrater proxified all commands from user to i2p router and replaces *.length in SESSION CREATE to user provided values, and adds prefix to session IDs

Updates

2026-08-13

Now garlicgrater remove sockets if client crashes, fixed = symbol parsing bug

Code

"""
TCP proxy credits

Source - https://stackoverflow.com/a/78557776
Posted by Estelle Poulin
Retrieved 2026-08-12, License - CC BY-SA 4.0
"""

import socket
import selectors

FORCE_INBOUND_LENGTH = 3
FORCE_OUTBOUND_LENGTH = 3

ID_PREFIX = 'isolated_' # Change this prefix per pseudonym

REAL_SAM_IP = '127.0.0.1'
REAL_SAM_PORT = 7656

FILTERED_SAM_LISTEN_IP = '127.0.0.1'
FILTERED_SAM_LISTEN_PORT = 7657

class TCPProxy:
    def __init__(
        self,
        listen_addr: str,
        listen_port: int,
        destination_addr: str,
        destination_port: int,
    ):
        self.listen_addr = listen_addr
        self.listen_port = listen_port
        self.destination_addr = destination_addr
        self.destination_port = destination_port
        self.selector = selectors.DefaultSelector()
        self.connections = {}

    @staticmethod
    def __sam_parse(sam_command):
        sam_command_splited = sam_command.split()
        command = f'{sam_command_splited.pop(0)} {sam_command_splited.pop(0)}'
        args = {}

        for arg_value in sam_command_splited:
            args[arg_value.split('=', 1)[0]] = arg_value.split('=', 1)[1]

        return (command, args)

    @staticmethod
    def __generate_sam_command(sam_command: tuple):
        command = sam_command[0]

        for arg, value in sam_command[1].items():
            command += f' {arg}={value}'

        command += '\n'
        return command.encode('utf-8')

    @staticmethod
    def __is_sam(sam_command):
        try:
            if sam_command.decode('utf-8').split()[0] in ['HELLO', 'NAMING', 'STREAM', 'SESSION']:
                return True
            return False
        except:
            return False

    def __process_sam_command(self, sam_command):
        parsed_sam_command = self.__sam_parse(sam_command.decode())

        if parsed_sam_command[0] == 'SESSION CREATE':
            parsed_sam_command[1]['inbound.length'] = FORCE_INBOUND_LENGTH
            parsed_sam_command[1]['outbound.length'] = FORCE_OUTBOUND_LENGTH

        if parsed_sam_command[1].get('ID') is not None:
            parsed_sam_command[1]['ID'] = f"""{ID_PREFIX}{parsed_sam_command[1]['ID']}"""

        return self.__generate_sam_command(parsed_sam_command)

    def on_accept(self, sock: socket.socket):
        conn, addr = sock.accept()

        # For each client connection we create a new server connection
        # as well. This lets us support multiple clients and not get the
        # data mixed up.
        send_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        send_sock.connect((self.destination_addr, self.destination_port))
        send_sock.setblocking(False)

        self.connections[conn] = send_sock

        # Start listening on the client -> proxy and the proxy -> server
        # sockets.
        self.selector.register(send_sock, selectors.EVENT_READ, self.on_data_from_client)
        self.selector.register(conn, selectors.EVENT_READ, self.on_data_from_server)

    def on_data_from_server(self, conn: socket.socket):
        try:
            data = conn.recv(1024)
            out_sock = self.connections[conn]

            if self.__is_sam(data):
                out_sock.send(self.__process_sam_command(data))
            else:
                out_sock.send(data)
        except:
            # Client has been killed, forcelly removing sockets

            self.selector.unregister(conn)
            self.selector.unregister(out_sock)
            out_sock.close()
            conn.close()
            del self.connections[conn]

    def on_data_from_client(self, conn: socket.socket):
        try:
            data = conn.recv(1024)

            # This is just a reverse lookup on the connections dict.
            return_sock = next(x for x, y in self.connections.items() if y is conn)
            return_sock.send(data)
        except:
            self.selector.unregister(conn)
            self.selector.unregister(return_sock)
            conn.close()
            return_sock.close()
            del self.connections[return_sock]

    def start(self):
        # Create a socket for the proxy and have it listen on the address/port you set.
        self.proxy_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.proxy_sock.bind((self.listen_addr, self.listen_port))
        self.proxy_sock.setblocking(False)
        self.proxy_sock.listen()

        # Start listening for incoming data on the socket.
        self.selector.register(self.proxy_sock, selectors.EVENT_READ, self.on_accept)

        while True:
            # Wait until one of the sockets has data.
            events = self.selector.select()
            for key, _ in events:
                # Call the on_* method associated with it.
                callback = key.data
                callback(key.fileobj)

proxy = TCPProxy(FILTERED_SAM_LISTEN_IP, FILTERED_SAM_LISTEN_PORT, REAL_SAM_IP, REAL_SAM_PORT)
proxy.start()
1 Like