File size: 2,555 Bytes
faa1bb6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """
Polyglot canary — module-level code executes on import.
Frameworks that add model dir to sys.path or exec_module this file:
- MLflow (sys.path injection from MLmodel code_path)
- SpeechBrain (foreign_class → exec_module on downloaded .py)
- evaluate (downloads .py from Hub → import_module)
- LiteLLM (spec_from_file_location → exec_module)
This file also serves as the custom_interface for SpeechBrain.
"""
import pathlib
import datetime
import json
import os
EVIDENCE_DIR = pathlib.Path("/tmp/polyglot_evidence")
EVIDENCE_DIR.mkdir(parents=True, exist_ok=True)
_caller = "unknown"
try:
import traceback
_stack = traceback.format_stack()
for frame in reversed(_stack):
if "mlflow" in frame.lower():
_caller = "mlflow"
break
elif "speechbrain" in frame.lower() or "hyperpyyaml" in frame.lower():
_caller = "speechbrain"
break
elif "evaluate" in frame.lower():
_caller = "evaluate"
break
elif "litellm" in frame.lower():
_caller = "litellm"
break
elif "lerobot" in frame.lower():
_caller = "lerobot"
break
except Exception:
pass
evidence = {
"event": "POLYGLOT_CANARY_FIRED",
"framework": _caller,
"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"pid": os.getpid(),
"mechanism": "module-level code execution on import",
}
marker = EVIDENCE_DIR / f"{_caller}.json"
with open(marker, "w") as f:
json.dump(evidence, f, indent=2)
(EVIDENCE_DIR / f"{_caller}.txt").write_text(
f"CANARY FIRED — {_caller}\n"
f"Time: {evidence['timestamp']}\n"
f"PID: {evidence['pid']}\n"
f"Vector: payload.py module-level code executed on import\n"
)
def _load_pyfunc(data_path):
"""MLflow pyfunc loader entry point."""
evidence["mechanism"] = "MLflow _load_pyfunc() called"
with open(EVIDENCE_DIR / "mlflow_pyfunc.json", "w") as f:
json.dump(evidence, f, indent=2)
return lambda x: x
class Canary:
"""Universal canary class — constructor writes additional evidence."""
def __init__(self, **kwargs):
rec = {**evidence, "kwargs": {k: repr(v)[:100] for k, v in kwargs.items()},
"mechanism": "Canary.__init__(**kwargs)"}
fw = kwargs.get("framework", _caller)
with open(EVIDENCE_DIR / f"{fw}_init.json", "w") as f:
json.dump(rec, f, indent=2)
def __call__(self, *args, **kwargs):
return args[0] if args else None
|