from __future__ import annotations import copy import json import re from pathlib import Path import pytest import yaml from src.services.notification_matrix import ( build_canonical_telegram_routing_runtime_readback, get_trusted_alert_canonical_route_context, load_canonical_telegram_routing_registry, resolve_canonical_telegram_route, ) EXPECTED_PRODUCTS = { "2026fifa", "agent-bounty-protocol", "awooogo", "awoooi", "bitan-pharmacy", "clawbot-openclaw", "momo-pro", "stockplatform-v2", "tsenyang-website", "vibework", "vtuber", } REPO_ROOT = Path(__file__).resolve().parents[3] def test_runtime_readback_separates_historical_source_state_from_live_gate() -> None: readback = build_canonical_telegram_routing_runtime_readback() assert readback["status"] == ( "runtime_policy_enforced_product_delivery_receipts_partial" ) assert readback["source_status"] == "source_policy_ready_runtime_not_applied" assert readback["source_generated_at"] == "2026-07-14T15:50:04+08:00" assert readback["runtime_generated_at"].endswith("+00:00") assert readback["runtime_enforcement"] == { "canonical_resolver_active": True, "gateway_pre_send_gate_active": True, "trusted_ingress_context_required": True, "unknown_route_decision": "block", "unknown_route_reason": "unknown_product_signal_or_severity", "raw_numeric_destination_count": 0, "destination_identity_exposed": False, "all_product_delivery_receipts_ready": False, } assert readback["rollups"]["runtime_allowed_route_count"] == 4 assert readback["rollups"]["runtime_blocked_route_count"] == 16 assert len(readback["products"]) == 11 assert len(readback["routes"]) == 20 assert readback["operation_boundaries"] == { "read_only": True, "telegram_send_performed": False, "bot_api_call_performed": False, "runtime_route_changed": False, "secret_value_read": False, "raw_identifier_included": False, } serialized = json.dumps(readback, ensure_ascii=False) assert '"chat_id"' not in serialized assert '"bot_token"' not in serialized def test_namespaced_product_ingress_route_is_explicit_and_policy_blocked() -> None: context = get_trusted_alert_canonical_route_context( { "awoooi_product_id": "momo-pro", "awoooi_signal_family": "crawler_and_data_freshness", "awoooi_route_severity": "P2", }, source="prometheus_alertmanager", source_severity="warning", ) assert context == { "product_id": "momo-pro", "signal_family": "crawler_and_data_freshness", "severity": "P2", "route_source": "prometheus_alertmanager", } assert resolve_canonical_telegram_route(**{ key: context[key] for key in ("product_id", "signal_family", "severity") })["decision"] == "block" @pytest.mark.parametrize( ("labels", "source", "severity"), [ ({}, "prometheus_alertmanager", "critical"), ( { "awoooi_product_id": "momo-pro", "awoooi_signal_family": "crawler_and_data_freshness", }, "untrusted_ingress", "critical", ), ( { "awoooi_product_id": "awoooi", "awoooi_signal_family": "incident_lifecycle", "awoooi_route_severity": "P1", }, "signoz_webhook", "critical", ), ( { "awoooi_product_id": "momo-pro", "awoooi_signal_family": "crawler_and_data_freshness", "awoooi_route_severity": "SEV-1", }, "prometheus_alertmanager", "critical", ), ], ) def test_product_ingress_route_fails_closed_without_complete_trusted_context( labels: dict[str, str], source: str, severity: str ) -> None: assert get_trusted_alert_canonical_route_context( labels, source=source, source_severity=severity, ) is None @pytest.mark.parametrize("rule_file", ["alerts-unified.yml", "alerts.yml"]) def test_product_alert_rules_carry_canonical_route_labels(rule_file: str) -> None: payload = yaml.safe_load((REPO_ROOT / "ops/monitoring" / rule_file).read_text()) rules = { rule["alert"]: rule["labels"] for group in payload["groups"] for rule in group.get("rules", []) if "alert" in rule } expected = { "OpenClawDown": ("clawbot-openclaw", "raw_monitoring_alert", "P1"), "MoWoooWorkDown": ("momo-pro", "scheduler_and_sales_pipeline", "P1"), "TsenyangWebsiteDown": ( "tsenyang-website", "public_site_and_agent_action_health", "P1", ), "StockWoooWorkDown": ( "stockplatform-v2", "market_data_and_recommendation_freshness", "P1", ), "BitanWoooWorkDown": ( "bitan-pharmacy", "container_and_product_health", "P1", ), "MomoScraperSuccessLow": ( "momo-pro", "crawler_and_data_freshness", "P2", ), } for alertname, route in expected.items(): labels = rules[alertname] assert ( labels["awoooi_product_id"], labels["awoooi_signal_family"], labels["awoooi_route_severity"], ) == route def test_specialized_openclaw_rule_cannot_fall_back_to_shared_sre() -> None: payload = yaml.safe_load( (REPO_ROOT / "k8s/monitoring/k3s-alerts-supplemental.yaml").read_text() ) rule = next( rule for group in payload["groups"] for rule in group.get("rules", []) if rule.get("alert") == "OpenClawDown" ) labels = rule["labels"] assert { key: labels[key] for key in ( "awoooi_product_id", "awoooi_signal_family", "awoooi_route_severity", ) } == { "awoooi_product_id": "clawbot-openclaw", "awoooi_signal_family": "raw_monitoring_alert", "awoooi_route_severity": "P2", } def test_active_source_has_no_raw_numeric_telegram_destination_defaults() -> None: candidate_paths = [ REPO_ROOT / "docker-compose.yml", REPO_ROOT / "k8s/awoooi-prod/04-configmap.yaml", REPO_ROOT / "scripts/ops/deploy-docker-health-monitor.sh", REPO_ROOT / "scripts/ops/docker-health-monitor.sh", *(REPO_ROOT / ".gitea/workflows").glob("*.y*ml"), ] raw_numeric_identifier = re.compile(r"-?\d{8,}") violations: list[str] = [] for path in candidate_paths: for line_number, line in enumerate(path.read_text().splitlines(), 1): if ( any(key in line for key in ("CHAT_ID", "USER_WHITELIST")) and raw_numeric_identifier.search(line) ): violations.append(f"{path.relative_to(REPO_ROOT)}:{line_number}") assert violations == [] def test_registry_covers_all_11_products_and_reconciles_rollups() -> None: registry = load_canonical_telegram_routing_registry() assert { product["product_id"] for product in registry["products"] } == EXPECTED_PRODUCTS assert registry["rollups"] == { "product_count": 11, "route_count": 20, "allowed_route_count": 4, "blocked_route_count": 13, "disabled_route_count": 2, "not_implemented_route_count": 1, "shared_sre_allowed_route_count": 4, "raw_numeric_destination_count": 0, "runtime_route_change_count": 0, "telegram_send_count": 0, } @pytest.mark.parametrize( "signal_family", [ "shared_infrastructure", "security_incident", "disaster_recovery", "incident_lifecycle", ], ) @pytest.mark.parametrize("severity", ["P0", "P1"]) def test_shared_sre_accepts_only_allowlisted_p0_p1_lifecycle( signal_family: str, severity: str ) -> None: decision = resolve_canonical_telegram_route("awoooi", signal_family, severity) assert decision["decision"] == "allow" assert decision["chat_alias"] == "awoooi_sre_war_room" assert decision["allowed_destination"] == ( "telegram_chat_alias:awoooi_sre_war_room" ) assert decision["receipt_backend"] == ( "awoooi.alert_operation_log+telegram_outbound_receipts" ) @pytest.mark.parametrize( ("signal_family", "severity"), [ ("product_raw_monitoring", "P2"), ("product_raw_monitoring", "P3"), ("marketing", "P1"), ], ) def test_shared_sre_blocks_raw_p2_p3_and_marketing( signal_family: str, severity: str ) -> None: decision = resolve_canonical_telegram_route("awoooi", signal_family, severity) assert decision["decision"] == "block" assert decision["allowed_destination"] == "blocked" @pytest.mark.parametrize("product_id", ["awooogo", "vibework", "vtuber"]) def test_disabled_or_not_implemented_products_never_resolve_to_egress( product_id: str, ) -> None: decision = resolve_canonical_telegram_route(product_id, "any-signal", "P0") assert decision["decision"] == "block" assert decision["egress_status"] in {"disabled", "not_implemented"} assert decision["allowed_destination"] == "blocked" @pytest.mark.parametrize( ("product_id", "signal_family", "severity"), [ ("unknown-product", "shared_infrastructure", "P0"), ("awoooi", "unknown-family", "P0"), ("awoooi", "shared_infrastructure", "P3"), ("awoooi", "shared_infrastructure", "SEV-1"), ], ) def test_unknown_route_dimensions_are_fail_closed( product_id: str, signal_family: str, severity: str ) -> None: decision = resolve_canonical_telegram_route(product_id, signal_family, severity) assert decision["decision"] == "block" assert decision["route_id"] is None assert decision["allowed_destination"] == "blocked" assert decision["reason"] == "unknown_product_signal_or_severity" def test_requested_destination_must_match_the_canonical_alias() -> None: decision = resolve_canonical_telegram_route( "awoooi", "security_incident", "P0", requested_destination="telegram_chat_alias:some_other_room", ) assert decision["decision"] == "block" assert decision["reason"] == "requested_destination_not_allowed" def test_explicit_empty_registry_is_invalid_instead_of_loading_default() -> None: with pytest.raises(ValueError, match="schema_version"): resolve_canonical_telegram_route( "awoooi", "security_incident", "P0", registry={}, ) def test_product_level_blocked_disposition_prevents_route_egress() -> None: decision = resolve_canonical_telegram_route( "agent-bounty-protocol", "a2a_task_lifecycle", "P1", ) assert decision["decision"] == "block" assert decision["egress_status"] == "blocked" assert decision["allowed_destination"] == "blocked" def test_registry_contains_only_symbolic_aliases_not_runtime_identifiers() -> None: registry = load_canonical_telegram_routing_registry() serialized = json.dumps(registry, ensure_ascii=False) assert '"chat_id"' not in serialized assert '"bot_token"' not in serialized for route in registry["routes"]: for field in ("bot_alias", "chat_alias", "allowed_destination"): assert not str(route[field]).lstrip("-").isdigit() def test_shared_monitoring_routes_bind_tsenyang_owner_and_destination_policy() -> None: registry = load_canonical_telegram_routing_registry() shared_routes = [ route for route in registry["routes"] if route["egress_status"] == "allowed" and route["scope"] == "shared" ] assert shared_routes assert {route["bot_alias"] for route in shared_routes} == {"tsenyang_bot"} owner_role = next( role for role in registry["destination_roles"] if role["chat_alias"] == "awoooi_sre_war_room" and role["bot_alias"] == "tsenyang_bot" ) assert owner_role["bot_owner"] == "sre_shared_runtime" assert owner_role["monitoring_owner"] is True assert owner_role["monitoring_egress_policy"] == "allowlist_only" @pytest.mark.parametrize( ("field", "value", "error"), [ ("bot_owner", "ai_agent_runtime", "bot ownership"), ("monitoring_owner", False, "not the monitoring owner"), ( "monitoring_egress_policy", "interaction_reply_only_no_monitoring_ownership", "not monitoring allowlisted", ), ], ) def test_tampered_shared_destination_ownership_is_rejected( tmp_path: Path, field: str, value: object, error: str, ) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) role = next( role for role in registry["destination_roles"] if role["chat_alias"] == "awoooi_sre_war_room" and role["bot_alias"] == "tsenyang_bot" ) role[field] = value _write_registry(tmp_path, registry) with pytest.raises(ValueError, match=error): load_canonical_telegram_routing_registry(tmp_path) def test_tampered_shared_route_cannot_switch_to_openclaw_bot(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) route = next( route for route in registry["routes"] if route["route_id"] == "awoooi-security-incident" ) route["bot_alias"] = "openclaw_bot" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="shared SRE P0/P1"): load_canonical_telegram_routing_registry(tmp_path) @pytest.mark.parametrize("visible_label", ["TsenYang", "小O_小龍蝦"]) def test_visible_labels_do_not_claim_proven_runtime_alias( tmp_path: Path, visible_label: str, ) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) role = next( role for role in registry["destination_roles"] if role["visible_label"] == visible_label ) assert role["runtime_alias_evidence"] == "visible_label_only" assert role["proven_runtime_alias"] == "none" role["proven_runtime_alias"] = role["chat_alias"] _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="visible label must not claim"): load_canonical_telegram_routing_registry(tmp_path) def test_direct_egress_scanners_cover_agent99_and_reboot_recovery() -> None: repo_root = Path(__file__).resolve().parents[3] scanner_paths = [ repo_root / "scripts/security/telegram-notification-egress-inventory.py", repo_root / "scripts/security/telegram-notification-egress-no-new-bypass-guard.py", ] for scanner_path in scanner_paths: source = scanner_path.read_text(encoding="utf-8") assert 'Path("agent99-control-plane.ps1")' in source assert 'Path("scripts/reboot-recovery")' in source assert '".ps1"' in source guarded_source = "\n".join( [ (repo_root / "agent99-control-plane.ps1").read_text(encoding="utf-8"), *( path.read_text(encoding="utf-8", errors="replace") for path in sorted((repo_root / "scripts/reboot-recovery").rglob("*")) if path.is_file() and path.suffix in {".ps1", ".sh", ".py"} ), ] ) assert "api.telegram.org/bot" not in guarded_source def test_tampered_shared_sre_route_is_rejected(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) route = next( route for route in registry["routes"] if route["route_id"] == "awoooi-shared-infrastructure" ) route["severity"].append("P3") _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="shared SRE P0/P1"): load_canonical_telegram_routing_registry(tmp_path) def test_tampered_allowed_route_without_receipt_is_rejected(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) route = next( route for route in registry["routes"] if route["egress_status"] == "allowed" ) route["receipt_backend"] = "none" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="receipt backend"): load_canonical_telegram_routing_registry(tmp_path) def test_tampered_unknown_destination_is_rejected(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) route = next( route for route in registry["routes"] if route["egress_status"] == "blocked" ) route["allowed_destination"] = "telegram_chat_alias:guessed_room" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="non-allowed route must be blocked"): load_canonical_telegram_routing_registry(tmp_path) def test_tampered_raw_chat_identifier_is_rejected(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) registry["routes"][0]["chat_alias"] = str(-(10**12)) _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="raw numeric runtime identifier"): load_canonical_telegram_routing_registry(tmp_path) def test_tampered_blocked_product_cannot_gain_an_allowed_route(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) route = next( route for route in registry["routes"] if route["route_id"] == "bitan-product-health" ) route.update( { "bot_alias": "bitan_bot", "chat_alias": "bitan_product_ops", "allowed_destination": "telegram_chat_alias:bitan_product_ops", "receipt_backend": "bitan.delivery_receipts", "egress_status": "allowed", } ) _write_registry(tmp_path, registry) with pytest.raises( ValueError, match="bitan-pharmacy must not have an allowed route" ): load_canonical_telegram_routing_registry(tmp_path) def test_recursive_guard_rejects_sensitive_runtime_key(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) registry["destination_roles"][0]["bot_token"] = "redacted-is-still-forbidden" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="forbidden sensitive runtime key"): load_canonical_telegram_routing_registry(tmp_path) @pytest.mark.parametrize( "sensitive_key", [ "api_key", "password", "credential", "private_key", "cookie", "session", ], ) def test_recursive_guard_rejects_extended_sensitive_runtime_keys( tmp_path: Path, sensitive_key: str, ) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) registry["destination_roles"][0][sensitive_key] = "redacted" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="forbidden sensitive runtime key"): load_canonical_telegram_routing_registry(tmp_path) def test_recursive_guard_rejects_secret_shaped_runtime_value(tmp_path: Path) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) registry["destination_roles"][0]["role"] = f"{10**7}:{'x' * 24}" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="secret-shaped runtime value"): load_canonical_telegram_routing_registry(tmp_path) @pytest.mark.parametrize( ("section", "field"), [ ("top", "status"), ("top", "destination_roles"), ("product", "owner"), ("route", "signal_family"), ("policy", "default_decision"), ("rollups", "route_count"), ("destination_role", "monitoring_egress_policy"), ("supersedes", "owner"), ("execution_boundaries", "telegram_send_allowed"), ], ) def test_runtime_validator_rejects_missing_fields_at_every_object_level( tmp_path: Path, section: str, field: str, ) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) target = _registry_object_for_section(registry, section) target.pop(field) _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="missing required fields"): load_canonical_telegram_routing_registry(tmp_path) @pytest.mark.parametrize( "section", [ "top", "product", "route", "policy", "rollups", "destination_role", "supersedes", "execution_boundaries", ], ) def test_runtime_validator_rejects_unexpected_fields_at_every_object_level( tmp_path: Path, section: str, ) -> None: registry = copy.deepcopy(load_canonical_telegram_routing_registry()) target = _registry_object_for_section(registry, section) target["unexpected_field"] = "must-fail" _write_registry(tmp_path, registry) with pytest.raises(ValueError, match="unexpected fields"): load_canonical_telegram_routing_registry(tmp_path) def test_schema_closes_every_data_object_and_validates_snapshot() -> None: import jsonschema repo_root = Path(__file__).resolve().parents[3] schema = json.loads( ( repo_root / "docs/schemas/telegram_canonical_routing_registry_v1.schema.json" ).read_text(encoding="utf-8") ) def assert_closed_objects(node: object, path: str = "schema") -> None: if isinstance(node, dict): if node.get("type") == "object": assert node.get("additionalProperties") is False, path for key, nested in node.items(): assert_closed_objects(nested, f"{path}.{key}") elif isinstance(node, list): for index, nested in enumerate(node): assert_closed_objects(nested, f"{path}[{index}]") assert_closed_objects(schema) jsonschema.Draft202012Validator.check_schema(schema) jsonschema.Draft202012Validator(schema).validate( load_canonical_telegram_routing_registry() ) def _registry_object_for_section(registry: dict, section: str) -> dict: if section == "top": return registry if section == "product": return next( product for product in registry["products"] if product["product_id"] == "awoooi" ) if section == "route": return registry["routes"][0] if section == "destination_role": return registry["destination_roles"][0] return registry[section] def _write_registry(directory: Path, registry: dict) -> None: (directory / "telegram-canonical-routing-registry.snapshot.json").write_text( json.dumps(registry), encoding="utf-8", )