feat(conduit): Add conduit auth functions

src/sentry/conduit/auth.py

@@ -0,0 +1,96 @@
+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.
+
+    Returns:
+        UUID string
+    """
+    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.
+
+    Args:
+        org_id: Sentry organization ID
+        channel_id: The channel UUID for the conduit stream
+        issuer: JWT claim for the issuer of the token
+        audience: JWT claim for the audience of the token
+        conduit_private_key: RSA private key
+
+    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 with token, channel_id, algorithm, and url
+    """
+    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",
+        tags={"org_id": org_id},
+        sample_rate=1.0,
+    )
+
+    return ConduitCredentials(
+        token=token,
+        channel_id=channel_id,
+        url=f"{gateway_url}/events/{org_id}",
+    )

src/sentry/conf/server.py

@@ -3171,3 +3171,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")

tests/sentry/conduit/test_auth.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}"