News : mitmproxy - part 001.

Mitmproxy is an interactive, open‑source proxy tool that lets you intercept, inspect, and modify HTTP and HTTPS traffic in real time. It acts as a "man‑in‑the‑middle" between your computer and the internet, making it possible to debug, test, or analyze how applications communicate online.

Why Python?

• Mitmproxy is built in Python and exposes a powerful addon API
• You can write custom scripts to automate tasks and traffic manipulation
• Block or rewrite requests and responses with flexible logic
• Inject headers or simulate server responses for testing
• Integrate with other Python tools for advanced automation

• Intercept and inspect HTTP and HTTPS traffic in real time
• Modify requests and responses dynamically with Python scripts
• Block specific hosts or URLs to prevent unwanted connections
• Inject custom headers into outgoing requests for debugging or control
• Rewrite response bodies (HTML, JSON, text) using regex or custom logic
• Log and save traffic flows for later analysis and replay
• Simulate server responses to test client behavior offline
• Automate testing of web applications and APIs with scripted rules
• Monitor performance metrics such as latency and payload size
• Integrate with other Python tools for advanced automation and analysis
• Use a trusted root certificate to decrypt and modify HTTPS traffic securely

Let's install:

pip install mitmproxy

Let's see the python script:

# addon.py
from mitmproxy import http
from mitmproxy import ctx
import re

BLOCKED_HOSTS = {
    "hyte.com",
    "ads.example.org",
}

REWRITE_RULES = [
    # Each rule: (pattern, replacement, content_type_substring)
    (re.compile(rb"Hello World"), b"Salut lume", "text/html"),
    (re.compile(rb"tracking", re.IGNORECASE), b"observare", "text"),
]

ADD_HEADERS = {
    "X-Debug-Proxy": "mitm",
    "X-George-Tool": "true",
}

class GeorgeProxy:
    def __init__(self):
        self.rewrite_count = 0

    def load(self, loader):
        ctx.log.info("GeorgeProxy addon loaded.")

    def request(self, flow: http.HTTPFlow):
        # Block specific hosts early
        host = flow.request.host
        if host in BLOCKED_HOSTS:
            flow.response = http.Response.make(
                403,
                b"Blocked by GeorgeProxy",
                {"Content-Type": "text/plain"}
            )
            ctx.log.warn(f"Blocked request to {host}")
            return

        # Add custom headers to outgoing requests
        for k, v in ADD_HEADERS.items():
            flow.request.headers[k] = v

        ctx.log.info(f"REQ {flow.request.method} {flow.request.url}")

    def response(self, flow: http.HTTPFlow):
        # Only process text-like contents
        ctype = flow.response.headers.get("Content-Type", "").lower()
        raw = flow.response.raw_content

        if raw and any(t in ctype for t in ["text", "html", "json"]):
            new_content = raw
            for pattern, repl, t in REWRITE_RULES:
                if t in ctype:
                    new_content, n = pattern.subn(repl, new_content)
                    self.rewrite_count += n

            if new_content != raw:
                flow.response.raw_content = new_content
                # Update Content-Length only if present
                if "Content-Length" in flow.response.headers:
                    flow.response.headers["Content-Length"] = str(len(new_content))
                ctx.log.info(f"Rewrote content ({ctype}); total matches: {self.rewrite_count}")

        ctx.log.info(f"RESP {flow.response.status_code} {flow.request.url}")

addons = [GeorgeProxy()]

Let's run it:

mitmdump -s addon.py
[21:46:04.435] Loading script addon.py
[21:46:04.504] GeorgeProxy addon loaded.
[21:46:04.506] HTTP(S) proxy listening at *:8080.
[21:46:18.547][127.0.0.1:52128] client connect
[21:46:18.593] REQ GET http://httpbin.org/get
[21:46:18.768][127.0.0.1:52128] server connect httpbin.org:80 (52.44.182.178:80)
[21:46:18.910] RESP 200 http://httpbin.org/get
127.0.0.1:52128: GET http://httpbin.org/get
              << 200 OK 353b
[21:46:19.019][127.0.0.1:52128] client disconnect
[21:46:19.021][127.0.0.1:52128] server disconnect httpbin.org:80 (52.44.182.178:80)

Let's see the result:

curl -x http://127.0.0.1:8080 http://httpbin.org/get
{
  "args": {},
  "headers": {
    "Accept": "*/*",
    "Host": "httpbin.org",
    "Proxy-Connection": "Keep-Alive",
    "User-Agent": "curl/8.13.0",
    "X-Amzn-Trace-Id": "Root=1-692c9f0b-7eaf43e61f276ee62b089933",
    "X-Debug-Proxy": "mitm",
    "X-George-Tool": "true"
  },
  "origin": "84.117.220.94",
  "url": "http://httpbin.org/get"
}

This means

The request successfully went through mitmproxy running on 127.0.0.1:8080. Your addon worked: it injected the custom headers (X-Debug-Proxy, X-George-Tool). The httpbin.org echoed back the request details, showing exactly what the server received.

Python Tornado - part 001.

Python Tornado is a high‑performance web framework and asynchronous networking library designed for extreme scalability and real‑time applications. Its standout capability is handling tens of thousands of simultaneous connections efficiently, thanks to non‑blocking I/O.

This is an open source project actively maintained and available on tornadoweb.org.

Python Tornado – Key Capabilities

• Massive Concurrency: Tornado can scale to tens of thousands of open connections without requiring huge numbers of threads.
• Non‑blocking I/O: Its asynchronous design makes it ideal for apps that need to stay responsive under heavy load.
• WebSockets Support: Built‑in support for WebSockets enables real‑time communication between clients and servers.
• Long‑lived Connections: Perfect for long polling, streaming, or chat applications where connections remain open for extended periods.
• Coroutines & Async/Await: Tornado integrates tightly with Python’s asyncio, allowing developers to write clean asynchronous code using coroutines.
• Versatile Use Cases: Beyond web apps, Tornado can act as an HTTP client/server, handle background tasks, or integrate with other services.

Tornado setup: The script creates a web server using the Tornado framework, listening on port 8888.

Route definition: A single route /form is registered, handled by the FormHandler class.

GET request: When you visit http://localhost:8888/form, the server responds with an HTML page (form.html) that contains a simple input form.

POST request: When the form is submitted, the post() method retrieves the value of the name field using self.get_argument("name").

Response: The server then sends back a personalized message

Let's see the script:

import tornado.ioloop
import tornado.web
import os

class FormHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("form.html")  # Render an HTML form

    def post(self):
        name = self.get_argument("name")
        self.write(f"Hello, {name}!")

def make_app():
    return tornado.web.Application([
        (r"/form", FormHandler),
    ],
    template_path=os.path.join(os.path.dirname(__file__), "templates")  # <-- aici
    )

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("Server pornit pe http://localhost:8888/form")
    tornado.ioloop.IOLoop.current().start()