"""A tiny fake Graylog 4.0 server used by the end-to-end tests. It intentionally mimics the response envelopes of the real 4.0 API (including the ``{"messages": [{"message": {...}}]}`` shape of the legacy search API) so the client code is exercised the same way it will be against edu-master. """ import base64 import json import re import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse TOKEN = "testtoken" EXPECTED_AUTH = "Basic " + base64.b64encode( ("%s:token" % TOKEN).encode("utf-8") ).decode("ascii") VERSION = "4.0.8+6b8c55d" MESSAGES = [ { "timestamp": "2026-10-17T15:00:00.000Z", "source": "10.180.254.17", "client_ip": "203.0.113.5", "status": 401, "request_method": "POST", "request_uri": "/api/login", "http_user_agent": "python-requests/2.31", "response_time": 0.011, "message": '203.0.113.5 - - "POST /api/login HTTP/1.1" 401 0.011', }, { "timestamp": "2026-10-17T15:00:05.000Z", "source": "10.180.254.17", "client_ip": "203.0.113.5", "status": 401, "request_method": "POST", "request_uri": "/api/login", "http_user_agent": "python-requests/2.31", "response_time": 0.009, "message": '203.0.113.5 - - "POST /api/login HTTP/1.1" 401 0.009', }, { "timestamp": "2026-10-17T15:00:10.000Z", "source": "10.180.254.17", "client_ip": "203.0.113.5", "status": 401, "request_method": "POST", "request_uri": "/api/login", "http_user_agent": "python-requests/2.31", "response_time": 0.010, "message": '203.0.113.5 - - "POST /api/login HTTP/1.1" 401 0.010', }, { "timestamp": "2026-10-17T15:00:15.000Z", "source": "10.180.254.17", "client_ip": "198.51.100.7", "status": 401, "request_method": "GET", "request_uri": "/admin", "http_user_agent": "Mozilla/5.0", "response_time": 0.021, "message": '198.51.100.7 - - "GET /admin HTTP/1.1" 401 0.021', }, { "timestamp": "2026-10-17T15:00:20.000Z", "source": "10.180.254.17", "client_ip": "198.51.100.7", "status": 403, "request_method": "GET", "request_uri": "/admin", "http_user_agent": "Mozilla/5.0", "response_time": 0.018, "message": '198.51.100.7 - - "GET /admin HTTP/1.1" 403 0.018', }, { "timestamp": "2026-10-17T15:00:25.000Z", "source": "10.180.254.17", "client_ip": "192.0.2.44", "status": 404, "request_method": "GET", "request_uri": "/wp-login.php", "http_user_agent": "sqlmap/1.7", "response_time": 0.006, "message": '192.0.2.44 - - "GET /wp-login.php HTTP/1.1" 404 0.006', }, { "timestamp": "2026-10-17T15:00:30.000Z", "source": "10.180.254.17", "client_ip": "192.0.2.44", "status": 404, "request_method": "GET", "request_uri": "/.env", "http_user_agent": "sqlmap/1.7", "response_time": 0.005, "message": '192.0.2.44 - - "GET /.env HTTP/1.1" 404 0.005', }, { "timestamp": "2026-10-17T15:00:35.000Z", "source": "10.180.254.17", "client_ip": "10.180.2.9", "status": 200, "request_method": "GET", "request_uri": "/health", "http_user_agent": "kube-probe/1.25", "response_time": 0.002, "message": '10.180.2.9 - - "GET /health HTTP/1.1" 200 0.002', }, ] def _build_judge_messages(): """Synthetic traffic for the judgement tests (brute force / scanner / legit).""" rows = [] def add(ip, count, status, uri, agent, method="GET"): for index in range(count): rows.append({ "timestamp": "2026-10-17T15:%02d:%02d.000Z" % (index // 60, index % 60), "source": "10.180.254.17", "client_ip": ip, "status": status, "request_method": method, "request_uri": uri, "http_user_agent": agent, "message": '%s - "%s %s HTTP/1.1" %d' % (ip, method, uri, status), }) add("203.0.113.66", 40, 401, "/api/login", "python-requests/2.31", "POST") add("192.0.2.99", 25, 404, "/wp-login.php", "sqlmap/1.7") add("198.51.100.200", 50, 200, "/index", "Mozilla/5.0") add("10.180.5.5", 30, 401, "/api/login", "Mozilla/5.0", "POST") return rows JUDGE_MESSAGES = _build_judge_messages() CLAUSE = re.compile(r'^([A-Za-z_][A-Za-z0-9_.]*):(\([^)]*\)|"[^"]*"|[^\s]+)$') def _clause_values(raw): raw = raw.strip() if raw.startswith("(") and raw.endswith(")"): return [part.strip().strip('"') for part in raw[1:-1].split(" OR ") if part.strip()] return [raw.strip('"')] def matches(message, query): """Very small subset of Lucene, enough to exercise the judge signal queries.""" query = (query or "*").strip() if query in ("", "*"): return True clause = CLAUSE.match(query) if clause is None: return True # unsupported syntax: do not filter (test double only) field, raw = clause.groups() value = message.get(field) if value is None: return False text = str(value).lower() for candidate in _clause_values(raw): # Mirror Lucene: `foo*` is a prefix match, `\/` is an escaped slash. needle = candidate.lower().replace("\\/", "/") if needle.endswith("*"): if needle[:-1] and text.startswith(needle[:-1]): return True continue if needle and needle in text: return True return False def aggregate(messages, body): """Minimal views-pivot implementation (one query, one search type).""" query = (body.get("queries") or [{}])[0] query_string = (query.get("query") or {}).get("query_string") or "*" filtered = [message for message in messages if matches(message, query_string)] stream_filter = query.get("filter") stream_ids = set() if isinstance(stream_filter, dict): for item in stream_filter.get("filters") or []: if isinstance(item, dict) and item.get("id"): stream_ids.add(item["id"]) if stream_ids: filtered = [message for message in filtered if set(message.get("streams") or []) & stream_ids] search_type = (query.get("search_types") or [{}])[0] series_id = ((search_type.get("series") or [{}])[0]).get("id", "c1") row_groups = search_type.get("row_groups") or [] column_groups = search_type.get("column_groups") or [] rows = [] total = 0 if row_groups: field = row_groups[0].get("field") limit = int(row_groups[0].get("limit") or 10) counts = {} for message in filtered: value = message.get(field) if value is None: continue counts[str(value)] = counts.get(str(value), 0) + 1 total = sum(counts.values()) ordered = sorted(counts.items(), key=lambda item: (-item[1], item[0]))[:limit] for key, count in ordered: rows.append({"key": [key], "source": "leaf", "values": [{"key": [series_id], "value": count, "rollup": True, "source": "row-leaf"}]}) rows.append({"key": [], "source": "rollup", "values": [{"key": [series_id], "value": total, "rollup": True, "source": "row-leaf"}]}) elif column_groups: buckets = {} for message in filtered: stamp = str(message.get("timestamp", "")) key = stamp[:16] + ":00.000Z" buckets[key] = buckets.get(key, 0) + 1 total = sum(buckets.values()) for key in sorted(buckets): rows.append({"key": [], "source": "leaf", "values": [{"key": [key, series_id], "value": buckets[key], "rollup": False, "source": "col-leaf"}]}) return {"id": search_type.get("id", "st1"), "type": "pivot", "total": total, "rows": rows} class FakeGraylog: def __init__(self, require_auth=True, messages=None, views_enabled=True): self.require_auth = require_auth self.messages = list(MESSAGES if messages is None else messages) self.views_enabled = views_enabled self.terms_size_param = "size" self.requests = [] owner = self class Handler(BaseHTTPRequestHandler): def log_message(self, *args): # silence test output pass def _send(self, code, payload): body = json.dumps(payload).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_POST(self): parsed = urlparse(self.path) path = parsed.path query = {key: values[0] for key, values in parse_qs(parsed.query).items()} length = int(self.headers.get("Content-Length") or 0) raw_body = self.rfile.read(length) if length else b"" owner.requests.append((path, query)) if owner.require_auth and self.headers.get("Authorization") != EXPECTED_AUTH: return self._send(401, {"message": "Unauthorized"}) if not owner.views_enabled: return self._send(404, {"message": "Sorry, Page Not Found"}) if path == "/api/views/search/sync": try: body = json.loads(raw_body or b"{}") except ValueError: return self._send(400, {"message": "invalid json"}) result = aggregate(owner.messages, body) return self._send(200, { "execution": {"done": True, "cancelled": False, "completed_exceptionally": False}, "results": {"q1": {"errors": [], "search_types": {"st1": result}}}, "search_id": "mock-search", }) return self._send(404, {"message": "not found: %s" % path}) def do_GET(self): parsed = urlparse(self.path) path = parsed.path query = {key: values[0] for key, values in parse_qs(parsed.query).items()} owner.requests.append((path, query)) if owner.require_auth and self.headers.get("Authorization") != EXPECTED_AUTH: return self._send(401, {"message": "Unauthorized"}) if path == "/api/system": return self._send(200, { "version": VERSION, "cluster_id": "6b8c55d0-dead-beef-0000-000000000001", "hostname": "edu-master", "is_processing": True, "timezone": "Asia/Shanghai", "locale": "zh_CN", }) if path == "/api/system/cluster/nodes": return self._send(200, {"total": 1, "nodes": [{ "node_id": "node-1", "hostname": "edu-master", "version": VERSION, "lifecycle": "RUNNING", "is_leader": True, "lb_status": "ALIVE", }]}) if path == "/api/system/indexer/cluster/health": return self._send(200, { "status": "green", "cluster_name": "graylog", "number_of_nodes": 1, }) if path == "/api/system/deflector": return self._send(200, {"index": "graylog_0", "is_up": True}) if path == "/api/system/fields": return self._send(200, {"fields": [ {"name": "client_ip", "type": "ip"}, {"name": "status", "type": "long"}, {"name": "request_uri", "type": "string"}, ]}) if path == "/api/system/license": return self._send(200, {"license": {"type": "opensource"}}) if path == "/api/system/inputs": return self._send(200, {"total": 1, "inputs": [{ "id": "5f0000000000000000000001", "title": "GELF UDP", "type": "org.graylog2.inputs.gelf.udp.GELFUDPInput", "global": True, "configuration": {"bind_address": "0.0.0.0", "port": 12201}, }]}) if path == "/api/streams": return self._send(200, {"total": 2, "streams": [ { "id": "5f00000000000000000000aa", "title": "业务访问日志", "description": "nginx access log", "disabled": False, "rules": [], }, { "id": "000000000000000000000001", "title": "Default Stream", "description": "default", "disabled": False, "rules": [], }, ]}) if path == "/api/system/indices/index_sets": return self._send(200, {"total": 1, "index_sets": [{ "id": "5f00000000000000000000bb", "title": "Default index set", "index_prefix": "graylog", "shards": 4, "replicas": 0, "rotation_strategy": "time", "rotation_strategy_configuration": {"type": "time", "period": "P1D"}, "retention_strategy": "delete", "retention_strategy_configuration": {"max_number_of_indices": 10}, }]}) if path == "/api/system/indexer/indices": return self._send(200, {"indices": [{ "index_name": "graylog_0", "index_set_id": "5f00000000000000000000bb", "size": 1048576, "begin": "2026-10-17T00:00:00.000Z", "end": "2026-10-18T00:00:00.000Z", }]}) if path == "/api/events/definitions": return self._send(200, { "total": 1, "count": 1, "page": 1, "per_page": 50, "event_definitions": [{ "id": "5f00000000000000000000cc", "title": "401 风暴", "priority": 3, "disabled": False, "config": {"type": "aggregation-v1"}, "notifications": [], }], }) if path == "/api/events/notifications": return self._send(200, {"total": 1, "notifications": [{ "id": "5f00000000000000000000dd", "title": "ban webhook", "config": {"type": "http-notification-v1", "url": "http://127.0.0.1:8080/ban"}, }]}) if path == "/api/events": return self._send(200, {"total_events": 0, "events": []}) if path == "/api/sidecars": return self._send(200, {"total": 0, "sidecars": []}) if path in ("/api/system/pipelines/pipeline", "/api/system/pipelines/rule"): return self._send(200, []) if path == "/api/views": return self._send(200, {"total": 0, "views": [], "page": 1, "per_page": 50}) if path == "/api/search/universal/relative": try: limit = int(query.get("limit", 100)) except ValueError: limit = 100 try: offset = int(query.get("offset", 0)) except ValueError: offset = 0 matched = [item for item in owner.messages if matches(item, query.get("query", "*"))] page = matched[offset:offset + limit] return self._send(200, { "messages": [{"message": dict(item), "index": "graylog_0"} for item in page], "total_results": len(matched), "time": 3, "built_query": "mock", }) if path == "/api/search/universal/terms": if owner.terms_size_param == "limit" and "size" in query: return self._send(400, {"message": "Parameter 'size' is not allowed"}) field = query.get("field", "source") param = "size" if "size" in query else ("limit" if "limit" in query else None) try: size = int(query.get(param, 50)) if param else 50 except ValueError: size = 50 counts = {} query_string = query.get("query", "*") for message in owner.messages: if not matches(message, query_string): continue value = message.get(field) if value is None: continue counts[str(value)] = counts.get(str(value), 0) + 1 terms = sorted( ({"term": key, "count": value} for key, value in counts.items()), key=lambda row: (-row["count"], row["term"]), )[:size] return self._send(200, { "field": field, "terms": terms, "missing": 0, "other": 0, "total": len(terms), }) if path == "/api/search/universal/histogram": return self._send(200, { "results": { "2026-10-17T15:00:00.000Z": {"count": 3}, "2026-10-17T15:01:00.000Z": {"count": 5}, }, "interval": "1m", "time": 2, }) return self._send(404, {"message": "not found: %s" % path}) self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.port = self.server.server_address[1] self.thread = None @property def url(self): return "http://127.0.0.1:%d" % self.port def start(self): self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() return self def stop(self): self.server.shutdown() self.server.server_close() if self.thread is not None: self.thread.join(timeout=5)