Metadata-Version: 2.4
Name: argos-python
Version: 1.1.0
Summary: argos Security SDK for Python
Author-email: argos Team <team@tachynoix.net>
License: MIT
Keywords: fraud-detection,middleware,sdk,security,threat-detection
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.9
Provides-Extra: all
Requires-Dist: django>=3.0.0; extra == 'all'
Requires-Dist: fastapi>=0.100.0; extra == 'all'
Requires-Dist: flask>=2.0.0; extra == 'all'
Requires-Dist: starlette>=0.27.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Provides-Extra: django
Requires-Dist: django>=3.0.0; extra == 'django'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100.0; extra == 'fastapi'
Requires-Dist: starlette>=0.27.0; extra == 'fastapi'
Provides-Extra: flask
Requires-Dist: flask>=2.0.0; extra == 'flask'
Description-Content-Type: text/markdown

# Argos Security SDK

<p align="center">
  <a href="https://pypi.org/project/argos-python/">
    <img src="https://img.shields.io/pypi/v/argos-python.svg" alt="PyPI Version">
  </a>
  <a href="https://pypi.org/project/argos-python/">
    <img src="https://img.shields.io/pypi/pyversions/argos-python.svg" alt="Python Versions">
  </a>
  <a href="https://opensource.org/licenses/MIT">
    <img src="https://img.shields.io/pypi/l/argos-python.svg" alt="License: MIT">
  </a>
</p>

A production-grade Python SDK for real-time threat detection and fraud prevention. Argos monitors every request, scores risk using ML models, and lets you block threats instantly or asynchronously.

## Features

- **AI-Powered Detection**: ML models analyze requests in real-time and return risk scores
- **Zero Latency Option**: Async mode queues events without blocking your application
- **Sync Enforcement**: Block malicious requests before they reach your business logic
- **Framework Middleware**: Drop-in middleware for FastAPI, Flask, Django, and Starlette
- **Circuit Breaker**: Built-in resilience with automatic recovery
- **Retry Logic**: Exponential backoff with jitter for failed requests
- **Blocklist Management**: Programmatic IP and user blocking
- **High Throughput**: Event queuing for peak traffic scenarios

## Installation

```bash
pip install argos-python

# With framework support
pip install argos-python[fastapi]    # FastAPI + Starlette
pip install argos-python[flask]      # Flask
pip install argos-python[django]     # Django
pip install argos-python[all]        # All frameworks
```

## Quick Start

```python
from argos import create_client

# Initialize the client
client = create_client(
    api_key="your-api-key",
)

# Ingest an event for analysis
event = client.ingest({
    "event_type": "login",
    "user_id": "user123",
    "ip_address": "192.168.1.1",
    "status": "success"
})

# Check the verdict
if event.signal == "BLOCK":
    print("Threat detected! Block this request.")
else:
    print("Request allowed.")
```

## Configuration

### Creating a Client

```python
from argos import create_client, ArgosClient, ArgosConfig

# Using the convenience function (recommended)
client = create_client(
    api_key="your-api-key",
    timeout=30.0,                    # Request timeout in seconds
    max_retries=3,                   # Maximum retry attempts
    retry_strategy="exponential",    # "exponential", "linear", or "none"
    retry_base_delay=0.5,            # Base delay between retries
    retry_max_delay=30.0,            # Maximum delay cap
    retry_jitter=True,               # Add randomness to delays
    circuit_breaker_threshold=5,     # Failures before opening circuit
    circuit_breaker_timeout=60.0,    # Seconds before attempting recovery
    queue_size=1000,                 # Maximum queued events
    auto_block_on_block=False,       # Auto-block IP on BLOCK verdict
    auto_block_ttl=3600.0,           # TTL for auto-blocked IPs (seconds)
    trusted_proxies=["10.0.0.0/8"],  # Trusted proxy CIDRs for IP extraction
)

# Or using the config class directly
config = ArgosConfig(
    api_key="your-api-key",
    # ... other options
)
client = ArgosClient(config)
```

### Configuration Options

| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `api_key` | `str` | Required | Your Argos API key |
| `timeout` | `float` | `30.0` | Request timeout in seconds |
| `max_retries` | `int` | `3` | Maximum retry attempts |
| `retry_strategy` | `str` | `"exponential"` | Retry strategy: `exponential`, `linear`, or `none` |
| `retry_base_delay` | `float` | `0.5` | Base delay for retries |
| `retry_max_delay` | `float` | `30.0` | Maximum delay between retries |
| `retry_jitter` | `bool` | `True` | Add random jitter to delays |
| `circuit_breaker_threshold` | `int` | `5` | Failures before opening circuit |
| `circuit_breaker_timeout` | `float` | `60.0` | Seconds before attempting recovery |
| `queue_size` | `int` | `1000` | Maximum events in queue |
| `auto_block_on_block` | `bool` | `False` | Automatically block IP on BLOCK verdict |
| `auto_block_ttl` | `float` | `3600.0` | TTL for auto-blocked IPs |
| `trusted_proxies` | `list[str]` | `None` | Trusted proxy CIDRs |

## Core API

### Ingesting Events

```python
# Single event ingestion
event = client.ingest(
    payload={
        "event_type": "login",           # Required: event type identifier
        "user_id": "user123",            # Recommended
        "ip_address": "192.168.1.1",     # Recommended
        "status": "success",             # Optional
        # ... any custom fields
    },
    metadata={
        "externalID": "req_123",         # External identifier
        "sessionID": "sess_456",         # Session identifier
        "environment_id": "env_789",     # Environment ID
        "client_ip": "192.168.1.1",      # Client IP (for auto-block)
        "source": "api",                 # Event source
    }
)

# Access event properties
print(event.id)              # Event ID from Argos
print(event.signal)          # Verdict: "BLOCK", "ALLOW", or "PENDING"
print(event.anomaly_score)   # ML risk score (0.0 - 1.0)
print(event.payload)         # Original payload
```

### Getting Decisions

```python
# Get decision for a specific event
decision = client.get_decision(event_id)

print(decision.verdict)  # "allow", "block", or "pending"
print(decision.reason)   # Reason for the decision

# Convenience properties
if decision.is_blocked:
    print("Request blocked!")
elif decision.is_allowed:
    print("Request allowed")
```

## Blocklist Management

```python
# Check if an IP is blocked
is_blocked, reason = client.is_blocked("192.168.1.1")
if is_blocked:
    print(f"IP blocked: {reason}")

# Check if a user is blocked
is_blocked, reason = client.is_user_blocked("user123")
if is_blocked:
    print(f"User blocked: {reason}")

# Check both IP and user at once
status = client.check_access(ip="192.168.1.1", user_id="user123")
print(status.blocked)           # Overall blocked status
print(status.ip_blocked)        # IP blocked status
print(status.user_blocked)      # User blocked status

# Block an IP
entry = client.block_ip(
    ip="192.168.1.1",
    environment_id="env_123",
    reason="Malicious activity detected"
)
print(f"Blocked IP: {entry.id}")

# Block a user
entry = client.block_user(
    user_id="user123",
    environment_id="env_123",
    reason="Account compromise suspected"
)
print(f"Blocked user: {entry.id}")

# Unblock an entry
client.unblock(entry_id="entry_123")

# Get all blocklist entries
entries = client.get_blocklist()
for entry in entries:
    print(f"{entry.type}: {entry.value} - {entry.reason}")
```

## Event Queuing

For high-throughput scenarios, queue events and flush them in batches:

```python
# Queue events without blocking
for request in requests:
    client.queue_event(
        payload={
            "event_type": "request",
            "path": request.url,
            "user_id": request.user_id,
        },
        metadata={
            "externalID": request.id,
            "environment_id": "env_123"
        }
    )

# Process all queued events
events = client.flush_queue()
print(f"Processed {len(events)} events")
```

## Async Usage

All client methods have async counterparts prefixed with `a`:

```python
import asyncio
from argos import create_client

async def main():
    client = create_client(api_key="your-api-key")
    
    # Async ingest
    event = await client.aingest({
        "event_type": "login",
        "user_id": "user123"
    })
    print(event.signal)
    
    # Async batch
    events = await client.aingest_batch([
        {"event_type": "login", "user_id": "user1"},
        {"event_type": "login", "user_id": "user2"},
    ])
    
    # Async blocklist checks
    is_blocked, reason = await client.ais_blocked("192.168.1.1")
    
    # Async blocklist management
    await client.ablock_ip("192.168.1.1", "env_123", "reason")
    
    # Async queuing
    await client.aqueue_event({"event_type": "request"})
    events = await client.aflush_queue()

asyncio.run(main())
```

