utils.py
Rama main · réplica de lectura del repositorio original.
"""piSynapse Utilities
Shared helpers: retry decorator, text processing.
"""
import asyncio
import logging
import random
import re
import time
from contextvars import ContextVar
from functools import wraps
logger = logging.getLogger("piSynapse")
# Request-scoped Nextcloud credentials (collective services). The dispatcher
# sets this per tool call from the caller's encrypted store; NC factories
# prefer it over the shared .env account. Task-contained by asyncio context
# semantics; every run_tool entry sets it explicitly (creds or None).
NC_CREDS: ContextVar[dict | None] = ContextVar("nc_creds", default=None)
# Whether the shared (admin) Nextcloud account may serve this call. Set from
# the caller's admin flag alongside NC_SHARED_OK — without it, a non-admin with
# no personal credentials would silently land on the admin's account.
NC_SHARED_OK: ContextVar[bool] = ContextVar("nc_shared_ok", default=False)
async def bind_nc_context(user_id: str | None) -> None:
"""Bind the per-user Nextcloud credential context for THIS task.
One shared contract for every NC-reading path (tool dispatcher,
widgets, ...): the caller's own credentials, and the shared account
for admins only — a credential-less non-admin must never silently
fall through to the shared account (invisibility hole). Call it in
the async task that runs the NC calls (directly or via
``asyncio.to_thread``, which propagates the context); ContextVars
do not cross tasks.
"""
from db import get_credential, get_user
NC_CREDS.set(await get_credential(user_id, "nextcloud"))
caller = await get_user(user_id) if user_id else None
NC_SHARED_OK.set(bool(caller is not None and caller.get("is_admin")))
def nc_cache_key() -> str:
"""Identity key for the process-global Nextcloud caches (calendar/tasks).
``calendar_ops`` and ``nextcloud_tasks`` keep module-level caches
(primary calendar, events, todos), but the Nextcloud identity of a call
is request-scoped (``NC_CREDS`` / ``NC_SHARED_OK``). Every cache entry
must be keyed by this identity or one user's calendar/todos are served
to another (2026-10-08 analysis).
Mirrors the client-selection logic in ``_get_nextcloud_client`` /
``_get_dav_client``: per-user creds -> ``u|...``, shared admin account
-> ``s|...``. ``"deny"`` marks calls with no usable identity — their
fetches fail before any cache write, so the key never populates and
nothing can ever be served to such a caller.
"""
from config import NEXTCLOUD_URL, NEXTCLOUD_USER
creds = NC_CREDS.get()
if creds and creds.get("url") and creds.get("password"):
return "u|" + str(creds.get("url") or "") + "|" + str(creds.get("user") or "")
if NC_SHARED_OK.get():
# Shared admin account. The key stays distinct from "deny" even when
# NEXTCLOUD_URL is unset — the client gate refuses the fetch anyway,
# so the entry never populates, and a runtime config change cannot
# alias one identity onto another.
return "s|" + str(NEXTCLOUD_URL or "") + "|" + str(NEXTCLOUD_USER or "")
return "deny"
def _is_retryable(exc: BaseException) -> bool:
"""Decide whether a failed attempt should be retried.
HTTP status errors are only retried for 429 (rate limit) and 5xx
(server-side/transient) — a 4xx like 404 or 403 is deterministic and
retrying would just delay the inevitable. Any other exception type is
treated as potentially transient (network blips, timeouts, DAV errors).
"""
try:
import httpx
except ImportError:
httpx = None
if httpx is not None and isinstance(exc, httpx.HTTPStatusError):
return exc.response.status_code == 429 or exc.response.status_code >= 500
return True
def retry(attempts: int = 2, delay: float = 1.0, backoff: float = 2.0, jitter: float = 0.1):
"""Decorator that retries a sync or async function on exception.
Works for both regular functions (runs in executor) and async functions.
Wait time grows exponentially: ``delay * backoff**attempt`` plus a small
random jitter to avoid thundering-herd retries. HTTP 4xx errors other
than 429 are not retried (see ``_is_retryable``). Re-raises the last
exception if all attempts fail.
"""
def decorator(fn):
def _wait_for(attempt: int):
return delay * (backoff ** attempt) + random.uniform(0, jitter * delay)
if asyncio.iscoroutinefunction(fn):
@wraps(fn)
async def async_wrapper(*args, **kwargs):
last_exc = None
for attempt in range(attempts):
try:
return await fn(*args, **kwargs)
except Exception as e:
last_exc = e
if attempt >= attempts - 1 or not _is_retryable(e):
break
wait = _wait_for(attempt)
logger.warning(f"{fn.__name__} attempt {attempt + 1} failed: {e}; retrying in {wait:.1f}s")
await asyncio.sleep(wait)
raise last_exc
return async_wrapper
else:
@wraps(fn)
def sync_wrapper(*args, **kwargs):
last_exc = None
for attempt in range(attempts):
try:
return fn(*args, **kwargs)
except Exception as e:
last_exc = e
if attempt >= attempts - 1 or not _is_retryable(e):
break
wait = _wait_for(attempt)
logger.warning(f"{fn.__name__} attempt {attempt + 1} failed: {e}; retrying in {wait:.1f}s")
time.sleep(wait)
raise last_exc
return sync_wrapper
return decorator
# -- Text Processing --
# Invisible/format characters frequently used as spam padding in email bodies
# (nbsp, figure space, soft hyphen, zero-width chars, combining marks, controls).
_INVISIBLE_RE = re.compile(
"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f\u00ad\u034f\u0350-\u036f\u061c\u115f\u1160"
"\u17b4-\u17b5\u180b-\u180d\u200b-\u200f\u2028\u2029\u2060-\u2064"
"\u206a-\u206f\ufeff]"
)
def clean_body_text(text: str) -> str:
"""Normalize email body text: strip invisible spam padding, collapse whitespace.
Reduces token usage dramatically when the body is passed to the LLM.
"""
if not text:
return ""
text = _INVISIBLE_RE.sub("", text)
return re.sub(r"\s+", " ", text).strip()
def decode_email_header(value) -> str:
"""Safely decode email header values (handles encoded UTF-8, ISO-8859, etc.)."""
if value is None:
return ""
from email.header import decode_header
parts = decode_header(value)
result = []
for part, enc in parts:
if isinstance(part, bytes):
result.append(part.decode(enc or "utf-8", errors="replace"))
else:
result.append(str(part))
return "".join(result)
def sanitize_imap_query(query: str) -> str:
"""Remove characters that would break IMAP search syntax.
Only allows alphanumeric characters, spaces, and basic safe punctuation.
IMAP search interprets quotes, parens, asterisks, and braces as operators —
all stripped to prevent injection (e.g. ``foo*) ALL`` leaking entire inbox).
"""
return re.sub(r"[^a-zA-Z0-9\s._@-]", "", query).strip()
def sanitize_external_text(text: str) -> str:
"""Sanitize external content (email, calendar, notes) before injection into LLM prompt.
Strips HTML tags, script injection vectors, and control characters.
This prevents XSS via malicious email subjects, calendar descriptions,
or note content that could be rendered as HTML by the frontend.
"""
if not text:
return ""
# Strip HTML tags
text = re.sub(r"<[^>]+>", "", text)
# Strip common script injection patterns
text = re.sub(r"javascript:", "", text, flags=re.IGNORECASE)
text = re.sub(r"on\w+\s*=", "", text, flags=re.IGNORECASE)
# Strip control characters (keep newlines and tabs)
text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text)
return text.strip()