feat(conduit): Add conduit auth functions (#101729)

Adds backend helpers for generating Conduit authentication credentials.
[Conduit](https://github.com/getsentry/conduit) is Sentry's real-time
delivery platform that enables scalable streaming from backend services
to frontend clients.

This PR provides the foundational utilities needed for API endpoints to
generate Conduit credentials (JWT tokens and channel IDs) that the
[conduit-client](https://github.com/getsentry/conduit-client) will use
to establish streaming connections.

**Changes**

- New utilities in `src/sentry/conduit/auth.py`
  - `generate_channel_id()`: UUID4 generation for stream identifiers
- `generate_conduit_token()`: RS256 signed JWT with org_id, channel_id,
and 10 minute expiration
- `get_conduit_credentials()`: Returns `{url, token, channel_id}` dict
for the client

- Configuration `src/sentry/conf/server.py`
  - `CONDUIT_PRIVATE_KEY` (required): RSA private key for JWT signing
  - `CONDUIT_GATEWAY_URL`: Gateway base URL
  - `CONDUIT_JWT_ISSUER` / `CONDUIT_JWT_AUDIENCE`: JWT claims

Note: Requires `CONDUIT_PRIVATE_KEY` environment variable to be set
before use.

Author: IanWoodard

src/sentry/conduit/__init__.py

@@ -0,0 +1,85 @@
+import time
+import uuid
+from typing import NamedTuple
+
+from django.conf import settings
+
+from sentry.utils import jwt, metrics
+
+TOKEN_TTL_SEC = 600  # 10 minutes
+
+
+class ConduitCredentials(NamedTuple):
+    token: str
+    channel_id: str
+    url: str
+
+
+def generate_channel_id() -> str:
+    """Generate a unique channel ID for a Conduit stream."""
+    return str(uuid.uuid4())
+
+
+def generate_conduit_token(
+    org_id: int,
+    channel_id: str,
+    issuer: str | None = None,
+    audience: str | None = None,
+    conduit_private_key: str | None = None,
+) -> str:
+    """
+    Generate a JWT token for Conduit authentication.
+
+    Optional parameters default to settings values if not provided.
+
+    Returns:
+        JWT token string
+    """
+    if issuer is None:
+        issuer = settings.CONDUIT_JWT_ISSUER
+    if audience is None:
+        audience = settings.CONDUIT_JWT_AUDIENCE
+    if conduit_private_key is None:
+        conduit_private_key = settings.CONDUIT_PRIVATE_KEY
+        if conduit_private_key is None:
+            raise ValueError("CONDUIT_PRIVATE_KEY not configured")
+
+    now = int(time.time())
+    exp = now + TOKEN_TTL_SEC
+    payload = {
+        "org_id": org_id,
+        "channel_id": channel_id,
+        "iat": now,
+        # Conduit only validates tokens on initial connection, not for stream lifetime
+        "exp": exp,
+        "iss": issuer,
+        "aud": audience,
+    }
+    return jwt.encode(payload, conduit_private_key, algorithm="RS256")
+
+
+def get_conduit_credentials(
+    org_id: int,
+    gateway_url: str | None = None,
+) -> ConduitCredentials:
+    """
+    Generate all credentials needed to connect to Conduit.
+
+    Returns:
+        ConduitCredentials containing authentication details
+    """
+    if gateway_url is None:
+        gateway_url = settings.CONDUIT_GATEWAY_URL
+    channel_id = generate_channel_id()
+    token = generate_conduit_token(org_id, channel_id)
+
+    metrics.incr(
+        "conduit.credentials.generated",
+        sample_rate=1.0,
+    )
+
+    return ConduitCredentials(
+        token=token,
+        channel_id=channel_id,
+        url=f"{gateway_url}/events/{org_id}",
+    )

src/sentry/conduit/auth.py

@@ -3177,3 +3177,8 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
     # the region API URL template is set to the ngrok host.
     SENTRY_OPTIONS["system.region-api-url-template"] = f"https://{{region}}.{ngrok_host}"
     SENTRY_FEATURES["system:multi-region"] = True
+
+CONDUIT_PRIVATE_KEY: str | None = os.getenv("CONDUIT_PRIVATE_KEY")
+CONDUIT_GATEWAY_URL: str = os.getenv("CONDUIT_GATEWAY_URL", "https://conduit.sentry.io")
+CONDUIT_JWT_ISSUER: str = os.getenv("CONDUIT_JWT_ISSUER", "sentry")
+CONDUIT_JWT_AUDIENCE: str = os.getenv("CONDUIT_JWT_AUDIENCE", "conduit")

src/sentry/conf/server.py

@@ -0,0 +1,184 @@
+import time
+from unittest.mock import patch
+
+import jwt as pyjwt
+import pytest
+
+from sentry.conduit.auth import generate_channel_id, generate_conduit_token, get_conduit_credentials
+from tests.sentry.utils.test_jwt import RS256_KEY, RS256_PUB_KEY
+
+
+def test_generate_channel_id_is_valid_uuid():
+    """Should generate a valid uuid."""
+    channel_id = generate_channel_id()
+
+    # UUID format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+    assert isinstance(channel_id, str)
+    assert len(channel_id) == 36  # Length of UUID
+    assert channel_id.count("-") == 4
+
+
+def test_generate_channel_id_is_unique():
+    """Should generate unique channel_ids."""
+    assert generate_channel_id() != generate_channel_id()
+
+
+def test_generate_conduit_token_is_valid_jwt():
+    """Should generate a valid JWT token with RS256."""
+    org_id = 123
+    channel_id = "ad342057-d66b-4ed4-ab01-3415dd2cb1ce"
+
+    token = generate_conduit_token(
+        org_id,
+        channel_id,
+        issuer="sentry",
+        audience="conduit",
+        conduit_private_key=RS256_KEY,
+    )
+
+    assert isinstance(token, str)
+    assert token.count(".") == 2
+
+    claims = pyjwt.decode(token, RS256_PUB_KEY, algorithms=["RS256"], audience="conduit")
+
+    assert claims["channel_id"] == channel_id
+    assert claims["org_id"] == org_id
+    assert claims["iss"] == "sentry"
+    assert claims["aud"] == "conduit"
+    assert "iat" in claims
+    assert "exp" in claims
+
+
+def test_generate_conduit_token_has_expiration():
+    """Token should expire in 10 minutes."""
+    org_id = 123
+    channel_id = "ad342057-d66b-4ed4-ab01-3415dd2cb1ce"
+
+    before_time = int(time.time())
+    token = generate_conduit_token(
+        org_id,
+        channel_id,
+        issuer="sentry",
+        audience="conduit",
+        conduit_private_key=RS256_KEY,
+    )
+    after_time = int(time.time())
+
+    claims = pyjwt.decode(
+        token,
+        RS256_PUB_KEY,
+        algorithms=["RS256"],
+        audience="conduit",
+        options={"verify_exp": False},
+    )
+
+    exp_time = claims["exp"]
+    iat_time = claims["iat"]
+
+    assert iat_time >= before_time
+    assert iat_time <= after_time
+    assert exp_time == iat_time + 600
+
+
+def test_generate_conduit_token_uses_settings():
+    """Should use settings when parameters are not provided."""
+    org_id = 123
+    channel_id = "ad342057-d66b-4ed4-ab01-3415dd2cb1ce"
+
+    with patch("sentry.conduit.auth.settings") as mock_settings:
+        mock_settings.CONDUIT_PRIVATE_KEY = RS256_KEY
+        mock_settings.CONDUIT_JWT_ISSUER = "test-issuer"
+        mock_settings.CONDUIT_JWT_AUDIENCE = "test-audience"
+
+        token = generate_conduit_token(
+            org_id,
+            channel_id,
+        )
+
+        claims = pyjwt.decode(
+            token,
+            RS256_PUB_KEY,
+            algorithms=["RS256"],
+            audience="test-audience",
+            options={"verify_exp": False},
+        )
+
+        assert claims["iss"] == "test-issuer"
+        assert claims["aud"] == "test-audience"
+
+
+def test_generate_conduit_token_raises_when_missing():
+    """Should raise an error if the private key is not configured."""
+    org_id = 123
+    channel_id = "ad342057-d66b-4ed4-ab01-3415dd2cb1ce"
+    with pytest.raises(ValueError, match="CONDUIT_PRIVATE_KEY not configured"):
+        generate_conduit_token(
+            org_id,
+            channel_id,
+        )
+
+
+def test_get_conduit_credentials_returns_all_credentials():
+    """Should return a url, token, and channel_id."""
+    gateway_url = "https://conduit.example.com"
+    with patch("sentry.conduit.auth.settings") as mock_settings:
+        mock_settings.CONDUIT_PRIVATE_KEY = RS256_KEY
+        mock_settings.CONDUIT_JWT_ISSUER = "sentry"
+        mock_settings.CONDUIT_JWT_AUDIENCE = "conduit"
+        mock_settings.CONDUIT_GATEWAY_URL = gateway_url
+
+        org_id = 123
+        result = get_conduit_credentials(org_id)
+
+        assert isinstance(result.token, str)
+        assert isinstance(result.channel_id, str)
+        assert isinstance(result.url, str)
+
+        assert str(org_id) in result.url
+        assert result.url == f"{gateway_url}/events/{org_id}"
+
+
+def test_get_conduit_credentials_uses_custom_url():
+    """Should use provided gateway_url instead of settings."""
+    gateway_url = "https://custom.conduit.io"
+    with patch("sentry.conduit.auth.settings") as mock_settings:
+        mock_settings.CONDUIT_PRIVATE_KEY = RS256_KEY
+        mock_settings.CONDUIT_JWT_ISSUER = "sentry"
+        mock_settings.CONDUIT_JWT_AUDIENCE = "conduit"
+
+        org_id = 123
+        result = get_conduit_credentials(org_id, gateway_url)
+
+        assert isinstance(result.token, str)
+        assert isinstance(result.channel_id, str)
+        assert isinstance(result.url, str)
+
+        assert str(org_id) in result.url
+        assert result.url == f"{gateway_url}/events/{org_id}"
+
+
+def test_get_conduit_credentials_token_is_valid():
+    """Generated token should be decodable with correct claims."""
+    gateway_url = "https://conduit.example.com"
+    with patch("sentry.conduit.auth.settings") as mock_settings:
+        mock_settings.CONDUIT_PRIVATE_KEY = RS256_KEY
+        mock_settings.CONDUIT_JWT_ISSUER = "sentry"
+        mock_settings.CONDUIT_JWT_AUDIENCE = "conduit"
+        mock_settings.CONDUIT_GATEWAY_URL = gateway_url
+
+        org_id = 123
+        result = get_conduit_credentials(org_id)
+
+        claims = pyjwt.decode(
+            result.token,
+            RS256_PUB_KEY,
+            algorithms=["RS256"],
+            audience="conduit",
+            options={"verify_exp": False},
+        )
+
+        assert claims["org_id"] == org_id
+        assert claims["channel_id"] == result.channel_id
+
+        assert str(org_id) in result.url
+        assert result.url == f"{gateway_url}/events/{org_id}"