Metadata-Version: 2.4
Name: cybat-sdk
Version: 0.2.0
Summary: Cybat SDK for Python — WSGI/ASGI middleware that reports HTTP traffic directly to the Cybat Intelligence Engine
License-Expression: MIT
Project-URL: Homepage, https://cybat.co
Project-URL: Documentation, https://cybat.co/#/docs
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# cybat-sdk (Python)

Flask, Django and ASGI (FastAPI / Starlette / any ASGI app) integration for the Cybat Intelligence Engine. Captures HTTP traffic your app already receives and reports it directly to Cybat — no GCP org-level log sink required.

## Install

```bash
pip install cybat-sdk
```

No required dependencies — Flask/Django/etc. are assumed already installed by your project; only the integration module for the framework you actually use imports it.

## Usage

### Flask

```python
from flask import Flask
from cybat_sdk.integrations.flask import init_app

app = Flask(__name__)
init_app(app,
    api_key=os.environ["CYBAT_API_KEY"],
    gcp_project_id="my-gcp-project",  # required (or aws_project_id) — also enables automatic blocking, see below
    service_name="checkout-api",      # optional, just a label on incidents
)
```

### Django

Django instantiates middleware classes itself with no way to pass constructor kwargs, so configuration comes from `settings.py` instead of a function call:

```python
# settings.py
MIDDLEWARE = [
    # ... your other middleware ...
    "cybat_sdk.integrations.django.CybatMiddleware",
]
CYBAT_API_KEY = os.environ["CYBAT_API_KEY"]
CYBAT_GCP_PROJECT_ID = "my-gcp-project"  # required (or CYBAT_AWS_PROJECT_ID) — also enables automatic blocking
```

