Skip to main content

Official Python SDK for LiteSOC - Security event tracking and threat detection

Project description

LiteSOC Python SDK

Official Python SDK for LiteSOC - Security event tracking and threat detection for your applications.

PyPI version Python Version CI Coverage License: MIT

Installation

pip install litesoc

Quick Start

from litesoc import LiteSOC, SecurityEvents

# Initialize the SDK
litesoc = LiteSOC(api_key="your-api-key")

# Track a login failure using the SecurityEvents enum
litesoc.track(SecurityEvents.AUTH_LOGIN_FAILED,
    actor_id="user_123",
    actor_email="user@example.com",
    user_ip="192.168.1.1",  # Required for Security Intelligence
    metadata={"reason": "invalid_password"}
)

# Get alerts from the Management API
alerts = litesoc.get_alerts(status="open", severity="high")

# Flush remaining events before shutdown
litesoc.flush()

Features

  • 26 standard security event types - Authentication, authorization, admin, data, and security events
  • Management API - Get alerts, resolve alerts, fetch events
  • Automatic batching - Events are batched for efficient delivery
  • Retry logic - Failed events are automatically retried
  • Type hints - Full type annotations for IDE support
  • Thread-safe - Safe to use across multiple threads
  • Context manager support - Use with with statement for automatic cleanup
  • Custom exceptions - Typed error handling with LiteSOCError, RateLimitError, etc.
  • 🗺️ GeoIP Enrichment - Automatic location data from IP addresses
  • 🛡️ Network Intelligence - VPN, Tor, Proxy & Datacenter detection
  • 📊 Threat Scoring - Auto-assigned severity (Low → Critical)

Security Intelligence (Automatic Enrichment)

When you provide user_ip, LiteSOC automatically enriches your events with:

🗺️ Geolocation

  • Country & City resolution
  • Latitude/Longitude coordinates
  • Interactive map visualization in dashboard

🛡️ Network Intelligence

  • VPN Detection - NordVPN, ExpressVPN, Surfshark, etc.
  • Tor Exit Nodes - Anonymizing network detection
  • Proxy Detection - HTTP/SOCKS proxy identification
  • Datacenter IPs - AWS, GCP, Azure, DigitalOcean, etc.

📊 Threat Scoring

Events are auto-classified by severity:

  • Low - Normal activity
  • Medium - Unusual patterns
  • High - Suspicious behavior
  • Critical - Active threats (triggers instant alerts)

Important: Always include user_ip for full Security Intelligence features.

Configuration Options

from litesoc import LiteSOC

litesoc = LiteSOC(
    api_key="your-api-key",      # Required
    endpoint="https://...",       # Custom API endpoint
    batching=True,                # Enable event batching (default: True)
    batch_size=10,                # Events before auto-flush (default: 10)
    flush_interval=5.0,           # Seconds between auto-flushes (default: 5.0)
    debug=False,                  # Enable debug logging (default: False)
    silent=True,                  # Fail silently on errors (default: True)
    timeout=5.0,                  # Request timeout in seconds (default: 5.0)
)

Tracking Events

Basic Usage

# Track any event type
litesoc.track("auth.login_failed",
    actor_id="user_123",
    actor_email="user@example.com",
    user_ip="192.168.1.1"
)

Using Actor Object

from litesoc import LiteSOC, Actor

litesoc = LiteSOC(api_key="your-api-key")

actor = Actor(id="user_123", email="user@example.com")
litesoc.track("auth.login_success", actor=actor, user_ip="192.168.1.1")

With Severity Level

from litesoc import EventSeverity

litesoc.track("security.suspicious_activity",
    actor_id="user_123",
    user_ip="192.168.1.1",
    severity=EventSeverity.CRITICAL,
    metadata={"reason": "impossible travel detected"}
)

With Metadata

litesoc.track("data.export",
    actor_id="user_123",
    user_ip="192.168.1.1",
    metadata={
        "file_type": "csv",
        "record_count": 1000,
        "export_reason": "monthly_report"
    }
)

