603 lines
19 KiB
Python
603 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any, Iterator
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
EXPORTER = ROOT / "scripts" / "backup" / "signoz-metadata-export.py"
|
|
VERIFIER = ROOT / "scripts" / "backup" / "verify-signoz-metadata-export.py"
|
|
RESTORE = ROOT / "scripts" / "backup" / "signoz-metadata-restore-drill.py"
|
|
POLICY = ROOT / "config" / "signoz" / "metadata-export-policy.json"
|
|
TRACE_ID = "trace-P0-OBS-002-metadata"
|
|
RUN_ID = "run-P0-OBS-002-metadata"
|
|
WORK_ITEM_ID = "P0-OBS-002"
|
|
TOKEN = "test-only-signoz-token-reference"
|
|
|
|
|
|
SOURCE_ASSETS: dict[str, Any] = {
|
|
"/api/v1/dashboards": {"data": [{"id": "dash-1", "title": "Ops"}]},
|
|
"/api/v1/rules": {"data": [{"id": "rule-1", "name": "Latency"}]},
|
|
"/api/v1/explorer/views": {"data": [{"id": "view-1", "name": "Errors"}]},
|
|
"/api/v1/roles": {"data": [{"id": "role-1", "name": "Viewer"}]},
|
|
"/api/v1/org/preferences": {"data": [{"name": "theme", "value": "dark"}]},
|
|
"/api/v1/user/preferences": {"data": [{"name": "timezone", "value": "UTC"}]},
|
|
"/api/v1/global/config": {"data": {"setupCompleted": True}},
|
|
}
|
|
|
|
|
|
class MetadataHandler(BaseHTTPRequestHandler):
|
|
assets: dict[str, Any] = {}
|
|
drift_path: str | None = None
|
|
get_counts: dict[str, int] = {}
|
|
post_count = 0
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
return
|
|
|
|
def _authorized(self) -> bool:
|
|
return self.headers.get("SIGNOZ-API-KEY") == TOKEN
|
|
|
|
def _send_json(self, status: int, value: Any) -> None:
|
|
payload = json.dumps(value, separators=(",", ":")).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
|
|
def do_GET(self) -> None:
|
|
if not self._authorized():
|
|
self._send_json(401, {"error": "unauthorized"})
|
|
return
|
|
path = self.path.split("?", 1)[0]
|
|
if path not in self.assets:
|
|
self._send_json(404, {"error": "not found"})
|
|
return
|
|
self.get_counts[path] = self.get_counts.get(path, 0) + 1
|
|
value = json.loads(json.dumps(self.assets[path]))
|
|
if self.drift_path == path and self.get_counts[path] >= 2:
|
|
value["data"].append({"id": "drift", "title": "changed"})
|
|
self._send_json(200, value)
|
|
|
|
def do_POST(self) -> None:
|
|
if not self._authorized():
|
|
self._send_json(401, {"error": "unauthorized"})
|
|
return
|
|
path = self.path.split("?", 1)[0]
|
|
if path not in self.assets or not isinstance(
|
|
self.assets[path].get("data"), list
|
|
):
|
|
self._send_json(404, {"error": "not found"})
|
|
return
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
value = json.loads(self.rfile.read(length))
|
|
if not isinstance(value, dict):
|
|
self._send_json(400, {"error": "object required"})
|
|
return
|
|
type(self).post_count += 1
|
|
restored = {"id": f"server-{type(self).post_count}", **value}
|
|
self.assets[path]["data"].append(restored)
|
|
self._send_json(201, {"data": restored})
|
|
|
|
|
|
@contextmanager
|
|
def metadata_server(
|
|
assets: dict[str, Any],
|
|
*,
|
|
drift_path: str | None = None,
|
|
) -> Iterator[tuple[str, type[MetadataHandler]]]:
|
|
handler = type("BoundMetadataHandler", (MetadataHandler,), {})
|
|
handler.assets = json.loads(json.dumps(assets))
|
|
handler.drift_path = drift_path
|
|
handler.get_counts = {}
|
|
handler.post_count = 0
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
host, port = server.server_address
|
|
yield f"http://{host}:{port}", handler
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def credential_file(tmp_path: Path, *, mode: int = 0o600) -> Path:
|
|
path = tmp_path / "signoz-api-key.ref"
|
|
path.write_text(f"{TOKEN}\n", encoding="ascii")
|
|
path.chmod(mode)
|
|
return path
|
|
|
|
|
|
def export_command(
|
|
*,
|
|
base_url: str,
|
|
output_dir: Path,
|
|
credential: Path | None,
|
|
apply: bool,
|
|
receipt_file: Path | None = None,
|
|
) -> list[str]:
|
|
command = [
|
|
sys.executable,
|
|
str(EXPORTER),
|
|
"--apply" if apply else "--check",
|
|
"--base-url",
|
|
base_url,
|
|
"--output-dir",
|
|
str(output_dir),
|
|
"--trace-id",
|
|
TRACE_ID,
|
|
"--run-id",
|
|
RUN_ID,
|
|
"--work-item-id",
|
|
WORK_ITEM_ID,
|
|
]
|
|
if credential is not None:
|
|
command.extend(["--credential-file", str(credential)])
|
|
if apply:
|
|
terminal_path = receipt_file or output_dir.with_name(
|
|
f"{output_dir.name}.terminal.json"
|
|
)
|
|
command.extend(["--receipt-file", str(terminal_path)])
|
|
return command
|
|
|
|
|
|
def create_bundle(tmp_path: Path) -> tuple[Path, Path]:
|
|
credential = credential_file(tmp_path)
|
|
bundle = tmp_path / "bundle"
|
|
with metadata_server(SOURCE_ASSETS) as (base_url, _):
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url=base_url,
|
|
output_dir=bundle,
|
|
credential=credential,
|
|
apply=True,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
return bundle, credential
|
|
|
|
|
|
def isolation_receipt(tmp_path: Path, target_base_url: str) -> Path:
|
|
policy = json.loads(POLICY.read_text(encoding="utf-8"))
|
|
value = {
|
|
"schema": "awoooi_signoz_metadata_restore_isolation_v1",
|
|
"trace_id": TRACE_ID,
|
|
"run_id": RUN_ID,
|
|
"work_item_id": WORK_ITEM_ID,
|
|
"target": {
|
|
"canonical_id": f"ephemeral-signoz-metadata-restore-{RUN_ID}",
|
|
"base_url_sha256": hashlib.sha256(
|
|
target_base_url.encode("utf-8")
|
|
).hexdigest(),
|
|
"image_ref": "signoz/signoz:v0.113.0",
|
|
"image_id": f"sha256:{'a' * 64}",
|
|
"image_present_locally": True,
|
|
"image_pull_performed": False,
|
|
"production_network_attached": False,
|
|
"production_volume_attached": False,
|
|
"ephemeral_volumes_only": True,
|
|
"cleanup_controller_armed": True,
|
|
"zero_residue_verifier_armed": True,
|
|
"network_scope": "loopback_and_internal_only",
|
|
"resource_label_key": policy["restore_isolation"]["resource_label_key"],
|
|
"resource_label_value": RUN_ID,
|
|
},
|
|
}
|
|
path = tmp_path / "isolation.json"
|
|
path.write_text(json.dumps(value), encoding="utf-8")
|
|
path.chmod(0o600)
|
|
return path
|
|
|
|
|
|
def restore_command(
|
|
*,
|
|
mode: str,
|
|
bundle: Path,
|
|
target_base_url: str,
|
|
credential: Path | None,
|
|
isolation: Path,
|
|
receipt_file: Path | None = None,
|
|
) -> list[str]:
|
|
command = [
|
|
sys.executable,
|
|
str(RESTORE),
|
|
mode,
|
|
"--bundle-dir",
|
|
str(bundle),
|
|
"--target-base-url",
|
|
target_base_url,
|
|
"--isolation-receipt",
|
|
str(isolation),
|
|
"--trace-id",
|
|
TRACE_ID,
|
|
"--run-id",
|
|
RUN_ID,
|
|
"--work-item-id",
|
|
WORK_ITEM_ID,
|
|
]
|
|
if credential is not None:
|
|
command.extend(["--credential-file", str(credential)])
|
|
if mode == "--apply":
|
|
terminal_path = receipt_file or bundle.parent / "restore-terminal.json"
|
|
command.extend(["--receipt-file", str(terminal_path)])
|
|
return command
|
|
|
|
|
|
def test_policy_is_explicitly_non_raw_and_non_completion() -> None:
|
|
policy = json.loads(POLICY.read_text(encoding="utf-8"))
|
|
assert policy["source_runtime"]["raw_database_access_allowed"] is False
|
|
assert policy["source_runtime"]["raw_volume_access_allowed"] is False
|
|
assert policy["source_runtime"]["github_supply_chain_allowed"] is False
|
|
assert policy["completion_contract"]["portable_export_alone_is_completion"] is False
|
|
assert (
|
|
policy["completion_contract"]["full_sqlite_completion_claim_allowed"] is False
|
|
)
|
|
assert {item["path"] for item in policy["forbidden_endpoints"]} >= {
|
|
"/api/v1/pats",
|
|
"/api/v1/domains",
|
|
"/api/v1/channels",
|
|
}
|
|
|
|
|
|
def test_check_mode_writes_nothing(tmp_path: Path) -> None:
|
|
output = tmp_path / "must-not-exist"
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url="https://signoz.invalid",
|
|
output_dir=output,
|
|
credential=None,
|
|
apply=False,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout)["terminal"] == "check_pass_no_write"
|
|
assert not output.exists()
|
|
|
|
|
|
def test_apply_and_independent_verifier_pass(tmp_path: Path) -> None:
|
|
bundle, _ = create_bundle(tmp_path)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(VERIFIER),
|
|
"--bundle-dir",
|
|
str(bundle),
|
|
"--expected-trace-id",
|
|
TRACE_ID,
|
|
"--expected-run-id",
|
|
RUN_ID,
|
|
"--expected-work-item-id",
|
|
WORK_ITEM_ID,
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout)["terminal"] == (
|
|
"pass_export_verified_restore_pending"
|
|
)
|
|
manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))
|
|
assert manifest["stable_read_count"] == 2
|
|
assert manifest["raw_sqlite_read"] is False
|
|
assert manifest["sensitive_value_persisted"] is False
|
|
assert manifest["completion_claim"] is False
|
|
assert oct((bundle / "manifest.json").stat().st_mode & 0o777) == "0o600"
|
|
|
|
|
|
def test_secret_key_fails_closed_without_bundle(tmp_path: Path) -> None:
|
|
assets = json.loads(json.dumps(SOURCE_ASSETS))
|
|
assets["/api/v1/global/config"]["data"]["clientSecret"] = "must-not-persist"
|
|
output = tmp_path / "bundle"
|
|
credential = credential_file(tmp_path)
|
|
with metadata_server(assets) as (base_url, _):
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url=base_url,
|
|
output_dir=output,
|
|
credential=credential,
|
|
apply=True,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "forbidden_key_detected" in result.stderr
|
|
assert "must-not-persist" not in result.stderr
|
|
assert not output.exists()
|
|
|
|
|
|
def test_source_drift_fails_closed_without_bundle(tmp_path: Path) -> None:
|
|
output = tmp_path / "bundle"
|
|
credential = credential_file(tmp_path)
|
|
with metadata_server(SOURCE_ASSETS, drift_path="/api/v1/dashboards") as (
|
|
base_url,
|
|
_,
|
|
):
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url=base_url,
|
|
output_dir=output,
|
|
credential=credential,
|
|
apply=True,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "source_metadata_drift_between_stable_reads" in result.stderr
|
|
assert not output.exists()
|
|
|
|
|
|
def test_tamper_breaks_independent_verifier(tmp_path: Path) -> None:
|
|
bundle, _ = create_bundle(tmp_path)
|
|
asset = bundle / "assets" / "dashboards.json"
|
|
asset.write_text('{"data":[]}\n', encoding="utf-8")
|
|
asset.chmod(0o600)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(VERIFIER),
|
|
"--bundle-dir",
|
|
str(bundle),
|
|
"--expected-trace-id",
|
|
TRACE_ID,
|
|
"--expected-run-id",
|
|
RUN_ID,
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "manifest_asset_hash_mismatch_dashboards" in result.stderr
|
|
|
|
|
|
def test_unexpected_bundle_file_breaks_independent_verifier(tmp_path: Path) -> None:
|
|
bundle, _ = create_bundle(tmp_path)
|
|
unexpected = bundle / "unlisted.json"
|
|
unexpected.write_text('{"not":"manifested"}\n', encoding="utf-8")
|
|
unexpected.chmod(0o600)
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(VERIFIER),
|
|
"--bundle-dir",
|
|
str(bundle),
|
|
"--expected-trace-id",
|
|
TRACE_ID,
|
|
"--expected-run-id",
|
|
RUN_ID,
|
|
],
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "bundle_tree_contains_unexpected_entries" in result.stderr
|
|
|
|
|
|
def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None:
|
|
bundle, credential = create_bundle(tmp_path)
|
|
target_assets = {
|
|
"/api/v1/dashboards": {"data": []},
|
|
"/api/v1/rules": {"data": []},
|
|
"/api/v1/explorer/views": {"data": []},
|
|
}
|
|
with metadata_server(target_assets) as (target_base_url, handler):
|
|
isolation = isolation_receipt(tmp_path, target_base_url)
|
|
check_result = subprocess.run(
|
|
restore_command(
|
|
mode="--check",
|
|
bundle=bundle,
|
|
target_base_url=target_base_url,
|
|
credential=credential,
|
|
isolation=isolation,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert check_result.returncode == 0, check_result.stderr
|
|
assert json.loads(check_result.stdout)["terminal"] == "check_pass_no_write"
|
|
assert handler.post_count == 0
|
|
|
|
apply_result = subprocess.run(
|
|
restore_command(
|
|
mode="--apply",
|
|
bundle=bundle,
|
|
target_base_url=target_base_url,
|
|
credential=credential,
|
|
isolation=isolation,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert apply_result.returncode == 0, apply_result.stderr
|
|
assert json.loads(apply_result.stdout)["terminal"] == (
|
|
"partial_degraded_cleanup_pending"
|
|
)
|
|
phase_receipts = json.loads(apply_result.stdout)["phase_receipts"]
|
|
assert [item["phase"] for item in phase_receipts] == [
|
|
"sensor_source",
|
|
"normalized_asset_identity",
|
|
"source_of_truth_diff",
|
|
"ai_decision",
|
|
"risk_policy_decision",
|
|
"check_mode",
|
|
"bounded_execution",
|
|
"independent_post_verifier",
|
|
"rollback",
|
|
"learning_writeback",
|
|
]
|
|
assert phase_receipts[-1]["terminal"] == "pending"
|
|
assert handler.post_count == 3
|
|
|
|
|
|
def test_restore_rejects_non_loopback_target_before_writes(tmp_path: Path) -> None:
|
|
bundle, credential = create_bundle(tmp_path)
|
|
target = "https://signoz.example.invalid"
|
|
isolation = isolation_receipt(tmp_path, target)
|
|
result = subprocess.run(
|
|
restore_command(
|
|
mode="--apply",
|
|
bundle=bundle,
|
|
target_base_url=target,
|
|
credential=credential,
|
|
isolation=isolation,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "restore_target_must_be_loopback" in result.stderr
|
|
|
|
|
|
def test_cleanup_verifier_requires_zero_labeled_residue(tmp_path: Path) -> None:
|
|
bundle, _ = create_bundle(tmp_path)
|
|
fake_bin = tmp_path / "bin"
|
|
fake_bin.mkdir()
|
|
docker = fake_bin / "docker"
|
|
docker.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
docker.chmod(0o755)
|
|
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
target = f"http://127.0.0.1:{port}"
|
|
isolation = isolation_receipt(tmp_path, target)
|
|
env = {**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"}
|
|
result = subprocess.run(
|
|
restore_command(
|
|
mode="--verify-cleanup",
|
|
bundle=bundle,
|
|
target_base_url=target,
|
|
credential=None,
|
|
isolation=isolation,
|
|
),
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout)["terminal"] == "pass_zero_residue_verified"
|
|
|
|
|
|
def test_credential_reference_mode_fails_before_network(tmp_path: Path) -> None:
|
|
credential = credential_file(tmp_path, mode=0o644)
|
|
output = tmp_path / "bundle"
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url="https://signoz.invalid",
|
|
output_dir=output,
|
|
credential=credential,
|
|
apply=True,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "credential_reference_mode_must_be_0400_or_0600" in result.stderr
|
|
assert not output.exists()
|
|
|
|
|
|
def test_credential_symlink_fails_before_network(tmp_path: Path) -> None:
|
|
credential = credential_file(tmp_path)
|
|
link = tmp_path / "credential-link"
|
|
link.symlink_to(credential)
|
|
output = tmp_path / "bundle"
|
|
result = subprocess.run(
|
|
export_command(
|
|
base_url="https://signoz.invalid",
|
|
output_dir=output,
|
|
credential=link,
|
|
apply=True,
|
|
),
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
assert "credential_reference_symlink_component_forbidden" in result.stderr
|
|
assert not output.exists()
|
|
|
|
|
|
def test_apply_failure_writes_terminal_receipt_without_secret(tmp_path: Path) -> None:
|
|
assets = json.loads(json.dumps(SOURCE_ASSETS))
|
|
secret_value = "must-not-appear-in-terminal-receipt"
|
|
assets["/api/v1/global/config"]["data"]["clientSecret"] = secret_value
|
|
output = tmp_path / "bundle"
|
|
receipt_file = tmp_path / "terminal.json"
|
|
credential = credential_file(tmp_path)
|
|
with metadata_server(assets) as (base_url, _):
|
|
command = export_command(
|
|
base_url=base_url,
|
|
output_dir=output,
|
|
credential=credential,
|
|
apply=True,
|
|
receipt_file=receipt_file,
|
|
)
|
|
result = subprocess.run(
|
|
command,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 1
|
|
terminal = json.loads(receipt_file.read_text(encoding="utf-8"))
|
|
assert terminal["terminal"] == "failed_export_not_closable"
|
|
assert secret_value not in receipt_file.read_text(encoding="utf-8")
|
|
assert not output.exists()
|