diff --git a/scripts/ops/apply_metrics_edge_policy.py b/scripts/ops/apply_metrics_edge_policy.py new file mode 100755 index 0000000..78d7d0c --- /dev/null +++ b/scripts/ops/apply_metrics_edge_policy.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Check or apply the source-owned EwoooC metrics edge policy.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from services.metrics_edge_controlled_apply_service import ( # noqa: E402 + apply_metrics_edge_plan, + build_metrics_edge_plan, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--nginx-config", type=Path, required=True) + parser.add_argument("--prometheus-config", type=Path, required=True) + parser.add_argument("--nginx-fragment", type=Path, required=True) + parser.add_argument("--prometheus-snippet", type=Path, required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--work-item-id", default="SEC-P0-001") + parser.add_argument("--apply", action="store_true") + parser.add_argument("--canary-timeout", type=int, default=90) + parser.add_argument("--receipt", type=Path) + args = parser.parse_args() + + plan = build_metrics_edge_plan( + nginx_config=args.nginx_config, + prometheus_config=args.prometheus_config, + nginx_fragment=args.nginx_fragment, + prometheus_snippet=args.prometheus_snippet, + run_id=args.run_id, + work_item_id=args.work_item_id, + ) + payload = apply_metrics_edge_plan( + plan, + canary_timeout=args.canary_timeout, + ) if args.apply else {key: value for key, value in plan.items() if not key.startswith("_")} + output = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + if args.receipt: + args.receipt.parent.mkdir(parents=True, exist_ok=True) + args.receipt.write_text(f"{output}\n", encoding="utf-8") + print(output) + return 0 if payload.get("status") not in {"rolled_back", "rollback_degraded"} else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/metrics_edge_controlled_apply_service.py b/services/metrics_edge_controlled_apply_service.py new file mode 100644 index 0000000..4425514 --- /dev/null +++ b/services/metrics_edge_controlled_apply_service.py @@ -0,0 +1,297 @@ +"""Controlled apply for the EwoooC metrics edge and Prometheus target.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import subprocess +import tempfile +import time +import urllib.request +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +POLICY = "ewoooc_metrics_edge_controlled_apply_v1" +NGINX_BEGIN = "# BEGIN MANAGED EWOOOC METRICS" +NGINX_END = "# END MANAGED EWOOOC METRICS" +PROMETHEUS_BEGIN = "# BEGIN MANAGED EWOOOC METRICS" +PROMETHEUS_END = "# END MANAGED EWOOOC METRICS" +IDENTITY_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{2,127}$") + + +def _managed_replace(text: str, begin: str, end: str, managed: str) -> str | None: + begin_count = text.count(begin) + end_count = text.count(end) + if begin_count == 0 and end_count == 0: + return None + if begin_count != 1 or end_count != 1: + raise ValueError(f"invalid managed marker count for {begin}") + start = text.index(begin) + finish = text.index(end, start) + len(end) + return f"{text[:start]}{managed}{text[finish:]}" + + +def _nginx_server_blocks(lines: list[str]) -> list[tuple[int, int]]: + blocks: list[tuple[int, int]] = [] + start: int | None = None + depth = 0 + for index, line in enumerate(lines): + code = line.split("#", 1)[0] + if start is None and re.match(r"^\s*server\s*\{", code): + start = index + depth = 0 + if start is None: + continue + depth += code.count("{") - code.count("}") + if depth == 0: + blocks.append((start, index + 1)) + start = None + if start is not None: + raise ValueError("unbalanced nginx server block") + return blocks + + +def render_nginx_config(config_text: str, fragment_text: str) -> str: + fragment = fragment_text.strip() + if "location = /metrics" not in fragment or "deny all;" not in fragment: + raise ValueError("metrics nginx fragment is incomplete") + indented_fragment = "\n".join(f" {line}" if line else "" for line in fragment.splitlines()) + managed = f"{NGINX_BEGIN}\n{indented_fragment}\n {NGINX_END}" + replaced = _managed_replace(config_text, NGINX_BEGIN, NGINX_END, managed) + if replaced is not None: + return replaced + + lines = config_text.splitlines(keepends=True) + candidates: list[tuple[int, int]] = [] + for start, end in _nginx_server_blocks(lines): + block = "".join(lines[start:end]) + if "server_name mo.wooo.work;" in block and re.search(r"\blisten\s+443\b", block): + candidates.append((start, end)) + if len(candidates) != 1: + raise ValueError(f"expected one mo.wooo.work TLS server, found {len(candidates)}") + + start, end = candidates[0] + location_indexes = [ + index + for index in range(start, end) + if re.match(r"^\s*location\s+/\s*\{", lines[index]) + ] + if len(location_indexes) != 1: + raise ValueError(f"expected one fallback location, found {len(location_indexes)}") + insert_at = location_indexes[0] + lines.insert(insert_at, f" {managed}\n\n") + return "".join(lines) + + +def render_prometheus_config(config_text: str, snippet_text: str) -> str: + snippet = snippet_text.rstrip() + if "job_name: 'ewoooc-app'" not in snippet or "metrics_path: /metrics" not in snippet: + raise ValueError("EwoooC Prometheus snippet is incomplete") + if not re.search(r"(?m)^scrape_configs:\s*$", config_text): + raise ValueError("Prometheus config has no scrape_configs section") + managed = f"{PROMETHEUS_BEGIN}\n{snippet}\n{PROMETHEUS_END}" + replaced = _managed_replace( + config_text, + PROMETHEUS_BEGIN, + PROMETHEUS_END, + managed, + ) + if replaced is not None: + return replaced + return f"{config_text.rstrip()}\n\n{managed}\n" + + +def _sha256(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def build_metrics_edge_plan( + *, + nginx_config: Path, + prometheus_config: Path, + nginx_fragment: Path, + prometheus_snippet: Path, + run_id: str, + work_item_id: str, +) -> dict[str, Any]: + if not IDENTITY_PATTERN.fullmatch(run_id) or not IDENTITY_PATTERN.fullmatch(work_item_id): + raise ValueError("run_id and work_item_id must use bounded safe characters") + nginx_before = nginx_config.read_text(encoding="utf-8") + prometheus_before = prometheus_config.read_text(encoding="utf-8") + nginx_after = render_nginx_config( + nginx_before, + nginx_fragment.read_text(encoding="utf-8"), + ) + prometheus_after = render_prometheus_config( + prometheus_before, + prometheus_snippet.read_text(encoding="utf-8"), + ) + return { + "policy": POLICY, + "trace_id": run_id, + "run_id": run_id, + "work_item_id": work_item_id, + "mode": "check", + "generated_at": datetime.now(timezone.utc).isoformat(), + "nginx": { + "path": str(nginx_config), + "before_sha256": _sha256(nginx_before), + "candidate_sha256": _sha256(nginx_after), + "changed": nginx_before != nginx_after, + }, + "prometheus": { + "path": str(prometheus_config), + "before_sha256": _sha256(prometheus_before), + "candidate_sha256": _sha256(prometheus_after), + "changed": prometheus_before != prometheus_after, + }, + "_candidate_text": { + "nginx": nginx_after, + "prometheus": prometheus_after, + }, + "controlled_apply_allowed": True, + "risk": "medium", + } + + +def _run(command: list[str], timeout: int = 30) -> subprocess.CompletedProcess[str]: + completed = subprocess.run( + command, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + if completed.returncode != 0: + message = (completed.stderr or completed.stdout or "command failed").strip() + raise RuntimeError(f"{command[0]} failed: {message[:500]}") + return completed + + +def _atomic_write(path: Path, content: str) -> None: + current = path.stat() + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=path.parent, + prefix=f".{path.name}.", + delete=False, + ) as handle: + handle.write(content) + temporary = Path(handle.name) + os.chmod(temporary, current.st_mode) + os.chown(temporary, current.st_uid, current.st_gid) + os.replace(temporary, path) + + +def _prometheus_target_health(timeout_seconds: int) -> str: + deadline = time.monotonic() + timeout_seconds + url = "http://127.0.0.1:9090/api/v1/targets?state=active" + last_health = "missing" + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(url, timeout=8) as response: + payload = json.load(response) + targets = ((payload.get("data") or {}).get("activeTargets") or []) + matching = [ + target for target in targets + if (target.get("labels") or {}).get("job") == "ewoooc-app" + ] + if matching: + last_health = str(matching[0].get("health") or "unknown") + if last_health == "up": + return last_health + except (OSError, ValueError, TypeError): + last_health = "query_error" + time.sleep(3) + return last_health + + +def apply_metrics_edge_plan(plan: dict[str, Any], *, canary_timeout: int = 90) -> dict[str, Any]: + candidate = plan.pop("_candidate_text") + nginx_path = Path(plan["nginx"]["path"]) + prometheus_path = Path(plan["prometheus"]["path"]) + suffix = re.sub(r"[^A-Za-z0-9_.-]", "_", plan["run_id"]) + nginx_backup = nginx_path.with_name(f"{nginx_path.name}.bak.{suffix}") + prometheus_backup = prometheus_path.with_name(f"{prometheus_path.name}.bak.{suffix}") + shutil.copy2(nginx_path, nginx_backup) + shutil.copy2(prometheus_path, prometheus_backup) + receipt = { + key: value for key, value in plan.items() + if not key.startswith("_") + } + receipt.update({ + "receipt_kind": "metrics_edge_controlled_apply_receipt", + "mode": "apply", + "status": "in_progress", + "backup_paths": { + "nginx": str(nginx_backup), + "prometheus": str(prometheus_backup), + }, + "rollback_performed": False, + "post_verifier": {}, + }) + try: + _atomic_write(nginx_path, candidate["nginx"]) + _atomic_write(prometheus_path, candidate["prometheus"]) + _run(["nginx", "-t"]) + _run([ + "docker", "exec", "prometheus", "promtool", "check", "config", + "/etc/prometheus/prometheus.yml", + ]) + _run(["systemctl", "reload", "nginx"]) + _run(["docker", "kill", "--signal=SIGHUP", "prometheus"]) + internal = _run([ + "curl", "-ksS", "--max-time", "20", "--resolve", + "mo.wooo.work:443:127.0.0.1", "-o", "/dev/null", "-w", "%{http_code}", + "https://mo.wooo.work/metrics", + ]) + internal_code = internal.stdout.strip() + target_health = _prometheus_target_health(max(10, canary_timeout)) + receipt["post_verifier"] = { + "nginx_config_valid": True, + "prometheus_config_valid": True, + "internal_metrics_http_code": internal_code, + "metrics_internal_canary": internal_code == "200", + "metrics_target_health": target_health, + } + if internal_code != "200" or target_health != "up": + raise RuntimeError( + f"post-verifier failed: internal={internal_code}, target={target_health}" + ) + receipt["status"] = "pass" + receipt["completed_at"] = datetime.now(timezone.utc).isoformat() + return receipt + except Exception as exc: + shutil.copy2(nginx_backup, nginx_path) + shutil.copy2(prometheus_backup, prometheus_path) + rollback_errors: list[str] = [] + for command in ( + ["nginx", "-t"], + ["systemctl", "reload", "nginx"], + ["docker", "kill", "--signal=SIGHUP", "prometheus"], + ): + try: + _run(command) + except Exception as rollback_exc: + rollback_errors.append(str(rollback_exc)) + receipt["status"] = "rolled_back" if not rollback_errors else "rollback_degraded" + receipt["rollback_performed"] = True + receipt["error"] = str(exc) + receipt["rollback_errors"] = rollback_errors + receipt["completed_at"] = datetime.now(timezone.utc).isoformat() + return receipt + + +__all__ = [ + "POLICY", + "apply_metrics_edge_plan", + "build_metrics_edge_plan", + "render_nginx_config", + "render_prometheus_config", +] diff --git a/tests/test_metrics_edge_controlled_apply_service.py b/tests/test_metrics_edge_controlled_apply_service.py new file mode 100644 index 0000000..ad7e8e5 --- /dev/null +++ b/tests/test_metrics_edge_controlled_apply_service.py @@ -0,0 +1,103 @@ +from pathlib import Path + +import pytest + + +NGINX_CONFIG = """server { + listen 80; + server_name mo.wooo.work; + return 301 https://$server_name$request_uri; +} + +server { + listen 443 ssl http2; + server_name mo.wooo.work; + + location / { + proxy_pass http://127.0.0.1:5003; + } +} +""" + +NGINX_FRAGMENT = """location = /metrics { + allow 127.0.0.1; + allow 172.16.0.0/12; + deny all; + proxy_pass http://127.0.0.1:5003/metrics; +} +""" + +PROMETHEUS_CONFIG = """global: + scrape_interval: 60s + +scrape_configs: + - job_name: existing + static_configs: + - targets: ['127.0.0.1:9090'] +""" + +PROMETHEUS_SNIPPET = """ - job_name: 'ewoooc-app' + metrics_path: /metrics + static_configs: + - targets: ['mo.wooo.work'] +""" + + +def test_render_nginx_config_targets_only_tls_server_and_is_idempotent(): + from services.metrics_edge_controlled_apply_service import render_nginx_config + + rendered = render_nginx_config(NGINX_CONFIG, NGINX_FRAGMENT) + second = render_nginx_config(rendered, NGINX_FRAGMENT) + + assert rendered == second + assert rendered.count("location = /metrics") == 1 + assert rendered.index("listen 443") < rendered.index("location = /metrics") + assert rendered.index("location = /metrics") < rendered.index("location / {") + + +def test_render_prometheus_config_is_idempotent(): + from services.metrics_edge_controlled_apply_service import render_prometheus_config + + rendered = render_prometheus_config(PROMETHEUS_CONFIG, PROMETHEUS_SNIPPET) + second = render_prometheus_config(rendered, PROMETHEUS_SNIPPET) + + assert rendered == second + assert rendered.count("job_name: 'ewoooc-app'") == 1 + + +def test_build_plan_is_no_write_and_commit_safe(tmp_path: Path): + from services.metrics_edge_controlled_apply_service import build_metrics_edge_plan + + paths = { + "nginx_config": tmp_path / "nginx.conf", + "prometheus_config": tmp_path / "prometheus.yml", + "nginx_fragment": tmp_path / "location.conf", + "prometheus_snippet": tmp_path / "scrape.yml", + } + paths["nginx_config"].write_text(NGINX_CONFIG, encoding="utf-8") + paths["prometheus_config"].write_text(PROMETHEUS_CONFIG, encoding="utf-8") + paths["nginx_fragment"].write_text(NGINX_FRAGMENT, encoding="utf-8") + paths["prometheus_snippet"].write_text(PROMETHEUS_SNIPPET, encoding="utf-8") + + plan = build_metrics_edge_plan( + **paths, + run_id="sec-p0-001-test", + work_item_id="SEC-P0-001", + ) + + assert plan["mode"] == "check" + assert plan["nginx"]["changed"] is True + assert plan["prometheus"]["changed"] is True + assert paths["nginx_config"].read_text(encoding="utf-8") == NGINX_CONFIG + + +def test_render_rejects_incomplete_templates(): + from services.metrics_edge_controlled_apply_service import ( + render_nginx_config, + render_prometheus_config, + ) + + with pytest.raises(ValueError, match="incomplete"): + render_nginx_config(NGINX_CONFIG, "location = /metrics {}") + with pytest.raises(ValueError, match="incomplete"): + render_prometheus_config(PROMETHEUS_CONFIG, "job_name: other")