Management API

The SDK provides methods to interact with the LiteSOC Management API:

Get Alerts

from litesoc import LiteSOC, LiteSOCAuthError, RateLimitError

litesoc = LiteSOC(api_key="your-api-key")

# Get all open alerts
alerts = litesoc.get_alerts(status="open")

# Get high severity alerts
alerts = litesoc.get_alerts(severity="high", limit=50)

for alert in alerts.get("alerts", []):
    print(f"Alert: {alert['id']} - {alert['type']}")

Get Single Alert

alert = litesoc.get_alert("alert_abc123")
print(f"Alert type: {alert['type']}")

Resolve Alert

# Resolve with notes
litesoc.resolve_alert(
    "alert_abc123",
    "fixed",
    notes="Issue was addressed in PR #123"
)

# Mark as false positive
litesoc.resolve_alert("alert_abc123", "false_positive")

Mark Alert Safe

litesoc.mark_alert_safe(
    "alert_abc123",
    notes="This is expected behavior from the test suite"
)

Get Events

events = litesoc.get_events(limit=50)

for event in events.get("events", []):
    print(f"Event: {event['type']} - {event['timestamp']}")

Get Single Event

event = litesoc.get_event("event_abc123")
print(f"Event type: {event['type']}")

SecurityEvents Enum

Use the SecurityEvents enum for type-safe event tracking:

from litesoc import LiteSOC, SecurityEvents

litesoc = LiteSOC(api_key="your-api-key")

# Use enum for type safety and IDE autocomplete
litesoc.track(SecurityEvents.AUTH_LOGIN_FAILED, actor_id="user_123")
litesoc.track(SecurityEvents.ADMIN_PRIVILEGE_ESCALATION, actor_id="admin_user")
litesoc.track(SecurityEvents.DATA_SENSITIVE_ACCESS, actor_id="user_123")

All 26 standard events are available:

  • AUTH_LOGIN_SUCCESS, AUTH_LOGIN_FAILED, AUTH_LOGOUT, etc.
  • AUTHZ_ROLE_CHANGED, AUTHZ_PERMISSION_GRANTED, etc.
  • ADMIN_PRIVILEGE_ESCALATION, ADMIN_USER_IMPERSONATION, etc.
  • DATA_BULK_DELETE, DATA_SENSITIVE_ACCESS, DATA_EXPORT
  • SECURITY_SUSPICIOUS_ACTIVITY, SECURITY_BRUTE_FORCE_DETECTED, etc.

Error Handling

The SDK provides custom exception classes for proper error handling:

from litesoc import (
    LiteSOC,
    LiteSOCError,
    LiteSOCAuthError,
    RateLimitError,
    PlanRestrictedError,
)

litesoc = LiteSOC(api_key="your-api-key")

try:
    alerts = litesoc.get_alerts()
except LiteSOCAuthError as e:
    print(f"Authentication failed: {e.message} (status: {e.status_code})")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except PlanRestrictedError as e:
    print(f"Feature requires {e.required_plan} plan")
except LiteSOCError as e:
    print(f"API error: {e.message}")

Convenience Methods

The SDK provides convenience methods for common security events:

# Track login failures
litesoc.track_login_failed("user_123", user_ip="192.168.1.1")

# Track login successes
litesoc.track_login_success("user_123", user_ip="192.168.1.1")

# Track privilege escalation (critical severity)
litesoc.track_privilege_escalation("admin_user", user_ip="192.168.1.1")

# Track sensitive data access (high severity)
litesoc.track_sensitive_access("user_123", "customer_pii_table", user_ip="192.168.1.1")

# Track bulk deletions (high severity)
litesoc.track_bulk_delete("admin_user", record_count=500, user_ip="192.168.1.1")

# Track role changes
litesoc.track_role_changed("user_123", old_role="viewer", new_role="admin", user_ip="192.168.1.1")

# Track access denied
litesoc.track_access_denied("user_123", resource="/admin/settings", user_ip="192.168.1.1")

Event Types