## Framework Middleware

### FastAPI

```python
from fastapi import FastAPI
from argos import create_client
from argos.middleware.fastapi import FastAPIMiddleware
from argos.middleware.base import MiddlewareConfig

app = FastAPI()
client = create_client(api_key="your-api-key")

# Configure middleware
config = MiddlewareConfig(
    mode="async",                    # "sync" blocks requests, "async" queues
    include_headers=True,            # Include request headers in events
    include_body=False,              # Include request body (careful with sensitive data)
    exclude_paths=["/health", "/metrics"],  # Paths to skip
    identity_header="X-User-ID",     # Header to extract user ID
)

app.add_middleware(FastAPIMiddleware, client=client, config=config)

@app.get("/protected")
def protected_route():
    return {"message": "This route is protected by Argos"}
```

### Flask

```python
from flask import Flask
from argos import create_client
from argos.middleware.flask import FlaskMiddleware
from argos.middleware.base import MiddlewareConfig

app = Flask(__name__)

# Create client with auto-block enabled (optional)
client = create_client(
    api_key="your-api-key",
    auto_block_on_block=True,  # Auto-block when ML returns BLOCK
)

# Configure middleware with blocklist checking
config = MiddlewareConfig(
    mode="sync",              # 'sync' blocks bad requests, 'async' queues
    include_headers=True,
    include_body=True,
)
FlaskMiddleware(app, client, config)

@app.route("/protected")
def protected_route():
    return {"message": "This route is protected by Argos"}
```

### Starlette

```python
from starlette.applications import Starlette
from starlette.middleware import Middleware
from argos import create_client
from argos.middleware.starlette import StarletteMiddleware

app = Starlette(
    routes=[...],
    middleware=[
        Middleware(
            StarletteMiddleware,
            client=create_client(api_key="your-api-key"),
            mode="async"
        )
    ]
)
```

### Django

Add to your `settings.py`:

```python
# settings.py
ARGOS_API_KEY = "your-api-key"
ARGOS_BASE_URL = "https://api.your-argos-deployment.com"  # Optional

# Add middleware (order matters!)
MIDDLEWARE = [
    # ... other middleware
    'argos.middleware.django.DjangoMiddleware',
]
```

Or configure with custom settings:

```python
# settings.py
ARGOS = {
    "api_key": "your-api-key",
    "base_url": "https://api.your-argos-deployment.com",
    "mode": "async",
    "exclude_paths": ["/health/", "/admin/"],
}
```

### Middleware Configuration Options

| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `mode` | `str` | `"async"` | `"sync"` waits for decision, `"async"` queues events |
| `include_headers` | `bool` | `True` | Include request headers in events |
| `include_body` | `bool` | `False` | Include request body (use carefully) |
| `exclude_paths` | `list[str]` | `["/health", "/metrics", "/_health"]` | Paths to skip |
| `identity_header` | `str` | `"X-User-ID"` | Header to extract user ID |
| `custom_payload_builder` | `Callable` | `None` | Custom function to build payload |

## Error Handling

```python
from argos import (
    ArgosError,
    ArgosAPIError,
    ArgosAuthenticationError,
    ArgosRateLimitError,
    ArgosTimeoutError,
    ArgosCircuitOpenError,
)

try:
    event = client.ingest({...})
except ArgosAuthenticationError as e:
    print(f"Invalid API key: {e.message}")
except ArgosRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ArgosTimeoutError as e:
    print(f"Request timed out: {e.message}")
except ArgosCircuitOpenError as e:
    print(f"Argos unavailable (circuit open): {e.message}")
except ArgosAPIError as e:
    print(f"API error ({e.status_code}): {e.message}")
except ArgosError as e:
    print(f"Generic error: {e.message}")
```

### Error Types

| Error | Description |
|-------|-------------|
| `ArgosError` | Base exception for all Argos errors |
| `ArgosAPIError` | Non-2xx API responses |
| `ArgosAuthenticationError` | Invalid API key (401) |
| `ArgosRateLimitError` | Rate limit exceeded (429), includes `retry_after` |
| `ArgosTimeoutError` | Request timeout |
| `ArgosCircuitOpenError` | Circuit breaker is open due to failures |
| `ArgosValidationError` | Invalid request data |
| `ArgosQueueFullError` | Event queue is full |

