224 lines
9.0 KiB
Python
224 lines
9.0 KiB
Python
#!/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 '<missing>'}")
|
|
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())
|