26 Standard Events (Primary)

These are the primary events for comprehensive security coverage:

Category Event Type Description
Auth auth.login_success Successful user login
Auth auth.login_failed Failed login attempt
Auth auth.logout User logout
Auth auth.password_reset Password reset completed
Auth auth.mfa_enabled MFA enabled on account
Auth auth.mfa_disabled MFA disabled on account
Auth auth.session_expired Session timeout/expiry
Auth auth.token_refreshed Token refresh
Authz authz.role_changed User role modified
Authz authz.permission_granted Permission assigned
Authz authz.permission_revoked Permission removed
Authz authz.access_denied Access denied event
Admin admin.privilege_escalation Admin privilege escalation
Admin admin.user_impersonation Admin impersonating user
Admin admin.settings_changed System settings modified
Admin admin.api_key_created New API key generated
Admin admin.api_key_revoked API key revoked
Admin admin.user_suspended User account suspended
Admin admin.user_deleted User account deleted
Data data.bulk_delete Bulk data deletion
Data data.sensitive_access PII/sensitive data accessed
Data data.export Data export operation
Security security.suspicious_activity Suspicious behavior detected
Security security.rate_limit_exceeded Rate limit triggered
Security security.ip_blocked IP address blocked
Security security.brute_force_detected Brute force attack detected

Extended Events (Backward Compatible)

Additional events for granular tracking:

  • auth.password_changed, auth.password_reset_requested, auth.mfa_challenge_success, auth.mfa_challenge_failed, auth.session_created
  • user.created, user.updated, user.deleted, user.email_changed, user.profile_updated
  • authz.role_assigned, authz.role_removed, authz.access_granted
  • admin.invite_sent, admin.invite_accepted, admin.member_removed
  • data.import, data.bulk_update, data.download, data.upload, data.shared
  • security.ip_unblocked, security.account_locked, security.impossible_travel, security.geo_anomaly
  • api.key_used, api.rate_limited, api.error, api.webhook_sent, api.webhook_failed
  • billing.subscription_created, billing.subscription_cancelled, billing.payment_succeeded, billing.payment_failed

Framework Integration

Flask

from flask import Flask, request, g
from litesoc import LiteSOC

app = Flask(__name__)
litesoc = LiteSOC(api_key="your-api-key")

@app.route("/login", methods=["POST"])
def login():
    user_ip = request.headers.get("X-Forwarded-For", request.remote_addr)
    
    # Attempt authentication
    user = authenticate(request.form["email"], request.form["password"])
    
    if user:
        litesoc.track_login_success(user.id, actor_email=user.email, user_ip=user_ip)
        return {"success": True}
    else:
        litesoc.track_login_failed(request.form["email"], user_ip=user_ip)
        return {"success": False}, 401

Django

from django.contrib.auth.signals import user_logged_in, user_login_failed
from django.dispatch import receiver
from litesoc import LiteSOC

litesoc = LiteSOC(api_key="your-api-key")

@receiver(user_logged_in)
def track_login_success(sender, request, user, **kwargs):
    user_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR"))
    litesoc.track_login_success(str(user.id), actor_email=user.email, user_ip=user_ip)

@receiver(user_login_failed)
def track_login_failure(sender, credentials, request, **kwargs):
    user_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR"))
    litesoc.track_login_failed(credentials.get("username", "unknown"), user_ip=user_ip)

FastAPI

from fastapi import FastAPI, Request, Depends
from litesoc import LiteSOC

app = FastAPI()
litesoc = LiteSOC(api_key="your-api-key")

@app.post("/login")
async def login(request: Request, credentials: LoginRequest):
    user_ip = request.headers.get("X-Forwarded-For", request.client.host)
    
    user = await authenticate(credentials.email, credentials.password)
    
    if user:
        litesoc.track_login_success(user.id, actor_email=user.email, user_ip=user_ip)
        return {"success": True}
    else:
        litesoc.track_login_failed(credentials.email, user_ip=user_ip)
        raise HTTPException(status_code=401)

Context Manager Support

from litesoc import LiteSOC

with LiteSOC(api_key="your-api-key") as litesoc:
    litesoc.track("auth.login_success", actor_id="user_123")
    # Events are automatically flushed when exiting the context

Queue Management

# Get current queue size
queue_size = litesoc.get_queue_size()

# Manually flush all events
litesoc.flush()

# Clear queue without sending
litesoc.clear_queue()

# Graceful shutdown
litesoc.shutdown()

Error Handling

By default, the SDK fails silently (silent=True). To catch errors:

litesoc = LiteSOC(api_key="your-api-key", silent=False)

try:
    litesoc.track("auth.login_failed", actor_id="user_123")
    litesoc.flush()
except Exception as e:
    print(f"Failed to track event: {e}")

Debug Mode

Enable debug logging to troubleshoot issues:

litesoc = LiteSOC(api_key="your-api-key", debug=True)
# Logs will be printed to stdout

Performance Tips

Timeout Configuration

The SDK uses a 5-second default timeout for all API requests. This prevents slow network conditions from blocking your application:

# Default: 5-second timeout
litesoc = LiteSOC(api_key="your-api-key")

# Custom timeout for the entire client
litesoc = LiteSOC(api_key="your-api-key", timeout=10.0)

# Per-request timeout override (takes precedence)
alerts = litesoc.get_alerts(timeout=2.0)  # 2-second timeout for this call
events = litesoc.get_events(timeout=3.0)  # 3-second timeout for this call

Graceful Timeout Handling

The track() method handles timeouts gracefully, returning False instead of raising an exception:

# track() returns True on success, False on timeout
success = litesoc.track("auth.login_success", actor_id="user_123")
if not success:
    print("Event tracking timed out, but application continues")

Batching for High-Throughput

For applications with high event volume, batching reduces network overhead:

litesoc = LiteSOC(
    api_key="your-api-key",
    batching=True,
    batch_size=50,          # Send after 50 events
    flush_interval=10.0,    # Or every 10 seconds
)

Development

Prerequisites

  • Python 3.9+
  • pip

Setup

# Clone the repository
git clone https://github.com/LiteSOC/litesoc-python.git
cd litesoc-python

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -e ".[dev]"

Running Tests

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run with coverage
pytest --cov=src/litesoc --cov-report=term-missing

# Run specific test file
pytest tests/test_litesoc.py -v

Code Quality

# Run linter
ruff check src/ tests/

# Run type checker
mypy src/

# Format code
ruff format src/ tests/

Building

# Build package
python -m build

# Install locally
pip install -e .

License

MIT License - see LICENSE for details.

Links

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

litesoc-2.2.0.tar.gz (33.0 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

litesoc-2.2.0-py3-none-any.whl (21.0 kB view details)

Uploaded Python 3

File details

Details for the file litesoc-2.2.0.tar.gz.

File metadata

  • Download URL: litesoc-2.2.0.tar.gz
  • Upload date:
  • Size: 33.0 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.14.3

File hashes

Hashes for litesoc-2.2.0.tar.gz
Algorithm Hash digest
SHA256 936353ef0884e7f509001c93b81668a77aadb6650e7cca75eecdd69bede8e57c
MD5 92a58b71b9db301b24189420011288d2
BLAKE2b-256 f53f438330e38e83f4e95b868a1d07c671d4226b4bab70db22ba3bc2774fbfd6

See more details on using hashes here.

File details

Details for the file litesoc-2.2.0-py3-none-any.whl.

File metadata

  • Download URL: litesoc-2.2.0-py3-none-any.whl
  • Upload date:
  • Size: 21.0 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.14.3

File hashes

Hashes for litesoc-2.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 ec2dea8ed83aba6077585bf7509a0959f6ef44707e6d513a55ea0f75e43b42e0
MD5 5214f5ddeb70c5a610cad7ad85cb0082
BLAKE2b-256 5d5e5c8b2184dc52d9229b19fb68e5fd15ae4208d67a69c26dd2954130a00e8c

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page