from http.server import BaseHTTPRequestHandler, HTTPServerfrom urllib.parse import parse_qs, urlparseimport json

USERS = {
1001: {“name”: “Alice”, “role”: “member”},
1002: {“name”: “Bob”, “role”: “admin”},
}
GROUPS = {
1: {“name”: “Demo Group”, “admins”: {1002}, “members”: {1001, 1002}, “banned”: {1001}}
}

PAGE = “””<!doctype html>
<html><head><meta charset=”utf-8"><title>Security Lab</title>
<style>body{font-family:system-ui;max-width:760px;margin:40px auto;padding:0 16px}
button{margin:4px;padding:8px}pre{background:#eee;padding:12px}</style></head>
<body><h1>Telegram-style Security Lab</h1>
<p>Fake local environment only.</p>
<button onclick=”call(‘/status?user_id=1001’)”>Check banned user</button>
<button onclick=”call(‘/vulnerable/unban?group_id=1&user_id=1001’)”>Run vulnerable PoC</button>
<button onclick=”call(‘/secure/unban?group_id=1&user_id=1001’)”>Try secure endpoint</button>
<pre id=”out”></pre>
<script>
async function call(u){let r=await fetch(u); document.querySelector(‘#out’).textContent=await r.text()}
</script></body></html>”””

def result(ok, message, **extra):
return {“ok”: ok, “message”: message, **extra}

class Handler(BaseHTTPRequestHandler):
def send_json(self, data, code=200):
raw = json.dumps(data, indent=2).encode()
self.send_response(code)
self.send_header(“Content-Type”, “application/json”)
self.send_header(“Content-Length”, str(len(raw)))
self.end_headers()
self.wfile.write(raw)

def do_GET(self):
p = urlparse(self.path)
q = parse_qs(p.query)

if p.path == “/”:
raw = PAGE.encode()
self.send_response(200)
self.send_header(“Content-Type”, “text/html; charset=utf-8”)
self.send_header(“Content-Length”, str(len(raw)))
self.end_headers()
self.wfile.write(raw)
return

Get Divyaanand’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

if p.path == “/status”:
uid = int(q.get(“user_id”, [“0”])[0])
self.send_json(result(True, “status”, user_id=uid,
banned=uid in GROUPS[1][“banned”]))
return

if p.path in (“/vulnerable/unban”, “/secure/unban”):
gid = int(q.get(“group_id”, [“0”])[0])
target = int(q.get(“user_id”, [“0”])[0])

if gid not in GROUPS:
self.send_json(result(False, “unknown group”), 404)
return

if p.path == “/vulnerable/unban”:
# INTENTIONALLY VULNERABLE:
# It checks that the target exists, but never verifies
# that the requester is an administrator.
GROUPS[gid][“banned”].discard(target)
self.send_json(result(
True,
“VULNERABLE: target was unbanned without authorization”,
target=target
))
return

# SECURE PATTERN:
# A real application must derive requester identity from a
# trusted authenticated session/token, not from user input.
requester = int(q.get(“requester_id”, [“1001”])[0])
if requester not in GROUPS[gid][“admins”]:
self.send_json(result(False, “forbidden: admin required”), 403)
return
GROUPS[gid][“banned”].discard(target)
self.send_json(result(True, “unbanned by authorized admin”,
requester=requester, target=target))
return

self.send_json(result(False, “not found”), 404)

def log_message(self, fmt, *args):
pass

if __name__ == “__main__”:
print(“Security lab running at http://127.0.0.1:8000")
HTTPServer((“127.0.0.1”, 8000), Handler).serve_forever()