loguard 2.0.7

Real-time attack detection and IP blocking for Python — FastAPI, Django, async support

LoGuard Python SDK

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

pip install loguard

With framework extras:

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

Quick Start

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

# 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

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

# 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}

# 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.

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.

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:

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

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 is running on your host, connect it for kernel-level IP blocking:

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
• Documentation
• PyPI