# Authentication (/connectors/cloudevents/how-to/authentication)



The CloudEvents connector has no authentication of its own. It shares the
[auth middleware](/connectors/reference/auth-and-security) of the shared HTTP server with
the REST, MCP, and A2A connectors, so a single JWT Bearer token secures every
`/ce/*` endpoint.

## Overview [#overview]

When authentication is enabled on the KubeMQ server, every CloudEvents request —
both publishing (`POST /ce/send/*`, `POST /ce/queue/*`) and subscribing
(`GET /ce/subscribe/*`) — must carry a valid JWT in the `Authorization` header:

```http
Authorization: Bearer <jwt-token>
```

Because `/ce/*` endpoints are standard HTTP (not JSON-RPC), an authentication
failure returns **HTTP 401 Unauthorized**, unlike the MCP and A2A gateways, which
return the JSON-RPC `-32010` error code. When authentication is disabled, requests
are accepted with synthetic anonymous claims and no header is required.

Token issuance and validation (shared secret vs OIDC provider, claim schema) are
configured server-wide. See [Auth & security](/connectors/reference/auth-and-security)
for the full model — public routes, CORS, origin validation, and TLS/mTLS.

## How it works [#how-it-works]

A CloudEvents request passes through the shared auth middleware before reaching the
CE handler; the verified `ClientID` claim flows downstream and becomes the KubeMQ
message identity.

<Mermaid
  chart="`