## Advanced Features

### Circuit Breaker

The SDK includes a circuit breaker that automatically opens after repeated failures to prevent cascade failures:

```python
config = ArgosConfig(
    api_key="your-api-key",
    circuit_breaker_threshold=5,    # Open after 5 consecutive failures
    circuit_breaker_timeout=60.0,   # Attempt recovery after 60 seconds
)

client = ArgosClient(config)

# When circuit is open, raises ArgosCircuitOpenError
try:
    event = client.ingest({...})
except ArgosCircuitOpenError:
    print("Argos is temporarily unavailable")
```

### Auto-Block

Automatically block IPs that receive a BLOCK verdict:

```python
config = ArgosConfig(
    api_key="your-api-key",
    auto_block_on_block=True,      # Enable auto-blocking
    auto_block_ttl=3600.0,         # Block for 1 hour
)

client = ArgosClient(config)

# When ingest returns BLOCK, IP is automatically blocked
event = client.ingest({
    "event_type": "login",
    "ip_address": "192.168.1.1",
}, metadata={
    "client_ip": "192.168.1.1",   # Required for auto-block
    "environment_id": "env_123",  # Required for auto-block
})
```

### Trusted Proxies

Properly extract client IP when behind a proxy or load balancer:

```python
config = ArgosConfig(
    api_key="your-api-key",
    trusted_proxies=[
        "10.0.0.0/8",      # Private network
        "172.16.0.0/12",   # Docker network
        "192.168.0.0/16",  # Local network
    ]
)

client = ArgosClient(config)
# X-Forwarded-For headers from trusted proxies will be used
```

### Custom Payload Builder

For complete control over the event payload:

```python
from argos.middleware.base import MiddlewareConfig

def custom_builder(request):
    return {
        "event_type": "api_request",
        "path": request.url.path,
        "method": request.method,
        "user_id": get_user_from_token(request),
        "ip_address": get_client_ip(request),
        "custom_field": "value",
    }

config = MiddlewareConfig(
    custom_payload_builder=custom_builder
)

app.add_middleware(FastAPIMiddleware, client=client, config=config)
```

### Creating Custom Middleware

You can create custom middleware for any Python framework by subclassing `BaseMiddleware`:

```python
from argos import create_client
from argos.middleware.base import BaseMiddleware, MiddlewareConfig
from argos.models import Decision

class MyFrameworkMiddleware(BaseMiddleware):
    """Custom middleware for any Python web framework."""

    def __init__(self, app, client, config=None):
        super().__init__(client, config)
        self.app = app  # Your framework's app

    def get_request_data(self, request):
        """Extract data from your framework's request object."""
        return {
            "method": request.method,
            "path": request.path,
            "headers": dict(request.headers),
        }

    def get_client_ip(self, request):
        """Extract client IP from request."""
        # Customize based on your framework
        return request.client_ip or request.headers.get("X-Forwarded-For", "unknown")

    def get_user_id(self, request):
        """Extract user ID from request."""
        # Customize based on your auth system
        return request.headers.get("X-User-ID")

    def process_request(self, request):
        """Process a request and return a Decision."""
        return self.config.mode == "sync"

    def __call__(self, environ, start_response):
        """WSGI interface example."""
        # Wrap your framework's request/response
        request = self._create_request_object(environ)
        
        if self.should_skip(request.path):
            return self.app(environ, start_response)
        
        # Check blocklist first
        blocked, reason = self.check_blocklist(request)
        if blocked:
            start_response("403 Forbidden", [("Content-Type", "application/json")])
            return [b'{"error": "blocked", "reason": "' + reason.encode() + b'"}']
        
        # Process based on mode
        if self.config.mode == "sync":
            decision = self.run_sync(request)
            if decision.verdict.lower() == "block":
                start_response("403 Forbidden", [("Content-Type", "application/json")])
                return [b'{"error": "blocked"}']
        else:
            self.run_async(request)
        
        # Add headers to response
        def custom_start_response(status, headers):
            headers.extend([
                ("X-Argos-Verdict", "allow"),
                ("X-Argos-Reason", "async_pending"),
            ])
            return start_response(status, headers)
        
        return self.app(environ, custom_start_response)
```

#### Minimal Custom Middleware Example

Here's a minimal example for a simple custom framework:

```python
from argos import create_client
from argos.models import Decision

class SimpleArgosMiddleware:
    """Minimal middleware for custom frameworks."""

    def __init__(self, app, api_key, mode="async", exclude_paths=None):
        self.app = app
        self.client = create_client(api_key=api_key)
        self.mode = mode
        self.exclude_paths = exclude_paths or ["/health", "/metrics"]

    def should_skip(self, path):
        return any(path.startswith(p) for p in self.exclude_paths)

    def get_client_ip(self, environ):
        """Extract IP from WSGI environ."""
        forwarded = environ.get("HTTP_X_FORWARDED_FOR")
        if forwarded:
            return forwarded.split(",")[0].strip()
        return environ.get("REMOTE_ADDR", "unknown")

    def __call__(self, environ, start_response):
        path = environ.get("PATH_INFO", "")

        if self.should_skip(path):
            return self.app(environ, start_response)

        # Build payload
        payload = {
            "event_type": environ.get("REQUEST_METHOD", "unknown").lower(),
            "path": path,
            "ip_address": self.get_client_ip(environ),
            "user_agent": environ.get("HTTP_USER_AGENT", ""),
        }

        if self.mode == "sync":
            # Wait for decision
            try:
                event = self.client.ingest(payload)
                if event.signal == "BLOCK":
                    start_response("403 Forbidden", [("Content-Type", "application/json")])
                    return [b'{"error": "blocked"}']
            except Exception:
                pass  # Allow on errors
        else:
            # Queue and continue
            try:
                self.client.queue_event(payload)
            except Exception:
                pass

        # Add Argos headers
        def wrapped_start_response(status, headers):
            headers.extend([
                ("X-Argos-Verdict", "allow"),
                ("X-Argos-Reason", "async_pending" if self.mode == "async" else "ok"),
            ])
            return start_response(status, headers)

        return self.app(environ, wrapped_start_response)
```

#### Using with Custom Framework

```python
# Example with a custom WSGI app
from my_framework import App

app = App()
wrapped_app = SimpleArgosMiddleware(
    app,
    api_key="your-api-key",
    mode="sync",  # or "async"
    exclude_paths=["/health", "/static"]
)

# Run with any WSGI server
if __name__ == "__main__":
    from wsgiref.simple_server import make_server
    server = make_server("localhost", 8000, wrapped_app)
    server.serve_forever()
```

#### Extending Existing Middleware

You can also extend the existing middleware classes to customize behavior:

```python
from argos.middleware.fastapi import FastAPIMiddleware
from argos.middleware.base import MiddlewareConfig

class CustomFastAPIMiddleware(FastAPIMiddleware):
    """Extended FastAPI middleware with custom behavior."""

    async def process_sync(self, request):
        """Override to add custom logic before/after processing."""
        # Custom pre-processing
        custom_data = await self.extract_custom_data(request)

        # Call parent processing
        decision = await super().process_sync(request)

        # Custom post-processing
        if decision.verdict.lower() == "block":
            await self.log_block_event(request, decision)

        return decision

    async def extract_custom_data(self, request):
        """Extract custom data for your use case."""
        return {"custom_field": "value"}

    async def log_block_event(self, request, decision):
        """Log blocked requests to your own logging system."""
        print(f"Blocked: {request.url.path} - {decision.reason}")

# Use it exactly like the regular middleware
app.add_middleware(CustomFastAPIMiddleware, client=client, config=config)
```

## Response Headers

When using middleware, responses include Argos headers:

```
X-Argos-Verdict: block
X-Argos-Reason: anomaly_score=0.85
```

## Health Check

```python
# Check if Argos API is healthy
health = client.health_check()
print(health)
```

## TypeScript/JavaScript SDK

For frontend and Node.js integrations, see the official Argos JavaScript SDK:

```bash
npm install @argos.dev/js-sdk
```

## License

MIT License - see [LICENSE](LICENSE) for details.

## Support

- Documentation: https://docs.argos.dev
- Issues: https://github.com/argosdev/argos-python/issues
- Email: team@tachynoix.net
