60 lines
2.7 KiB
Python
60 lines
2.7 KiB
Python
"""Canonical schema fingerprints and conservative classification."""
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
SCHEMA_SUPPORTED = "SCHEMA_SUPPORTED"
|
|
SCHEMA_CHANGED = "SCHEMA_CHANGED"
|
|
SCHEMA_UNKNOWN = "SCHEMA_UNKNOWN"
|
|
_SHA256 = re.compile(r"\A[0-9a-f]{64}\Z")
|
|
_REDACTED_WORDS = ("PASSWORD", "SECRET", "TOKEN", "CREDENTIAL", "AUTH", "BLOB", "SOURCE", "DEFAULT", "BODY")
|
|
_UNORDERED_COLLECTIONS = {"tables", "relations", "fields", "keys", "indexes", "foreign_keys", "triggers", "procedures", "procedure_parameters", "packages", "generators", "dependencies", "monitoring", "events"}
|
|
_POSITIONAL_COLLECTIONS = {"columns", "segments", "field_order", "positions"}
|
|
|
|
|
|
def _canonical(value: Any, key: str | None = None) -> Any:
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
str(name): _canonical(item, str(name).lower())
|
|
for name, item in sorted(value.items(), key=lambda pair: str(pair[0]))
|
|
if not any(word in str(name).upper() for word in _REDACTED_WORDS)
|
|
}
|
|
if isinstance(value, (list, tuple, set, frozenset)):
|
|
items = [_canonical(item) for item in value]
|
|
if key in _UNORDERED_COLLECTIONS and key not in _POSITIONAL_COLLECTIONS:
|
|
return sorted(items, key=lambda item: json.dumps(item, ensure_ascii=True, sort_keys=True, separators=(",", ":")))
|
|
return items
|
|
if isinstance(value, (bytes, bytearray, memoryview)):
|
|
return "<redacted>"
|
|
if isinstance(value, (str, int, float, bool)) or value is None:
|
|
return value if not isinstance(value, str) or len(value) <= 1024 else "<redacted>"
|
|
return "<redacted>"
|
|
|
|
|
|
def build_fingerprint(metadata: dict) -> str:
|
|
"""Return SHA-256 of recursively canonicalized relevant metadata."""
|
|
payload = json.dumps(_canonical(metadata), ensure_ascii=True, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def classify_schema(metadata: dict, expected_fingerprints, *, mapping_ambiguous: bool = False) -> str:
|
|
"""Classify metadata against a validated allowlist of expected fingerprints."""
|
|
if mapping_ambiguous:
|
|
return SCHEMA_UNKNOWN
|
|
if metadata.get("incomplete") or metadata.get("errors") or metadata.get("unavailable"):
|
|
return SCHEMA_UNKNOWN
|
|
try:
|
|
allowlist = (
|
|
{expected_fingerprints}
|
|
if isinstance(expected_fingerprints, str)
|
|
else set(expected_fingerprints)
|
|
)
|
|
except Exception:
|
|
return SCHEMA_UNKNOWN
|
|
if not allowlist or not all(isinstance(item, str) and _SHA256.fullmatch(item) for item in allowlist):
|
|
return SCHEMA_UNKNOWN
|
|
return SCHEMA_SUPPORTED if build_fingerprint(metadata) in allowlist else SCHEMA_CHANGED
|