Every option has a `CYBAT_*` settings equivalent — see [Options](#options) below for the full list; the Django setting name is `CYBAT_` + the option name upper-cased (`retry_backoff_ms` → `CYBAT_RETRY_BACKOFF_MS`).

### FastAPI / Starlette / any ASGI app

```python
from cybat_sdk.integrations.asgi import CybatASGIMiddleware

app.add_middleware(CybatASGIMiddleware,
    api_key=os.environ["CYBAT_API_KEY"],
    gcp_project_id="my-gcp-project",
)
```

This speaks the raw ASGI protocol (`scope`/`receive`/`send`) directly rather than wrapping a framework-specific base class, so it works correctly with streaming responses too, and needs no FastAPI/Starlette import. `add_middleware(cls, **options)` instantiates `cls(app, **options)`, which is exactly this class's constructor — no adapter needed. For a bare ASGI app without that helper: `app = CybatASGIMiddleware(app, api_key=...)`.

Create an API key from the Intelligence Dashboard → **API Keys**. It's shown once at creation — store it as a secret (`CYBAT_API_KEY`), it can't be retrieved again.

## Multi-cloud routing

Set **exactly one** of `gcp_project_id` or `aws_project_id` — required, and never both on the same SDK instance. Which one you set decides where data is sent, automatically:

| You set | Data goes to | Status |
|---------|--------------|--------|
| `gcp_project_id` | `gcp-ingest.cybat.co` | Live |
| `aws_project_id` | `aws-ingest.cybat.co` | **Not yet implemented** — batches will fail to send until AWS-based ingestion ships |

An app that runs in both clouds needs two SDK instances, one per cloud — not one instance with both fields set (the constructor raises if you do that, since it's ambiguous which cloud's infrastructure should be enforced against). The constructor also raises if **neither** is set — there's no default cloud anymore, you must say which one this app runs on.

## ⚠️ Automatic blocking also requires the right IAM grant on `gcp_project_id`

Cybat classifies and alerts on threats as soon as `gcp_project_id` is set. But **automatically blocking** a malicious IP (VPC Firewall / Cloud Armor) additionally requires granting Cybat access to that specific project — without the grant below, threats are still detected and you still get alerted, but nothing gets blocked automatically.

To enable blocking:
1. Set `gcp_project_id` to the GCP project your app's infrastructure lives in (already required above).
2. Grant `soc-ai-agent@cybat-public.iam.gserviceaccount.com` the `roles/compute.securityAdmin` and `roles/compute.networkViewer` roles on that project (a Cloud Shell command for this is shown in the Intelligence Dashboard onboarding flow).

## ⚠️ Behind a load balancer or reverse proxy?

Each integration resolves the client IP from the framework's own notion of it (Flask's `request.remote_addr`, Django's `request.META["REMOTE_ADDR"]`, the ASGI `scope["client"]`), which reflects the *direct* TCP peer. If your app sits behind your own load balancer, CDN, or reverse proxy, configure that framework's standard mechanism for trusting `X-Forwarded-For` — otherwise every request reports the proxy's IP instead of the real client, and Cybat won't be able to tell your traffic apart:

- **Flask**: wrap your WSGI app with Werkzeug's `ProxyFix`.
- **Django**: set `SECURE_PROXY_SSL_HEADER` / put a proxy-aware middleware before Cybat's in `MIDDLEWARE`, per Django's deployment docs.
- **ASGI**: use Starlette's/Uvicorn's `--proxy-headers --forwarded-allow-ips` (or `ProxyHeadersMiddleware`).

## Options

All three integrations take the same options (Flask/ASGI as keyword arguments, Django as `CYBAT_*` settings):

```python
{
    "api_key": str,              # required
    "gcp_project_id": str,       # required (unless aws_project_id set) — routes to GCP, enables automatic blocking — see above
    "aws_project_id": str,       # required (unless gcp_project_id set) — routes to AWS — NOT YET LIVE, see "Multi-cloud routing"
    "service_name": str,         # optional label shown on incidents
    "ingest_url": str,           # overrides cloud auto-detection entirely; default depends on gcp_project_id/aws_project_id
    "flush_interval_ms": 2000,
    "max_batch_size": 100,       # flushes immediately at this size
    "max_retries": 2,            # a batch is dropped, not queued forever, after this
    "retry_backoff_ms": 500,     # base delay before the first retry; doubles each retry (exponential)
    "status_filter": "all",      # "all" | "errors" | a callable (status: int) -> bool
    "capture_body": False,       # send a redacted excerpt of the request body — see "Request bodies & privacy"
    "body_max_bytes": 2048,      # default 2048, max 8192 — cap on the body excerpt after redaction
    "body_status_filter": "all", # capture bodies only for matching responses
}
```

`status_filter` controls which captured requests actually get sent:
- `"all"` (default) — every request, regardless of status code.
- `"errors"` — only 4xx/5xx. Use this to cut ingestion volume/cost on high-traffic apps where 2xx telemetry isn't needed — Cybat still classifies and alerts on threats using only error responses, the same signal the old GCP Log Sink integration filtered down to.
- A callable — full control, e.g. `lambda status: status != 304` to drop cache-hit noise but keep everything else, 2xx included.

## Request bodies & privacy

Off by default. Set `capture_body=True` to include the request body in each event — that's where SQL/command injection, credential stuffing and prompt-injection against LLM-backed endpoints actually show up, so it meaningfully improves detection. It also means request content leaves your infrastructure, so **redaction runs in your process, before an event is even buffered, and can't be turned off** once capture is on:

| What | Becomes |
|------|---------|
| Emails | `j***n@example.com` |
| Card numbers (Luhn-checked, 13–19 digits) | `XXXX-XXXX-XXXX-1234` |
| JWTs, `Bearer …` / `Basic …` tokens | `[REDACTED jwt]`, `Bearer [REDACTED]` |
| Provider keys (AWS, Google, GitHub, Slack, Stripe) and PEM private keys | `[REDACTED key]` |
| The value of any field named like a secret (`password`, `token`, `secret`, `authorization`, `cookie`, `session`, `api_key`, `card`, `cvv`, `ssn`, `iban`…), in JSON, form bodies and query strings | `[REDACTED len=13 sus=quote,comment]` |

A redacted secret is replaced by a **content-free shape**, not blanked: its length plus which kinds of attack characters it contained (`quote`, `comment`, `sqlkw`, `angle`, `semicolon`, `traversal`, `cmdchar`, `template`, `nullbyte`). That's deliberate — if you blanked a `password` field entirely you'd also blank the SQL injection an attacker typed into it. With the shape, Cybat can still see "a 14-character password containing quote, comment and SQL-keyword characters" without ever seeing the value.

- JSON, form-encoded and text bodies only. Multipart, binary, and empty bodies are always skipped, regardless of `capture_body`.
- The excerpt is capped at `body_max_bytes` after redaction, centered on the most attack-looking part rather than just the first N bytes, and flagged `bodyTruncated` (on the wire) when it was cut.
- If redaction raises for any reason, no body is sent — never an unredacted one.
- `body_status_filter="errors"` captures bodies only for 4xx/5xx responses, which keeps volume (and what you send) down while still covering most attacks.

**URLs, referers and user agents are redacted the same way whether or not `capture_body` is on** — query strings routinely carry tokens and emails (`/reset?token=…&email=…`). Secret-named query values become the same shape, so `?password=' OR 1=1--` still reads as an injection attempt.

Redaction is pattern-based. It covers the common cases above but can't know that a field called `notes` happens to contain something private — if a route handles data you don't want to leave your infrastructure at all, leave `capture_body` off, or use `body_status_filter` / a `status_filter` callable to exclude it.

## Behavior

- Never raises into your app. A network error, a rejected request, or a misconfigured key results in a dropped batch and a printed warning, not a crash.
- Buffers in memory (thread-safe) and flushes on a background daemon thread every `flush_interval_ms` — there's no persistent queue in the SDK itself. If the process exits between flushes, buffered events are lost (matches the durability model of application logs in general). `collector.stop()` stops the background thread on shutdown; it does not force a final flush — call `collector.flush()` first if you need one.
- Uses only the standard library (`urllib`) for the HTTP call — no extra dependency.
- One HTTP event captured per request: method, URL, status, remote IP, user agent, referer, latency, response size, timestamp (plus a redacted body excerpt if `capture_body` is on).
- A failed send retries `max_retries` times with exponential backoff (`retry_backoff_ms`, `retry_backoff_ms * 2`, `retry_backoff_ms * 4`, ...) before the batch is dropped for good. Retries are scheduled on a background timer thread and never block the request that triggered them.

## Wire format

See [`../README.md`](../README.md) for the shared batch/auth contract every language SDK implements. Field names on the wire (`remoteIp`, `latencyMs`, `bodyTruncated`, ...) are camelCase per that shared contract even though this package's own Python API (`gcp_project_id`, `status_filter`, ...) is snake_case — that boundary is deliberate: the wire format is a cross-language contract, the options are this language's idiomatic API.

## Testing

```bash
pip install -e . pytest flask django
python -m pytest tests/ -q
```

The default run (78 tests) never touches the network — it exercises `CybatCollector`, both framework integrations and the redaction contract against a local `http.server` instance (`_RecordingServer` in `tests/test_collector.py`) or a raw ASGI harness, never a mock of the HTTP transport itself.

`tests/test_live_gateway.py` is separate: it sends real batches through the real, deployed ingest API Gateway (`https://gcp-ingest.cybat.co` by default) using a real `x-api-key`, to prove the SDK's wire format is actually accepted end-to-end — not just by the local fake server. It's skipped automatically unless you set an API key:

```bash
CYBAT_LIVE_TEST_API_KEY=<a real key from the Intelligence Dashboard> \
  python -m pytest tests/test_live_gateway.py -v
```

Optionally set `CYBAT_LIVE_TEST_INGEST_URL` to point at a staging gateway instead of production. These tests only ever send synthetic, boring `GET`/`200` events against an obviously-fake path with no `gcp_project_id` set, so they can't trip real threat detection or enforcement on whatever org the key belongs to — they do still create a handful of real events for that org, so don't run them against a production customer's key. CI never runs this file (no `CYBAT_LIVE_TEST_API_KEY` is configured there), so a revoked or expired key never breaks a release.

## Publishing (maintainers)

CI is `.github/workflows/publish-python.yml`, using PyPI Trusted Publishing (OIDC) — no API token stored anywhere. One-time setup on [pypi.org](https://pypi.org) *before* the first release (PyPI lets you register a "pending publisher" for a project that doesn't exist yet):

1. **pypi.org → Your projects → Publishing → Add a pending publisher**, with:
   - PyPI project name: `cybat-sdk`
   - Owner: `CybatInc`, Repository name: `SDK`, Workflow name: `publish-python.yml`, Environment name: `pypi`
2. **GitHub repo → Settings → Environments → New environment**, name it `pypi` (must match exactly). Optionally add Required reviewers here for a manual approval gate before every publish — the OIDC equivalent of the 2FA-gated staging step `publish-node.yml` uses for npm.

To release:

```bash
# bump "version" in pyproject.toml, commit that change, then tag THAT commit
git add pyproject.toml && git commit -m "chore: bump cybat-sdk to 0.1.1"
git tag 0.1.1
git push origin main --tags
```

Same bare `X.Y.Z` tag format as `@cybat/sdk-node`'s releases. GitHub Actions does **not** evaluate `paths` filters on tag pushes, so what actually keeps one tag from triggering both workflows is a `check-changes` job in each workflow that diffs the tagged commit against its parent (`git diff HEAD^ HEAD -- python/` here, `node/` for Node's) — every other job is skipped, not failed, when it doesn't touch that package's directory. Tag the version-bump commit itself, not an unrelated one, or that diff will be empty and the whole run will just skip. The `test` job runs the full test suite before building (the live-gateway tests in `tests/test_live_gateway.py` are skipped here too, same as any local run without `CYBAT_LIVE_TEST_API_KEY` — CI never touches the real gateway), then `publish` fails outright if the tag doesn't match `pyproject.toml`'s version.
