148 lines
5.4 KiB
Python
148 lines
5.4 KiB
Python
"""llm-ingestion-guard — a write-time defensive layer for LLM ingestion pipelines.
|
|
|
|
Query-time guardrails guard the answer; this library guards the *artifact*. It
|
|
packages the ingestion-side security contract — sanitize -> fence -> tool-less
|
|
quarantined transform -> per-stage capability isolation -> scan output before
|
|
persist -> fail-secure — as composable, stdlib-first, framework-agnostic code.
|
|
|
|
The library never makes the model call itself (no SDK is imported by the core),
|
|
so the public surface is a **toolkit plus two bookends** around the caller's
|
|
tool-less transform (BRIEF §6):
|
|
|
|
prepared = prepare_input(untrusted_content) # §6 steps 1-2: sanitize + fence
|
|
output = your_model(prepared.fenced) # §6 step 3: tool-less, caller's job
|
|
decision = screen_output(output, policy) # §6 steps 6-7: scan + dispose
|
|
|
|
if decision.disposition is Disposition.FAIL_SECURE:
|
|
raise SystemExit # halt + alert; never persist (§6 steps 7-8)
|
|
|
|
The individual detectors (:func:`sanitize`, :func:`scan_lexicon`,
|
|
:func:`scan_output`, ...), the contract asserters (:func:`assert_tool_less`,
|
|
:func:`scoped_env`, ...) and the disposition machinery are all exported for
|
|
pipelines that compose the checklist themselves. See ``docs/PLAN.md`` for the
|
|
build order and ``docs/BRIEF.md`` §6 for the contract this wiring encodes.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from .report import Finding, Report, Severity, Source, severity_rank
|
|
from .sanitize import sanitize, SanitizeResult
|
|
from .entropy import scan_entropy, EntropyResult, DecodedBlob
|
|
from .lexicon import scan_lexicon, load_lexicon, LexiconPattern
|
|
from .fence import fence, FenceResult
|
|
from .neutralize import neutralize, NeutralizeResult
|
|
from .output import scan_output, scan_secret_egress
|
|
from .disposition import (
|
|
decide,
|
|
guard,
|
|
Policy,
|
|
Trust,
|
|
Provenance,
|
|
Disposition,
|
|
DispositionResult,
|
|
PRESET_TRUSTED_SOURCE,
|
|
PRESET_USER_UPLOAD,
|
|
)
|
|
from .contract import (
|
|
assert_tool_less,
|
|
assert_credential_allowlist,
|
|
credential_env_names,
|
|
scoped_env,
|
|
ContractViolation,
|
|
)
|
|
from .grounding import (
|
|
SourceGroundingCheck,
|
|
no_grounding_check,
|
|
DEFAULT_GROUNDING_CHECK,
|
|
)
|
|
from . import okf
|
|
|
|
__version__ = "0.2.0"
|
|
|
|
|
|
# --- §6 bookends: the two library-side halves around the transform ---------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PreparedInput:
|
|
"""Untrusted content made ready for a tool-less transform (§6 steps 1-2).
|
|
|
|
``fenced`` is the sanitized, spotlight-fenced text to hand the model.
|
|
``nonce`` is the per-call fence delimiter — a caller may check for it in the
|
|
output to detect a fence breakout. ``report`` merges the sanitize and fence
|
|
findings so the caller can gate on the *input* too, not only the output.
|
|
"""
|
|
|
|
fenced: str
|
|
nonce: str
|
|
report: Report
|
|
|
|
|
|
def prepare_input(text: str, source: Source = Source.INPUT) -> PreparedInput:
|
|
"""Sanitize then fence untrusted ``text`` (BRIEF §6 steps 1-2).
|
|
|
|
Strips carrier classes first (:func:`sanitize`), then spotlight-fences the
|
|
cleaned payload in a randomized per-call delimiter (:func:`fence`) — sanitize
|
|
*before* fence so a carrier can never smuggle a forged delimiter. The merged
|
|
report carries both steps' findings; ``prepare_input`` itself renders no
|
|
disposition (design principle 4 — the caller decides).
|
|
"""
|
|
sanitized = sanitize(text, source)
|
|
fenced = fence(sanitized.text, source)
|
|
report = Report()
|
|
report.extend(sanitized.report.findings)
|
|
report.extend(fenced.report.findings)
|
|
return PreparedInput(fenced=fenced.text, nonce=fenced.nonce, report=report)
|
|
|
|
|
|
def screen_output(
|
|
text: str,
|
|
policy: Policy,
|
|
*,
|
|
provenance: Provenance = Provenance.PROSE,
|
|
transform_failed: bool = False,
|
|
) -> DispositionResult:
|
|
"""Scan the emitted ``text`` and dispose it, failing *closed* (§6 steps 6-7).
|
|
|
|
Runs :func:`scan_output` (lexicon + entropy + decode-and-rescan + secret
|
|
egress) under :func:`guard`, so a scanner that errors on crafted input yields
|
|
``FAIL_SECURE`` rather than an auto-persist — an un-scannable artifact is
|
|
never committed. ``transform_failed=True`` (the caller's model call raised or
|
|
fell back) plus any finding is treated as a probable forced-fallback attack
|
|
and also halts, regardless of trust tier.
|
|
"""
|
|
return guard(
|
|
lambda: scan_output(text, source=Source.OUTPUT),
|
|
policy,
|
|
provenance=provenance,
|
|
transform_failed=transform_failed,
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"__version__",
|
|
# shared types
|
|
"Finding", "Report", "Severity", "Source", "severity_rank",
|
|
# input-side detectors + result types
|
|
"sanitize", "SanitizeResult",
|
|
"scan_entropy", "EntropyResult", "DecodedBlob",
|
|
"scan_lexicon", "load_lexicon", "LexiconPattern",
|
|
"fence", "FenceResult",
|
|
"neutralize", "NeutralizeResult",
|
|
# output-side
|
|
"scan_output", "scan_secret_egress",
|
|
# disposition
|
|
"decide", "guard", "Policy", "Trust", "Provenance",
|
|
"Disposition", "DispositionResult",
|
|
"PRESET_TRUSTED_SOURCE", "PRESET_USER_UPLOAD",
|
|
# contract asserters
|
|
"assert_tool_less", "assert_credential_allowlist",
|
|
"credential_env_names", "scoped_env", "ContractViolation",
|
|
# grounding seam
|
|
"SourceGroundingCheck", "no_grounding_check", "DEFAULT_GROUNDING_CHECK",
|
|
# §6 bookends
|
|
"prepare_input", "screen_output", "PreparedInput",
|
|
# OKF adapter (v0.2) — the format-specific layer, as its own namespace
|
|
"okf",
|
|
]
|