| """ |
| 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 |
|
|