graph LR
CALLER[&#x22;CloudEvents client&#x22;]
AUTH[&#x22;Auth middleware<br/>verify Bearer JWT&#x22;]
CE[&#x22;CloudEvents connector&#x22;]
ARR[&#x22;Array&#x22;]
BROKER[&#x22;Message Broker&#x22;]

CALLER -- &#x22;Authorization: Bearer&#x22; --> AUTH
AUTH -- &#x22;claims.ClientID&#x22; --> CE
AUTH -. &#x22;401 if invalid&#x22; .-> CALLER
CE --> ARR
ARR --> BROKER

class CALLER client
class AUTH,ARR,BROKER broker
class CE connector
`"
/>

*The shared auth middleware verifies the Bearer token, then hands the resolved ClientID to the CloudEvents connector.*

## ClientID resolution [#clientid-resolution]

The KubeMQ `ClientID` attached to a published message is resolved with authentication
taking priority over the event payload:

| Authentication        | ClientID source                                |
| --------------------- | ---------------------------------------------- |
| Disabled              | CloudEvent `source` attribute                  |
| Enabled (valid token) | JWT claims `ClientID` — **overrides** `source` |

When auth is enabled, the CloudEvent `source` is still preserved as the `ce_source`
tag (so the event round-trips intact), but the KubeMQ `ClientID` is set from the
verified token. This ensures identity is controlled by the authentication system
rather than by client-supplied data. The same override applies to the `client_id`
query parameter on queue and subscribe endpoints.

## Authenticated publish [#authenticated-publish]

Add the `Authorization` header to any send request. The token format and header are
identical across all CloudEvents endpoints — events, events-store, queues, commands,
and queries.

```bash
curl -X POST http://localhost:9090/ce/send/event \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -H "Content-Type: application/cloudevents+json" \
  -d '{
    "specversion": "1.0",
    "type": "com.example.order.created",
    "source": "order-service",
    "subject": "orders",
    "data": {"order_id": "12345", "amount": 99.99}
  }'
```

<Tabs groupId="language" items="['Go','Python','JavaScript','Java','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    req, _ := http.NewRequest("POST", base+"/ce/send/event", strings.NewReader(string(body)))
    req.Header.Set("Content-Type", "application/cloudevents+json")
    req.Header.Set("Authorization", "Bearer "+os.Getenv("JWT_TOKEN"))

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        log.Fatal("send event:", err)
    }
    defer resp.Body.Close()
    ```
  </Tab>

  <Tab value="Python">
    ```python
    headers, body = to_structured(event)
    headers["Authorization"] = f"Bearer {os.environ['JWT_TOKEN']}"

    resp = requests.post(
        f"{base}/ce/send/event",
        data=body,
        headers=dict(headers),
        timeout=10,
    )
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    const message = HTTP.structured(event);
    const resp = await fetch(`${base}/ce/send/event`, {
      method: 'POST',
      headers: {
        ...(message.headers as Record<string, string>),
        Authorization: `Bearer ${process.env.JWT_TOKEN}`,
      },
      body: message.body as string,
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java
    HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(base + "/ce/send/event"))
            .POST(HttpRequest.BodyPublishers.ofByteArray(body))
            .header("Content-Type", "application/cloudevents+json")
            .header("Authorization", "Bearer " + System.getenv("JWT_TOKEN"))
            .build();

    HttpResponse<String> response = httpClient.send(request,
            HttpResponse.BodyHandlers.ofString());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using var content = new ByteArrayContent(eventBytes.ToArray());
    content.Headers.ContentType = MediaTypeHeaderValue.Parse(contentType.ToString());

    using var httpClient = new HttpClient();
    httpClient.DefaultRequestHeaders.Authorization =
        new AuthenticationHeaderValue("Bearer", Environment.GetEnvironmentVariable("JWT_TOKEN"));

    var resp = await httpClient.PostAsync($"{base_}/ce/send/event", content);
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    headers, body = sdk.encode_event(event, structured_format: "json")
    uri = URI("#{base}/ce/send/event")
    Net::HTTP.start(uri.host, uri.port) do |http|
      req = Net::HTTP::Post.new(uri)
      req["Content-Type"] = headers["Content-Type"]
      req["Authorization"] = "Bearer #{ENV.fetch('JWT_TOKEN')}"
      req.body = body
      res = http.request(req)
    end
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    let body = serde_json::to_string(&event)?;
    let resp = client
        .post(format!("{}/ce/send/event", base))
        .header("Content-Type", "application/cloudevents+json")
        .header("Authorization", format!("Bearer {}", env::var("JWT_TOKEN")?))
        .body(body)
        .send()
        .await?;
    ```
  </Tab>
</Tabs>

## Authenticated subscribe [#authenticated-subscribe]

SSE subscriptions are long-lived `GET` requests; the Bearer token is sent once when
the stream is opened.

```bash
curl -N "http://localhost:9090/ce/subscribe/events?client_id=my-client&channel=orders" \
  -H "Authorization: Bearer $JWT_TOKEN" \
  -H "Accept: text/event-stream"
```

<Tabs groupId="language" items="['Go','Python','JavaScript','Java','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    req, _ := http.NewRequest("GET", sseURL, nil)
    req.Header.Set("Accept", "text/event-stream")
    req.Header.Set("Cache-Control", "no-cache")
    req.Header.Set("Authorization", "Bearer "+os.Getenv("JWT_TOKEN"))

    resp, err := http.DefaultClient.Do(req)
    ```
  </Tab>

  <Tab value="Python">
    ```python
    with requests.get(sse_url, stream=True, timeout=None,
                      headers={"Accept": "text/event-stream",
                               "Cache-Control": "no-cache",
                               "Authorization": f"Bearer {os.environ['JWT_TOKEN']}"}) as resp:
        for raw_line in resp.iter_lines(decode_unicode=True):
            ...
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    const es = new EventSource(sseUrl, {
      headers: { Authorization: `Bearer ${process.env.JWT_TOKEN}` },
    });
    es.addEventListener('cloudevent', (evt: MessageEvent) => {
      console.log(JSON.parse(evt.data));
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java
    HttpURLConnection conn = (HttpURLConnection) new URL(sseUrl).openConnection();
    conn.setRequestMethod("GET");
    conn.setRequestProperty("Accept", "text/event-stream");
    conn.setRequestProperty("Cache-Control", "no-cache");
    conn.setRequestProperty("Authorization", "Bearer " + System.getenv("JWT_TOKEN"));
    conn.setDoInput(true);
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using var request = new HttpRequestMessage(HttpMethod.Get, sseUrl);
    request.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue("text/event-stream"));
    request.Headers.Authorization =
        new AuthenticationHeaderValue("Bearer", Environment.GetEnvironmentVariable("JWT_TOKEN"));

    using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead);
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    Net::HTTP.start(uri.host, uri.port) do |http|
      req = Net::HTTP::Get.new(uri)
      req["Accept"] = "text/event-stream"
      req["Cache-Control"] = "no-cache"
      req["Authorization"] = "Bearer #{ENV.fetch('JWT_TOKEN')}"
      http.request(req) do |resp|
        ...
      end
    end
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    let stream = client
        .get(&sub_url)
        .header("Accept", "text/event-stream")
        .header("Cache-Control", "no-cache")
        .header("Authorization", format!("Bearer {}", env::var("JWT_TOKEN")?))
        .send()
        .await?
        .bytes_stream();
    ```
  </Tab>
</Tabs>

## Running without authentication [#running-without-authentication]

When authentication is disabled in the server configuration, the CloudEvents
connector accepts every request with synthetic anonymous claims:

* No `Authorization` header is required.
* The CloudEvent `source` attribute is used directly as the KubeMQ `ClientID`.
* The examples throughout the CloudEvents docs assume this mode unless stated otherwise.

<Callout type="warn">
  Running without authentication is for development and trusted networks only. Enable
  JWT authentication before exposing the CloudEvents connector in production.
</Callout>

## Related [#related]

<Cards>
  <Card title="Auth & security" href="/connectors/reference/auth-and-security" description="Shared JWT model, public routes, CORS, origin validation, and TLS/mTLS across all connectors." />

  <Card title="Channel resolution" href="/connectors/cloudevents/how-to/channel-resolution" description="How subject, the channel query param, and ClientID claims resolve the target channel." />

  <Card title="CE to KubeMQ mapping" href="/connectors/cloudevents/reference/ce-to-kubemq-mapping" description="Attribute-to-tag mapping, including how source and the ClientID claim relate." />
</Cards>