Metadata-Version: 2.5
Name: whiteout-ai
Version: 0.7.0
Summary: Whiteout AI Governance SDK — wrap AI API calls with policy enforcement
Project-URL: Homepage, https://groovysec.com
Project-URL: Documentation, https://groovysec.com/docs/admin-guides/developers/python-sdk/
Author-email: Groovy Security <engineering@groovysec.com>
License-Expression: Apache-2.0
License-File: LICENSE
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25.0
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.20.0; extra == 'anthropic'
Provides-Extra: auto
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2.0; extra == 'langchain'
Provides-Extra: openai
Requires-Dist: openai>=1.0.0; extra == 'openai'
Description-Content-Type: text/markdown

# Whiteout AI Governance SDK

Wrap AI API calls with Whiteout policy enforcement. Three lines of code to add governance to any Python AI application.

## Install

```bash
pip install whiteout-ai
```

## Custom AI Apps: `AppGuard` (0.2+)

For your own LLM applications registered under **Integrations → Custom AI Apps**.
Authenticates with the app key from the setup wizard, with no registration step, and
checks **both** the prompt and the model's reply.

```bash
pip install "whiteout-ai>=0.2"
export WHITEOUT_APP_KEY=wo_app_prod_...
export WHITEOUT_BASE_URL=https://api.whiteout.groovysec.com   # the default
```

```python
from openai import OpenAI
from whiteout_ai import AppGuard, WhiteoutBlockedError

guard = AppGuard()
client = guard.openai(OpenAI())            # or guard.anthropic(Anthropic())

try:
    reply = client.chat.completions.create(
        model="gpt-5", messages=messages,
        whiteout={"end_user": user.id},     # optional; never sent to OpenAI
    )
except WhiteoutBlockedError as e:
    print(e.stage, e.reason, e.violated_policies)   # stage: "input" or "output"
```

The wrapped client is the same object with `chat.completions.create` (or
`messages.create`) governed. Everything else passes straight through.

**What gets scanned.** By default this is the latest user message and the
system prompt (0.5+). Whiteout checks an unchanged system prompt once, not on
every call. Pass `scan=["user", "system", "context", "tool"]` to the constructor,
or per call via `whiteout={"scan": [...]}`, to also catch sensitive data in
retrieved documents (`context=`) and tool results. Pass `scan=["user"]` to leave
the system prompt out.

**Checking without a wrapper:**

```python
d = guard.check_input(messages, context=retrieved_docs, model="gpt-5", provider="openai")
reply = call_your_model(messages)
guard.check_output(d.evaluation_id, reply)    # raises WhiteoutBlockedError on block
```

**Streaming.** `stream=True` works for both providers.

| `stream_mode` | Behaviour |
|---|---|
| `"buffered"` (default) | The whole response is held until the output check passes, so nothing flagged reaches the user. |
| `"passthrough"` | Chunks are yielded as they arrive, with a check every `check_every` characters (default 2000) plus a final check. It stops at the first flagged checkpoint. Text already shown cannot be retracted, so use buffered mode where you enforce. |

**Fail behaviour.** If Whiteout is unreachable, or the key is revoked (logged at
ERROR), the SDK applies your policy group's fail behaviour from
`GET /v1/guard/config`. That setting is cached for 5 minutes and revalidated with an ETag.
Pass `fail_open=True/False` to override it. With no config yet, the SDK fails open.

**Async.** `AsyncAppGuard` has the same surface (`await guard.check_input(...)`,
`guard.openai(AsyncOpenAI())`, `async for` over streams). Use it as
`async with AsyncAppGuard() as guard:` so pending background checks drain on exit.

**LangChain.**

```python
from whiteout_ai.langchain import WhiteoutCallbackHandler   # pip install "whiteout-ai[langchain]"
llm = ChatOpenAI(callbacks=[WhiteoutCallbackHandler(AppGuard())])
```

A block raises out of the chain. Streaming chains are checked when the
completion finishes.

**Other options:**
- `block_output=False` makes output checks fire-and-forget (observe the reply without blocking it).
- `send_events([...])` reports calls after the fact (observe-only, batched).
- `timeout` sets the request timeout in seconds (default 10).

## No code changes: auto-instrumentation (0.3+)

```bash
pip install "whiteout-ai[auto]"
export WHITEOUT_APP_KEY=wo_app_prod_...
whiteout-ai run python app.py            # or: whiteout-ai run -- gunicorn app:server
whiteout-ai check                        # key set? which clients will be governed?
```

In a container where you can't change the start command, put the bootstrap on the path instead:
`PYTHONPATH="$(whiteout-ai bootstrap-path)"`. It runs any `sitecustomize` you already have.

