A high-performance, multithreaded HTTP/1.1 server written entirely from scratch using Python's standard library.
Zero frameworks. Zero external dependencies. Pure sockets and bytes.
I built this project to deeply understand the mechanics of network programming, concurrency, and the HTTP protocol. By bypassing high-level abstractions like Flask or FastAPI, I engineered the underlying systems that power them—from raw TCP socket management to regex-based routing and thread-safe rate limiting.
-
Zero-Copy Routing (Regex Trie): Implemented an ergonomic, Flask-style decorator API (
@router.get) backed by compiled Regular Expressions with named capture groups for$O(1)$ path parameter extraction. -
Thread-Safe Concurrency: Utilizes a
ThreadPoolExecutorto handle concurrent connections without blocking the main event loop, carefully managing shared state (like metrics) usingthreading.Lock(). -
Sliding Window Rate Limiter: Built a highly accurate rate limiting engine using
$O(1)$ collections.dequeoperations to protect against burst traffic and simple DDoS attempts. - Memory-Bounded Parsing: The custom HTTP protocol parser enforces strict buffer limits (4MB) to prevent Out-Of-Memory (OOM) attacks from malicious clients streaming infinite headers.
-
Path Traversal Protection: Implemented secure static file serving that aggressively sanitizes file paths, preventing
../directory traversal vulnerabilities.
To validate the server's concurrency model, I built a custom, multithreaded load-testing tool (benchmarks/load_test.py).
Despite being a low-level engine, the API is designed to be highly ergonomic and familiar to developers used to modern micro-frameworks.
from core.server import HTTPServer
from core.router import Router
from core.request import Request
from core.response import Response
router = Router()
# 1. Regex Path Parameters
@router.get("/users/{user_id}")
def get_user(request: Request) -> Response:
user_id = request.path_params["user_id"]
return Response.json({"id": user_id, "status": "active"})
# 2. Query Parameters & Headers
@router.post("/echo")
def echo_data(request: Request) -> Response:
return Response.json({
"received_query": request.query_params,
"client_agent": request.headers.get("user-agent", "Unknown")
})
# 3. Start the multithreaded engine
if __name__ == "__main__":
server = HTTPServer(host="127.0.0.1", port=8080, router=router, max_workers=20)
server.start()Infrastructure code requires infrastructure-level testing. The tests/ suite uses Python's standard unittest / pytest to guarantee stability:
- Concurrency Testing: dynamically binds to ephemeral OS ports to launch parallel threads, proving the server won't deadlock under race conditions.
- Protocol Fuzzing: The parser is unit-tested against malformed request lines, missing
\r\nboundaries, and invalid header cases - CI Compatibility: socket-based concurrency tests run in normal local environments and GitHub Actions; restricted sandboxes may block socket creation.
The "Engine" (server.py, protocol.py) knows nothing about the "Application" logic (router.py).