From a03b2259efab05fed12098ad004a0ebda4b9b752 Mon Sep 17 00:00:00 2001 From: Rob Emery Date: Thu, 15 Oct 2026 16:56:38 +0100 Subject: [PATCH] Porting to python so it's easier to implement a full switch-emulator with state --- tools/httpd_sim.py | 504 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 504 insertions(+) create mode 100755 tools/httpd_sim.py diff --git a/tools/httpd_sim.py b/tools/httpd_sim.py new file mode 100755 index 00000000..f785aec2 --- /dev/null +++ b/tools/httpd_sim.py @@ -0,0 +1,504 @@ +#!/usr/bin/env python3 +"""Serve files from the current directory and provide the example device API.""" + +import os +import secrets +import time +import re +from email.parser import BytesParser +from email.policy import default +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from urllib.parse import parse_qs, unquote, urlsplit + +HOST = "0.0.0.0" +PORT = 8081 +PASSWORD = "1234" + +PORTS = 6 +NSFP = 1 +N_COUNTERS = 52 +VLAN_MAX_ID = 4094 +L2_MAX_TRANSFER = 5 + +ROOT = Path.cwd().resolve() + +phys_to_log_port = [] + +if PORTS == 9: + phys_to_log_port = [ 0, 1, 2, 3, 4, 5, 6, 7, 8 ] +else: + phys_to_log_port = [ 4, 5, 6, 7, 3, 8 ] + +l2_entries = [ + { "mac": "00:01:2f:00:00:01", "vlan":"001", "type":'l', "port":7, "idx":0x10 }, + { "mac": "00:06:78:00:00:02", "vlan":"001", "type":'l', "port":7, "idx":0x20 }, + { "mac": "00:01:2e:00:00:03", "vlan":"001", "type":'s', "port":7, "idx":0x30 }, + { "mac": "60:7d:09:00:00:04", "vlan":"001", "type":'l', "port":7, "idx":0x40 }, + { "mac": "1c:2a:a3:00:00:05", "vlan":"001", "type":'l', "port":7, "idx":0x50 }, + { "mac": "1c:2a:a3:00:00:06", "vlan":"001", "type":'s', "port":7, "idx":0x60 }, + { "mac": "02:e0:4c:00:00:07", "vlan":"001", "type":'l', "port":7, "idx":0x70 }, + { "mac": "00:e0:4c:00:00:08", "vlan":"001", "type":'l', "port":4, "idx":0x80 }, + { "mac": "1c:2a:a3:00:00:09", "vlan":"001", "type":'l', "port":9, "idx":0x90 }, # CPU-Port + { "mac": "3c:8c:f8:00:00:0a", "vlan":"001", "type":'l', "port":7, "idx":0xa0 }, + { "mac": "00:01:2e:00:00:0b", "vlan":"001", "type":'l', "port":7, "idx":0xb0 }, + { "mac": "00:01:2f:00:01:01", "vlan":"001", "type":'l', "port":7, "idx":0xc0 } +] + +state = { + "lldp_enabled": False, + "vlans": {1: {"active": True, "name": ""}} +} + +cmd_history = [] +tx_g = [0] * PORTS +rx_g = [0] * PORTS +tx_b = [0] * PORTS +rx_b = [0] * PORTS +last_called = int(time.time()) + +sessions = set() +uploaded_config = None + + +class Handler(SimpleHTTPRequestHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, directory=str(ROOT), **kwargs) + + def api_handlers(self,query): + return { + "/status.json": self._status, + "/eee.json": self._eee, + "/bandwidth.json": self._bandwidth, + "/information.json": self._information, + "/mirror.json": self._mirror, + "/lag.json": self._lag, + "/l2.json": lambda: self._l2(query), + "/mtu.json": self._mtu, + "/vlan.json": lambda: self._vlan(query), + "/vlanlist": self._vlanlist, + "/lldp.json": self._lldp, + "/cmd_log": self._cmd_log, + "/counters.json": self._counters, + } + + def _send(self, status, body=b"", content_type="text/plain", + headers=()): + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + for name, value in headers: + self.send_header(name, value) + self.end_headers() + if body: + self.wfile.write(body) + + def _json(self, value, headers=()): + import json + + body = json.dumps(value, separators=(",", ":")).encode("utf-8") + return self._send( + 200, body, "application/json", headers=headers + ) + + def _redirect(self, location, headers=()): + return self._send( + 302, headers=(("Location", location), *headers) + ) + + def _authenticated(self): + for item in self.headers.get("Cookie", "").split(";"): + name, _, value = item.strip().partition("=") + if name == "session" and value in sessions: + return True + return False + + def _protected(self): + if not self._authenticated(): + self._send(401, b"Unauthorized\n") + return True + return False + + @staticmethod + def _int_param(query, name): + try: + return int(query.get(name, ["0"])[0], 0) + except ValueError: + return 0 + + def _api_response(self, path, query): + handler = self.api_handlers(query).get(path) + return handler() if handler else {} + + def do_GET(self): + global uploaded_config + + parsed = urlsplit(self.path) + path = unquote(parsed.path) + query = parse_qs(parsed.query) + + if path in self.api_handlers(None).keys(): + if self._protected(): + return + result = self._api_response(path, query) + if isinstance(result, tuple): + body, content_type = result + return self._send(200, body, content_type) + return self._json( + result, + headers=(("Cache-Control", "no-cache"),) + if path in ("/status.json", "/counters.json") + else (), + ) + + if path == "/config" and uploaded_config is not None: + if self._protected(): + return + return self._send(200, uploaded_config, "text/plain") + + if not self._authenticated() and path not in ("/login.html", "/favicon.ico"): + return self._redirect("/login.html") + + # Resolve files under ROOT only; reject path traversal. + relative = path.lstrip("/") or "index.html" + target = (ROOT / relative).resolve() + if target != ROOT and ROOT not in target.parents: + return self._send(404, b"Not found\n") + + if not target.is_file(): + return self._send(404, b"Not found\n") + + try: + data = target.read_bytes() + except OSError: + return self._send(404, b"Not found\n") + + import mimetypes + + mime = mimetypes.guess_type(str(target))[0] or "application/octet-stream" + return self._send( + 200, + data, + f"{mime}", + headers=(("Cache-Control", "max-age=60, must-revalidate"),), + ) + + def do_POST(self): + global uploaded_config + + path = urlsplit(self.path).path + + if path == "/login": + form = self._read_urlencoded_form() + if form.get("pwd", [""])[0] == PASSWORD: + token = secrets.token_urlsafe(32) + sessions.add(token) + return self._redirect( + "/index.html", + headers=(( + "Set-Cookie", + f"session={token}; Path=/; SameSite=Strict; HttpOnly", + ),), + ) + return self._redirect("/login.html") + + if self._protected(): + return + + if path == "/cmd": + form = self._read_urlencoded_form() + command = form.get("cmd", [""])[0] + cmd_history.append(command) + + self.parse_cmd(command); + + return self._send( + 200, + b" Upload OK" + b"

