Fix critical security vulnerability in Stripe webhook handler

SECURITY_AUDIT.md

@@ -0,0 +1,25 @@
+# Security Audit Findings
+
+## Critical Issues
+
+### 1. Stripe Webhook Signature Bypass (FIXED)
+
+**Severity:** Critical
+**Description:** The Stripe webhook handler `_try_construct_event` was configured to accept payloads signed with the Test Secret even when the application was running in Production mode. This would allow an attacker knowing the Test Secret to forge events (e.g., subscription creation) against the Production environment.
+**Fix:** The logic in `src/api/routes/payments/webhooks.py` was updated to strictly enforce secret usage based on the environment (`DEV_ENV`). Production only uses `STRIPE_WEBHOOK_SECRET`, and non-production environments use `STRIPE_TEST_WEBHOOK_SECRET`.
+
+## Low/Medium Issues
+
+### 1. `alert_admin` DB Session Handling
+**Severity:** Low
+**Description:** The `alert_admin` tool manually manages a database session obtained from `get_db_session` generator. While it correctly closes it, the generator remains suspended until garbage collection.
+**Recommendation:** Ensure `get_db_session` is used as a context manager if possible, or refactor tool to use `scoped_session` similar to `agent_stream_endpoint`.
+
+### 2. Missing Rate Limiting on Webhooks
+**Severity:** Low
+**Description:** While signature verification prevents unauthorized payloads, the webhook endpoint is still exposed to DoS attacks.
+**Recommendation:** Implement rate limiting (e.g., via Nginx or application middleware) for the webhook endpoint.
+
+### 3. Usage Reset Logic
+**Severity:** Low
+**Description:** The `handle_usage_reset_webhook` trusts `invoice.payment_succeeded` events to reset usage. Ensure idempotency keys are handled to prevent double-processing if Stripe retries webhooks (though resetting to 0 is idempotent-ish).

alembic/versions/33ae457b2ddf_add_referral_columns.py

@@ -5,6 +5,7 @@
 Create Date: 2025-12-26 10:37:46.325765
 
 """
+
 from typing import Sequence, Union
 
 from alembic import op
@@ -13,26 +14,28 @@
 from sqlalchemy.ext.declarative import declarative_base
 
 # revision identifiers, used by Alembic.
-revision: str = '33ae457b2ddf'
-down_revision: Union[str, Sequence[str], None] = '8b9c2e1f4c1c'
+revision: str = "33ae457b2ddf"
+down_revision: Union[str, Sequence[str], None] = "8b9c2e1f4c1c"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 
 # Define a minimal model for data migration
 Base = declarative_base()
 
+
 class Profile(Base):
-    __tablename__ = 'profiles'
+    __tablename__ = "profiles"
     user_id = sa.Column(sa.UUID, primary_key=True)
     referral_code = sa.Column(sa.String)
     referral_count = sa.Column(sa.Integer)
 
+
 def upgrade() -> None:
     """Upgrade schema."""
     # 1. Add columns as nullable first
-    op.add_column('profiles', sa.Column('referral_code', sa.String(), nullable=True))
-    op.add_column('profiles', sa.Column('referrer_id', sa.UUID(), nullable=True))
-    op.add_column('profiles', sa.Column('referral_count', sa.Integer(), nullable=True))
+    op.add_column("profiles", sa.Column("referral_code", sa.String(), nullable=True))
+    op.add_column("profiles", sa.Column("referrer_id", sa.UUID(), nullable=True))
+    op.add_column("profiles", sa.Column("referral_count", sa.Integer(), nullable=True))
 
     # 2. Backfill existing rows with 0 count
     bind = op.get_bind()
@@ -45,10 +48,12 @@ def upgrade() -> None:
     # 3. Alter columns
     # referral_code stays nullable=True
     # referral_count becomes nullable=False
-    op.alter_column('profiles', 'referral_count', nullable=False)
+    op.alter_column("profiles", "referral_count", nullable=False)
 
     # 4. Create unique constraint and index
-    op.create_unique_constraint("uq_profiles_referral_code", "profiles", ["referral_code"])
+    op.create_unique_constraint(
+        "uq_profiles_referral_code", "profiles", ["referral_code"]
+    )
     op.create_index("ix_profiles_referral_code", "profiles", ["referral_code"])
 
     # Add foreign key for referrer_id
@@ -62,6 +67,6 @@ def downgrade() -> None:
     op.drop_constraint("fk_profiles_referrer_id", "profiles", type_="foreignkey")
     op.drop_index("ix_profiles_referral_code", table_name="profiles")
     op.drop_constraint("uq_profiles_referral_code", "profiles", type_="unique")
-    op.drop_column('profiles', 'referral_count')
-    op.drop_column('profiles', 'referrer_id')
-    op.drop_column('profiles', 'referral_code')
+    op.drop_column("profiles", "referral_count")
+    op.drop_column("profiles", "referrer_id")
+    op.drop_column("profiles", "referral_code")

pyproject.toml

@@ -71,7 +71,8 @@ exclude = [
     "src/api/auth/",
     "src/utils/integration/",
     "src/stripe/",
-    "scripts/"
+    "scripts/",
+    "tests/security/test_stripe_webhook_security.py"
 ]
 ignore_names = [
     "apply_referral",

src/api/routes/payments/webhooks.py

@@ -28,20 +28,23 @@ def _try_construct_event(payload: bytes, sig_header: str | None) -> dict:
     """
 
     def _secrets() -> Iterable[str]:
-        primary = (
-            global_config.STRIPE_WEBHOOK_SECRET
-            if global_config.DEV_ENV == "prod"
-            else global_config.STRIPE_TEST_WEBHOOK_SECRET
-        )
-        secondary = (
-            global_config.STRIPE_TEST_WEBHOOK_SECRET
-            if global_config.DEV_ENV == "prod"
-            else global_config.STRIPE_WEBHOOK_SECRET
-        )
-        if primary:
-            yield primary
-        if secondary and secondary != primary:
-            yield secondary
+        # STRICTLY enforce secret usage based on environment
+        # Prod env MUST use Prod secret.
+        # Non-prod env MUST use Test secret.
+        # No fallbacks across environments.
+
+        if global_config.DEV_ENV == "prod":
+            if global_config.STRIPE_WEBHOOK_SECRET:
+                yield global_config.STRIPE_WEBHOOK_SECRET
+            else:
+                logger.error("STRIPE_WEBHOOK_SECRET not configured in PROD")
+        else:
+            if global_config.STRIPE_TEST_WEBHOOK_SECRET:
+                yield global_config.STRIPE_TEST_WEBHOOK_SECRET
+            else:
+                logger.warning(
+                    f"STRIPE_TEST_WEBHOOK_SECRET not configured in {global_config.DEV_ENV}"
+                )
 
     if not sig_header:
         raise HTTPException(status_code=400, detail="Missing stripe-signature header")

src/db/utils/users.py

@@ -4,13 +4,14 @@
 import uuid
 from loguru import logger
 
+
 def ensure_profile_exists(
     db: Session,
     user_uuid: uuid.UUID,
     email: str | None = None,
     username: str | None = None,
     avatar_url: str | None = None,
-    is_approved: bool = False
+    is_approved: bool = False,
 ) -> Profiles:
     """
     Ensure a profile exists for the given user UUID.
@@ -27,7 +28,7 @@ def ensure_profile_exists(
                 email=email,
                 username=username,
                 avatar_url=avatar_url,
-                is_approved=is_approved
+                is_approved=is_approved,
             )
             db.add(profile)
         # No need for explicit commit/refresh as db_transaction handles commit,

tests/security/test_stripe_webhook_security.py

@@ -0,0 +1,77 @@
+import os
+import sys
+import pytest
+import stripe
+import time
+from unittest.mock import patch
+from fastapi import HTTPException
+
+# We need to add the root to sys.path to import src
+sys.path.append(os.getcwd())
+
+# Import the module under test
+from src.api.routes.payments.webhooks import _try_construct_event
+
+
+def test_stripe_webhook_vuln_repro():
+    """
+    Verify that _try_construct_event REJECTS a payload signed with the
+    TEST secret when configured for PROD.
+    """
+
+    # Setup test secrets
+    PROD_SECRET = "whsec_prod_secret_12345"
+    TEST_SECRET = "whsec_test_secret_67890"
+
+    # Mock global_config to simulate PROD environment
+    with patch("src.api.routes.payments.webhooks.global_config") as mock_config:
+        mock_config.configure_mock(
+            DEV_ENV="prod",
+            STRIPE_WEBHOOK_SECRET=PROD_SECRET,
+            STRIPE_TEST_WEBHOOK_SECRET=TEST_SECRET,
+        )
+
+        # Create a payload
+        payload = b'{"id": "evt_test", "object": "event"}'
+        timestamp = int(time.time())
+        header = f"t={timestamp},v1=dummy_signature"
+
+        # The function under test
+        try:
+            with patch("stripe.Webhook.construct_event") as mock_construct:
+                # This mock simulates:
+                # - Fails if secret is PROD_SECRET (simulating bad signature because payload signed with TEST)
+                # - Succeeds if secret is TEST_SECRET
+                def side_effect(payload, sig_header, secret):
+                    if secret == TEST_SECRET:
+                        return {"type": "test_event"}
+                    raise stripe.error.SignatureVerificationError(
+                        "Invalid signature", sig_header=sig_header, http_body=payload
+                    )
+
+                mock_construct.side_effect = side_effect
+
+                _try_construct_event(payload, header)
+
+                # If we get here, it means it accepted it
+                pytest.fail(
+                    "Vulnerability STILL PRESENT: _try_construct_event accepted TEST secret in PROD mode!"
+                )
+
+        except HTTPException as e:
+            # If it raises HTTPException(400), it means it rejected it (secure behavior)
+            assert e.status_code == 400
+            assert e.detail == "Invalid signature"
+            print(
+                "\n[SUCCESS] _try_construct_event rejected TEST secret in PROD mode."
+            )
+        except Exception as e:
+            pytest.fail(f"Unexpected error: {e}")
+
+
+if __name__ == "__main__":
+    # Manually run the test function if executed as script
+    try:
+        test_stripe_webhook_vuln_repro()
+    except Exception as e:
+        print(e)