LogShield CLI
#python#automation#microsaas
Abbatti la tua spesa SaaS da 300 $/mese con LogShield CLI: un log shipper e oscuratore di PII in Go senza dipendenze. Trasmetti in streaming log compressi con zstd direttamente su S3/R2 per pochissimi centesimi, generando alert di incidente istantanei su Slack/Discord in caso di panic 5xx. Include binari multi-arch, Docker sidecar e configurazioni systemd pronte all'uso.
Leggi l'approfondimento tecnicoTaglia i Costi SaaS di $300/mese: Redazione PII Locale e S3 Streaming con LogShield CLImain.pymain.py — estratto d'esempio nello stile del tool
#!/usr/bin/env python3
"""LogShield CLI - Zero-dependency log shipping, PII redaction, and alerting.
LogShield CLI provides ultra-lightweight telemetry management by redacting
sensitive PII patterns, compressing logs using native stream handling, and
shipping structured payloads directly to S3/R2 compatible storage while
evaluating real-time alert rules for instant operational notifications.
Features:
- Zero-dependency PII redaction (email, IPv4, API tokens, JWTs)
- Streaming zstd/gzip archive generation for low-cost object storage
- Direct S3 / Cloudflare R2 signature v4 API log forwarding
- Incident alerting triggers for 5xx panics and anomaly thresholds
- Production-ready sidecar configuration and systemd generator
Usage:
$ logshield run --config /etc/logshield/config.json --target s3://my-bucket/logs
$ cat app.log | logshield redact --preset strict | logshield ship --endpoint r2
"""
from dataclasses import dataclass, field
from enum import Enum, auto
from typing import Dict, List, Optional, Pattern, Set, Tuple
import re
import sys
import time
@dataclass(frozen=True)
class RedactionRule:
"""Defines a regex compilation rule and replacement string for PII masking."""
name: str
pattern: Pattern[str]
replacement: str = "[REDACTED]"
class LogSeverity(Enum):
"""Standard log severity levels mapped to alert escalation thresholds."""
DEBUG = auto()
INFO = auto()
WARN = auto()
ERROR = auto()
FATAL = auto()
class LogStreamProcessor:
"""Core engine handling stream parsing, inline PII filtering, and buffer dispatch."""
def __init__(self, rules: List[RedactionRule], max_buffer_size_mb: int = 16) -> None:
self.rules: List[RedactionRule] = rules
self.max_buffer_bytes: int = max_buffer_size_mb * 1024 * 1024
self._buffer: List[str] = []
self._current_size_bytes: int = 0
# ... the full file ships with your download