552 lines
22 KiB
Python
Executable File
552 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Shared fail-closed primitives for the SigNoz metadata backup lane.
|
|
|
|
The contract intentionally works only through authenticated HTTP APIs. It
|
|
never opens the SigNoz SQLite database or a Docker volume and never accepts a
|
|
secret value on the command line.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
DEFAULT_POLICY = REPO_ROOT / "config" / "signoz" / "metadata-export-policy.json"
|
|
SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
|
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"}
|
|
|
|
|
|
class ContractError(ValueError):
|
|
"""A bounded policy, transport, or artifact contract failure."""
|
|
|
|
|
|
class _NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(
|
|
self,
|
|
req: urllib.request.Request,
|
|
fp: Any,
|
|
code: int,
|
|
msg: str,
|
|
headers: Any,
|
|
newurl: str,
|
|
) -> None:
|
|
return None
|
|
|
|
|
|
def utc_now() -> str:
|
|
return (
|
|
datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
|
|
)
|
|
|
|
|
|
def canonical_json_bytes(value: Any) -> bytes:
|
|
return (
|
|
json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
+ "\n"
|
|
).encode("utf-8")
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def read_json_object(path: Path, *, label: str) -> dict[str, Any]:
|
|
require_no_symlink_components(path, label=label)
|
|
try:
|
|
metadata = path.stat()
|
|
if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > 1024 * 1024:
|
|
raise ContractError(f"{label}_size_or_type_invalid")
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise ContractError(f"{label}_read_failed") from exc
|
|
if not isinstance(value, dict):
|
|
raise ContractError(f"{label}_not_object")
|
|
return value
|
|
|
|
|
|
def _normalized_key(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]", "", value.casefold())
|
|
|
|
|
|
def require_no_symlink_components(path: Path, *, label: str) -> None:
|
|
if not path.is_absolute():
|
|
path = Path.cwd() / path
|
|
current = path
|
|
while True:
|
|
try:
|
|
metadata = current.lstat()
|
|
except OSError as exc:
|
|
raise ContractError(f"{label}_path_component_unavailable") from exc
|
|
if stat.S_ISLNK(metadata.st_mode):
|
|
raise ContractError(f"{label}_symlink_component_forbidden")
|
|
if current == current.parent:
|
|
break
|
|
current = current.parent
|
|
|
|
|
|
def load_policy(path: Path) -> dict[str, Any]:
|
|
policy = read_json_object(path, label="policy")
|
|
if policy.get("schema") != "awoooi_signoz_metadata_export_policy_v1":
|
|
raise ContractError("unsupported_policy_schema")
|
|
|
|
runtime = policy.get("source_runtime")
|
|
if not isinstance(runtime, dict):
|
|
raise ContractError("policy_source_runtime_missing")
|
|
if runtime.get("raw_database_access_allowed") is not False:
|
|
raise ContractError("policy_raw_database_must_be_forbidden")
|
|
if runtime.get("raw_volume_access_allowed") is not False:
|
|
raise ContractError("policy_raw_volume_must_be_forbidden")
|
|
if runtime.get("github_supply_chain_allowed") is not False:
|
|
raise ContractError("policy_github_supply_chain_must_be_forbidden")
|
|
if runtime.get("minimum_python_version") != "3.10":
|
|
raise ContractError("policy_minimum_python_version_invalid")
|
|
|
|
limits = policy.get("limits")
|
|
if not isinstance(limits, dict):
|
|
raise ContractError("policy_limits_missing")
|
|
for key in (
|
|
"request_timeout_seconds",
|
|
"max_response_bytes",
|
|
"max_asset_count",
|
|
"max_items_per_asset",
|
|
"max_restore_actions",
|
|
"stable_read_count",
|
|
):
|
|
value = limits.get(key)
|
|
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
|
|
raise ContractError(f"policy_limit_invalid_{key}")
|
|
if limits["stable_read_count"] != 2:
|
|
raise ContractError("policy_requires_exactly_two_stable_reads")
|
|
|
|
authentication = policy.get("authentication")
|
|
if not isinstance(authentication, dict):
|
|
raise ContractError("policy_authentication_missing")
|
|
if authentication.get("header_name") != "SIGNOZ-API-KEY":
|
|
raise ContractError("policy_auth_header_not_fixed")
|
|
if authentication.get("secret_value_in_arguments_allowed") is not False:
|
|
raise ContractError("policy_secret_argument_must_be_forbidden")
|
|
if authentication.get("accepted_file_modes") != ["0400", "0600"]:
|
|
raise ContractError("policy_credential_modes_invalid")
|
|
|
|
assets = policy.get("assets")
|
|
if not isinstance(assets, list) or not assets:
|
|
raise ContractError("policy_assets_missing")
|
|
if len(assets) > limits["max_asset_count"]:
|
|
raise ContractError("policy_asset_count_exceeds_limit")
|
|
|
|
forbidden_paths = {
|
|
item.get("path")
|
|
for item in policy.get("forbidden_endpoints", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
asset_ids: set[str] = set()
|
|
asset_paths: set[str] = set()
|
|
for asset in assets:
|
|
if not isinstance(asset, dict):
|
|
raise ContractError("policy_asset_not_object")
|
|
asset_id = asset.get("id")
|
|
endpoint = asset.get("endpoint")
|
|
if not isinstance(asset_id, str) or not SAFE_ASSET_ID.fullmatch(asset_id):
|
|
raise ContractError("policy_asset_id_unsafe")
|
|
if asset_id in asset_ids:
|
|
raise ContractError(f"policy_asset_duplicate_{asset_id}")
|
|
asset_ids.add(asset_id)
|
|
if not isinstance(endpoint, str) or not endpoint.startswith("/api/v1/"):
|
|
raise ContractError(f"policy_asset_endpoint_unsafe_{asset_id}")
|
|
if "?" in endpoint or "#" in endpoint or ".." in endpoint:
|
|
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:
|
|
raise ContractError(f"policy_asset_endpoint_duplicate_{asset_id}")
|
|
asset_paths.add(endpoint)
|
|
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}")
|
|
if asset.get("classification") not in {"portable", "export_only"}:
|
|
raise ContractError(f"policy_asset_classification_invalid_{asset_id}")
|
|
if asset.get("classification") == "portable":
|
|
restore = asset.get("restore")
|
|
if not isinstance(restore, dict):
|
|
raise ContractError(f"policy_restore_missing_{asset_id}")
|
|
if restore.get("method") not in {"POST", "PUT", "PATCH"}:
|
|
raise ContractError(f"policy_restore_method_invalid_{asset_id}")
|
|
if restore.get("strategy") != "collection_items":
|
|
raise ContractError(f"policy_restore_strategy_invalid_{asset_id}")
|
|
if restore.get("endpoint") != endpoint:
|
|
raise ContractError(f"policy_restore_endpoint_mismatch_{asset_id}")
|
|
if asset.get("response_kind") != "collection":
|
|
raise ContractError(f"policy_portable_asset_not_collection_{asset_id}")
|
|
|
|
completion = policy.get("completion_contract")
|
|
if not isinstance(completion, dict):
|
|
raise ContractError("policy_completion_contract_missing")
|
|
if completion.get("portable_export_alone_is_completion") is not False:
|
|
raise ContractError("policy_export_must_not_claim_completion")
|
|
if completion.get("full_sqlite_completion_claim_allowed") is not False:
|
|
raise ContractError("policy_full_sqlite_claim_must_be_forbidden")
|
|
if completion.get("durable_terminal_receipt_required_for_apply") is not True:
|
|
raise ContractError("policy_durable_apply_receipt_must_be_required")
|
|
if completion.get("failure_terminal_receipt_required") is not True:
|
|
raise ContractError("policy_failure_receipt_must_be_required")
|
|
restore_isolation = policy.get("restore_isolation")
|
|
if not isinstance(restore_isolation, dict):
|
|
raise ContractError("policy_restore_isolation_missing")
|
|
if restore_isolation.get("image_id_sha256_required") is not True:
|
|
raise ContractError("policy_restore_image_id_must_be_required")
|
|
|
|
forbidden_names = policy.get("forbidden_key_names")
|
|
if not isinstance(forbidden_names, list) or not forbidden_names:
|
|
raise ContractError("policy_forbidden_keys_missing")
|
|
normalized = [_normalized_key(str(item)) for item in forbidden_names]
|
|
if not all(normalized) or len(set(normalized)) != len(normalized):
|
|
raise ContractError("policy_forbidden_keys_invalid")
|
|
for pattern in policy.get("forbidden_value_patterns", []):
|
|
try:
|
|
re.compile(str(pattern))
|
|
except re.error as exc:
|
|
raise ContractError("policy_forbidden_value_pattern_invalid") from exc
|
|
return policy
|
|
|
|
|
|
def validate_identity(value: str, *, label: str) -> str:
|
|
if not SAFE_ID.fullmatch(value):
|
|
raise ContractError(f"{label}_unsafe")
|
|
return value
|
|
|
|
|
|
def validate_base_url(value: str, *, loopback_only: bool = False) -> str:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
if parsed.username or parsed.password or parsed.query or parsed.fragment:
|
|
raise ContractError("base_url_contains_forbidden_components")
|
|
if parsed.path not in {"", "/"}:
|
|
raise ContractError("base_url_path_must_be_empty")
|
|
host = (parsed.hostname or "").casefold()
|
|
is_loopback = host in LOOPBACK_HOSTS
|
|
if loopback_only and not is_loopback:
|
|
raise ContractError("restore_target_must_be_loopback")
|
|
if parsed.scheme == "http" and not is_loopback:
|
|
raise ContractError("plain_http_allowed_only_for_loopback")
|
|
if parsed.scheme not in {"http", "https"} or not host:
|
|
raise ContractError("base_url_scheme_or_host_invalid")
|
|
try:
|
|
parsed.port
|
|
except ValueError as exc:
|
|
raise ContractError("base_url_port_invalid") from exc
|
|
return value.rstrip("/")
|
|
|
|
|
|
def validate_token_reference(path: Path, *, read_value: bool) -> str | None:
|
|
if not path.is_absolute():
|
|
raise ContractError("credential_reference_must_be_absolute")
|
|
require_no_symlink_components(path, label="credential_reference")
|
|
try:
|
|
metadata = path.lstat()
|
|
except OSError as exc:
|
|
raise ContractError("credential_reference_unavailable") from exc
|
|
if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
|
|
raise ContractError("credential_reference_not_regular_file")
|
|
if metadata.st_uid != os.geteuid():
|
|
raise ContractError("credential_reference_owner_mismatch")
|
|
if stat.S_IMODE(metadata.st_mode) not in ALLOWED_TOKEN_MODES:
|
|
raise ContractError("credential_reference_mode_must_be_0400_or_0600")
|
|
if metadata.st_size <= 0 or metadata.st_size > 4096:
|
|
raise ContractError("credential_reference_size_invalid")
|
|
if not read_value:
|
|
return None
|
|
try:
|
|
raw = path.read_bytes()
|
|
except OSError as exc:
|
|
raise ContractError("credential_reference_read_failed") from exc
|
|
token = raw.rstrip(b"\r\n")
|
|
if not token or len(token) > 4096 or any(byte < 33 or byte > 126 for byte in token):
|
|
raise ContractError("credential_reference_value_shape_invalid")
|
|
return token.decode("ascii")
|
|
|
|
|
|
def scan_forbidden(value: Any, policy: dict[str, Any]) -> None:
|
|
forbidden_keys = {
|
|
_normalized_key(str(item)) for item in policy["forbidden_key_names"]
|
|
}
|
|
value_patterns = [
|
|
re.compile(str(item)) for item in policy.get("forbidden_value_patterns", [])
|
|
]
|
|
|
|
def walk(item: Any, pointer: str) -> None:
|
|
if isinstance(item, dict):
|
|
for key, child in item.items():
|
|
if not isinstance(key, str):
|
|
raise ContractError("json_object_key_not_string")
|
|
if _normalized_key(key) in forbidden_keys:
|
|
raise ContractError(
|
|
f"forbidden_key_detected_at_{pointer or 'root'}"
|
|
)
|
|
walk(child, f"{pointer}/{key}")
|
|
elif isinstance(item, list):
|
|
for index, child in enumerate(item):
|
|
walk(child, f"{pointer}/{index}")
|
|
elif isinstance(item, str):
|
|
if any(pattern.search(item) for pattern in value_patterns):
|
|
raise ContractError(
|
|
f"forbidden_value_pattern_detected_at_{pointer or 'root'}"
|
|
)
|
|
|
|
walk(value, "")
|
|
|
|
|
|
def resolve_pointer(value: Any, pointer: str) -> Any:
|
|
if pointer == "":
|
|
return value
|
|
if not pointer.startswith("/"):
|
|
raise ContractError("response_pointer_invalid")
|
|
current = value
|
|
for raw_part in pointer[1:].split("/"):
|
|
part = raw_part.replace("~1", "/").replace("~0", "~")
|
|
if isinstance(current, dict) and part in current:
|
|
current = current[part]
|
|
else:
|
|
raise ContractError("response_pointer_not_found")
|
|
return current
|
|
|
|
|
|
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']}")
|
|
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']}")
|
|
return len(selected)
|
|
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']}")
|
|
restore = asset.get("restore")
|
|
if not isinstance(restore, dict):
|
|
raise ContractError(f"asset_not_portable_{asset['id']}")
|
|
strip_fields = set(restore.get("strip_top_level_fields", []))
|
|
normalized: list[dict[str, Any]] = []
|
|
for item in selected:
|
|
if not isinstance(item, dict):
|
|
raise ContractError(f"asset_collection_item_invalid_{asset['id']}")
|
|
normalized.append(
|
|
{key: value for key, value in item.items() if key not in strip_fields}
|
|
)
|
|
return sorted(normalized, key=lambda item: canonical_json_bytes(item))
|
|
|
|
|
|
class ApiClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
header_name: str,
|
|
token: str,
|
|
timeout_seconds: int,
|
|
max_response_bytes: int,
|
|
) -> None:
|
|
self.base_url = validate_base_url(base_url)
|
|
self.header_name = header_name
|
|
self.token = token
|
|
self.timeout_seconds = timeout_seconds
|
|
self.max_response_bytes = max_response_bytes
|
|
self.opener = urllib.request.build_opener(_NoRedirect())
|
|
|
|
def _url(self, endpoint: str) -> 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}"
|
|
|
|
def request_json(self, endpoint: str) -> Any:
|
|
request = urllib.request.Request(
|
|
self._url(endpoint),
|
|
method="GET",
|
|
headers={
|
|
self.header_name: self.token,
|
|
"Accept": "application/json",
|
|
"User-Agent": "AWOOOI-P0-OBS-002-metadata-export/1",
|
|
},
|
|
)
|
|
try:
|
|
with self.opener.open(request, timeout=self.timeout_seconds) as response:
|
|
status_code = int(response.status)
|
|
content_type = response.headers.get_content_type()
|
|
payload = response.read(self.max_response_bytes + 1)
|
|
except urllib.error.HTTPError as exc:
|
|
raise ContractError(f"api_http_status_{exc.code}") from exc
|
|
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
raise ContractError("api_transport_failed") from exc
|
|
if status_code != 200:
|
|
raise ContractError(f"api_http_status_{status_code}")
|
|
if content_type != "application/json":
|
|
raise ContractError("api_content_type_not_json")
|
|
if len(payload) > self.max_response_bytes:
|
|
raise ContractError("api_response_size_exceeded")
|
|
try:
|
|
return json.loads(payload)
|
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
|
raise ContractError("api_response_json_invalid") from exc
|
|
|
|
def request_write(self, method: str, endpoint: str, payload: Any) -> None:
|
|
if method not in {"POST", "PUT", "PATCH"}:
|
|
raise ContractError("api_write_method_forbidden")
|
|
body = canonical_json_bytes(payload)
|
|
if len(body) > self.max_response_bytes:
|
|
raise ContractError("api_request_size_exceeded")
|
|
request = urllib.request.Request(
|
|
self._url(endpoint),
|
|
method=method,
|
|
data=body,
|
|
headers={
|
|
self.header_name: self.token,
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "AWOOOI-P0-OBS-002-metadata-restore/1",
|
|
},
|
|
)
|
|
try:
|
|
with self.opener.open(request, timeout=self.timeout_seconds) as response:
|
|
status_code = int(response.status)
|
|
response_payload = response.read(self.max_response_bytes + 1)
|
|
except urllib.error.HTTPError as exc:
|
|
raise ContractError(f"api_write_http_status_{exc.code}") from exc
|
|
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
raise ContractError("api_write_transport_failed") from exc
|
|
if status_code not in {200, 201, 202, 204}:
|
|
raise ContractError(f"api_write_http_status_{status_code}")
|
|
if len(response_payload) > self.max_response_bytes:
|
|
raise ContractError("api_write_response_size_exceeded")
|
|
|
|
|
|
def receipt(
|
|
*,
|
|
trace_id: str,
|
|
run_id: str,
|
|
work_item_id: str,
|
|
phase: str,
|
|
terminal: str,
|
|
detail: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"schema": "awoooi_signoz_metadata_receipt_v1",
|
|
"trace_id": trace_id,
|
|
"run_id": run_id,
|
|
"work_item_id": work_item_id,
|
|
"observed_at": utc_now(),
|
|
"phase": phase,
|
|
"terminal": terminal,
|
|
"detail": detail,
|
|
}
|
|
|
|
|
|
def write_private(path: Path, payload: bytes) -> None:
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
descriptor = os.open(path, flags, 0o600)
|
|
try:
|
|
os.fchmod(descriptor, 0o600)
|
|
with os.fdopen(descriptor, "wb", closefd=True) as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except BaseException:
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def validate_new_private_destination(path: Path, *, label: str) -> None:
|
|
if not path.is_absolute():
|
|
raise ContractError(f"{label}_path_must_be_absolute")
|
|
require_no_symlink_components(path.parent, label=f"{label}_parent")
|
|
try:
|
|
parent_metadata = path.parent.lstat()
|
|
except OSError as exc:
|
|
raise ContractError(f"{label}_parent_unavailable") from exc
|
|
if not stat.S_ISDIR(parent_metadata.st_mode):
|
|
raise ContractError(f"{label}_parent_not_directory")
|
|
if parent_metadata.st_uid != os.geteuid():
|
|
raise ContractError(f"{label}_parent_owner_mismatch")
|
|
if path.exists() or path.is_symlink():
|
|
raise ContractError(f"{label}_path_exists")
|
|
|
|
|
|
def write_new_private_atomic(path: Path, payload: bytes, *, label: str) -> None:
|
|
validate_new_private_destination(path, label=label)
|
|
temporary = path.with_name(f".{path.name}.partial.{os.getpid()}")
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
try:
|
|
descriptor = os.open(temporary, flags, 0o600)
|
|
try:
|
|
with os.fdopen(descriptor, "wb", closefd=True) as stream:
|
|
stream.write(payload)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
except BaseException:
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
os.replace(temporary, path)
|
|
try:
|
|
parent_descriptor = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(parent_descriptor)
|
|
finally:
|
|
os.close(parent_descriptor)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
try:
|
|
temporary.unlink()
|
|
except FileNotFoundError:
|
|
pass
|