Command executed successully

", + "text/html", + ) + + if path in ("/upload", "/config"): + if self.headers.get_content_type() != "multipart/form-data": + return self._send(400, b"Expected multipart/form-data\n") + + parts = self._read_multipart() + if parts is None: + return self._send(400, b"Bad multipart request\n") + + file_data = next( + ( + data for content_type, data in parts + if content_type == "application/octet-stream" + ), + None, + ) + if file_data is None: + return self._send( + 400, b"No application/octet-stream part\n" + ) + + if path == "/config": + uploaded_config = file_data + + return self._send( + 200, + b" Upload OK" + b"

File uploaded successfully

", + "text/html", + ) + + return self._send(404, b"Not found\n") + + def _read_urlencoded_form(self): + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length).decode("utf-8", errors="replace") + return parse_qs(raw) + + def _read_multipart(self): + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) + content_type = self.headers.get("Content-Type", "") + message = BytesParser(policy=default).parsebytes( + b"Content-Type: " + + content_type.encode("utf-8") + + b"\r\nMIME-Version: 1.0\r\n\r\n" + + raw + ) + + if not message.is_multipart(): + return None + + parts = [] + for part in message.iter_parts(): + parts.append(( + part.get_content_type(), + part.get_payload(decode=True) or b"", + )) + return parts + + def parse_cmd(self, cmd): + pieces = cmd.split(" ") + handler = self.get_parsers().get(pieces[0]) + return handler(pieces[1:]) if handler else {} + + def get_parsers(self): + return { + "vlan": self._parse_vlan, + "lldp": self._parse_lldp + } + + def _parse_vlan(self, pieces): #vlan 2 name 6 3 8t or vlan 2 6 3 8t or vlan 2 d + global state + vid = int(pieces[1]); + startIndex=2 + if pieces[2] == "d": + return remove_vlan(vid) + elif re.match(pieces[2], '^[A-Za-z]+$'): + name = pieces[2] + startIndex=3 + + state.vlans[vid] = {"active":True, "name":name, "untagged":[], "tagged":[]} + # list of ports in either 3 or 5t form + for i in pieces[startIndex:]: + if pieces[i][-1] == "t": + state.vlans[vid].tagged.append(pieces[i]) + else: + state.vlans[vid].untagged.append(pieces[i]) + + def _parse_lldp(self,pieces): + global state + state.lldp_enabled = (pieces[0] == "on") + + def _status(self): + global last_called + now = max(int(time.time()), last_called + 1) + elapsed = now - last_called + ports = [] + + for i in range(1, PORTS + 1): + n = i - 1 + if i % 2: + rate = 2_400_000_000 if i == 1 else 950_000_000 + tx_g[n] += rate * elapsed + rx_g[n] += rate * elapsed + tx_b[n] += rate * elapsed // 10_000_000 + rx_b[n] += rate * elapsed // 10_000_000 + + item = { + "portNum": i, + "logPort": phys_to_log_port[n], + "isSFP": int(i > PORTS - NSFP), + "enabled": int(i % 4 != 0), + "link": (5 if i == 1 else 2) if i % 2 else 0, + "txG": f"0x{tx_g[n]:016x}", + "txB": f"0x{tx_b[n]:016x}", + "rxG": f"0x{rx_g[n]:016x}", + "rxB": f"0x{rx_b[n]:016x}", + } + + if i >= PORTS - NSFP: + item.update({ + "sfp_vendor": "OEM", + "sfp_model": "10G-SFP+", + "sfp_serial": "12345678", + "sfp_options": "0x68", + "sfp_temp": f"0x{0x28fb + secrets.randbelow(100):04x}", + "sfp_vcc": f"0x{0x7eda + secrets.randbelow(100):04x}", + "sfp_txbias": f"0x{0x0d24 + secrets.randbelow(100):04x}", + "sfp_txpower": f"0x{0x14bd + secrets.randbelow(100):04x}", + "sfp_rxpower": "0x0000", + "sfp_laser": "0x0000", + }) + else: + item["adv"] = ( + "100000" if i == 1 + else "000011" if i == 2 + else "000100" + ) + + ports.append(item) + + last_called = now + return ports + + def _eee(self): + return [ + { + "portNum": i, + "isSFP": int(i >= 5), + "eee": "00000010", + "eee_lp": "00000100", + "active": int(i % 2 == 1), + } + for i in range(1, PORTS + 1) + ] + + def _bandwidth(self): + ports = [] + for i in range(1, PORTS + 1): + ingress_limited = i % 2 == 1 + egress_limited = i % 4 != 0 + ports.append({ + "portNum": i, + "iLimited": int(ingress_limited), + "iFC": int(i % 4 != 0) if ingress_limited else 0, + "iBW": f"{0x100 if ingress_limited else 0xfffff:08x}", + "eLimited": int(egress_limited), + "eBW": f"{0x1000 if egress_limited else 0xfffff:08x}", + }) + return ports + + def _mirror(self): + return { + "mPort": 1, + "enabled": 1, + "mirror_tx": f"{0b000110:016b}", + "mirror_rx": f"{0b000010:016b}", + } + + def _lag(self): + return [ + { + "lagNum": i, + "members": f"{(0x30 << (i * 2)) & 0x1f8:016b}", + "hash": "0x7e", + } + for i in range(4) + ] + + def _mtu(self): + return [ + { + "portNum": i, + "mtu": f"0x{16383 if i % 2 else 1522:04x}", + } + for i in range(1, PORTS + 1) + ] + + def _lldp(self): + return { + "on": 1, + "portStatus": [ + { + "portNum": i, + "logPort": i - 1, + "name": "", + "permitted": 1, + } + for i in range(1, PORTS + 1) + ], + } + + def _counters(self): + return [f"0x{0x1234 + i:016x}" for i in range(N_COUNTERS)] + + def _vlanlist(self): + global state + vlan_list = [] + for i in state.get("vlans").keys(): + if i in state.get("vlans") and state.get("vlans")[i]["active"]: + vlan_list.append({ + "id": i, + "name": state.get("vlans")[i]["name"], + }) + return {"mgmt": 1, "vlan": vlan_list} + + def _vlan(self,query): + global state + vlan_id = self._int_param(query, "vid") + if vlan_id == 1: + return { + "members": "0x0207ffff", + "name": "native", + "pvid": "0x01ff", + } + + vlan = state.get("vlans").get(vlan_id) + if vlan and vlan.get("active"): + return { + "name": vlan["name"], + "members": "0x0000", + "pvid": "0x0000", + } + return {} + + def _l2(self,query): + idx = self._int_param(query, "idx") + entries = sorted(l2_entries, key=lambda entry: entry["idx"]) + start = next( + (n for n, entry in enumerate(entries) if entry["idx"] >= idx), + len(entries), + ) + result = [] + for entry in entries[start:start + L2_MAX_TRANSFER]: + result.append({ + "mac": entry["mac"], + "vlan": entry["vlan"], + "type": "l" if entry.get("type") == "l" else "s", + "port": entry["port"], + "idx": f"{entry['idx']:04x}", + }) + return result + + def _cmd_log(self): + text = "\n".join(cmd_history) + if cmd_history: + text += "\n" + return text.encode("utf-8"), "text/plain" + + def _information(self): + return { + "ip_address":"192.168.10.247", + "ip_gateway":"192.168.2.22", + "ip_netmask":"255.255.255.0", + "mac_address":"1c:2a:a3:23:00:02", + "sw_ver":"0.1-sim", + "hw_ver":"SWGT024-V2.0" + } + + +if __name__ == "__main__": + print(f"Serving {ROOT} on http://{HOST}:{PORT}") + ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()