1095 lines
43 KiB
Python
1095 lines
43 KiB
Python
#!/usr/bin/env python3
|
||
"""Offline unit tests for stack_status_lib (no network)."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
sys.path.insert(0, str(ROOT / "site-gen"))
|
||
|
||
import stack_status_lib as L # noqa: E402
|
||
|
||
|
||
def _disable_extra_planes() -> None:
|
||
"""Keep offline unit tests off dig/TLS/catalog network work."""
|
||
L.SERVICES_ENABLED = False
|
||
L.MAIL_ENABLED = False
|
||
L.CERT_PROBES_ENABLED = False
|
||
L.ORPHAN_PROBES_ENABLED = False
|
||
L.MAIL_PORTS_ENABLED = False
|
||
L.ORPHAN_HTTP_ENABLED = False
|
||
L.PORT_SCAN_ENABLED = False
|
||
L.DELIVER_ENABLED = False
|
||
|
||
|
||
_disable_extra_planes()
|
||
|
||
|
||
class AggregateTests(unittest.TestCase):
|
||
def test_all_green_ok(self):
|
||
members = [{"light": "green"}] * 3
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "green")
|
||
self.assertEqual(intensity, "ok")
|
||
|
||
def test_taler_software_from_config(self):
|
||
body = json.dumps(
|
||
{"name": "taler-exchange", "build_version": "1.6.15-dev.1", "version": "39:0:5"}
|
||
).encode()
|
||
items = L.software_from_taler_config(body, host="exchange.demo.taler.net", role="exchange")
|
||
self.assertEqual(len(items), 1)
|
||
self.assertEqual(items[0]["package"], "taler-exchange")
|
||
self.assertEqual(items[0]["version"], "1.6.15-dev.1")
|
||
colored = L.colorize_software(
|
||
items + L.software_from_server_header("nginx/1.24.0", host="exchange.demo.taler.net"),
|
||
vulns={"hits": [{"package": "nginx", "version": "1.24.0", "severity": "HIGH"}]},
|
||
)
|
||
by = {c["package"]: c for c in colored}
|
||
self.assertEqual(by["taler-exchange"]["light"], "green")
|
||
self.assertEqual(by["nginx"]["light"], "orange")
|
||
|
||
def test_deliverability_stale_and_fail(self):
|
||
now = 1_000_000.0
|
||
old_en = L.DELIVER_ENABLED
|
||
L.DELIVER_ENABLED = True
|
||
try:
|
||
state = {
|
||
"domains": {
|
||
"example.test": {
|
||
"ok": True,
|
||
"light": "green",
|
||
"reason": "rcpt_accepted",
|
||
"ts": now - 100,
|
||
"mx": "mx.example.test",
|
||
"rcpt": "postmaster@example.test",
|
||
}
|
||
}
|
||
}
|
||
got = L.ensure_deliverability(
|
||
"example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False
|
||
)
|
||
self.assertTrue(got["ok"])
|
||
self.assertEqual(got["light"], "green")
|
||
# stale success → yellow (prevent re-probe by raising interval)
|
||
state["domains"]["example.test"]["ts"] = now - (L.DELIVER_STALE_S + 10)
|
||
old_i = L.DELIVER_INTERVAL_S
|
||
L.DELIVER_INTERVAL_S = L.DELIVER_STALE_S + 1000
|
||
try:
|
||
got = L.ensure_deliverability(
|
||
"example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False
|
||
)
|
||
finally:
|
||
L.DELIVER_INTERVAL_S = old_i
|
||
self.assertFalse(got["ok"])
|
||
self.assertEqual(got["light"], "yellow")
|
||
self.assertTrue(got["stale"])
|
||
# hard fail stored
|
||
state["domains"]["example.test"] = {
|
||
"ok": False,
|
||
"light": "orange",
|
||
"reason": "rcpt_reject:550",
|
||
"ts": now - 60,
|
||
"mx": "mx.example.test",
|
||
"rcpt": "postmaster@example.test",
|
||
}
|
||
old_i = L.DELIVER_INTERVAL_S
|
||
L.DELIVER_INTERVAL_S = 10_000
|
||
try:
|
||
got = L.ensure_deliverability(
|
||
"example.test", mx_hosts=["mx.example.test"], now=now, state=state, force=False
|
||
)
|
||
finally:
|
||
L.DELIVER_INTERVAL_S = old_i
|
||
self.assertEqual(got["light"], "orange")
|
||
self.assertIn("rcpt_reject", got["reason"])
|
||
finally:
|
||
L.DELIVER_ENABLED = old_en
|
||
|
||
def test_published_currency_behind(self):
|
||
self.assertEqual(L.normalize_build_version("1.6.15-dev.1"), "1.6.15~dev1")
|
||
pub = {
|
||
"enabled": True,
|
||
"packages": {"taler-merchant": "1.6.22-0+trixie", "taler-exchange": "1.6.15~dev1-0+trixie"},
|
||
}
|
||
items = L.apply_published_currency(
|
||
[
|
||
{"package": "taler-merchant", "version": "1.6.21", "light": "green"},
|
||
{"package": "taler-exchange", "version": "1.6.15-dev.1", "light": "green"},
|
||
],
|
||
pub,
|
||
)
|
||
by = {i["package"]: i for i in items}
|
||
self.assertEqual(by["taler-merchant"]["currency_status"], "behind")
|
||
self.assertEqual(by["taler-merchant"]["behind"], "1.6.22")
|
||
self.assertEqual(by["taler-exchange"]["currency_status"], "current")
|
||
self.assertGreater(L.deb_version_cmp("1.6.22-0+trixie", "1.6.21-0+trixie"), 0)
|
||
|
||
def test_port_whitelist_and_suspicious(self):
|
||
rules = L.parse_ports_whitelist # noqa: ensure exists
|
||
self.assertTrue(callable(rules))
|
||
wl = [{"pattern": "*.hacktivism.ch", "ports": [22, 8080, 9200]}, {"pattern": "*", "ports": [8080]}]
|
||
self.assertIn(22, L.whitelist_ports_for_host("taler.hacktivism.ch", wl))
|
||
self.assertIn(9200, L.whitelist_ports_for_host("git.hacktivism.ch", wl))
|
||
self.assertIn(8080, L.whitelist_ports_for_host("anything.example", wl))
|
||
# Intentional koopa public ports must not be suspicious / orange
|
||
koopa = L.assess_host_ports(
|
||
"git.hacktivism.ch",
|
||
ports=[22, 80, 443, 8080, 9200, 3306],
|
||
whitelist_rules=wl,
|
||
open_ports=[22, 80, 443, 8080, 9200],
|
||
)
|
||
self.assertEqual(koopa["suspicious"], [])
|
||
self.assertTrue(koopa["ok"])
|
||
self.assertEqual(koopa["light"], "green")
|
||
got = L.assess_host_ports(
|
||
"edge.example",
|
||
ports=[22, 80, 443, 3306],
|
||
whitelist_rules=[],
|
||
open_ports=[80, 443, 3306],
|
||
)
|
||
self.assertEqual([s["port"] for s in got["suspicious"]], [3306])
|
||
self.assertEqual(got["suspicious"][0]["severity"], "hard")
|
||
self.assertEqual(got["light"], "orange")
|
||
got2 = L.assess_host_ports(
|
||
"edge.example",
|
||
ports=[22, 80, 443],
|
||
whitelist_rules=[{"pattern": "edge.example", "ports": [22]}],
|
||
open_ports=[22, 443],
|
||
)
|
||
self.assertEqual(got2["suspicious"], [])
|
||
self.assertTrue(got2["ok"])
|
||
soft = L.assess_host_ports(
|
||
"edge.example",
|
||
ports=[22, 80, 443],
|
||
whitelist_rules=[],
|
||
open_ports=[22, 443],
|
||
)
|
||
self.assertEqual(soft["suspicious"][0]["severity"], "soft")
|
||
self.assertTrue(soft["ok"]) # soft does not fail ok
|
||
|
||
def test_orphan_skips_siblings_and_wildcards(self):
|
||
known = {"demo.taler.net", "backend.demo.taler.net"}
|
||
orphans = L.collect_orphan_hosts(
|
||
known_hosts=known,
|
||
tls_sans=["*.demo.taler.net", "donations.demo.taler.net", "gone.example.invalid"],
|
||
probe_fn=lambda url: {"ok": False, "code": 0, "error": "URLError"},
|
||
)
|
||
hosts = [o["host"] for o in orphans]
|
||
self.assertNotIn("demo.taler.net", hosts)
|
||
self.assertNotIn("donations.demo.taler.net", hosts) # sibling
|
||
# gone.example.invalid may appear only if dig says no DNS — skip assert on network
|
||
self.assertTrue(all(not h.startswith("*.") for h in hosts))
|
||
|
||
def test_total_score_thresholds(self):
|
||
# 10 green of 10 → green
|
||
s = L.compute_total_score([{"light": "green"}] * 10)
|
||
self.assertEqual(s["green"], 10)
|
||
self.assertEqual(s["total"], 10)
|
||
self.assertEqual(s["light"], "green")
|
||
# 9/10 → blue (≥0.85)
|
||
s = L.compute_total_score([{"light": "green"}] * 9 + [{"light": "orange"}])
|
||
self.assertEqual(s["light"], "blue")
|
||
# 6/10 → yellow
|
||
s = L.compute_total_score([{"light": "green"}] * 6 + [{"light": "red"}] * 4)
|
||
self.assertEqual(s["light"], "yellow")
|
||
# 4/10 → orange
|
||
s = L.compute_total_score([{"light": "green"}] * 4 + [{"light": "red"}] * 6)
|
||
self.assertEqual(s["light"], "orange")
|
||
# 1/10 → red
|
||
s = L.compute_total_score([{"light": "green"}] + [{"light": "orange"}] * 9)
|
||
self.assertEqual(s["light"], "red")
|
||
# 0 green mixed → white (gone)
|
||
s = L.compute_total_score([{"light": "red"}, {"light": "orange"}])
|
||
self.assertEqual(s["green"], 0)
|
||
self.assertEqual(s["light"], "white")
|
||
# all black → black
|
||
s = L.compute_total_score([{"light": "black"}] * 3)
|
||
self.assertEqual(s["light"], "black")
|
||
|
||
def test_breakdown_counts(self):
|
||
members = [
|
||
{"role": "bank", "host": "a", "light": "green", "reason": "ok"},
|
||
{"role": "exchange", "host": "b", "light": "green", "reason": "ok"},
|
||
{"role": "merchant", "host": "c", "light": "orange", "reason": "http_501"},
|
||
{"role": "bank", "host": "IBAN", "light": "blue", "reason": "ebics_iban"},
|
||
]
|
||
bd = L.aggregate_breakdown(members)
|
||
self.assertEqual(bd["by_light"].get("green"), 2)
|
||
self.assertEqual(bd["by_light"].get("orange"), 1)
|
||
self.assertNotIn("blue", bd["by_light"]) # skipped informational
|
||
self.assertEqual(bd["worst"], "orange")
|
||
self.assertEqual(bd["counted"], 3)
|
||
|
||
def test_blue_present_ok(self):
|
||
members = [{"light": "green"}, {"light": "blue"}, {"light": "blue"}]
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "blue")
|
||
self.assertEqual(intensity, "ok")
|
||
|
||
def test_one_yellow_muted(self):
|
||
members = [
|
||
{"light": "green"},
|
||
{"light": "yellow"},
|
||
{"light": "green"},
|
||
]
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "yellow")
|
||
self.assertEqual(intensity, "muted")
|
||
|
||
def test_one_red_muted(self):
|
||
members = [
|
||
{"light": "green"},
|
||
{"light": "red"},
|
||
{"light": "green"},
|
||
]
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "red")
|
||
self.assertEqual(intensity, "muted")
|
||
|
||
def test_all_red_full(self):
|
||
members = [{"light": "red"}] * 3
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "red")
|
||
self.assertEqual(intensity, "full")
|
||
|
||
def test_all_black_full(self):
|
||
members = [{"light": "black"}] * 3
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "black")
|
||
self.assertEqual(intensity, "full")
|
||
|
||
def test_mixed_orange_green_muted(self):
|
||
members = [
|
||
{"light": "orange"},
|
||
{"light": "green"},
|
||
{"light": "green"},
|
||
]
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "orange")
|
||
self.assertEqual(intensity, "muted")
|
||
|
||
def test_all_yellow_still_muted(self):
|
||
# Full punch only for red/black total outage
|
||
members = [{"light": "yellow"}] * 3
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "yellow")
|
||
self.assertEqual(intensity, "muted")
|
||
|
||
def test_weakest_wins_black_over_red(self):
|
||
members = [
|
||
{"light": "red"},
|
||
{"light": "black"},
|
||
{"light": "green"},
|
||
]
|
||
light, intensity = L.aggregate_stack(members)
|
||
self.assertEqual(light, "black")
|
||
self.assertEqual(intensity, "muted")
|
||
|
||
|
||
class ClassifyTests(unittest.TestCase):
|
||
def test_ok_green(self):
|
||
c = L.classify_probe({"ok": True, "code": 200, "elapsed": 0.2, "error": None}, now=1000.0)
|
||
self.assertEqual(c["light"], "green")
|
||
self.assertIsNone(c["down_since"])
|
||
|
||
def test_slow_yellow(self):
|
||
c = L.classify_probe({"ok": True, "code": 200, "elapsed": 3.0, "error": None}, now=1000.0)
|
||
self.assertEqual(c["light"], "yellow")
|
||
|
||
def test_acute_red(self):
|
||
c = L.classify_probe(
|
||
{"ok": False, "code": 0, "elapsed": 1.0, "error": "URLError"},
|
||
prev={"ok": True},
|
||
now=1000.0,
|
||
acute_s=90,
|
||
)
|
||
self.assertEqual(c["light"], "red")
|
||
self.assertEqual(c["down_since"], 1000.0)
|
||
|
||
def test_long_black(self):
|
||
c = L.classify_probe(
|
||
{"ok": False, "code": 0, "elapsed": 1.0, "error": "URLError"},
|
||
prev={"ok": False, "down_since": 1000.0},
|
||
now=1200.0,
|
||
acute_s=90,
|
||
)
|
||
self.assertEqual(c["light"], "black")
|
||
self.assertEqual(c["down_since"], 1000.0)
|
||
|
||
def test_timeout_orange_when_fresh(self):
|
||
c = L.classify_probe(
|
||
{"ok": False, "code": 0, "elapsed": 5.0, "error": "TimeoutError"},
|
||
prev={"ok": True},
|
||
now=1000.0,
|
||
acute_s=90,
|
||
)
|
||
self.assertEqual(c["light"], "orange")
|
||
|
||
|
||
class EndpointTests(unittest.TestCase):
|
||
def test_legal_missing_is_medium_orange(self):
|
||
for label in ("privacy", "terms"):
|
||
c = L.classify_endpoint_probe(
|
||
{"ok": False, "code": 404, "elapsed": 0.1, "error": "http_404", "body": b""},
|
||
expect=(200,),
|
||
severity="medium",
|
||
now=1000.0,
|
||
)
|
||
self.assertEqual(c["light"], "orange", label)
|
||
self.assertFalse(c["expected"], label)
|
||
|
||
def test_summarize_dots(self):
|
||
eps = [
|
||
{"light": "green", "skipped": False},
|
||
{"light": "yellow", "skipped": False},
|
||
{"light": "blue", "skipped": True},
|
||
]
|
||
s = L.summarize_endpoints(eps)
|
||
self.assertEqual(s["ok"], 1)
|
||
self.assertEqual(s["soft"], 1)
|
||
self.assertEqual(s["skipped"], 1)
|
||
self.assertEqual(s["light"], "yellow")
|
||
self.assertEqual(s["dots"], ["green", "yellow"]) # skips omitted from strip
|
||
|
||
def test_probe_role_endpoints_intro_skip(self):
|
||
calls = []
|
||
|
||
def probe(url: str):
|
||
calls.append(url)
|
||
return {"ok": True, "code": 200, "elapsed": 0.05, "error": None, "body": b"{}", "url": url}
|
||
|
||
eps = L.probe_role_endpoints(
|
||
"bank",
|
||
"bank.example.test",
|
||
has_intro=False,
|
||
probe_fn=probe,
|
||
now=1000.0,
|
||
)
|
||
ids = {e["id"]: e for e in eps}
|
||
self.assertTrue(ids["intro"]["skipped"])
|
||
self.assertEqual(ids["intro"]["reason"], "no_intro")
|
||
self.assertFalse(ids["config"]["skipped"])
|
||
self.assertTrue(any(u.endswith("/config") for u in calls))
|
||
self.assertFalse(any("/intro/" in u for u in calls))
|
||
|
||
def test_custom_intro_stacks_only(self):
|
||
self.assertTrue(
|
||
L.stack_has_custom_intro({"currency": "GOA", "id": "hacktivism.ch", "kind": "taler"})
|
||
)
|
||
self.assertTrue(
|
||
L.stack_has_custom_intro(
|
||
{"currency": "TESTPAYSAN", "id": "stage.lefrancpaysan.ch", "kind": "taler"}
|
||
)
|
||
)
|
||
self.assertFalse(
|
||
L.stack_has_custom_intro({"currency": "KUDOS", "id": "demo.taler.net", "kind": "taler"})
|
||
)
|
||
self.assertFalse(
|
||
L.stack_has_custom_intro({"currency": "CHF", "id": "taler-ops.ch", "kind": "taler"})
|
||
)
|
||
self.assertFalse(
|
||
L.stack_has_custom_intro(
|
||
{"currency": "", "id": "lefrancpaysan.ch", "kind": "static_pending"}
|
||
)
|
||
)
|
||
|
||
def test_server_header_parse(self):
|
||
self.assertEqual(L.guess_http_package("nginx/1.22.1"), "nginx")
|
||
self.assertEqual(L.extract_http_version("nginx/1.22.1"), "1.22.1")
|
||
self.assertEqual(L.guess_http_package("Caddy"), "caddy")
|
||
|
||
def test_vuln_raises_box_colour(self):
|
||
members = [
|
||
{"role": "exchange", "host": "ex.example", "light": "green", "reason": "ok"},
|
||
{"role": "merchant", "host": "m.example", "light": "green", "reason": "ok"},
|
||
]
|
||
vulns = {
|
||
"enabled": True,
|
||
"worst": "CRITICAL",
|
||
"light": "red",
|
||
"hits": [
|
||
{
|
||
"id": "CVE-2099-0001",
|
||
"severity": "CRITICAL",
|
||
"package": "nginx",
|
||
"version": "1.0.0",
|
||
"host": "ex.example",
|
||
}
|
||
],
|
||
}
|
||
light, intensity, members2 = L.apply_vuln_to_stack("green", "ok", members, vulns)
|
||
self.assertEqual(light, "red")
|
||
self.assertEqual(intensity, "muted")
|
||
self.assertEqual(members2[-1]["role"], "cve")
|
||
|
||
def test_network_zones_and_ssh(self):
|
||
self.assertEqual(
|
||
L.classify_network_zone(
|
||
{"id": "hacktivism.ch", "currency": "GOA", "local": True, "kind": "taler"}
|
||
)["zone"],
|
||
"goa-local",
|
||
)
|
||
self.assertEqual(
|
||
L.classify_network_zone(
|
||
{"id": "demo.taler.net", "currency": "KUDOS", "local": False, "kind": "taler"}
|
||
)["zone"],
|
||
"public-demo",
|
||
)
|
||
targets = L.ssh_targets_for_stack(
|
||
{
|
||
"id": "hacktivism.ch",
|
||
"currency": "GOA",
|
||
"local": True,
|
||
"kind": "taler",
|
||
"merchant": "taler.hacktivism.ch",
|
||
}
|
||
)
|
||
self.assertTrue(targets["expected"])
|
||
self.assertTrue(any("hacktivism" in h for h in targets["hosts"]))
|
||
|
||
def fake_ssh(host: str):
|
||
return {
|
||
"ok": True,
|
||
"light": "green",
|
||
"reason": "SSH-2.0-OpenSSH_test",
|
||
"host": host,
|
||
"elapsed": 0.05,
|
||
"banner": "SSH-2.0-OpenSSH_test",
|
||
}
|
||
|
||
net = L.build_network_status(
|
||
{
|
||
"id": "hacktivism.ch",
|
||
"currency": "GOA",
|
||
"local": True,
|
||
"kind": "taler",
|
||
"merchant": "taler.hacktivism.ch",
|
||
},
|
||
members=[{"light": "green", "reason": "ok"}],
|
||
ssh_probe_fn=fake_ssh,
|
||
)
|
||
self.assertEqual(net["zone"], "goa-local")
|
||
self.assertTrue(net["ssh"]["ok"])
|
||
self.assertIn("hacktivism", net["ssh"]["host"])
|
||
|
||
def test_tops_stage_ssh_is_env_mapped_hidden(self):
|
||
st = {
|
||
"id": "stage.taler-ops.ch",
|
||
"currency": "CHF",
|
||
"kind": "taler",
|
||
"local": False,
|
||
"merchant": "stage.my.taler-ops.ch",
|
||
"exchange": "exchange.stage.taler-ops.ch",
|
||
}
|
||
self.assertEqual(L.classify_network_zone(st)["zone"], "tops-stage")
|
||
with mock.patch.dict(
|
||
os.environ, {"TALER_MONITORD_SSH_TOPS": "betel.taler-systems.com"}
|
||
):
|
||
targets = L.ssh_targets_for_stack(st)
|
||
self.assertEqual(targets["hosts"], ["betel.taler-systems.com"])
|
||
self.assertTrue(targets.get("hide_host"))
|
||
self.assertNotIn("stage.taler-ops.ch", targets["hosts"])
|
||
|
||
def fake_ssh(host: str):
|
||
self.assertEqual(host, "betel.taler-systems.com")
|
||
return {
|
||
"ok": True,
|
||
"light": "green",
|
||
"reason": "SSH-2.0-OpenSSH_test",
|
||
"host": host,
|
||
"elapsed": 0.05,
|
||
"banner": "SSH-2.0-OpenSSH_test",
|
||
}
|
||
|
||
net = L.build_network_status(
|
||
st, members=[{"light": "green", "reason": "ok"}], ssh_probe_fn=fake_ssh
|
||
)
|
||
self.assertTrue(net["ssh"]["ok"])
|
||
# UI/JSON must not advertise env-mapped host
|
||
self.assertEqual(net["ssh"]["host"], "")
|
||
self.assertEqual(net["ssh"]["candidates"], [])
|
||
|
||
def test_tops_stage_ssh_without_env_uses_public_fqdn(self):
|
||
st = {
|
||
"id": "stage.taler-ops.ch",
|
||
"currency": "CHF",
|
||
"kind": "taler",
|
||
"local": False,
|
||
"merchant": "stage.my.taler-ops.ch",
|
||
"exchange": "exchange.stage.taler-ops.ch",
|
||
}
|
||
cleared = {
|
||
k: ""
|
||
for k in (
|
||
"TALER_MONITORD_SSH_TOPS",
|
||
"TALER_MONITORD_SSH_FP",
|
||
"TALER_MONITORD_SSH_GOA",
|
||
"TALER_MONITORD_SSH_TOPS_PROD",
|
||
)
|
||
}
|
||
with mock.patch.dict(os.environ, cleared):
|
||
targets = L.ssh_targets_for_stack(st)
|
||
self.assertEqual(targets["hosts"][0], "stage.taler-ops.ch")
|
||
self.assertFalse(targets.get("hide_host"))
|
||
|
||
def test_fp_stage_ssh_is_env_mapped_hidden(self):
|
||
st = {
|
||
"id": "stage.lefrancpaysan.ch",
|
||
"currency": "TESTPAYSAN",
|
||
"kind": "taler",
|
||
"local": False,
|
||
"merchant": "stage.monnaie.lefrancpaysan.ch",
|
||
"exchange": "stage.exchange.lefrancpaysan.ch",
|
||
"bank": "stage.bank.lefrancpaysan.ch",
|
||
}
|
||
self.assertEqual(L.classify_network_zone(st)["zone"], "fp-stage")
|
||
with mock.patch.dict(
|
||
os.environ, {"TALER_MONITORD_SSH_FP": "ov-1a5ee8.infomaniak.ch"}
|
||
):
|
||
targets = L.ssh_targets_for_stack(st)
|
||
self.assertEqual(targets["hosts"], ["ov-1a5ee8.infomaniak.ch"])
|
||
self.assertTrue(targets.get("hide_host"))
|
||
self.assertNotIn("lefrancpaysan", " ".join(targets["hosts"]))
|
||
|
||
def fake_ssh(host: str):
|
||
self.assertEqual(host, "ov-1a5ee8.infomaniak.ch")
|
||
return {
|
||
"ok": True,
|
||
"light": "green",
|
||
"reason": "SSH-2.0-OpenSSH_test",
|
||
"host": host,
|
||
"elapsed": 0.05,
|
||
"banner": "SSH-2.0-OpenSSH_test",
|
||
}
|
||
|
||
net = L.build_network_status(
|
||
st, members=[{"light": "green", "reason": "ok"}], ssh_probe_fn=fake_ssh
|
||
)
|
||
self.assertTrue(net["ssh"]["ok"])
|
||
self.assertEqual(net["ssh"]["host"], "")
|
||
self.assertEqual(net["ssh"]["candidates"], [])
|
||
|
||
def test_load_monitoring_env_expands_home(self):
|
||
with tempfile.TemporaryDirectory() as td:
|
||
p = Path(td) / "env"
|
||
p.write_text(
|
||
"TALER_MONITORD_SSH_TOPS=betel.example\n"
|
||
"SUITE_DIR=$HOME/git/taler-monitoring\n"
|
||
"# comment\n"
|
||
"export TALER_MONITORD_SSH_FP=fp.example\n",
|
||
encoding="utf-8",
|
||
)
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
for k in (
|
||
"TALER_MONITORD_SSH_TOPS",
|
||
"TALER_MONITORD_SSH_FP",
|
||
"SUITE_DIR",
|
||
):
|
||
os.environ.pop(k, None)
|
||
got = L.load_monitoring_env(path=p)
|
||
self.assertEqual(got, p)
|
||
self.assertEqual(os.environ["TALER_MONITORD_SSH_TOPS"], "betel.example")
|
||
self.assertEqual(os.environ["TALER_MONITORD_SSH_FP"], "fp.example")
|
||
self.assertTrue(os.environ["SUITE_DIR"].endswith("/git/taler-monitoring"))
|
||
|
||
def test_commercial_prod_tops_only(self):
|
||
self.assertTrue(
|
||
L.stack_is_commercial_prod(
|
||
{"id": "taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False}
|
||
)
|
||
)
|
||
self.assertEqual(
|
||
L.legal_doc_severity(
|
||
{"id": "taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False}
|
||
),
|
||
"medium",
|
||
)
|
||
self.assertFalse(
|
||
L.stack_is_commercial_prod(
|
||
{"id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False}
|
||
)
|
||
)
|
||
self.assertEqual(
|
||
L.legal_doc_severity(
|
||
{"id": "stage.taler-ops.ch", "currency": "CHF", "kind": "taler", "local": False}
|
||
),
|
||
"soft",
|
||
)
|
||
self.assertFalse(
|
||
L.stack_is_commercial_prod(
|
||
{"id": "hacktivism.ch", "currency": "GOA", "kind": "taler", "local": True}
|
||
)
|
||
)
|
||
self.assertFalse(
|
||
L.stack_is_commercial_prod(
|
||
{"id": "demo.taler.net", "currency": "KUDOS", "kind": "taler", "local": False}
|
||
)
|
||
)
|
||
|
||
|
||
class CurrencySpecTests(unittest.TestCase):
|
||
def test_summarize_chf(self):
|
||
cs = {
|
||
"name": "Swiss francs",
|
||
"currency": "CHF",
|
||
"alt_unit_names": {"0": "Fr.", "-2": "Rp."},
|
||
"common_amounts": ["CHF:5", "CHF:10", "CHF:25", "CHF:50"],
|
||
}
|
||
info = L.summarize_currency_spec(cs)
|
||
self.assertEqual(info["currency"], "CHF")
|
||
self.assertEqual(info["symbol"], "Fr.")
|
||
self.assertEqual(info["units_label"], "Fr. / Rp.")
|
||
self.assertEqual(info["denoms"], ["5", "10", "25", "50"])
|
||
|
||
def test_denoms_from_keys_when_common_empty(self):
|
||
cs = {"currency": "FLODOS", "alt_unit_names": {"0": "FLODOS"}, "common_amounts": []}
|
||
body = b'{"denominations":[{"value":"FLODOS:1"},{"value":"FLODOS:5"},{"value":"FLODOS:0"}]}'
|
||
info = L.summarize_currency_spec(cs, keys_body=body)
|
||
self.assertEqual(info["denoms"], ["1", "5"])
|
||
|
||
|
||
class WireBackendTests(unittest.TestCase):
|
||
def test_payto_methods(self):
|
||
blob = b'{"accounts":[{"payto_uri":"payto://iban/CH9300762011623852957"},{"payto_uri":"payto://x-taler-bank/bank.example/ex"}]}'
|
||
self.assertEqual(L.payto_wire_methods(blob), {"iban", "x-taler-bank"})
|
||
|
||
def test_classify_ebics_iban_only(self):
|
||
self.assertEqual(L.classify_bank_backend({"iban"}), "ebics")
|
||
|
||
def test_classify_xtaler_wins_mixed(self):
|
||
self.assertEqual(L.classify_bank_backend({"iban", "x-taler-bank"}), "x-taler-bank")
|
||
|
||
def test_classify_conf_skip(self):
|
||
self.assertEqual(L.classify_bank_backend(set(), bank_configured=False), "ebics")
|
||
self.assertFalse(L.bank_host_configured("-"))
|
||
self.assertFalse(L.bank_host_configured("ebics"))
|
||
self.assertTrue(L.bank_host_configured("bank.example.test"))
|
||
|
||
|
||
class ParseAndBuildTests(unittest.TestCase):
|
||
def setUp(self):
|
||
_disable_extra_planes()
|
||
|
||
def test_parse_stage_fp(self):
|
||
conf = ROOT / "domains.conf"
|
||
stacks = L.parse_domains_conf(conf)
|
||
ids = {s["id"] for s in stacks}
|
||
self.assertIn("stage.lefrancpaysan.ch", ids)
|
||
stage = next(s for s in stacks if s["id"] == "stage.lefrancpaysan.ch")
|
||
self.assertEqual(stage["bank"], "stage.bank.lefrancpaysan.ch")
|
||
self.assertEqual(stage["merchant"], "stage.monnaie.lefrancpaysan.ch")
|
||
self.assertEqual(stage["currency"], "TESTPAYSAN")
|
||
self.assertEqual(stage["kind"], "taler")
|
||
prod = next(s for s in stacks if s["id"] == "lefrancpaysan.ch")
|
||
self.assertEqual(prod["kind"], "static_pending")
|
||
self.assertEqual(prod["currency"], "")
|
||
|
||
def test_ebics_skips_bank_probe(self):
|
||
stacks = [
|
||
{
|
||
"id": "tops.example",
|
||
"bank": "bank.tops.example",
|
||
"exchange": "exchange.tops.example",
|
||
"merchant": "my.tops.example",
|
||
"currency": "CHF",
|
||
"kind": "taler",
|
||
"local": False,
|
||
}
|
||
]
|
||
probed = []
|
||
|
||
def probe(url: str):
|
||
probed.append(url)
|
||
if "/keys" in url:
|
||
body = b'{"accounts":[{"payto_uri":"payto://iban/CH6808573105529100001?receiver-name=X"}]}'
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url}
|
||
if "bank." in url:
|
||
return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "body": b"", "url": url}
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url}
|
||
|
||
doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False)
|
||
st = doc["stacks"][0]
|
||
self.assertEqual(st["bank_backend"], "ebics")
|
||
self.assertEqual(st["light"], "green")
|
||
self.assertEqual(st["intensity"], "ok")
|
||
roles = {m["role"]: m for m in st["members"]}
|
||
self.assertEqual(roles["bank"]["reason"], "ebics_iban")
|
||
self.assertEqual(roles["bank"]["light"], "blue")
|
||
self.assertEqual(roles["bank"]["host"], "IBAN")
|
||
self.assertTrue(all("bank.tops" not in u for u in probed))
|
||
|
||
def test_xtaler_bank_still_probed(self):
|
||
stacks = [
|
||
{
|
||
"id": "demo.example",
|
||
"bank": "bank.demo.example",
|
||
"exchange": "exchange.demo.example",
|
||
"merchant": "backend.demo.example",
|
||
"currency": "KUDOS",
|
||
"kind": "taler",
|
||
"local": False,
|
||
}
|
||
]
|
||
|
||
def probe(url: str):
|
||
if "/keys" in url:
|
||
body = b'{"accounts":[{"payto_uri":"payto://x-taler-bank/bank.demo.example/exchange"}]}'
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url}
|
||
if "bank." in url:
|
||
return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "body": b"", "url": url}
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url}
|
||
|
||
doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False)
|
||
st = doc["stacks"][0]
|
||
self.assertEqual(st["bank_backend"], "x-taler-bank")
|
||
self.assertEqual(st["light"], "red")
|
||
self.assertEqual(st["intensity"], "muted")
|
||
roles = {m["role"]: m for m in st["members"]}
|
||
self.assertEqual(roles["bank"]["light"], "red")
|
||
|
||
def test_domains_conf_bank_dash(self):
|
||
with tempfile.TemporaryDirectory() as td:
|
||
conf = Path(td) / "domains.conf"
|
||
conf.write_text(
|
||
"ebics.example - exchange.ebics.example my.ebics.example CHF 0 0 en de-CH\n",
|
||
encoding="utf-8",
|
||
)
|
||
stacks = L.parse_domains_conf(conf)
|
||
self.assertEqual(len(stacks), 1)
|
||
self.assertFalse(stacks[0]["bank_configured"])
|
||
self.assertEqual(stacks[0]["bank"], "")
|
||
|
||
def test_landing_stats_summary(self):
|
||
stacks = [
|
||
{
|
||
"id": "stage.example",
|
||
"bank": "bank.stage.example",
|
||
"exchange": "exchange.stage.example",
|
||
"merchant": "monnaie.stage.example",
|
||
"currency": "TESTPAYSAN",
|
||
"kind": "taler",
|
||
"local": False,
|
||
}
|
||
]
|
||
stats_body = json.dumps(
|
||
{
|
||
"ok": True,
|
||
"currency": "TESTPAYSAN",
|
||
"generated_at_unix": 5000.0 - 120,
|
||
"generated_at_human": "just now",
|
||
"source": "stage-shared",
|
||
"bank_accounts": {"total": 36, "users": 34},
|
||
"withdraws": {"count": 45, "total_amount": "TESTPAYSAN:12"},
|
||
}
|
||
).encode()
|
||
|
||
cfg_body = json.dumps(
|
||
{
|
||
"currency": "TESTPAYSAN",
|
||
"currency_specification": {
|
||
"name": "TESTPAYSAN stage currency",
|
||
"currency": "TESTPAYSAN",
|
||
"alt_unit_names": {"0": "TESTPAYSAN", "-2": "ct"},
|
||
"common_amounts": [
|
||
"TESTPAYSAN:5",
|
||
"TESTPAYSAN:10",
|
||
"TESTPAYSAN:20",
|
||
"TESTPAYSAN:50",
|
||
"TESTPAYSAN:100",
|
||
],
|
||
},
|
||
}
|
||
).encode()
|
||
|
||
def probe(url: str):
|
||
if url.endswith("/intro/stats.json") and "bank.stage" in url:
|
||
return {
|
||
"ok": True,
|
||
"code": 200,
|
||
"elapsed": 0.05,
|
||
"error": None,
|
||
"body": stats_body,
|
||
"url": url,
|
||
}
|
||
if "/keys" in url:
|
||
body = b'{"accounts":[{"payto_uri":"payto://x-taler-bank/bank.stage.example/exchange"}]}'
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": body, "url": url}
|
||
if url.endswith("/config") and "exchange.stage" in url:
|
||
return {
|
||
"ok": True,
|
||
"code": 200,
|
||
"elapsed": 0.05,
|
||
"error": None,
|
||
"body": cfg_body,
|
||
"url": url,
|
||
}
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "body": b"{}", "url": url}
|
||
|
||
doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False)
|
||
st = doc["stacks"][0]
|
||
self.assertIsNotNone(st.get("stats"))
|
||
self.assertTrue(st["stats"]["present"])
|
||
self.assertEqual(st["stats"]["age_s"], 120)
|
||
self.assertIn("accounts 36", st["stats"]["lines"])
|
||
self.assertFalse(st["stats"]["stale"])
|
||
self.assertEqual(st["currency"], "TESTPAYSAN")
|
||
self.assertEqual(st["currency_info"]["symbol"], "TESTPAYSAN")
|
||
self.assertEqual(st["currency_info"]["units_label"], "TESTPAYSAN / ct")
|
||
self.assertEqual(st["currency_info"]["denoms"], ["5", "10", "20", "50", "100"])
|
||
|
||
def test_static_pending_softens_api_down(self):
|
||
stacks = [
|
||
{
|
||
"id": "lefrancpaysan.ch",
|
||
"bank": "bank.lefrancpaysan.ch",
|
||
"exchange": "exchange.lefrancpaysan.ch",
|
||
"merchant": "monnaie.lefrancpaysan.ch",
|
||
"currency": "",
|
||
"kind": "static_pending",
|
||
"local": False,
|
||
}
|
||
]
|
||
|
||
def probe(url: str):
|
||
if url.rstrip("/").endswith("lefrancpaysan.ch"):
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "url": url}
|
||
return {"ok": False, "code": 502, "elapsed": 0.1, "error": "http_502", "url": url}
|
||
|
||
doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False)
|
||
st = doc["stacks"][0]
|
||
self.assertEqual(st["kind"], "static_pending")
|
||
self.assertEqual(st["light"], "blue")
|
||
self.assertEqual(st["intensity"], "ok")
|
||
roles = {m["role"]: m for m in st["members"]}
|
||
self.assertEqual(roles["site"]["light"], "green")
|
||
self.assertEqual(roles["bank"]["reason"], "no_prod_yet")
|
||
self.assertEqual(roles["bank"]["light"], "blue")
|
||
|
||
def test_build_with_mock_probe(self):
|
||
stacks = [
|
||
{
|
||
"id": "example.test",
|
||
"bank": "bank.example.test",
|
||
"exchange": "exchange.example.test",
|
||
"merchant": "merch.example.test",
|
||
"currency": "X",
|
||
"local": False,
|
||
}
|
||
]
|
||
|
||
def probe(url: str):
|
||
if "exchange" in url:
|
||
return {"ok": False, "code": 0, "elapsed": 0.1, "error": "URLError", "url": url}
|
||
return {"ok": True, "code": 200, "elapsed": 0.1, "error": None, "url": url}
|
||
|
||
doc = L.build_stack_status(stacks, probe_fn=probe, now=5000.0, public_only=False)
|
||
self.assertEqual(len(doc["stacks"]), 1)
|
||
st = doc["stacks"][0]
|
||
self.assertEqual(st["light"], "red")
|
||
self.assertEqual(st["intensity"], "muted")
|
||
self.assertIn("breakdown", st)
|
||
self.assertEqual(doc.get("schema"), L.SCHEMA)
|
||
|
||
def test_atomic_write_roundtrip(self):
|
||
with tempfile.TemporaryDirectory() as td:
|
||
path = Path(td) / "stack-status.json"
|
||
doc = {"schema": 1, "stacks": []}
|
||
L.atomic_write_json(path, doc)
|
||
self.assertTrue(path.is_file())
|
||
got = L.read_status(path)
|
||
self.assertEqual(got["schema"], 1)
|
||
|
||
def test_parse_services_conf(self):
|
||
with tempfile.TemporaryDirectory() as td:
|
||
conf = Path(td) / "services-catalog.conf"
|
||
conf.write_text(
|
||
"bugs.gnunet.org bugs.gnunet.org MantisBT firefly\n"
|
||
"erp.stage.taler-systems.com erp.stage.taler-systems.com ERPnext pixel\n",
|
||
encoding="utf-8",
|
||
)
|
||
rows = L.parse_services_conf(conf)
|
||
self.assertEqual(len(rows), 2)
|
||
self.assertEqual(rows[0]["service_label"], "MantisBT")
|
||
self.assertEqual(rows[1]["zone"], "pixel")
|
||
|
||
|
||
class HtmlRenderTests(unittest.TestCase):
|
||
def test_degraded_why_and_cve_link(self):
|
||
import stack_status_html as H
|
||
|
||
doc = {
|
||
"schema": L.SCHEMA,
|
||
"ts": time.time(),
|
||
"acute_down_s": 90,
|
||
"stacks": [
|
||
{
|
||
"id": "my.taler-ops.ch",
|
||
"light": "orange",
|
||
"intensity": "muted",
|
||
"currency": "CHF",
|
||
"bank_backend": "ebics",
|
||
"kind": "taler",
|
||
"breakdown": {"by_light": {"green": 4, "orange": 3}, "worst": "orange", "counted": 7},
|
||
"members": [
|
||
{"role": "bank", "host": "IBAN", "light": "blue", "reason": "ebics_iban"},
|
||
{"role": "exchange", "host": "exchange.taler-ops.ch", "light": "green", "reason": "ok"},
|
||
{
|
||
"role": "merchant",
|
||
"host": "my.taler-ops.ch",
|
||
"light": "orange",
|
||
"reason": "http_501",
|
||
},
|
||
],
|
||
"vulns": {
|
||
"enabled": True,
|
||
"worst": "HIGH",
|
||
"hits": [
|
||
{
|
||
"id": "CVE-2024-1234",
|
||
"severity": "HIGH",
|
||
"package": "nginx",
|
||
"version": "1.24.0",
|
||
"summary": "test vuln",
|
||
"host": "my.taler-ops.ch",
|
||
}
|
||
],
|
||
"servers": [
|
||
{"host": "my.taler-ops.ch", "package": "nginx", "version": "1.24.0", "server": "nginx/1.24.0"}
|
||
],
|
||
},
|
||
"network": {"zone": "tops-prod", "ssh": {"expected": False}},
|
||
"tls": {"days_left": 60, "sans": ["my.taler-ops.ch"]},
|
||
}
|
||
],
|
||
"services": [],
|
||
"mail": [],
|
||
"orphans": [],
|
||
}
|
||
html = H.render_html(doc)
|
||
self.assertIn("4× OK", html)
|
||
self.assertIn("3× Degraded", html)
|
||
self.assertIn("osv.dev/vulnerability/CVE-2024-1234", html)
|
||
self.assertIn("openStack", html)
|
||
self.assertIn("aria-label=\"nginx\"", html)
|
||
self.assertIn("0/1 green", html) # orange stack → not green
|
||
self.assertIn("Gone", html) # or Degraded label in score — 0 green → white/Gone
|
||
|
||
def test_git_ls_remote_soft_probe_mocked(self):
|
||
calls: list[list[str]] = []
|
||
|
||
def fake_run(cmd, **kwargs):
|
||
calls.append(list(cmd))
|
||
|
||
class R:
|
||
returncode = 0
|
||
stdout = b"abc123\trefs/heads/main\n"
|
||
stderr = b""
|
||
|
||
return R()
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
with mock.patch.dict(
|
||
os.environ, {"TALER_MONITORD_STATE": td, "TALER_MONITORD_GIT": "1"}
|
||
):
|
||
L.refresh_runtime_flags_from_env()
|
||
with mock.patch("stack_status_lib.subprocess.run", side_effect=fake_run):
|
||
got = L.collect_git_probe("git.example", now=1000.0, force=True)
|
||
self.assertTrue(got["ok"])
|
||
self.assertEqual(got["light"], "green")
|
||
self.assertGreaterEqual(len(calls), 2) # https + ssh
|
||
n = len(calls)
|
||
with mock.patch("stack_status_lib.subprocess.run", side_effect=fake_run):
|
||
cached = L.collect_git_probe("git.example", now=1100.0, force=False)
|
||
self.assertTrue(cached["ok"])
|
||
self.assertEqual(len(calls), n) # interval cache hit
|
||
|
||
def test_result_index_optional_by_colour_and_cmd(self):
|
||
import stack_status_html as H
|
||
|
||
doc = {
|
||
"schema": L.SCHEMA,
|
||
"ts": time.time(),
|
||
"acute_down_s": 90,
|
||
"stacks": [
|
||
{
|
||
"id": "stage.example",
|
||
"light": "yellow",
|
||
"intensity": "muted",
|
||
"currency": "CHF",
|
||
"kind": "taler",
|
||
"members": [
|
||
{
|
||
"role": "exchange",
|
||
"host": "exchange.stage.example",
|
||
"light": "green",
|
||
"reason": "ok",
|
||
"endpoints": [
|
||
{
|
||
"id": "config",
|
||
"path": "/config",
|
||
"url": "https://exchange.stage.example/config",
|
||
"light": "green",
|
||
"reason": "ok",
|
||
"skipped": False,
|
||
},
|
||
{
|
||
"id": "keys",
|
||
"path": "/keys",
|
||
"url": "https://exchange.stage.example/keys",
|
||
"light": "yellow",
|
||
"reason": "slow",
|
||
"skipped": False,
|
||
},
|
||
],
|
||
},
|
||
{
|
||
"role": "bank",
|
||
"host": "IBAN",
|
||
"light": "blue",
|
||
"reason": "ebics_iban",
|
||
},
|
||
],
|
||
"network": {
|
||
"zone": "tops-stage",
|
||
"ssh": {
|
||
"expected": True,
|
||
"ok": True,
|
||
"light": "green",
|
||
"host": "",
|
||
"candidates": [],
|
||
"reason": "SSH-2.0-OpenSSH_test",
|
||
},
|
||
},
|
||
}
|
||
],
|
||
"services": [],
|
||
"mail": [],
|
||
}
|
||
off = H.render_html(doc)
|
||
self.assertNotIn('id="result-index"', off)
|
||
|
||
colors = H._render_result_index(doc, mode="colors")
|
||
self.assertIn('id="result-index"', colors)
|
||
self.assertIn("all probes by colour", colors)
|
||
self.assertIn("exchange/config", colors)
|
||
self.assertIn("exchange/keys", colors)
|
||
self.assertIn("bank", colors)
|
||
self.assertIn("ssh", colors)
|
||
self.assertNotIn("ri-cmd", colors)
|
||
|
||
cmd = H._render_result_index(doc, mode="cmd")
|
||
self.assertIn("with command", cmd)
|
||
self.assertIn("GET https://exchange.stage.example/config", cmd)
|
||
self.assertIn("GET https://exchange.stage.example/keys", cmd)
|
||
self.assertIn("TCP", cmd)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|