Self-custody payouts + audit hardening (re-land of #26 remainder)

backend/.env.example

@@ -3,3 +3,9 @@ SECRET_KEY=change-me-to-a-long-random-string
 ALGORITHM=HS256
 ACCESS_TOKEN_EXPIRE_MINUTES=1440
 FRONTEND_URL=http://localhost:3000
+
+# --- Optional ---
+# Enables AI normalization of free-text "Other" strengths (deterministic fallback when unset).
+# ANTHROPIC_API_KEY=sk-ant-...
+# Hard ceiling (sats) on a single team payout; a safety net against a fat-finger amount.
+# PAYOUT_MAX_SATS=5000000

backend/app/api/v1/payouts.py

@@ -5,11 +5,14 @@
 from sqlalchemy.orm import Session
 
 from app.api.deps import get_current_user, get_db
+from app.core.config import settings
 from app.models.allocation import Allocation
 from app.models.payout import Payout, PayoutItem
 from app.models.team import Team
 from app.models.user import User
-from app.schemas.payout import PayoutCreate, PayoutRetry, PayoutOut
+from app.schemas.payout import (
+    PayoutCreate, PayoutOut, PayoutItemResult, PayoutItemFailed,
+)
 from app.services.event_service import assert_allocation_organizer
 from app.services import payout_service
 
@@ -38,6 +41,14 @@ def create_payout(
     if not team:
         raise HTTPException(status_code=404, detail="Team not found in this allocation")
 
+    # Spend ceiling: reject an implausibly large amount before touching a wallet.
+    if req.total_sats > settings.PAYOUT_MAX_SATS:
+        raise HTTPException(
+            status_code=422,
+            detail=f"total_sats {req.total_sats} exceeds the payout ceiling "
+                   f"of {settings.PAYOUT_MAX_SATS} sats",
+        )
+
     # Idempotency: refuse a second payout for a team that already has one, so a
     # double-click or a client retry after a timeout can never pay winners twice.
     if db.query(Payout).filter(
@@ -67,20 +78,49 @@ def create_payout(
             status_code=status.HTTP_409_CONFLICT,
             detail="This team has already been paid; retry the existing payout instead.",
         )
-    payout = payout_service.execute_payout(db, payout, splits, req.nwc)
+    # Self-custody: create pending items; the browser pays and reports each result.
+    # The server never receives a spend credential.
+    payout = payout_service.create_pending(db, payout, splits)
     return _payout_out(db, payout)
 
 
-@router.post("/payouts/{payout_id}/retry", response_model=PayoutOut)
-def retry_payout(
+def _get_item(db: Session, payout_id: UUID, item_id: UUID, user_id: UUID) -> tuple[Payout, PayoutItem]:
+    """Load a payout + one of its items, asserting the caller is the organizer."""
+    payout = db.query(Payout).filter(Payout.id == payout_id).first()
+    if not payout:
+        raise HTTPException(status_code=404, detail="Payout not found")
+    assert_allocation_organizer(db, payout.allocation_id, user_id)
+    item = db.query(PayoutItem).filter(
+        PayoutItem.id == item_id, PayoutItem.payout_id == payout_id
+    ).first()
+    if not item:
+        raise HTTPException(status_code=404, detail="Payout item not found")
+    return payout, item
+
+
+@router.post("/payouts/{payout_id}/items/{item_id}/result", response_model=PayoutOut)
+def report_item_result(
     payout_id: UUID,
-    req: PayoutRetry,
+    item_id: UUID,
+    req: PayoutItemResult,
     db: Session = Depends(get_db),
     current_user: User = Depends(get_current_user),
 ):
-    payout = db.query(Payout).filter(Payout.id == payout_id).first()
-    if not payout:
-        raise HTTPException(status_code=404, detail="Payout not found")
-    assert_allocation_organizer(db, payout.allocation_id, current_user.id)
-    payout = payout_service.retry_failed(db, payout, req.nwc)
+    """Self-custody: the browser reports a completed send; the server verifies the preimage."""
+    payout, item = _get_item(db, payout_id, item_id, current_user.id)
+    payout = payout_service.record_item_result(db, payout, item, req.bolt11, req.preimage)
+    return _payout_out(db, payout)
+
+
+@router.post("/payouts/{payout_id}/items/{item_id}/failed", response_model=PayoutOut)
+def report_item_failed(
+    payout_id: UUID,
+    item_id: UUID,
+    req: PayoutItemFailed,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Self-custody: the browser reports a send that produced no preimage."""
+    payout, item = _get_item(db, payout_id, item_id, current_user.id)
+    payout = payout_service.record_item_failed(db, payout, item, req.error)
     return _payout_out(db, payout)

backend/app/core/config.py

@@ -20,6 +20,9 @@ def cors_origins(self) -> list[str]:
     # request URL FastAPI sees (http, internal host) differs from the signed URL.
     # When unset, the live request URL is used (correct for local/dev).
     PUBLIC_API_URL: str | None = None
+    # Hard ceiling on a single team payout (sats). A safety net against a
+    # fat-finger (extra zeros) — the request is refused before anything is sent.
+    PAYOUT_MAX_SATS: int = 5_000_000
     # Optional: enables AI normalization of free-text "Other" strengths.
     # When unset, allocation falls back to a deterministic slug per Other entry.
     ANTHROPIC_API_KEY: str | None = None

backend/app/schemas/payout.py

@@ -6,14 +6,29 @@
 class PayoutCreate(BaseModel):
     team_id: UUID
     total_sats: int = Field(gt=0)
-    nwc: str = Field(min_length=1)
+    # Self-custody: the server never receives a spend credential. It creates pending
+    # items and the browser performs the NIP-47 send, reporting each result back.
     # Optional per-member address overrides: {str(participant_id): "name@domain"}.
     # Lets the organizer fill/correct a missing address in the payout modal.
     addresses: Optional[dict[str, str]] = None
 
 
+class PayoutItemResult(BaseModel):
+    """A browser-performed send to report for one item."""
+    bolt11: str = Field(min_length=1)
+    preimage: str = Field(min_length=1)
+
+
+class PayoutItemFailed(BaseModel):
+    error: str = Field(min_length=1)
+
+
 class PayoutRetry(BaseModel):
     nwc: str = Field(min_length=1)
+    # Optional per-member address corrections applied to failed items before retry:
+    # {str(participant_id): "name@domain"}. Lets an organizer recover a payout that
+    # failed because an address was wrong, without creating a new payout.
+    addresses: Optional[dict[str, str]] = None
 
 
 class PayoutItemOut(BaseModel):

backend/app/services/categorization_service.py

@@ -17,12 +17,32 @@
 
 logger = logging.getLogger(__name__)
 
+# Cap participants per AI call so a batch's tool-use output cannot exceed
+# _MAX_TOKENS and truncate — a truncated tool call would fail to parse and
+# silently drop the whole batch to the deterministic fallback.
+_BATCH_SIZE = 25
+_MAX_TOKENS = 4096
+
+_SYSTEM_INSTRUCTIONS = (
+    "You normalize each participant's free-text strength into exactly one of a "
+    "fixed set of categories so a downstream deterministic engine can compare "
+    "them. You never assign teams. Map each participant to the single best-fit "
+    "category. If a participant's text is too vague, empty, or unrelated to every "
+    "category, OMIT them from the assignments instead of guessing — an omitted "
+    "participant is handled by a deterministic fallback and an organizer can set "
+    "their category by hand."
+)
+
 
 def _slug(text: str) -> str:
     s = re.sub(r"[^a-z0-9]+", "_", text.strip().lower()).strip("_")
     return s or "other"
 
 
+def _category_catalog() -> str:
+    return ", ".join(f"{v} ({STRENGTH_LABELS[v]})" for v in CONCRETE_STRENGTHS)
+
+
 def _pending(db: Session, event_id: UUID) -> list[Participant]:
     return (
         db.query(Participant)
@@ -36,12 +56,12 @@ def _pending(db: Session, event_id: UUID) -> list[Participant]:
     )
 
 
-def _classify(event: Event, participants: list[Participant]) -> dict[str, str]:
-    """Call Claude; return {participant_id: concrete_category}. Raises on failure."""
-    import anthropic
+def _build_request(event: Event, participants: list[Participant]) -> dict:
+    """Build the Messages API kwargs. Pure (no network) so it is unit-testable.
 
-    client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
-    categories = ", ".join(f"{v} ({STRENGTH_LABELS[v]})" for v in CONCRETE_STRENGTHS)
+    The static instructions + category catalog go in a cacheable system block;
+    the per-event context and participant list go in the user message.
+    """
     people = "\n".join(f"- id={p.id}: {p.strength_other}" for p in participants)
     tool = {
         "name": "assign_categories",
@@ -64,29 +84,47 @@ def _classify(event: Event, participants: list[Participant]) -> dict[str, str]:
             "required": ["assignments"],
         },
     }
-    msg = client.messages.create(
-        model=settings.CATEGORIZATION_MODEL,
-        max_tokens=1024,
-        tools=[tool],
-        tool_choice={"type": "tool", "name": "assign_categories"},
-        messages=[{
+    return {
+        "model": settings.CATEGORIZATION_MODEL,
+        "max_tokens": _MAX_TOKENS,
+        "system": [{
+            "type": "text",
+            "text": f"{_SYSTEM_INSTRUCTIONS}\n\nAvailable categories: {_category_catalog()}",
+            "cache_control": {"type": "ephemeral"},
+        }],
+        "tools": [tool],
+        "tool_choice": {"type": "tool", "name": "assign_categories"},
+        "messages": [{
             "role": "user",
             "content": (
                 f"Event: {event.title}\nDescription: {event.description or '(none)'}\n\n"
-                f"Available categories: {categories}\n\n"
-                f"Map each participant's free-text strength to the best category.\n{people}"
+                f"Map each participant's free-text strength to the best category, "
+                f"omitting any that are too unclear to place.\n{people}"
             ),
         }],
-    )
+    }
+
+
+def _parse_assignments(content_blocks) -> dict[str, str]:
+    """Extract {id: category} from tool_use blocks, dropping out-of-taxonomy values."""
     out: dict[str, str] = {}
-    for block in msg.content:
+    for block in content_blocks:
         if getattr(block, "type", None) == "tool_use":
-            for a in block.input["assignments"]:
-                if a["category"] in CONCRETE_STRENGTHS:
+            for a in block.input.get("assignments", []):
+                if a.get("category") in CONCRETE_STRENGTHS and a.get("id"):
                     out[a["id"]] = a["category"]
     return out
 
 
+def _classify(event: Event, participants: list[Participant]) -> dict[str, str]:
+    """Call Claude for one batch; return {participant_id: concrete_category}. Raises on failure."""
+    import anthropic
+
+    client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
+    msg = client.messages.create(**_build_request(event, participants))
+    return _parse_assignments(msg.content)
+
+
 def normalize_pending(db: Session, event_id: UUID) -> dict[str, int]:
     """Fill normalized_strength for un-normalized Other entries. Never raises.
 
@@ -101,11 +139,14 @@ def normalize_pending(db: Session, event_id: UUID) -> dict[str, int]:
 
     mapping: dict[str, str] = {}
     if settings.ANTHROPIC_API_KEY:
-        try:
-            mapping = _classify(event, pending)
-        except Exception as exc:  # noqa: BLE001 — AI is best-effort
-            logger.warning("Categorization AI failed, using fallback: %s", exc)
-            mapping = {}
+        # Batch so output can't truncate, and so one failing batch only forces its
+        # own members to the fallback rather than dumping everyone.
+        for i in range(0, len(pending), _BATCH_SIZE):
+            batch = pending[i:i + _BATCH_SIZE]
+            try:
+                mapping.update(_classify(event, batch))
+            except Exception as exc:  # noqa: BLE001 — AI is best-effort, per batch
+                logger.warning("Categorization AI batch failed, using fallback: %s", exc)
 
     for p in pending:
         ai_cat = mapping.get(str(p.id))

backend/app/services/nostr_service.py

@@ -22,26 +22,44 @@
 _BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
 
 
+def _bech32_polymod(values: list[int]) -> int:
+    """BIP-173 checksum polymod over hrp-expansion + data (incl. checksum)."""
+    generator = (0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3)
+    chk = 1
+    for value in values:
+        top = chk >> 25
+        chk = ((chk & 0x1FFFFFF) << 5) ^ value
+        for i in range(5):
+            chk ^= generator[i] if ((top >> i) & 1) else 0
+    return chk
+
+
+def _bech32_hrp_expand(hrp: str) -> list[int]:
+    return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp]
+
+
 def bech32_decode(bech: str) -> tuple[str, bytes]:
     """Decode a bech32 `npub`/`nsec` to (hrp, 32-byte key).
 
-    Minimal decoder: splits on the last '1', drops the 6-char checksum, and
-    converts the 5-bit data groups to 8-bit bytes. Sufficient for npub/nsec.
+    Validates the BIP-173 checksum so a single mistyped character is rejected
+    rather than silently decoded to a different (wrong) key. Sufficient for
+    npub/nsec (bech32, not bech32m).
     """
     bech = bech.strip().lower()
     pos = bech.rfind("1")
-    if pos < 1:
+    if pos < 1 or pos + 7 > len(bech):  # need a non-empty hrp + 6-char checksum
         raise ValueError("invalid bech32 string")
     hrp = bech[:pos]
     try:
         data = [_BECH32_CHARSET.index(c) for c in bech[pos + 1:]]
     except ValueError as exc:
         raise ValueError("invalid bech32 character") from exc
-    data = data[:-6]  # drop checksum
+    if _bech32_polymod(_bech32_hrp_expand(hrp) + data) != 1:
+        raise ValueError("invalid bech32 checksum")
     acc = 0
     bits = 0
     out = bytearray()
-    for value in data:
+    for value in data[:-6]:  # drop the now-verified 6-char checksum
         acc = (acc << 5) | value
         bits += 5
         if bits >= 8: