Foundationপ্রথম নীতি থেকে
LEVEL 10অ্যাডভান্সড~২৮ ঘণ্টাPython

Hardened HTTP Server

Hardened HTTP Server

একটা bare, দুর্বল Python HTTP server থেকে শুরু করে ধাপে ধাপে TLS, input validation, rate limiting, security headers, আর process-level sandboxing বসিয়ে একটা প্রকৃত hardened server বানানো — module-এর ২০টা লেসনের প্রায় সবগুলো mitigation একটাই কোডবেসে একসাথে প্রয়োগ করে।

মাইলস্টোন

আগে যা পড়া দরকার

কেন এই প্রজেক্ট

উনিশ-কুড়িটা লেসন জুড়ে module-টা একই কাঠামো বারবার চালিয়েছে — একটা আক্রমণ, একটা mitigation, আলাদা আলাদা কোডে। বাস্তব জীবনে এই আক্রমণগুলো কখনো একটার পর একটা সাজানো আসে না — একটা সিঙ্গেল প্রোডাকশন HTTP server-কেই একই সময়ে TLS downgrade, SQL injection, credential-stuffing burst, XSS, আর একটা compromised process থেকে host-এ pivot করার চেষ্টা — সবকিছুর বিরুদ্ধেই টিকে থাকতে হয়। এই প্রজেক্টের কাজ ঠিক সেই সংশ্লেষণ — একটাই কোডবেসে, একটার পর একটা স্তর বসিয়ে, প্রতিটা স্তর বসানোর আগে আর পরে সত্যিই আক্রমণ চালিয়ে দেখা যে স্তরটা আসলে কী থামায়, কী থামায় না।

শুরুটা ইচ্ছাকৃতভাবে বাজে — একটা raw string-concatenated SQL query, কোনো TLS নেই, কোনো rate limit নেই, root-এর মতোই unrestricted process। প্রতিটা পরের ধাপ সেই একই server-কে একটা নির্দিষ্ট trust boundary-তে শক্ত করে — আর প্রতিটা ধাপের পরেই আগের ধাপের আক্রমণ আবার চালিয়ে প্রমাণ করা হবে সেটা এখন ব্যর্থ হচ্ছে। শেষে secure-coding-review লেসনের trust-boundary checklist দিয়ে নিজের বানানো সার্ভারটাকেই review করা হবে — এটাই সবচেয়ে সৎ পরীক্ষা, কারণ এখানে কোনো instructor-এর বসানো “ইচ্ছাকৃত bug” নেই, শুধু আপনার নিজের সাত ধাপের সিদ্ধান্ত।

যা সত্যিই চালানো হয়েছে, যা ডিজাইন হিসেবে দেওয়া। এই লেসনের প্রতিটা Python কোড-ব্লক আসলে চালিয়ে তার real output নেওয়া হয়েছে (macOS/Darwin পরিবেশে) — server চালু করে curl/openssl দিয়ে সত্যিকারের request পাঠানো হয়েছে। ব্যতিক্রম একটাই — লাইভ kernel-attached seccomp-bpf filter, যার জন্য Linux ও prctl()/seccomp(2) লাগে, এই এজেন্ট-পরিবেশে সম্ভব না (sandboxing-isolation লেসনেও এই একই সীমা স্বীকার করা হয়েছে)। সেই অংশে syscall-decision logic-টা একটা সিমুলেশন হিসেবে সত্যিই চালানো হয়েছে, আর কোথায় সেটা লাইভ filter থেকে আলাদা তা স্পষ্টভাবে লেবেল করা আছে।

ধাপে ধাপে

১. Bare minimal server — ইচ্ছাকৃতভাবে দুর্বল

শুরুর বিন্দু — stdlib-এর http.server দিয়ে দুইটা route, একটা sqlite3-backed “notes” টেবিল, আর injection-attacks লেসনের সেই একই ভুল — user input সরাসরি SQL query-string-এ concatenate করা:

# step1_server.py -- bare minimal HTTP server, DELIBERATELY vulnerable
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

DB = sqlite3.connect(":memory:", check_same_thread=False)
DB.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, title TEXT, body TEXT)")
DB.executemany(
    "INSERT INTO notes (title, body) VALUES (?, ?)",
    [
        ("shopping", "eggs, milk, bread"),
        ("secret-plan", "launch codes: 4471-9921"),
        ("todo", "renew TLS cert, patch server"),
    ],
)
DB.commit()

class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        pass

    def do_GET(self):
        parsed = urlparse(self.path)
        if parsed.path == "/ping":
            self._send(200, "pong")
            return
        if parsed.path == "/notes":
            qs = parse_qs(parsed.query)
            title = qs.get("title", [""])[0]
            # VULNERABLE: raw string concatenation -- ধাপ ৩-এ ফিক্স হবে
            query = f"SELECT id, title, body FROM notes WHERE title = '{title}'"
            try:
                rows = DB.execute(query).fetchall()
            except sqlite3.Error as e:
                self._send(500, f"db error: {e}")
                return
            self._send(200, str(rows))
            return
        self._send(404, "not found")

    def _send(self, code, text):
        body = text.encode()
        self.send_response(code)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

if __name__ == "__main__":
    HTTPServer(("127.0.0.1", 8091), Handler).serve_forever()

সার্ভার চালিয়ে সত্যিই curl দিয়ে হিট করা হলো:

$ python3 step1_server.py &
$ curl -s http://127.0.0.1:8091/ping
pong
$ curl -s "http://127.0.0.1:8091/notes?title=shopping"
[(1, 'shopping', 'eggs, milk, bread')]
$ curl -s "http://127.0.0.1:8091/notes?title=%27%20OR%20%271%27%3D%271"
[(1, 'shopping', 'eggs, milk, bread'), (2, 'secret-plan', 'launch codes: 4471-9921'), (3, 'todo', 'renew TLS cert, patch server')]

শেষ request-এ title প্যারামিটার ছিল ' OR '1'='1 — injection-attacks লেসনের একেবারে সেই ক্লাসিক payload। secret-plan note-টা কোনো authentication ছাড়াই leak হয়ে গেল। এই একটা vulnerable endpoint-ই বাকি সাত ধাপের ভিত্তি — প্রতিটা পরের ধাপ এই একই server-কে একটা নতুন দিক থেকে শক্ত করবে।

Checkpoint: সার্ভার নিজে চালিয়ে দেখুন /notes?title=shopping একটা row ফেরত দেয়, আর ' OR '1'='1 payload তিনটা row-ই ফেরত দেয় — leak-টা নিজের চোখে দেখা জরুরি, পরের ধাপে এই ঠিক এই আচরণটাই বন্ধ হবে তার প্রমাণ দরকার হবে।

২. TLS termination — ECDHE-only, forward-secret

tls-security লেসনের কেন্দ্রীয় শিক্ষা — ephemeral (EC)DHE ছাড়া কোনো session ভবিষ্যতের key-leak থেকে নিরাপদ না। প্রথমে একটা self-signed EC certificate বানানো হলো (real command, real চালানো):

openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:prime256v1 \
  -keyout server.key -out server.crt -days 365 -nodes \
  -subj "/CN=localhost"
$ openssl x509 -in server.crt -noout -text | grep -E "Signature Algorithm|Public Key Algorithm"
Signature Algorithm: ecdsa-with-SHA256
    Public Key Algorithm: id-ecPublicKey
    Signature Algorithm: ecdsa-with-SHA256

এবার Python-এর ssl module দিয়ে সেই cert socket-এর ওপর wrap করা — minimum_version দিয়ে TLS 1.0/1.1 বাদ, আর set_ciphers দিয়ে শুধু ECDHE-suite অনুমোদন (static RSA key-transport সম্পূর্ণ বাদ, ঠিক forward-secrecy অংশের experiment যা প্রমাণ করেছিল সেই যুক্তি অনুসরণ করে):

# step2_tls_server.py -- ECDHE-only, TLS 1.0/1.1 disabled
import ssl, socket

def make_context():
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(certfile="server.crt", keyfile="server.key")
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2   # SSLv3/TLS1.0/1.1 বাদ
    # ECDHE-only, forward-secret suite -- static RSA key-transport কখনো না
    ctx.set_ciphers("ECDHE+AESGCM:ECDHE+CHACHA20:!aNULL:!eNULL:!RSA")
    ctx.options |= ssl.OP_NO_COMPRESSION
    return ctx

def serve():
    ctx = make_context()
    raw = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    raw.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    raw.bind(("127.0.0.1", 8443))
    raw.listen(5)
    with ctx.wrap_socket(raw, server_side=True) as tls_sock:
        while True:
            try:
                conn, addr = tls_sock.accept()
            except (ssl.SSLError, OSError):
                continue
            with conn:
                conn.recv(4096)
                conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\npong\n")

if __name__ == "__main__":
    serve()

সার্ভারটা সত্যিই চালিয়ে openssl s_client দিয়ে negotiate করা cipher যাচাই করা হলো:

$ python3 step2_tls_server.py &
$ echo -e "GET / HTTP/1.0\r\n\r\n" | openssl s_client -connect 127.0.0.1:8443 2>/dev/null | grep -E "Protocol|Cipher"
New, TLSv1.2, Cipher is ECDHE-ECDSA-AES256-GCM-SHA384
Protocol: TLSv1.2
    Protocol  : TLSv1.2
    Cipher    : ECDHE-ECDSA-AES256-GCM-SHA384

Cipher is ECDHE-ECDSA-AES256-GCM-SHA384 — negotiated suite-এর নামের মধ্যেই ECDHE আছে, মানে ephemeral key exchange, forward secrecy guaranteed। এবার একই server-এর বিরুদ্ধে জোর করে TLS 1.1 চালানোর চেষ্টা:

$ echo -e "GET / HTTP/1.0\r\n\r\n" | openssl s_client -connect 127.0.0.1:8443 -tls1_1 2>&1 | grep error
error:0A0000BF:SSL routines:tls_setup_handshake:no protocols available

minimum_version = TLSv1_2 সেট করার ফল হাতেনাতে — handshake শুরুই হতে পারল না। POODLE-জাতীয় downgrade attack-এর সবচেয়ে সহজ প্রতিরোধ হলো প্রথমে দুর্বল version-টাকেই server-side অফার-তালিকা থেকে বাদ দেওয়া।

Reverse-proxy বিকল্প। Production-এ প্রায়ই এই TLS-termination কাজটা application কোড না করে nginx/Caddy করে। সমতুল্য nginx config-এ ssl_protocols TLSv1.2 TLSv1.3; আর ssl_ciphers ECDHE... দিয়ে ঠিক একই দুইটা সিদ্ধান্ত (পুরনো version বাদ, শুধু ECDHE suite) নেওয়া হয় — নীতিটা এক, শুধু enforcement-এর জায়গা বদলায়। এই প্রজেক্টে application-স্তরেই রাখা হয়েছে যাতে সিদ্ধান্তগুলো কোডে দৃশ্যমান থাকে।

Checkpoint: openssl s_client-এর আউটপুটে Cipher লাইনে ECDHE থাকা নিশ্চিত করুন, আর -tls1_1/-ssl3 flag দিয়ে হ্যান্ডশেক ব্যর্থ হওয়া নিজের চোখে দেখুন।

৩. Input validation ও injection fix

আসল fix — injection-attacks লেসনের ঠিক সেই parameterized-query প্যাটার্ন, plus একটা validation middleware-স্তর যা handler-এ পৌঁছানোর আগেই malformed input প্রত্যাখ্যান করে:

# step3_fix_and_validate.py
import re

TITLE_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$")

def validate_title(title):
    if not isinstance(title, str) or not TITLE_RE.match(title):
        raise ValueError("invalid title: length 1-64, only [A-Za-z0-9._-]")
    return title

def lookup_safe(title):
    validate_title(title)                       # ১ম গেট: format/length
    return DB.execute(                            # ২য় গেট: parameterized query
        "SELECT id, title, body FROM notes WHERE title = ?", (title,)
    ).fetchall()

দুইটা গেটই আলাদাভাবে test করা হলো — একই ' OR '1'='1 payload আগের ধাপের naive query-তে যা করেছিল, এবার তার বিরুদ্ধে:

== আগে (naive, ধাপ ১-এর কোড) ==
  lookup_naive("' OR '1'='1") -> [(1, 'shopping', ...), (2, 'secret-plan', ...), (3, 'todo', ...)]
  leak হলো: True

== পরে (validation + parameterized query) ==
  lookup_safe("' OR '1'='1") -> ValueError: invalid title: length 1-64, only [A-Za-z0-9._-]

== validation middleware -- normal ও oversized input ==
  lookup_safe('shopping')      -> [(1, 'shopping', 'eggs, milk, bread')]
  lookup_safe('x'*100)         -> ValueError: invalid title: length 1-64, only [A-Za-z0-9._-]

লক্ষ্য করুন প্রথম গেটটাই (regex) এই নির্দিষ্ট payload আটকে দিয়েছে, কারণ payload-এ apostrophe আছে যা allowlist-এ নেই। কিন্তু validation কোনো structural গ্যারান্টি না — regex লেখায় ভুল হতে পারে, edge case miss হতে পারে। তাই আসল প্রশ্ন: validation বাদ দিয়ে শুধু parameterized query একাই কি যথেষ্ট? সরাসরি চালিয়ে দেখা হলো:

== structural প্রমাণ: validation বাদ দিয়ে শুধু parameterization-ই যথেষ্ট কিনা ==
  lookup_safe_no_validation("' OR '1'='1") -> []
  leak হলো: False

হ্যাঁ — parameterized query একাই যথেষ্ট, কারণ injection-attacks লেসনের misconception-অংশের যুক্তি এখানেও খাটে: ? placeholder-এ বসা মান কখনোই SQL syntax হিসেবে parse হয় না, escaping লাগে না। তাহলে validation middleware কেন রাখা হলো? কারণ defense in depth — parameterization SQL-চ্যানেলটা বন্ধ করে, কিন্তু length/format validation আরও বিস্তৃত ক্ষতি ঠেকায় (oversized payload দিয়ে memory চাপ, malformed input দিয়ে অন্য কোনো downstream consumer-এর ওপর চাপ) — দুইটা ভিন্ন boundary-র ভিন্ন প্রশ্ন, একটা আরেকটার বিকল্প না।

Checkpoint: দুইটা আলাদা test চালান — lookup_safe (validation + parameterization) আর lookup_safe_no_validation (শুধু parameterization) — দুইটাতেই payload ব্যর্থ হওয়া উচিত, কিন্তু ব্যর্থ হওয়ার কারণ (ValueError বনাম খালি ফলাফল) আলাদা — সেই পার্থক্যটা ব্যাখ্যা করতে পারা মানেই দুইটা গেট আসলে কী আলাদাভাবে করছে তা বোঝা।

৪. Rate limiting — token bucket

এখনো পর্যন্ত সার্ভার একজন client-কে সীমাহীন request পাঠাতে দেয় — credential-stuffing বা resource-exhaustion-এর জন্য এটাই খোলা দরজা। একটা real token-bucket limiter, per-client-IP:

# step4_server.py -- token-bucket rate limiter, প্রতিটা client-এর নিজস্ব bucket
import time, threading
from http.server import BaseHTTPRequestHandler, HTTPServer

class TokenBucket:
    def __init__(self, capacity, refill_per_sec):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_per_sec = refill_per_sec
        self.last = time.monotonic()
        self.lock = threading.Lock()

    def allow(self):
        with self.lock:
            now = time.monotonic()
            self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.refill_per_sec)
            self.last = now
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

buckets = {}
LOCK = threading.Lock()

def allowed(ip):
    with LOCK:
        b = buckets.setdefault(ip, TokenBucket(5, 2))   # capacity=5, refill=2/sec
    return b.allow()

class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        pass

    def do_GET(self):
        if not allowed(self.client_address[0]):
            body = b"429 Too Many Requests\n"
            self.send_response(429)
            self.send_header("Retry-After", "1")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        body = b"pong\n"
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

if __name__ == "__main__":
    HTTPServer(("127.0.0.1", 8092), Handler).serve_forever()

সার্ভার চালিয়ে সত্যিই আট-টা পরপর curl request পাঠানো হলো:

$ python3 step4_server.py &
$ for i in $(seq 1 8); do
    curl -s -o /dev/null -w "request #$i -> HTTP %{http_code}\n" http://127.0.0.1:8092/
  done
request #1 -> HTTP 200
request #2 -> HTTP 200
request #3 -> HTTP 200
request #4 -> HTTP 200
request #5 -> HTTP 200
request #6 -> HTTP 429
request #7 -> HTTP 429
request #8 -> HTTP 429

capacity=5 মানে প্রথম পাঁচটা burst-এ চলে যায় (bucket ভর্তি ছিল), তারপরের প্রতিটাই 429 — refill rate (২ token/sec)-এর চেয়ে দ্রুত এলে প্রত্যাখ্যাত। এক সেকেন্ড অপেক্ষা করে আবার পাঠালে bucket আংশিক ভরে 200 ফিরে আসে (পুরোপুরি আলাদা Python session-এ, pure token-bucket টেস্টে, যাচাই করা হয়েছিল)।

Rate limiter পাইপলাইনে কোথায় বসবে? দুইটা ভিন্ন স্তরে দুইটা ভিন্ন কাজ:

  • Authentication-এর আগে, একটা coarse per-IP limiter — এই প্রজেক্টেই বসানো হলো, do_GET-এর একেবারে শুরুতে, এমনকি request parse হওয়ার আগেই। উদ্দেশ্য সস্তা — unauthenticated flood বা connection-exhaustion ঠেকানো, তাই এটাকে যতটা সম্ভব সস্তা আর যতটা সম্ভব আগে রাখা হয়। এই স্তর ব্যর্থ (bypass) হলেও পরের স্তর এখনও আছে।
  • Authentication-এর পরে, একটা stricter per-account limiter — credential-stuffing-এর মতো আক্রমণে attacker হাজারটা ভিন্ন IP থেকে একই username-এ password-guess চালাতে পারে; per-IP limiter সেখানে অকার্যকর, কারণ প্রতিটা IP নিজে নিজে limit-এর নিচে থাকে। per-username বা per-account limiter (login attempt counter) সেই gap ভরাট করে।

দুইটা একে অপরের বিকল্প না — প্রথমটা সস্তা, সবার জন্য, resource-protection; দ্বিতীয়টা দামী (state track করতে হয়), নির্দিষ্ট, identity-protection।

Checkpoint: নিজের burst-test চালিয়ে অন্তত একটা 429 দেখুন, তারপর Retry-After header অনুযায়ী অপেক্ষা করে পরের request 200 ফিরে আসা নিশ্চিত করুন।

৫. Security headers ও session hardening

xss-csrf-ssrf আর authentication লেসনের mitigation-গুলো একসাথে বসানো — প্রতিটা response-এ একগুচ্ছ hardening header, আর session cookie-তে তিনটা flag:

# step5_headers.py -- security headers + hardened session cookie
SECURITY_HEADERS = {
    "Content-Security-Policy": "default-src 'self'; script-src 'self'; frame-ancestors 'none'",
    "X-Content-Type-Options": "nosniff",
    "X-Frame-Options": "DENY",
    "Referrer-Policy": "no-referrer",
    "Strict-Transport-Security": "max-age=63072000; includeSubDomains",
    "Permissions-Policy": "geolocation=(), microphone=(), camera=()",
}

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = b"pong\n"
        self.send_response(200)
        for k, v in SECURITY_HEADERS.items():
            self.send_header(k, v)
        self.send_header(
            "Set-Cookie",
            "session=8f3a...e21c; HttpOnly; Secure; SameSite=Lax; Max-Age=3600",
        )
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

সার্ভার চালিয়ে http.client দিয়ে সত্যিই headers পড়া হলো:

HTTP 200
  Content-Security-Policy: default-src 'self'; script-src 'self'; frame-ancestors 'none'
  X-Content-Type-Options: nosniff
  X-Frame-Options: DENY
  Referrer-Policy: no-referrer
  Strict-Transport-Security: max-age=63072000; includeSubDomains
  Permissions-Policy: geolocation=(), microphone=(), camera=()
  Set-Cookie: session=8f3a...e21c; HttpOnly; Secure; SameSite=Lax; Max-Age=3600
  Content-Length: 5

প্রতিটা header একটা নির্দিষ্ট trust boundary রক্ষা করে, module-এর আগের লেসনের ভাষায়:

Header/flagকী ঠেকায়কেন
Content-Security-PolicyXSS-এর blast radiusscript-এর উৎস 'self'-এ সীমিত করলে inject করা <script> external payload load করতে পারে না (xss-csrf-ssrf লেসন)
X-Content-Type-Options: nosniffMIME-sniffing-ভিত্তিক XSSbrowser response-এর declared Content-Type বাইপাস করে “অনুমান” করে ফাইল টাইপ ঠিক করতে পারবে না
X-Frame-Options: DENYClickjackingসাইটটা অন্য কোনো origin-এর <iframe>-এ embed হতে পারবে না
Strict-Transport-SecurityDowngrade-to-plain-HTTPbrowser-কে বাধ্য করে ভবিষ্যতের প্রতিটা visit HTTPS-এই যেতে, প্রথম MITM-redirect-টাও আটকায়
HttpOnly (cookie)XSS দিয়ে session চুরিdocument.cookie-তে JavaScript থেকে session ID পড়া যায় না (authentication লেসন)
Secure (cookie)plaintext-এ session leakcookie শুধু TLS connection-এ পাঠানো হয়, ধাপ ২-এর TLS ছাড়া এই flag-এর কোনো অর্থই থাকত না
SameSite=Lax (cookie)CSRFcross-site request-এ cookie সংযুক্তই হয় না — xss-csrf-ssrf লেসনের CSRF-defense দুই-স্তরের একটা

লক্ষ্য করুন Secure flag-এর সারিটা — এটা প্রমাণ করে কেন ধাপগুলোর ক্রম গুরুত্বপূর্ণ। ধাপ ২ (TLS) ছাড়া Secure flag বসানো অর্থহীন হতো, কারণ কানেকশনটাই কখনো HTTPS হতো না।

Checkpoint: headers প্রিন্ট করে প্রতিটা row ওপরের টেবিলের সাথে নিজে মিলিয়ে দেখুন — কোনো একটা header বাদ দিলে ঠিক কোন আক্রমণটা আবার খুলে যায় তা বলতে পারা এই ধাপের আসল পরীক্ষা।

৬. Process-level sandboxing

এতক্ষণ প্রতিটা স্তর request-কে target করেছে — এই ধাপ process নিজেকে target করে। sandboxing-isolation লেসনের ভাষায়: namespace/cgroup কী দেখা যায় আর কতটুকু resource ব্যবহার হয় তা নিয়ন্ত্রণ করে, কিন্তু কোন syscall চালানো যায় তা না — আর ঠিক সেই ফাঁকটাই আজ বন্ধ করা হবে, সেই সাথে root-privilege আর resource-ceiling।

Non-root user-এ চালানো। Production deployment-এ সার্ভার প্রসেসটা কখনোই root-এ চলবে না — একটা dedicated, unprivileged system user (useradd --system --no-create-home hardened-http) বানিয়ে, systemd unit বা Docker USER directive দিয়ে সেই UID-তে exec করা হয়:

# systemd unit-এর অংশ (বা Dockerfile-এর USER directive)
User=hardened-http
Group=hardened-http
NoNewPrivileges=true

NoNewPrivileges=true ঠিক sandboxing-isolation লেসনের সেই prctl(PR_SET_NO_NEW_PRIVS, 1) — SUID-binary exec করে privilege escalate করার পথ বন্ধ করে দেয়।

Resource ceiling — সত্যিই চালিয়ে verify করা। resource.setrlimit POSIX-স্ট্যান্ডার্ড, macOS/Linux দুটোতেই সত্যিই কাজ করে — এই অংশটা আসলেই চালানো হয়েছে:

# step6_rlimit.py -- server startup-এ resource ceiling
import resource, os

print(f"current UID   : {os.getuid()}")
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (64, hard))   # server-এর দরকারের চেয়ে অনেক কম

handles = []
try:
    for i in range(200):
        handles.append(open(os.devnull))
except OSError as e:
    print(f"{i} নম্বর open()-এ ব্যর্থ: {e}")
finally:
    for h in handles:
        h.close()
current UID   : 501
61 নম্বর open()-এ ব্যর্থ: [Errno 24] Too many open files: '/dev/null'

RLIMIT_NOFILE কে ৬৪-তে নামানোর পরে ৬১তম open()-এই process ব্যর্থ হলো — একটা compromised process যদি হাজার হাজার file descriptor খুলে resource-exhaustion চেষ্টা করে, সেই আক্রমণের সিলিং এখন আগে থেকেই বসানো।

seccomp — সততার সাথে লেবেল করা। লাইভ seccomp-bpf filter বসাতে prctl(PR_SET_NO_NEW_PRIVS, 1) আর seccomp(SECCOMP_SET_MODE_FILTER, ...) লাগে, যা শুধু Linux-এ চলে — এই macOS agent sandbox-এ সেটা সত্যিই bind/attach করা সম্ভব না। নিচের policy আর decision-logic real Linux deployment-এর জন্য একটা concrete design — logic-টা এখানে সত্যিই চালানো হয়েছে (Python simulation হিসেবে), শুধু kernel-এ real BPF bytecode হিসেবে installed হয়নি।

সার্ভারের normal flow-এ ঠিক কোন syscall-গুলো লাগে (epoll, socket, TLS-এর জন্য getrandom, ফাইল খোলা) তার একটা allowlist, আর একটা post-RCE attacker যা করার চেষ্টা করত তার বিরুদ্ধে সিমুলেশন:

# seccomp_policy_sim.py -- Linux-এ real seccomp-bpf-এ যেভাবে deploy হবে,
# সেই decision logic-ই এখানে সিমুলেট করা -- macOS agent sandbox-এ লাইভ
# kernel filter bind করা সম্ভব না (sandboxing-isolation লেসনের একই সীমা)
SCMP_ACT_ALLOW = "ALLOW"
SCMP_ACT_KILL  = "KILL_PROCESS"

ALLOWED = {
    "epoll_wait", "epoll_ctl", "accept4", "read", "readv", "recvfrom",
    "write", "writev", "sendto", "openat", "close", "newfstatat", "lseek",
    "mmap", "munmap", "brk", "socket", "bind", "listen", "rt_sigaction",
    "rt_sigprocmask", "futex", "clone", "getrandom", "exit", "exit_group",
}

def decide(name):
    return SCMP_ACT_ALLOW if name in ALLOWED else SCMP_ACT_KILL

normal_flow = ["socket", "bind", "listen", "epoll_wait", "accept4",
               "readv", "getrandom", "openat", "sendto", "close"]
post_rce_attacker = ["read", "mmap", "execve", "ptrace", "mount", "socket"]

print("== normal request/TLS-handshake flow ==")
for sc in normal_flow:
    print(f"  {sc:12s} -> {decide(sc)}")

print("\n== parser-bug দিয়ে RCE পাওয়ার পর pivot করার চেষ্টা ==")
killed = False
for sc in post_rce_attacker:
    if killed:
        print(f"  {sc:12s} -> (never reached, process already dead)")
        continue
    v = decide(sc)
    print(f"  {sc:12s} -> {v}")
    if v == SCMP_ACT_KILL:
        killed = True
== normal request/TLS-handshake flow ==
  socket       -> ALLOW
  bind         -> ALLOW
  listen       -> ALLOW
  epoll_wait   -> ALLOW
  accept4      -> ALLOW
  readv        -> ALLOW
  getrandom    -> ALLOW
  openat       -> ALLOW
  sendto       -> ALLOW
  close        -> ALLOW

== parser-bug দিয়ে RCE পাওয়ার পর pivot করার চেষ্টা ==
  read         -> ALLOW
  mmap         -> ALLOW
  execve       -> KILL_PROCESS
  ptrace       -> (never reached, process already dead)
  mount        -> (never reached, process already dead)
  socket       -> (never reached, process already dead)

read, mmap — দুইটাই normal flow-এরও অংশ, তাই allowed; কিন্তু execve (একটা shell স্পন করার প্রথম পদক্ষেপ) allowlist-এ নেই বলে সঙ্গে সঙ্গে kill, আর তারপরের ptrace/mount/reverse-shell socket কল কখনো পৌঁছায়ই না। এই একই allowlist Linux-এ deploy করতে হলে libseccomp দিয়ে একটা OCI-স্টাইল JSON profile বানিয়ে Docker/podman-এর --security-opt seccomp=profile.json flag-এ পাস করতে হবে — sandboxing-isolation লেসনের build অংশে এই ঠিক এই ফরম্যাটটাই তৈরি করা হয়েছিল।

Capability bounding। সার্ভার যদি port ৮০৯১-এর মতো একটা high port-এ চলে (এই প্রজেক্টে তাই করা হয়েছে), তার কোনো বাড়তি Linux capability লাগে না — bounding set সম্পূর্ণ খালি রাখা যায় (--cap-drop=ALL)। শুধু production-এ port ৪৪৩-এ সরাসরি bind করতে হলে (১০২৪-এর নিচে) CAP_NET_BIND_SERVICE একটামাত্র bit লাগবে — sandboxing-isolation লেসনের সেই ৪১-বিট granularity-র বাস্তব প্রয়োগ, পুরো root না বানিয়ে।

Checkpoint: resource.setrlimit-এর কোড নিজে চালিয়ে limit-এর পরে OSError দেখুন। তারপর seccomp সিমুলেশনে ALLOWED সেট থেকে getrandom সরিয়ে দেখুন — TLS handshake-এর জন্য দরকারি একটা syscall বাদ পড়লে normal flow-ও KILL_PROCESS পায়, যা মনে করিয়ে দেয় allowlist বানানো একটা careful, empirical কাজ, আন্দাজে করার কাজ না।

৭. Final code review pass — নিজের সার্ভারের ওপর trust-boundary checklist

secure-coding-review লেসনের checklist-টা এবার নিজের ছয় ধাপে বানানো সার্ভারের ওপর, লাইনে-লাইনে প্রয়োগ করা হলো — প্রতিটা boundary, আর কী verify করা হয়েছে:

Trust boundaryপ্রশ্নএই সার্ভারে যাচাই
Untrusted input → interpreterডেটা কি syntax-string-এ concatenate হচ্ছে?title parameter-এর প্রতিটা path parameterized query দিয়ে যায় (ধাপ ৩); কোনো f-string SQL অবশিষ্ট নেই — গোটা handler গ্রেপ করে যাচাই করা হয়েছে
Identity claim → পরবর্তী requestSession token সঠিকভাবে verify হচ্ছে?এই প্রজেক্টের scope-এ authentication যোগ করা হয়নি (note lookup পাবলিক) — production version-এ authentication লেসনের session-cookie মডেল যোগ করতে হবে, এটা একটা explicit, নথিভুক্ত সীমা
Client ↔ server চ্যানেলChannel authenticated ও confidential, downgrade-প্রতিরোধী?TLS 1.2+ enforced, ECDHE-only, openssl s_client-এ verify করা হয়েছে (ধাপ ২)
Secret data → storage/transmissionSession cookie ঠিক flag-এ পাঠানো হচ্ছে?HttpOnly+Secure+SameSite=Lax — headers real response-এ দেখা গেছে (ধাপ ৫)
Attacker-controlled size → memoryInput length bound-checked?TITLE_RE-এ hard {1,64} ceiling (ধাপ ৩); RLIMIT_NOFILE দিয়ে file-descriptor exhaustion-এর সিলিং (ধাপ ৬)
Volumetric abuse → availabilityBurst/flood-এর বিরুদ্ধে কোনো সিলিং আছে?per-IP token bucket, 429 real curl-test-এ verified (ধাপ ৪)
Deployed process → হোস্টProcess least-privilege-এ চলছে, syscall/capability সীমিত?non-root user + NoNewPrivileges + resource ceiling সত্যিই চালানো হয়েছে; seccomp allowlist ডিজাইন করা আছে, Linux deployment-এ activate করতে হবে

