Metadata-Version: 2.4
Name: loguard
Version: 2.0.7
Summary: Real-time attack detection and IP blocking for Python — FastAPI, Django, async support
Project-URL: Homepage, https://logguard.io
Project-URL: Documentation, https://docs.logguard.io/sdk/python
Project-URL: Repository, https://github.com/logguard/logguard-python
License: MIT
Keywords: alerts,blacklist,monitoring,sdk,security
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
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
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25.0
Provides-Extra: dev
Requires-Dist: httpx; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: respx; extra == 'dev'
Provides-Extra: django
Requires-Dist: django>=3.2; extra == 'django'
Provides-Extra: fastapi
Requires-Dist: starlette>=0.27.0; extra == 'fastapi'
Description-Content-Type: text/markdown

# LoGuard Python SDK

[![PyPI version](https://img.shields.io/pypi/v/loguard.svg)](https://pypi.org/project/loguard/)
[![Python](https://img.shields.io/pypi/pyversions/loguard.svg)](https://pypi.org/project/loguard/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)

Real-time security monitoring for Python applications. Detect attacks, block malicious IPs, define custom alert rules — with a single SDK that works across FastAPI, Django, and any Python backend.

## Installation

```bash
pip install loguard
```

With framework extras:

```bash
pip install loguard[fastapi]   # FastAPI / Starlette middleware
pip install loguard[django]    # Django middleware
```

## Quick Start

```python
import os
from loguard import monitor

monitor.init(
    api_key=os.environ["LOGUARD_API_KEY"],
    base_url=os.environ.get("LOGUARD_BASE_URL", "https://loguard.org"),
    env=os.environ.get("LOGUARD_ENV", "production"),
)
```

### Track events

```python
# Synchronous — waits for server response, returns IngestResult
result = monitor.event(
    type="login_failed",
    ip="1.2.3.4",
    path="/api/login",
    status_code=401,
    user_id="user_123",           # optional
    meta={"method": "POST"},      # optional
)
print(result)
# IngestResult(ok=True, inserted=1, alerts_fired=0, plan='pro')

# Non-blocking — queues internally, never raises, ideal for production
monitor.event_fire_and_forget(
    type="http_request",
    ip="1.2.3.4",
    path="/api/users",
    status_code=200,
)

# Async
result = await monitor.aevent(
    type="login_failed",
    ip="1.2.3.4",
    path="/api/login",
    status_code=401,
)

# Batch — multiple events in one request
result = monitor.event_batch([
    {"type": "http_request", "ip": "1.2.3.4", "path": "/",         "status_code": 200},
    {"type": "login_failed",  "ip": "1.2.3.4", "path": "/login",    "status_code": 401},
    {"type": "http_request", "ip": "5.6.7.8", "path": "/.env",     "status_code": 404},
])
print(f"accepted={result.inserted} alerts={result.alerts_fired}")

# Async batch
result = await monitor.aevent_batch([...])
```

### Event types

| `type` | When to use |
|---|---|
| `http_request` | Any incoming HTTP request |
| `login_failed` | Failed authentication (401) |
| `login_success` | Successful login |
| `forbidden` | Access denied (403) |
| `waf_block` | Request blocked by WAF |
| `bot_detected` | Identified bot traffic |

## Integrations

### FastAPI / Starlette

```python
import os
from fastapi import FastAPI
from loguard import monitor
from loguard.integrations.fastapi import LoGuardMiddleware

app = FastAPI()

monitor.init(
    api_key=os.environ["LOGUARD_API_KEY"],
    base_url=os.environ.get("LOGUARD_BASE_URL", "https://loguard.org"),
)

app.add_middleware(
    LoGuardMiddleware,
    track_statuses={400, 401, 403, 404, 429, 500, 502, 503},
    enforce_blacklist=True,        # returns 403 for blocked IPs before hitting routes
    get_user_id=lambda r: r.state.user_id if hasattr(r.state, "user_id") else None,
)
```

Blocked IPs receive `HTTP 403 {"detail": "Forbidden", "reason": "..."}` before the request reaches your route handlers.

### Django

```python
# settings.py
MIDDLEWARE = [
    "loguard.integrations.django.LogguardMiddleware",
    # ... other middleware
]

LOGUARD_ENFORCE_BLACKLIST = True    # block blacklisted IPs at middleware level
LOGUARD_TRACK_ALL         = False   # True = track all requests, not just errors
LOGUARD_TRACK_STATUSES    = {400, 401, 403, 404, 429, 500, 502, 503}
```

```python
# apps.py
import os
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = "myapp"

    def ready(self):
        from loguard import monitor
        monitor.init(
            api_key=os.environ["LOGUARD_API_KEY"],
            base_url=os.environ.get("LOGUARD_BASE_URL", "https://loguard.org"),
            env=os.environ.get("LOGUARD_ENV", "production"),
        )
```

## Alert Rules

Define custom rules — the server evaluates them on every ingest and fires alerts when conditions match.

```python
from loguard import monitor
from loguard.models import AlertRule, AlertCondition

# Brute force: >10 failed logins/min from same IP
rule = monitor.alerts.create(AlertRule(
    name="Brute force",
    conditions=[
        AlertCondition(field="type",            op="eq", value="login_failed"),
        AlertCondition(field="rate_per_minute", op="gt", value=10),
    ],
    severity="high",          # "low" | "medium" | "high" | "critical"
    actions=["notify", "block"],  # "notify" | "block" | "log"
    logic="and",              # "and" (all conditions) | "or" (any condition)
    cooldown_sec=300,         # min seconds between repeated alerts for same IP
))
print(rule.id)

# Scanner: any request to sensitive paths
rule2 = monitor.alerts.create(AlertRule(
    name="Path scanner",
    conditions=[
        AlertCondition(field="path", op="in", value=["/.env", "/.git/config", "/admin"]),
    ],
    severity="medium",
    actions=["notify"],
))

# Manage
rules = monitor.alerts.list()
rule.enabled = False
monitor.alerts.update(rule)
monitor.alerts.delete(rule.id)

# Async variants: acreate / alist / aupdate / adelete
```

### Condition fields and operators

| Field | Type | Description |
|---|---|---|
| `type` | string | Event type (`login_failed`, `http_request`, …) |
| `ip` | string | Source IP address |
| `path` | string | Request path |
| `status_code` | int | HTTP status code |
| `user_id` | string | Authenticated user ID |
| `rate_per_minute` | int | Request rate per minute from same IP |
| `rate_per_hour` | int | Request rate per hour from same IP |

| Operator | Description |
|---|---|
| `eq` / `neq` | Equal / not equal |
| `gt` / `gte` / `lt` / `lte` | Numeric comparison |
| `contains` / `startswith` / `endswith` | String matching |
| `regex` | Regular expression |
| `in` / `not_in` | Value in list |

## Blacklist

Block or flag IPs, users, CIDR ranges, paths, and user agents. Checks use a 60-second TTL cache — safe to call on every request.

```python
from datetime import datetime, timedelta, timezone
from loguard.models import BlacklistEntry

# Block an IP permanently
monitor.blacklist.block_ip("185.220.101.1", reason="Known scanner")

# Block with expiry
expires = (datetime.now(timezone.utc) + timedelta(hours=24)).isoformat()
monitor.blacklist.block_ip("1.2.3.4", reason="Brute force", expires_at=expires)

# Block a CIDR range
monitor.blacklist.block_cidr("185.220.0.0/16", reason="Tor exit nodes")

# Block a user
monitor.blacklist.block_user("user_abc123", reason="Fraud", expires_at=expires)

# Flag (mark but don't block)
monitor.blacklist.flag_ip("9.9.9.9", reason="Suspicious activity")

# Full control via BlacklistEntry
entry = monitor.blacklist.add(BlacklistEntry(
    type="user_agent",
    value="sqlmap",
    reason="Attack tool",
    action="block",
))

# Check (with 60s TTL cache)
result = monitor.blacklist.check(type="ip", value="185.220.101.1")
if result["blacklisted"]:
    print(result["action"])   # "block" | "flag" | "alert"
    print(result["reason"])

# List, update, remove
entries = monitor.blacklist.list(type="ip", enabled_only=True)
monitor.blacklist.remove(entry.id)

# Async variants: aadd / alist / aremove / acheck
result = await monitor.blacklist.acheck(type="ip", value="1.2.3.4")
```

### Blacklist entry types

| `type` | `value` example | Description |
|---|---|---|
| `ip` | `"1.2.3.4"` | Single IPv4 address |
| `cidr` | `"185.220.0.0/16"` | IP range |
| `user_id` | `"user_abc123"` | Authenticated user |
| `path` | `"/admin"` | Request path prefix |
| `user_agent` | `"sqlmap"` | User-Agent substring |

## IngestResult

Every `monitor.event()` call returns an `IngestResult`:

```python
result = monitor.event(type="login_failed", ip="1.2.3.4", path="/login", status_code=401)

result.ok            # bool — request accepted
result.inserted      # int  — events accepted by server
result.dropped       # int  — events dropped (quota or plan limit)
result.alerts_fired  # int  — alerts triggered by this batch
result.alerts        # List[AlertOut] — alert details
result.plan          # str  — current plan ("free", "pro", "business", …)
result.usage_info    # UsageInfo — quota usage for current month

print(result.usage_info)
# UsageInfo(used=48851/1000000, remaining=951149, month='2026-06')

if result.usage_info.is_near_limit:
    print("Approaching monthly quota")
```

## Error Handling

```python
from loguard.exceptions import (
    LogguardAuthError,        # invalid or missing API key
    LogguardQuotaError,       # monthly event quota exceeded
    LogguardConnectionError,  # server unreachable or 5xx
    LogguardValidationError,  # invalid event data
    LogguardNotFoundError,    # alert rule / blacklist entry not found
    LogguardConflictError,    # duplicate entry
)

try:
    monitor.event(type="login_failed", ip="1.2.3.4", path="/login", status_code=401)
except LogguardQuotaError:
    pass   # quota exceeded — handle gracefully
except LogguardConnectionError:
    pass   # server unreachable — fail open
except LogguardAuthError:
    raise  # bad API key — fail loud
```

`event_fire_and_forget()` never raises — safe to use in any middleware without try/except.

## Kernel Firewall (optional)

If [LoGuard Daemon](https://loguard.org/daemon) is running on your host, connect it for kernel-level IP blocking:

```python
monitor.init(api_key=os.environ["LOGUARD_API_KEY"])
monitor.init_firewall(sock_path="/var/run/loguard.sock")

# block_ip() now also triggers kernel-level NF_DROP automatically
monitor.blacklist.block_ip("1.2.3.4", reason="SQL_INJECTION")

# Direct firewall access
monitor.firewall.block_ip("1.2.3.4", duration=3600)
monitor.firewall.block_country("KP")
stats = monitor.firewall.get_stats()
```

`init_firewall()` is fail-safe — if the daemon is not running, it silently disables kernel blocking and the SDK continues working normally.

## Environment Variables

| Variable | Default | Description |
|---|---|---|
| `LOGUARD_API_KEY` | — | Your project API key (required) |
| `LOGUARD_BASE_URL` | `https://loguard.org` | API base URL |
| `LOGUARD_ENV` | `production` | Environment tag (`production`, `staging`, `development`) |

## Links

- [Dashboard](https://loguard.org)
- [Documentation](https://docs.loguard.org/sdk/python)
- [PyPI](https://pypi.org/project/loguard/)