wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(signoz): scope explorer metadata exports

Browse Source
This commit is contained in:
Your Name
2026-07-22 19:43:47 +08:00
parent 7a66401d22
commit 15bebe3edc
8 changed files with 385 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
scripts/backup/signoz-metadata-export.py
View File

@@ -101,7 +101,10 @@ def stable_read(
documents: dict[str, Any] = {}
counts: dict[str, int] = {}
for asset in policy["assets"]:
document = client.request_json(asset["endpoint"])
document = client.request_json(
asset["endpoint"],
query_parameters=asset.get("query_parameters"),
)
counts[asset["id"]] = validate_asset_document(document, asset, policy)
documents[asset["id"]] = document
pass_documents.append(documents)
@@ -142,6 +145,7 @@ def create_bundle(
"id": asset_id,
"classification": asset["classification"],
"endpoint": asset["endpoint"],
"query_parameters": asset.get("query_parameters", {}),
"file": f"assets/{asset_id}.json",
"sha256": sha256_bytes(payload),
"size_bytes": len(payload),

5
scripts/backup/signoz-metadata-restore-drill.py
View File

@@ -239,7 +239,10 @@ def read_target_stable(
for _ in range(2):
current: dict[str, list[dict[str, Any]]] = {}
for asset, _ in assets:
document = client.request_json(asset["endpoint"])
document = client.request_json(
asset["endpoint"],
query_parameters=asset.get("query_parameters"),
)
validate_asset_document(document, asset, policy)
current[asset["id"]] = semantic_items(document, asset)
snapshots.append(current)

87
scripts/backup/signoz_metadata_contract.py
View File

@@ -28,6 +28,8 @@ SAFE_ASSET_ID = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
ALLOWED_TOKEN_MODES = {0o400, 0o600}
LOOPBACK_HOSTS = {"127.0.0.1", "::1", "localhost"}
EXPLORER_VIEW_SOURCES = {"logs", "meter", "metrics", "traces"}
EMPTY_COLLECTION_SENTINELS = {"null", "empty_object"}
class ContractError(ValueError):
@@ -95,6 +97,19 @@ def _normalized_key(value: str) -> str:
return re.sub(r"[^a-z0-9]", "", value.casefold())
def validate_query_parameters(endpoint: str, value: Any) -> dict[str, str]:
if value is None:
return {}
if not isinstance(value, dict):
raise ContractError("api_query_parameters_invalid")
if endpoint != "/api/v1/explorer/views" or set(value) != {"sourcePage"}:
raise ContractError("api_query_parameters_not_allowlisted")
source_page = value.get("sourcePage")
if not isinstance(source_page, str) or source_page not in EXPLORER_VIEW_SOURCES:
raise ContractError("api_query_source_page_invalid")
return {"sourcePage": source_page}
def require_no_symlink_components(path: Path, *, label: str) -> None:
if not path.is_absolute():
path = Path.cwd() / path
@@ -167,7 +182,7 @@ def load_policy(path: Path) -> dict[str, Any]:
if isinstance(item, dict)
}
asset_ids: set[str] = set()
asset_paths: set[str] = set()
asset_paths: set[tuple[str, tuple[tuple[str, str], ...]]] = set()
for asset in assets:
if not isinstance(asset, dict):
raise ContractError("policy_asset_not_object")
@@ -184,13 +199,34 @@ def load_policy(path: Path) -> dict[str, Any]:
raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}")
if endpoint in forbidden_paths:
raise ContractError(f"policy_forbidden_endpoint_allowlisted_{asset_id}")
if endpoint in asset_paths:
try:
query_parameters = validate_query_parameters(
endpoint, asset.get("query_parameters")
)
except ContractError as exc:
raise ContractError(f"policy_asset_query_invalid_{asset_id}") from exc
endpoint_identity = (endpoint, tuple(sorted(query_parameters.items())))
if endpoint_identity in asset_paths:
raise ContractError(f"policy_asset_endpoint_duplicate_{asset_id}")
asset_paths.add(endpoint)
asset_paths.add(endpoint_identity)
if asset.get("method") != "GET":
raise ContractError(f"policy_asset_export_must_be_get_{asset_id}")
if asset.get("response_kind") not in {"collection", "document"}:
raise ContractError(f"policy_asset_response_kind_invalid_{asset_id}")
empty_sentinels = asset.get("empty_collection_sentinels", [])
if (
not isinstance(empty_sentinels, list)
or any(not isinstance(item, str) for item in empty_sentinels)
or len(empty_sentinels) != len(set(empty_sentinels))
or any(item not in EMPTY_COLLECTION_SENTINELS for item in empty_sentinels)
):
raise ContractError(f"policy_empty_collection_sentinels_invalid_{asset_id}")
if empty_sentinels and (
asset.get("response_kind") != "collection"
or endpoint != "/api/v1/explorer/views"
or not query_parameters
):
raise ContractError(f"policy_empty_collection_sentinels_unsafe_{asset_id}")
if asset.get("classification") not in {"portable", "export_only"}:
raise ContractError(f"policy_asset_classification_invalid_{asset_id}")
if asset.get("classification") == "portable":
@@ -337,30 +373,51 @@ def resolve_pointer(value: Any, pointer: str) -> Any:
return current
def resolve_collection(value: Any, asset: dict[str, Any]) -> list[Any]:
selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
if isinstance(selected, list):
return selected
empty_sentinels = set(asset.get("empty_collection_sentinels", []))
if selected is None and "null" in empty_sentinels:
return []
if isinstance(selected, dict) and not selected and "empty_object" in empty_sentinels:
return []
raise ContractError(f"asset_collection_shape_invalid_{asset['id']}")
def validate_collection_scope(
selected: list[dict[str, Any]], asset: dict[str, Any]
) -> None:
query_parameters = asset.get("query_parameters", {})
if not query_parameters:
return
source_page = query_parameters.get("sourcePage")
if any(item.get("sourcePage") != source_page for item in selected):
raise ContractError(f"asset_collection_scope_invalid_{asset['id']}")
def validate_asset_document(
value: Any,
asset: dict[str, Any],
policy: dict[str, Any],
) -> int:
scan_forbidden(value, policy)
selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
if asset["response_kind"] == "collection":
if not isinstance(selected, list):
raise ContractError(f"asset_collection_shape_invalid_{asset['id']}")
selected = resolve_collection(value, asset)
if len(selected) > policy["limits"]["max_items_per_asset"]:
raise ContractError(f"asset_item_limit_exceeded_{asset['id']}")
if any(not isinstance(item, dict) for item in selected):
raise ContractError(f"asset_collection_item_invalid_{asset['id']}")
validate_collection_scope(selected, asset)
return len(selected)
selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
if not isinstance(selected, dict):
raise ContractError(f"asset_document_shape_invalid_{asset['id']}")
return 1
def semantic_items(value: Any, asset: dict[str, Any]) -> list[dict[str, Any]]:
selected = resolve_pointer(value, str(asset.get("response_pointer", "")))
if not isinstance(selected, list):
raise ContractError(f"asset_collection_shape_invalid_{asset['id']}")
selected = resolve_collection(value, asset)
restore = asset.get("restore")
if not isinstance(restore, dict):
raise ContractError(f"asset_not_portable_{asset['id']}")
@@ -369,6 +426,8 @@ def semantic_items(value: Any, asset: dict[str, Any]) -> list[dict[str, Any]]:
for item in selected:
if not isinstance(item, dict):
raise ContractError(f"asset_collection_item_invalid_{asset['id']}")
validate_collection_scope(selected, asset)
for item in selected:
normalized.append(
{key: value for key, value in item.items() if key not in strip_fields}
)
@@ -392,16 +451,18 @@ class ApiClient:
self.max_response_bytes = max_response_bytes
self.opener = urllib.request.build_opener(_NoRedirect())
def _url(self, endpoint: str) -> str:
def _url(self, endpoint: str, query_parameters: Any = None) -> str:
if not endpoint.startswith("/api/v1/") or any(
item in endpoint for item in ("..", "?", "#")
):
raise ContractError("api_endpoint_unsafe")
return f"{self.base_url}{endpoint}"
query = validate_query_parameters(endpoint, query_parameters)
suffix = f"?{urllib.parse.urlencode(sorted(query.items()))}" if query else ""
return f"{self.base_url}{endpoint}{suffix}"
def request_json(self, endpoint: str) -> Any:
def request_json(self, endpoint: str, *, query_parameters: Any = None) -> Any:
request = urllib.request.Request(
self._url(endpoint),
self._url(endpoint, query_parameters),
method="GET",
headers={
self.header_name: self.token,

217
scripts/backup/tests/test_signoz_metadata_export_contract.py
View File

@@ -7,6 +7,7 @@ import socket
import subprocess
import sys
import threading
import urllib.parse
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
@@ -30,7 +31,18 @@ SOURCE_ASSETS: dict[str, Any] = {
"status": "success",
"data": {"rules": [{"id": "rule-1", "name": "Latency"}]},
},
"/api/v1/explorer/views": {"data": [{"id": "view-1", "name": "Errors"}]},
"/api/v1/explorer/views?sourcePage=logs": {
"data": [
{
"id": "view-1",
"name": "Errors",
"sourcePage": "logs",
}
]
},
"/api/v1/explorer/views?sourcePage=meter": {"data": None},
"/api/v1/explorer/views?sourcePage=metrics": {"data": {}},
"/api/v1/explorer/views?sourcePage=traces": {"data": []},
"/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"}]},
@@ -63,12 +75,13 @@ class MetadataHandler(BaseHTTPRequestHandler):
self._send_json(401, {"error": "unauthorized"})
return
path = self.path.split("?", 1)[0]
if path not in self.assets:
asset_key = self.path if self.path in self.assets else path
if asset_key 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:
self.get_counts[asset_key] = self.get_counts.get(asset_key, 0) + 1
value = json.loads(json.dumps(self.assets[asset_key]))
if self.drift_path in {path, asset_key} and self.get_counts[asset_key] >= 2:
value["data"].append({"id": "drift", "title": "changed"})
self._send_json(200, value)
@@ -77,16 +90,7 @@ class MetadataHandler(BaseHTTPRequestHandler):
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
data = self.assets[path].get("data")
collection = (
data.get("rules")
if path == "/api/v1/rules" and isinstance(data, dict)
else data
)
if not isinstance(collection, list):
if path != "/api/v1/explorer/views" and path not in self.assets:
self._send_json(404, {"error": "not found"})
return
length = int(self.headers.get("Content-Length", "0"))
@@ -94,6 +98,32 @@ class MetadataHandler(BaseHTTPRequestHandler):
if not isinstance(value, dict):
self._send_json(400, {"error": "object required"})
return
if path == "/api/v1/explorer/views":
source_page = value.get("sourcePage")
if source_page not in {"logs", "meter", "metrics", "traces"}:
self._send_json(400, {"error": "sourcePage required"})
return
asset_key = (
f"{path}?{urllib.parse.urlencode({'sourcePage': source_page})}"
)
if asset_key not in self.assets:
self._send_json(404, {"error": "not found"})
return
data = self.assets[asset_key].get("data")
if data is None or data == {}:
self.assets[asset_key]["data"] = []
data = self.assets[asset_key]["data"]
collection = data
else:
data = self.assets[path].get("data")
collection = (
data.get("rules")
if path == "/api/v1/rules" and isinstance(data, dict)
else data
)
if not isinstance(collection, list):
self._send_json(404, {"error": "not found"})
return
type(self).post_count += 1
restored = {"id": f"server-{type(self).post_count}", **value}
collection.append(restored)
@@ -137,6 +167,7 @@ def export_command(
credential: Path | None,
apply: bool,
receipt_file: Path | None = None,
policy: Path | None = None,
) -> list[str]:
command = [
sys.executable,
@@ -155,6 +186,8 @@ def export_command(
]
if credential is not None:
command.extend(["--credential-file", str(credential)])
if policy is not None:
command.extend(["--policy", str(policy)])
if apply:
terminal_path = receipt_file or output_dir.with_name(
f"{output_dir.name}.terminal.json"
@@ -267,6 +300,53 @@ def test_policy_is_explicitly_non_raw_and_non_completion() -> None:
alert_rules = next(item for item in policy["assets"] if item["id"] == "alert_rules")
assert alert_rules["endpoint"] == "/api/v1/rules"
assert alert_rules["response_pointer"] == "/data/rules"
explorer_views = [
item for item in policy["assets"] if item["id"].startswith("explorer_views_")
]
assert {item["query_parameters"]["sourcePage"] for item in explorer_views} == {
"logs",
"meter",
"metrics",
"traces",
}
assert all(item["response_pointer"] == "/data" for item in explorer_views)
assert all(
item["empty_collection_sentinels"] == ["null", "empty_object"]
for item in explorer_views
)
def test_policy_rejects_unhashable_empty_collection_sentinel_as_contract_error(
tmp_path: Path,
) -> None:
policy_value = json.loads(POLICY.read_text(encoding="utf-8"))
explorer = next(
item
for item in policy_value["assets"]
if item["id"] == "explorer_views_logs"
)
explorer["empty_collection_sentinels"] = [{}]
malformed_policy = tmp_path / "malformed-policy.json"
malformed_policy.write_text(json.dumps(policy_value), encoding="utf-8")
result = subprocess.run(
export_command(
base_url="https://signoz.invalid",
output_dir=tmp_path / "must-not-exist",
credential=None,
apply=False,
policy=malformed_policy,
),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=10,
check=False,
)
assert result.returncode == 1
assert "policy_empty_collection_sentinels_invalid_explorer_views_logs" in (
result.stderr
)
assert "Traceback" not in result.stderr
def test_check_mode_writes_nothing(tmp_path: Path) -> None:
@@ -322,6 +402,108 @@ def test_apply_and_independent_verifier_pass(tmp_path: Path) -> None:
assert oct((bundle / "manifest.json").stat().st_mode & 0o777) == "0o600"
def test_explorer_views_are_source_scoped_and_empty_sentinels_are_bounded(
tmp_path: Path,
) -> None:
credential = credential_file(tmp_path)
bundle = tmp_path / "bundle"
with metadata_server(SOURCE_ASSETS) as (base_url, handler):
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
expected_keys = {
f"/api/v1/explorer/views?sourcePage={source_page}"
for source_page in ("logs", "meter", "metrics", "traces")
}
assert expected_keys <= set(handler.get_counts)
assert all(handler.get_counts[key] == 2 for key in expected_keys)
assert "/api/v1/explorer/views" not in handler.get_counts
manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8"))
view_counts = {
item["id"]: item["item_count"]
for item in manifest["assets"]
if item["id"].startswith("explorer_views_")
}
assert view_counts == {
"explorer_views_logs": 1,
"explorer_views_meter": 0,
"explorer_views_metrics": 0,
"explorer_views_traces": 0,
}
view_queries = {
item["id"]: item["query_parameters"]
for item in manifest["assets"]
if item["id"].startswith("explorer_views_")
}
assert view_queries == {
"explorer_views_logs": {"sourcePage": "logs"},
"explorer_views_meter": {"sourcePage": "meter"},
"explorer_views_metrics": {"sourcePage": "metrics"},
"explorer_views_traces": {"sourcePage": "traces"},
}
def test_explorer_views_nonempty_scalar_fails_closed(tmp_path: Path) -> None:
assets = json.loads(json.dumps(SOURCE_ASSETS))
assets["/api/v1/explorer/views?sourcePage=metrics"]["data"] = "unexpected"
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 "asset_collection_shape_invalid_explorer_views_metrics" in result.stderr
assert not output.exists()
def test_explorer_view_item_must_match_scoped_source_page(tmp_path: Path) -> None:
assets = json.loads(json.dumps(SOURCE_ASSETS))
assets["/api/v1/explorer/views?sourcePage=logs"]["data"][0][
"sourcePage"
] = "traces"
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 "asset_collection_scope_invalid_explorer_views_logs" in result.stderr
assert not output.exists()
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"
@@ -429,7 +611,10 @@ def test_restore_check_and_semantic_apply_are_bounded(tmp_path: Path) -> None:
target_assets = {
"/api/v1/dashboards": {"data": []},
"/api/v1/rules": {"status": "success", "data": {"rules": []}},
"/api/v1/explorer/views": {"data": []},
"/api/v1/explorer/views?sourcePage=logs": {"data": []},
"/api/v1/explorer/views?sourcePage=meter": {"data": None},
"/api/v1/explorer/views?sourcePage=metrics": {"data": {}},
"/api/v1/explorer/views?sourcePage=traces": {"data": []},
}
with metadata_server(target_assets) as (target_base_url, handler):
isolation = isolation_receipt(tmp_path, target_base_url)

4
scripts/backup/verify-signoz-metadata-export.py
View File

@@ -150,6 +150,10 @@ def verify_assets(
raise ContractError(f"manifest_asset_file_mismatch_{asset_id}")
if recorded.get("endpoint") != asset["endpoint"]:
raise ContractError(f"manifest_asset_endpoint_mismatch_{asset_id}")
if recorded.get("query_parameters") != asset.get("query_parameters", {}):
raise ContractError(
f"manifest_asset_query_parameters_mismatch_{asset_id}"
)
if recorded.get("classification") != asset["classification"]:
raise ContractError(f"manifest_asset_classification_mismatch_{asset_id}")