যা এই review-এ পাওয়া গেছে — আর যা বাকি। Checklist সৎভাবে চালালে একটা ফাঁক স্পষ্ট বেরিয়ে আসে — এই সার্ভারে authentication/authorization boundary নেই, কারণ scope ইচ্ছাকৃতভাবে “note lookup পাবলিক” রাখা হয়েছিল module-এর আগের ছয়টা লেসনের ওপর ফোকাস রাখতে। secure-coding-review লেসনের শিক্ষা এখানেই সবচেয়ে স্পষ্ট — একটা review-এর কাজ শুধু bug খুঁজে ফিক্স করা না, যা এখনো cover হয়নি তা স্পষ্টভাবে বলা। একটা “clean” review যেটা কোনো ফাঁক স্বীকার করে না, সেটা review-ই না।

Checkpoint: নিজে টেবিলের প্রতিটা row-এ গিয়ে আসল কোড খুলে দেখান কোন লাইন সেই যাচাইটা করে — “আমি মনে করি এটা ঠিক আছে” যথেষ্ট না, সঠিক লাইন-নম্বর বা function-নাম উদ্ধৃত করতে পারা।

নিজেকে চ্যালেঞ্জ করুন

  1. একটা WAF-স্টাইল request-pattern filter যোগ করুন — path/query-তে পরিচিত SQLi/XSS payload-শেপ (যেমন ' OR, <script) দেখলে rate limiter-এর আগেই একটা cheap pre-filter হিসেবে block করুন, আর আলোচনা করুন কেন এটা parameterized query-র বিকল্প না, শুধু আরেকটা স্তর
  2. Rate limiter-কে সত্যিকারের load দিয়ে টেস্ট করুন — একটা concurrent client script (threading বা asyncio) দিয়ে শত শত request একসাথে পাঠিয়ে দেখুন token-bucket-এর Lock race condition ছাড়া সঠিক আচরণ করছে কিনা
  3. ফিনিশড সার্ভারটা একটা automated scanner-এ (nikto, OWASP ZAP baseline scan) চালান, ফলাফলের প্রতিটা finding-কে false-positive/ real-gap হিসেবে triage করুন — যা “already covered by ধাপ N” তা উল্লেখ করুন
  4. পুরো সার্ভারটাকে একটা hardened Dockerfile-এ containerize করুন — multi-stage build, non-root USER, --cap-drop=ALL + --cap-add=NET_BIND_SERVICE (শুধু প্রয়োজন হলে), আর ধাপ ৬-এর seccomp JSON profile সত্যিই --security-opt seccomp= দিয়ে Linux-এ চালিয়ে verify করুন
  5. mTLS যোগ করুন — client certificate যাচাই বাধ্যতামূলক করে দেখুন কোন internal-service-to-service সিনারিওতে এটা password-এর চেয়ে ভালো fit
  6. get_note-স্টাইল একটা owner-scoped endpoint যোগ করুন (session + authorization বসিয়ে) — তারপর secure-coding-review লেসনের batch-test প্যাটার্নে IDOR payload-এর একটা সেট চালিয়ে ০/N leak প্রমাণ করুন

যা এখানে শিখলেন, পরে যেখানে লাগবে

এখানেপরে
ECDHE-only, downgrade-resistant TLS configProduction load balancer/reverse-proxy-এর TLS policy লেখা বা audit করা
Parameterized query + validation middleware, একই সময়ে দুই স্তরেযেকোনো ORM/framework-এ raw-query escape-hatch দেখলে সন্দেহপ্রবণ হওয়া
Token-bucket rate limiter, pipeline-এ তার অবস্থানAPI gateway-তে rate-limit policy ডিজাইন, coarse বনাম per-account স্তর আলাদা করা
Security headers + session cookie flag-এর যুক্তিBrowser-facing যেকোনো সার্ভিসে একটা baseline hardening checklist প্রয়োগ
Non-root + rlimit + seccomp allowlist একসাথেContainer/Kubernetes-এ securityContext, seccompProfile, capabilities.drop কনফিগার করা
নিজের কোডে trust-boundary checklist প্রয়োগ, ফাঁক স্বীকার করাপ্রতিটা ভবিষ্যৎ pull-request review-এ habitual reading mode হিসেবে