Metadata-Version: 2.4
Name: cat-session-sdk
Version: 1.2.3
Summary: Python SDK for CAT Session behavioral security API
License: MIT
Project-URL: Homepage, https://session.artrofimov.xyz
Project-URL: Repository, https://gitlab.com/cat-platform/session-sdk.git
Keywords: security,session,anomaly-detection,behavioral-analytics
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# cat-session-sdk

Python SDK for [CAT Session](https://session.artrofimov.xyz) — behavioral
session verification without ML, training data, or retraining.

## Install

```bash
pip install cat-session-sdk
```

## Quick start

```python
from cat_session_sdk import CATClient

cat = CATClient(api_key="your_key")  # get key at session.artrofimov.xyz

score = cat.score(
    entity_id="user_123",
    endpoint="/api/payment",
    status_code=200,
)

if score and score.alert:
    print(score.which_channel)
    # "behavioral drift detected"        → session hijack, insider
    # "population deviation detected"    → credential stuffing, bot
    print(score.continuation_state)
    # "STABLE" | "DRIFTING" | "COLLAPSED" | "CRITICAL"
```

## Middleware (score every request, zero latency impact)

```python
from fastapi import FastAPI, Request
from cat_session_sdk import CATClient
import asyncio

app = FastAPI()
cat = CATClient(api_key="your_key")

@app.middleware("http")
async def cat_middleware(request: Request, call_next):
    response = await call_next(request)
    user_id = request.headers.get("x-user-id")
    if user_id:
        # Non-blocking — adds 0ms to response time
        asyncio.create_task(cat.score(
            entity_id=user_id,
            endpoint=str(request.url.path),
            status_code=response.status_code,
            error=response.status_code >= 400,
        ))
    return response
```

## Enterprise log adapters

```python
from cat_session_sdk import CATClient
from cat_session.adapters.okta import stream_okta_log

cat = CATClient(api_key="your_key")

for event in stream_okta_log("okta_export.json"):
    score = cat.score(
        entity_id=event.entity_id,
        endpoint=event.endpoint,
        status_code=event.status_code,
        error=event.error,
        timestamp_ms=event.timestamp_ms,
    )
    if score and score.alert:
        print(f"{event.entity_id}: {score.which_channel}")
```

Adapters included: `okta`, `cloudtrail`, `windows_evtx`

## SessionScore fields

| Field | Type | Description |
|---|---|---|
| `score` | float 0–1 | Session health. Below 0.55 triggers alert. |
| `alert` | bool | True if action required. |
| `regime` | str | `normal` · `transition` · `breakdown` · `anomalous` |
| `continuation_state` | str | `STABLE` · `DRIFTING` · `COLLAPSED` · `CRITICAL` |
| `which_channel` | str | What fired the alert. |
| `events_scored` | int | Total events processed for this entity. |

Convenience properties: `score.behavioral_rhythm`, `score.access_pattern`,
`score.request_outcome`, `score.navigation_consistency`

## Production notes

**Fails open.** If the API is unreachable, `score()` returns `None`.
Never raises exceptions into your request path.

**State is server-side.** Works correctly across multiple pods,
serverless functions, and auto-scaling groups — no Redis required.
`user_123` hitting Pod A then Pod B produces the same result as
hitting one pod 200 times.

**Score after warmup.** Returns `None` for the first ~30 events while
building the behavioral baseline. Treat `None` as no signal.

## Get an API key

Free tier — 10,000 events/month, no credit card required.

→ [session.artrofimov.xyz/signup](https://session.artrofimov.xyz/signup)

## Links

- [Documentation](https://session.artrofimov.xyz/session/docs)
- [Interactive sandbox](https://session.artrofimov.xyz/session/sandbox)
- [Benchmarks](https://session.artrofimov.xyz/session/benchmarks)
