infrasynth-backend-kit/infrasynth/shared/exceptions.py
jcv-dev 551b42eab5 feat: production-hardening pass across the kit
Close the gaps between the documented contract (API-STANDARD, TENANCY,
ENTITLEMENTS) and the implementation, and remove committed build artifacts.

Security:
- verify + process inbound webhooks (HMAC/handler verify, size limit,
  timestamp tolerance, idempotency via InboundEvent.external_id)
- real 2FA login flow (pre-auth challenge; tokens only after verify/recovery)
- wire HybridPermission into security/audit views; add API-key rotate and
  users/<id>/permissions|roles endpoints
- tenant-scoped throttling on by default; webhook replay protection
- verify MercadoPago webhook signatures
- login brute-force guard, configurable password policy, real ALTCHA PoW

Correctness:
- apply verified billing webhooks idempotently (subscription/entitlement/
  invoice/PaymentTransaction); scheduled payment lifecycle jobs
- capture audit update diffs automatically; add audit retention purge
- working notification retries, per-channel rate limits, log retention
- pluggable virus scanner, upload-size limit, pipeline toggle
- feature rollout %/environment targeting; settings-driven registrations
- workflow guards (instance cap, route depth, self-assignment, clone on re-entry)
- wire every previously-dead INFRASYNTH_* setting; drop truly dead ones

Delivery:
- README + CHANGELOG; CI format check + coverage gate
- keep test media out of the tree; untrack .coverage, __pycache__,
  egg-info, docs/ and invoice artifacts
2026-09-24 10:41:21 -05:00

147 lines
4.5 KiB
Python

"""Zero-Django exception hierarchy for the InfraSynth kit.
These exceptions are raised by services and mapped to the namespaced error
envelope (``API-STANDARD.md`` §5) by ``infrasynth.api.exceptions``. They carry a
stable machine-readable ``code``, an HTTP ``status``, and a list of ``details``
so no caller has to parse a message.
This module must never import Django (``infrasynth.shared`` is zero-Django).
"""
from __future__ import annotations
from typing import Any
__all__ = [
"AppError",
"AuthError",
"ConflictError",
"EntitlementError",
"NotFoundError",
"RateLimitError",
"ServerError",
"ValidationAppError",
"ENTITLEMENT_APP_NOT_OWNED",
"ENTITLEMENT_EXPIRED",
"ENTITLEMENT_LIMIT_REACHED",
"ENTITLEMENT_PLAN_UPGRADE_REQUIRED",
"ENTITLEMENT_REVOKED",
"ENTITLEMENT_SUBSCRIPTION_PAST_DUE",
"ENTITLEMENT_TENANT_SUSPENDED",
]
# --- Entitlement error codes (ENTITLEMENTS.md §7) ---------------------------
ENTITLEMENT_APP_NOT_OWNED = "ENTITLEMENT_APP_NOT_OWNED"
ENTITLEMENT_PLAN_UPGRADE_REQUIRED = "ENTITLEMENT_PLAN_UPGRADE_REQUIRED"
ENTITLEMENT_LIMIT_REACHED = "ENTITLEMENT_LIMIT_REACHED"
ENTITLEMENT_SUBSCRIPTION_PAST_DUE = "ENTITLEMENT_SUBSCRIPTION_PAST_DUE"
ENTITLEMENT_EXPIRED = "ENTITLEMENT_EXPIRED"
ENTITLEMENT_TENANT_SUSPENDED = "ENTITLEMENT_TENANT_SUSPENDED"
ENTITLEMENT_REVOKED = "ENTITLEMENT_REVOKED"
class AppError(Exception):
"""Base class for every expected, mappable application error."""
default_code: str = "SERVER_ERROR"
default_message: str = "An unexpected error occurred."
default_status: int = 500
def __init__(
self,
message: str | None = None,
*,
code: str | None = None,
status: int | None = None,
details: list[dict[str, Any]] | None = None,
) -> None:
self.code = code or self.default_code
self.message = message or self.default_message
self.status = status or self.default_status
self.details: list[dict[str, Any]] = list(details) if details else []
super().__init__(self.message)
def to_dict(self) -> dict[str, Any]:
"""Serializable error body (the ``error`` member of the envelope)."""
return {"code": self.code, "message": self.message, "details": self.details}
class AuthError(AppError):
"""Authentication or authorization failure (``AUTH_*``)."""
default_code = "AUTH_ERROR"
default_message = "Authentication failed."
default_status = 401
class EntitlementError(AppError):
"""Commercial access failure (``ENTITLEMENT_*``).
Extra keyword arguments (``app``, ``plan``, ``feature``, ``limit``,
``current``, ``max``) are folded into a single ``details`` entry, matching
``ENTITLEMENTS.md`` §7::
raise EntitlementError(
code=ENTITLEMENT_PLAN_UPGRADE_REQUIRED, app="helpdesk", feature="tickets"
)
"""
default_code = ENTITLEMENT_APP_NOT_OWNED
default_message = "This workspace does not have access to the requested app."
default_status = 402
def __init__(
self,
message: str | None = None,
*,
code: str | None = None,
status: int | None = None,
details: list[dict[str, Any]] | None = None,
**context: Any,
) -> None:
merged = list(details) if details else []
extra = {key: value for key, value in context.items() if value is not None}
if extra:
merged.append(extra)
super().__init__(message, code=code, status=status, details=merged)
class ValidationAppError(AppError):
"""Request payload failure (``VALIDATION_*``)."""
default_code = "VALIDATION_ERROR"
default_message = "The request payload is invalid."
default_status = 400
class NotFoundError(AppError):
"""Resource not found (also used for cross-tenant access)."""
default_code = "NOT_FOUND"
default_message = "The requested resource was not found."
default_status = 404
class ConflictError(AppError):
"""State or idempotency conflict (``CONFLICT_*``)."""
default_code = "CONFLICT_ERROR"
default_message = "The request conflicts with the current state."
default_status = 409
class RateLimitError(AppError):
"""Throttling failure (``RATE_LIMIT_*``)."""
default_code = "RATE_LIMIT_EXCEEDED"
default_message = "Too many requests."
default_status = 429
class ServerError(AppError):
"""Unhandled/internal error (``SERVER_*``). Never leaks a stack trace."""
default_code = "SERVER_ERROR"
default_message = "An unexpected error occurred."
default_status = 500