Initial commit to show off.

Makefile

@@ -0,0 +1,29 @@
+BIN = ./.venv/bin/
+PYTHON = ./.venv/bin/python
+PIP = ./.venv/bin/pip
+APP_NAME=zamenyat
+
+all: install build
+
+ensure_repo:
+	-@git init 
+
+ensure_env: ensure_repo
+	-@python3 -m venv .venv 
+
+install: ensure_env
+	$(PIP) install -e . 
+
+format: 
+	$(PIP) install shed 
+	. $(BIN)/activate && shed
+
+build: 
+	$(MAKE) format
+	$(PIP) install build
+	$(PYTHON) -m build 
+
+run:  
+	$(BIN)$(APP_NAME) --host="0.0.0.0" --port=3046 --upstream-host="127.0.0.1" --upstream-port=9999
+
+

README.md

@@ -1,3 +1,4 @@
-# zamenyat
+# Zamenyat
+ 
+HTTP bridge configurable to replace the content you want to see replaced. This can be used to have a real version and anonymous version for a website for example like I did. This site exists in the retoor version and under my real name for example.
 
-HTTP bridge configurable to replace the content you want to see replaced. This can be used to have a real version and anonymous version for a website for example like I did. This site exists in the retoor version and under my real name for example.

pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"

setup.cfg

@@ -0,0 +1,24 @@
+[metadata]
+name = zamenyat
+version = 1.0.0
+description = Replaces sensitive information served by HTTP.
+author = retoor
+author_email = retoor@molodetz.nl
+license = MIT
+long_description = file: README.md
+long_description_content_type = text/markdown
+
+[options]
+packages = find:
+package_dir =
+    = src
+python_requires = >=3.7
+install_requires =
+    app @ git+https://retoor.molodetz.nl/retoor/app
+
+[options.packages.find]
+where = src
+
+[options.entry_points]
+console_scripts =
+    zamenyat = zamenyat.__main__:main

src/zamenyat/__main__.py

@@ -0,0 +1,35 @@
+import argparse
+from zamenyat.app import Application
+
+parser = argparse.ArgumentParser(description="Zamenyat sensitive content replacer.")
+
+parser.add_argument(
+    "--host",
+    required=True,
+    type=str
+)
+parser.add_argument(
+    "--port",
+    required=True,
+    type=int
+)
+parser.add_argument(
+    "--upstream-host",
+    required=True,
+    type=str
+)
+parser.add_argument(
+    "--upstream-port",
+    required=True,
+    type=int
+)
+
+
+def main():
+    args = parser.parse_args()
+    app = Application(upstream_host=args.upstream_host, upstream_port=args.upstream_port)
+    app.serve(host=args.host,port=args.port)
+
+if __name__ == '__main__':
+    main()
+

src/zamenyat/app.py

@@ -0,0 +1,134 @@
+from app.app import Application as BaseApplication, get_timestamp
+import asyncio
+from concurrent.futures import ThreadPoolExecutor as Executor
+import time 
+
+ZAMENYAT_THREAD_COUNT = 500
+ZAMENYAT_BUFFER_SIZE = 4096*2
+ZAMENYAT_HEADER_MAX_LENGTH = 4096*2
+
+class Application:
+
+    def __init__(self, upstream_host, upstream_port, *args, **kwargs):
+        self.upstream_host = upstream_host
+        self.upstream_port = upstream_port
+        self.server = None
+        self.host = None
+        self.port = None
+        self.executor = None
+        self.buffer_size = ZAMENYAT_BUFFER_SIZE
+        self.header_max_length = ZAMENYAT_HEADER_MAX_LENGTH
+        self.connection_count = 0
+        self.total_connection_count = 0
+        super().__init__(*args, **kwargs)
+    
+    async def get_headers(self, reader):
+        data = b''
+        headers = None
+        while True:
+            chunk = await reader.read(self.buffer_size)
+            if not chunk:
+                break
+            data += chunk
+            if len(data) > self.header_max_length:
+                break
+            headers_end = data.find(b'\r\n\r\n')
+            if headers_end:
+                headers = data[:headers_end]
+                data = data[:headers_end + 4]
+                break
+        if not headers:
+            return None, None, None
+        header_dict = {}
+        req_resp, *headers = headers.split(b"\r\n")
+        for header_line in headers:
+            key, *value = header_line.split(b": ")
+            key = key.decode()
+            value = ": ".join([value.decode() for value in value])
+            header_dict[key] = int(value) if value.isdigit() else value 
+        return req_resp.decode(), header_dict, data
+
+    def header_dict_to_bytes(self, req_resp, headers):
+        header_list = [req_resp]
+        for key, value in headers.items():
+            header_list.append("{}: {}".format(key, value))
+        header_list.append("\r\n")
+        return ("\r\n".join(header_list)).encode()
+
+    async def stream(self, reader,writer):
+        try:
+            while True:
+                req_resp, headers, data = await self.get_headers(reader)        
+                if not headers:
+                    break
+                if 'Content-Length' in headers:
+                    while not len(data) == headers['Content-Length']:
+                        chunk_size = headers['Content-Length'] - len(data) if self.buffer_size > headers['Content-Length'] - len(data) else self.buffer_size
+                        chunk = await reader.read(chunk_size)
+                        if not chunk:
+                            data = None
+                            break
+                        data += chunk
+                print(self.header_dict_to_bytes(req_resp,headers).decode())
+                writer.write(self.header_dict_to_bytes(req_resp, headers))
+                #await writer.drain()
+                if data:
+                    print(data)
+                while data:
+                    chunk_size = self.buffer_size if len(data) > self.buffer_size else len(data)
+                    
+
+                    chunk = data[:chunk_size]
+                    
+
+                    writer.write(chunk)
+                    data = data[chunk_size:]
+                    await writer.drain()
+                if not headers.get('Connection') == 'keep-alive': # and not headers.get('Upgrade-Insecure-Requests'):
+                    break
+                
+        except asyncio.CancelledError:
+            pass
+        finally:
+            pass
+            #writer.close()
+            #await writer.wait_closed()
+    
+    async def handle_client(self,reader,writer):
+        self.connection_count += 1
+        self.total_connection_count += 1
+        connection_nr = self.connection_count
+
+
+        upstream_reader, upstream_writer = await asyncio.open_connection(self.upstream_host, self.upstream_port)
+        time_start = time.time()
+        print(f"Connected to upstream #{self.total_connection_count} server {self.upstream_host}:{self.upstream_port} #{connection_nr} Time: {get_timestamp()}")
+        tasks = [
+            self.stream(upstream_reader, writer),
+            self.stream(reader, upstream_writer)
+        ]
+        await asyncio.gather(*tasks)
+        time_end = time.time()
+        time_duration = time_end - time_start
+        print(f"Disconnected upstream #{self.total_connection_count} server {self.upstream_host}:{self.upstream_port} #{connection_nr} Duration: {time_duration:.5f}s")
+        self.connection_count -= 1
+
+    def upgrade_executor(self, thread_count):
+        self.executor = Executor(max_workers=thread_count)
+        loop = asyncio.get_running_loop()
+        loop.set_default_executor(self.executor)
+        return self.executor
+
+    async def serve_async(self, host,port):
+        self.upgrade_executor(ZAMENYAT_THREAD_COUNT)
+        self.host = host
+        self.port = port
+        self.server = await asyncio.start_server(self.handle_client, self.host, self.port)
+        async with self.server:
+            await self.server.serve_forever()
+
+    def serve(self, host, port):
+        try:
+            asyncio.run(self.serve_async(host,port))
+        except KeyboardInterrupt:    
+            print("Shutted down server")