**What gets governed:**
- `openai`: chat completions and the Responses API, sync and async.
- `anthropic`: `messages.create`, sync and async. `messages.stream()` is checked on input, and its final text is recorded.
- boto3 `bedrock-runtime`: Converse and InvokeModel (input and output) and the streaming variants (input). This runs through botocore's event system.
- `openai` fine-tuning (0.6+): training-file uploads (`files.create(purpose="fine-tune")`) and `fine_tuning.jobs.create`, sync and async. See [Fine-tuning and training data](#fine-tuning-and-training-data-06).
- Embedding calls (0.7+): `openai` `embeddings.create` (sync and async) and Bedrock `invoke_model` with an embedding model (Titan, Cohere, Nova embeddings). See [Vector stores and embeddings](#vector-stores-and-embeddings-07).

A library is patched when your app imports it, so there's no startup cost for clients you don't use.

**Environment variables:**

| Variable | Values |
|---|---|
| `WHITEOUT_MODE` | `enforce` (default) or `observe`. Observe checks and records but never raises. |
| `WHITEOUT_STREAM_MODE` | Stream mode for governed calls |
| `WHITEOUT_SCAN` | Which parts of the call to scan |
| `WHITEOUT_INCLUDE` / `WHITEOUT_EXCLUDE` | `openai`, `anthropic`, `bedrock` |
| `WHITEOUT_DISABLE=1` | Turns it off completely |
| `WHITEOUT_FINETUNE` | `record` (default), `block` or `off` — fine-tuning uploads and jobs |
| `WHITEOUT_FINETUNE_SCAN_BYTES` | How much of each training file is checked (default `1000000`, max `20000000`; `0` records without checking) |
| `WHITEOUT_EMBEDDINGS` | `record` (default), `block` or `off` — embedding calls |

**Limits:** raw HTTP calls to a model API aren't seen. If a declared model host is called without Guard events, the AI Footprint still flags it.

## Fine-tuning and training data (0.6+)

Auto-instrumentation also watches an app fine-tuning an OpenAI model (OWASP LLM04 data poisoning, LLM02 sensitive data in training sets):

- **Training-file uploads** — `client.files.create(file=..., purpose="fine-tune")`. Before the upload, the file's text is checked with the Guard input check in chunks of up to 100,000 characters (the first 1 MB by default). Each JSONL line is reduced to its string values, so the check sees the examples, not the JSON. After the upload a *training data upload* event links those checks to the new file id.
- **Fine-tuning jobs** — `client.fine_tuning.jobs.create(...)` reports a *fine-tune job* event: base model, training and validation file ids, suffix, and whether this process checked the training file (and flagged it).

Your file is never consumed: bytes and paths are read directly, a seekable file object is read and rewound, and a one-shot stream is buffered and the buffer is uploaded instead. Events carry metadata only (file name and size, ids, how much was checked) — never training content.

By default nothing is stopped: the checks and events are recorded, and flagged training data shows in the app's Activity and as OWASP LLM04 evidence. With `WHITEOUT_FINETUNE=block`, an upload whose check comes back `block` (your app's policy in enforce mode, or fail-closed when Whiteout is unreachable) raises `WhiteoutBlockedError` with `stage="training_data"` and is not sent. `WHITEOUT_MODE=observe` never blocks.

The app key needs the `guard:events` scope (on by default) for the events. Not covered: the multi-part Uploads API (`client.uploads`, files over 512 MB) and training files uploaded outside the app — their jobs are still reported, marked as not checked. AWS Bedrock fine-tuning and model imports are recorded from CloudTrail by Bedrock ingest instead.

## Vector stores and embeddings (0.7+)

Documents you embed end up in front of the model later, so check them on the way in (OWASP LLM08, and LLM04 poisoning at ingest).

**Screen documents before indexing.** `screen_documents` runs each document through your app's data policy and Whiteout's prompt-injection rules:

```python
screen = guard.screen_documents(chunks)          # str, dict or LangChain Document
vector_store.add_documents(screen.clean())       # what to index
for d in screen.quarantined:
    log.warning("kept out of the index: %s %s", d.id, d.reasons)
```

Each document comes back with an `action`:

- `pass`: index it unchanged.
- `redact`: sensitive values were blanked. `d.redacted_text` holds the text, and `clean()` returns a copy of your document carrying it.
- `quarantine`: keep it out. The reason is planted instructions (`injection`), or sensitive data that couldn't be redacted (`redaction` says why). Pass `redact=False` to quarantine sensitive documents instead of rewriting them.

`clean()` keeps your order and never changes your objects. In `warn` or `monitor` mode, and for audit-only orgs, every document passes and `would_action` shows what enforce mode would do. The SDK batches for you: the API takes 100 documents and 2,000,000 characters per request, and 200,000 characters per document. A document over that size isn't sent; it comes back `quarantine` with reason `too_large`. When Whiteout can't be reached, your app's fail behaviour decides: fail-open passes every document (`fail_open`), and fail-closed quarantines them (`check_unavailable`). `AsyncAppGuard.screen_documents` works the same way. API: `POST /v1/guard/documents`.

**Embedding calls are recorded.** With auto-instrumentation, `openai` `embeddings.create` and Bedrock `invoke_model` on an embedding model are recorded. Their input text is checked with the Guard input check (shown as *Embedding call* in the app's Activity). One check covers the first 200,000 characters of a call; token-id inputs can't be checked, and vectors aren't. By default nothing is stopped. With `WHITEOUT_EMBEDDINGS=block`, a call whose check comes back `block` raises `WhiteoutBlockedError` with `stage="embedding"`. `WHITEOUT_MODE=observe` never blocks. Bedrock embedding calls used to go through the full prompt check (including blocking in enforce mode). From 0.7 they follow `WHITEOUT_EMBEDDINGS`.

## Prompt injection: quarantine (0.3+)

When your org has Prompt Injection Defense on, a tool result or system prompt carrying instructions can be *quarantined*. Wrapped clients, and auto-instrumentation, replace that turn with a short notice before the model sees it. Tool calls stay paired, and your own objects are never changed.

Context documents you send as `context=` can't be rewritten inside your prompt, so drop them yourself:

```python
d = guard.check_input(messages, context=docs, scan=["user", "context"])
docs = d.clean_context(docs)             # the documents Whiteout didn't quarantine
```

Replies are checked for injected instructions too. If your org blocks such content, `check_output` blocks.

## Unsafe output and system-prompt leaks (0.4+)

Treat the model's reply as untrusted input to whatever renders or runs it. Every
output check also returns:

- `output_findings`: script/HTML, SQL statements, shell commands (pipe-to-shell,
  reverse shells, disk wipes are high), links or images whose URL carries data, and
  `system_prompt_leak`. Each finding has `kind`, `severity`, `rule` and an `excerpt`.
  SQL or shell shown inside a fenced code block is reported at `low`.
- `safe_text`: the reply HTML-escaped, with external images removed and links
  reduced to text plus the visible URL. Insert it into HTML as-is.

```python
guard = AppGuard(canary=True, block_unsafe_output=["script_html", "system_prompt_leak"])
client = guard.openai(OpenAI())
reply = client.chat.completions.create(model="gpt-5", messages=messages)
out = guard.last_output                     # this call's output decision
html_safe = out.safe_text
if out.system_prompt_leak: ...
```

- Findings never block on their own. `block_unsafe_output=[kinds]` (or `True` for all)
  makes medium/high findings of those kinds fail the output check under your app's
  mode (enforce blocks, warn warns, monitor records).
- **Leak check.** Wrapped clients send the call's system prompt with the output
  check (it already went to Whiteout with the input check). Whiteout compares it with
  the reply and drops it; it is never stored. `leak_threshold=` (words, default 12)
  tunes it, and `leak_check=False` turns it off. Without a wrapper, pass
  `check_output(..., system_prompt=...)`.
- **Canary.** `canary=True` appends a random token to the system prompt sent to the
  model. A reply that contains it is a certain leak, and the SDK flags it even if
  Whiteout is unreachable.

See `docs/guides/handling-model-output.md` for encoding by sink.

## The employee's own policies (0.3+)

If the app is set to *the employee's own policies* in the admin console, pass the signed-in employee's OIDC `id_token`. Whiteout then applies their group's policies, just as it does in ChatGPT:

```python
guard.check_input(messages, user_token=session["id_token"])
client.chat.completions.create(model=..., messages=..., whiteout={"user_token": session["id_token"]})
```

## Infrastructure workloads: `WhiteoutGuard`

`WhiteoutGuard` is unchanged. It uses an enrollment token and the `/infra-agent` routes,
and suits workloads enrolled as infrastructure resources.

## Quick Start

```python
from whiteout_ai import WhiteoutGuard

guard = WhiteoutGuard(
    token="whiteout_enroll_...",
    backend_url="https://your-org.api.whiteout.groovysec.com",
)

# Option 1: Evaluate a prompt directly
result = guard.evaluate(
    prompt="Summarize this patient record...",
    ai_provider="openai",
    model="gpt-4o",
)
# result = {"decision": "block", "reason": "Contains PHI", ...}

# Option 2: Wrap an OpenAI call
import openai
response = guard.wrap(
    openai.chat.completions.create,
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello world"}],
)

# Option 3: Catch blocked prompts
from whiteout_ai import WhiteoutBlockedError
try:
    guard.evaluate(prompt="SSN 123-45-6789")
except WhiteoutBlockedError as e:
    print(f"Blocked: {e.reason}")
    print(f"Policies: {e.violated_policies}")
```

## Configuration

| Parameter | Required | Default | Description |
|-----------|----------|---------|-------------|
| `token` | Yes | - | Enrollment token from Whiteout admin |
| `backend_url` | Yes | - | Your org's Whiteout API URL |
| `fail_open` | No | `True` | Allow calls when backend unreachable |
| `timeout` | No | `10.0` | Backend call timeout (seconds) |
| `mode` | No | `sdk` | Deployment mode identifier |

## Features

- Automatic provider detection (OpenAI, Anthropic, Google, Mistral)
- Fast checks cached locally (provider block, model allowlist)
- Remote compliance evaluation for content policies
- Output governance (response scanning in background)
- Fail-open by default (never breaks your app)
- Thread-safe
