#!/usr/bin/env python3 import base64 import hashlib import hmac import json import os import re import secrets import sys import time from dataclasses import dataclass from typing import Any, Dict, Optional from urllib.parse import urlparse import requests @dataclass class FlareConfig: src_url: str get_base_url: str post_url: str wrapper_sh_url: str wrapper_bash_url: str wrapper_ps1_url: str dashboard_url: str flare_html_url: str events_url: str topic: str token: str secret: str def strip_query_fragment(url: str) -> str: return url.split("#", 1)[0].split("?", 1)[0] def derive_get_base(src_url: str) -> str: clean = strip_query_fragment(src_url).rstrip("/") for suffix in ("/wrapper.sh", "/wrapper.bash", "/wrapper.ps1"): if clean.endswith(suffix): return clean[: -len(suffix)] return clean.rsplit("/", 1)[0] def parse_wrapper_exports(wrapper_text: str) -> Dict[str, str]: exports: Dict[str, str] = {} patterns = [ re.compile(r'^export\s+(FLARE_[A-Z_]+)="([^"]*)"$', re.MULTILINE), re.compile(r'^\$env:(FLARE_[A-Z_]+)\s*=\s*"([^"]*)"$', re.MULTILINE), ] for pattern in patterns: for key, value in pattern.findall(wrapper_text): exports[key] = value return exports def load_config_from_src(src_url: str) -> FlareConfig: get_base_url = derive_get_base(src_url) wrapper = requests.get(src_url, timeout=10) wrapper.raise_for_status() exports = parse_wrapper_exports(wrapper.text) post_url = (exports.get("FLARE_URL") or "").rstrip("/") topic = exports.get("FLARE_TOPIC", "") token = exports.get("FLARE_TOKEN", "") secret = exports.get("FLARE_SECRET", "") return FlareConfig( src_url=src_url, get_base_url=get_base_url, post_url=post_url, wrapper_sh_url=f"{get_base_url}/wrapper.sh", wrapper_bash_url=f"{get_base_url}/wrapper.bash", wrapper_ps1_url=f"{get_base_url}/wrapper.ps1", dashboard_url=f"{get_base_url}/", flare_html_url=f"{get_base_url}/flare.html", events_url=f"{get_base_url}/events", topic=topic, token=token, secret=secret, ) def create_signed_payload(topic: str, title: str, detail: Optional[Dict[str, Any]] = None, secret: str = "") -> Dict[str, Any]: detail = detail or {} content = { "complete": 0, "detail": detail, "error": 0, "host": "test-runner", "seq": int(time.time() * 1000) & 0xFFFFFFFF, "timestamp": int(time.time()), "title": title, "topic": topic, } envelope: Dict[str, Any] = {"content": content} if secret: canon = json.dumps(content, sort_keys=True, separators=(",", ":"), ensure_ascii=False) nonce = secrets.token_hex(16) mac = hmac.new(secret.encode("utf-8"), (canon + nonce).encode("utf-8"), hashlib.sha256).hexdigest() envelope["nonce"] = nonce envelope["hmac"] = mac return { "topic": topic, "title": title, "message": json.dumps(envelope, separators=(",", ":"), ensure_ascii=False), "tags": ["test"], "priority": 3, } def auth_headers(token: str, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]: headers = dict(extra or {}) if token: headers["Authorization"] = f"Bearer {token}" return headers def run_case( description: str, method: str, url: str, *, expected_status: int, headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, str]] = None, json_body: Optional[Any] = None, data: Optional[Any] = None, stream: bool = False, ) -> bool: try: response = requests.request( method=method, url=url, headers=headers, params=params, json=json_body, data=data, timeout=8, stream=stream, ) ok = response.status_code == expected_status status = "PASS" if ok else f"FAIL ({response.status_code})" print(f"{status:<12} {method:<7} {response.request.path_url:<46} {description}") if stream: response.close() return ok except Exception as exc: print(f"ERROR {method:<7} {url:<46} {description}: {exc}") return False def main() -> int: src_url = os.environ.get("__FLARE_SRC", "").strip() if not src_url: print("ERROR: __FLARE_SRC is required", file=sys.stderr) return 1 cfg = load_config_from_src(src_url) root = f"{urlparse(cfg.get_base_url).scheme}://{urlparse(cfg.get_base_url).netloc}" bad_post = root + "/definitely-not-real-post-path" print("Resolved Flare test target") print(f" src : {cfg.src_url}") print(f" get base : {cfg.get_base_url}") print(f" post target : {cfg.post_url}") print(f" dashboard : {cfg.dashboard_url}") print(f" events : {cfg.events_url}") print(f" topic : {cfg.topic or ''}") print(f" token present : {'yes' if cfg.token else 'no'}") print(f" secret present: {'yes' if cfg.secret else 'no'}") print() print(f"{'RESULT':<12} {'METHOD':<7} {'PATH':<46} DESCRIPTION") print("-" * 100) all_ok = True all_ok &= run_case("dashboard index", "GET", cfg.dashboard_url, expected_status=200) all_ok &= run_case("dashboard HTML with extra query", "GET", cfg.flare_html_url, expected_status=200, params={"junk": "1"}) all_ok &= run_case("wrapper.sh fetch", "GET", cfg.wrapper_sh_url, expected_status=200) all_ok &= run_case("wrapper.bash fetch", "GET", cfg.wrapper_bash_url, expected_status=200) all_ok &= run_case("wrapper.ps1 fetch", "GET", cfg.wrapper_ps1_url, expected_status=200) all_ok &= run_case("root path stays dark", "GET", root + "/", expected_status=404) all_ok &= run_case("weird GET path under root", "GET", root + "/definitely-not-real", expected_status=404) all_ok &= run_case("unsupported HEAD currently returns 501", "HEAD", cfg.dashboard_url, expected_status=501) all_ok &= run_case("CORS preflight on POST target", "OPTIONS", cfg.post_url, expected_status=204) all_ok &= run_case("events without auth rejected", "GET", cfg.events_url, expected_status=401, params={"topic": cfg.topic}, stream=True) all_ok &= run_case("events with bearer auth", "GET", cfg.events_url, expected_status=200, headers=auth_headers(cfg.token), params={"topic": cfg.topic}, stream=True) all_ok &= run_case("events history via since=all", "GET", cfg.events_url, expected_status=200, headers=auth_headers(cfg.token), params={"topic": cfg.topic, "since": "all"}, stream=True) if cfg.token: auth_b64 = base64.b64encode(f"Bearer {cfg.token}".encode("utf-8")).decode("ascii") all_ok &= run_case("events with auth query param", "GET", cfg.events_url, expected_status=200, params={"topic": cfg.topic, "auth": auth_b64}, stream=True) all_ok &= run_case("POST missing auth rejected", "POST", cfg.post_url, expected_status=401, json_body={"topic": cfg.topic}) all_ok &= run_case("POST bad bearer rejected", "POST", cfg.post_url, expected_status=401, headers=auth_headers("definitely-wrong-token"), json_body={"topic": cfg.topic}) all_ok &= run_case("POST invalid JSON rejected", "POST", cfg.post_url, expected_status=400, headers=auth_headers(cfg.token, {"Content-Type": "application/json"}), data="{not-json") all_ok &= run_case("POST missing topic rejected", "POST", cfg.post_url, expected_status=400, headers=auth_headers(cfg.token), json_body={"title": "no topic"}) all_ok &= run_case("POST wrong path rejected", "POST", bad_post, expected_status=404, headers=auth_headers(cfg.token), json_body={"topic": cfg.topic}) if cfg.secret: all_ok &= run_case("POST bad HMAC rejected", "POST", cfg.post_url, expected_status=400, headers=auth_headers(cfg.token), json_body=create_signed_payload(cfg.topic, "bad hmac", {"status": "bad"}, secret="wrong-secret")) all_ok &= run_case("POST valid signed event", "POST", cfg.post_url, expected_status=200, headers=auth_headers(cfg.token), json_body=create_signed_payload(cfg.topic, "valid signed", {"status": "ok"}, secret=cfg.secret)) all_ok &= run_case("POST signed flare-summary event", "POST", cfg.post_url, expected_status=200, headers=auth_headers(cfg.token), json_body=create_signed_payload(cfg.topic, "flare-summary", {"phase": "http_test", "status": "summary-ok"}, secret=cfg.secret)) else: all_ok &= run_case("POST unsigned event in no-secret mode", "POST", cfg.post_url, expected_status=200, headers=auth_headers(cfg.token), json_body=create_signed_payload(cfg.topic, "unsigned allowed", {"status": "ok"}, secret="")) all_ok &= run_case("PUT unsupported method returns 501", "PUT", cfg.post_url, expected_status=501, headers=auth_headers(cfg.token), json_body={"topic": cfg.topic}) print("-" * 100) if all_ok: print("All HTTP tests passed.") return 0 print("One or more HTTP tests failed.") return 1 if __name__ == "__main__": raise SystemExit(main())