--- name: azion-stream-request-and-waf-records-to-a-siem description: >- Send every request to an application, with the WAF fields of the same request, to a SIEM in one log line, and confirm the delivery. --- # Stream request and WAF records to a SIEM You send every request to an application, with the WAF fields of the same request, to a security information and event management (SIEM) platform from Azion Console or the Azion API. To send only the requests WAF analyzed, refer to [Stream WAF events to a SIEM](/en/documentation/guides/application-security/firewall-and-waf/integrate-siems/). A [Data Stream](/en/documentation/platform/data-stream/) stream on the *Applications* data source, shaped by the *Applications + WAF Event Collector* preset, carries the request, the response, and the client address with the WAF score, matches, and block flag of the same request. One log line then answers a question about one request, whether WAF flagged it or not. The cost is volume: the preset sends 51 keys, the most of the five presets, and Data Stream is billed on requests and data transfer. ```mermaid %%{init: {"layout": "dagre", "themeVariables": {"fontSize": "13px"}, "flowchart": {"nodeSpacing": 12, "rankSpacing": 12, "padding": 6, "wrappingWidth": 70, "minNodeWidth": 40, "useMaxWidth": true}}}%% flowchart TD Req["A request reaches the workload"] --> Source["Applications data source"] Source --> Filter["Workload filter: only the chosen workloads"] Filter --> Template["Template 184: request and WAF fields in one line"] Template --> Out["Connector: the SIEM's endpoint"] Out --> Siem["SIEM"] Out --> Rte["Real-Time Events records each send and its status"] ``` 1. Each request to a chosen workload becomes an event of the *Applications* data source. 2. The workload filter keeps only the workloads you choose, and leaves the account's other streams active. 3. Template `184` renders each event as one log line with the request fields and the WAF fields. 4. The stream sends the lines to the SIEM's endpoint in batches, and Real-Time Events records each send with the status the endpoint returned. --- ## Prerequisites ### Console - An account user with the **Edit Data Stream** permission. For the permissions, refer to [Stream settings](/en/documentation/platform/data-stream/stream-settings/#permissions). - A workload whose firewall applies a WAF rule set with a *Set WAF* rule. To build it, refer to [Apply a rule set to every request](/en/documentation/guides/application-security/firewall-and-waf/apply-rule-set/). - An endpoint that your SIEM reads. The examples use an Apache Kafka cluster, with the host and port of its servers and one topic. For the other endpoints, refer to [Endpoints](/en/documentation/platform/data-stream/endpoints/). - Access to Azion Console. Refer to [Access Azion Console](/en/documentation/guides/platform/account-and-billing/how-to-access-azion-console/). ### API - An account user with the **Edit Data Stream** permission. For the permissions, refer to [Stream settings](/en/documentation/platform/data-stream/stream-settings/#permissions). - A workload whose firewall applies a WAF rule set with a *Set WAF* rule. To build it, refer to [Apply a rule set to every request](/en/documentation/guides/application-security/firewall-and-waf/apply-rule-set/). - An endpoint that your SIEM reads. The examples use an Apache Kafka cluster, with the host and port of its servers and one topic. For the other endpoints, refer to [Endpoints](/en/documentation/platform/data-stream/endpoints/). - A [personal token](/en/documentation/guides/platform/account-and-billing/personal-tokens/) and the ID of the workload. The examples send the requests of the workload `` to the topic `azion.requests` on `kafka1.example.com:9092` and `kafka2.example.com:9092`. Replace them with your workload and endpoint. --- ## Create the stream The stream takes its scope from a workload filter rather than sampling. A stream carries exactly one of the two, and saving an active stream with sampling, at any rate including `100`, deactivates every other stream on the account with no error. A filter holds up to 600 workloads, and a workload created later is not collected until you add it. ### Console To create the stream in Azion Console: 1. **Open Data Stream** Access [Azion Console](https://console.azion.com/) > **Data Stream**. 2. **Select + Stream** 3. **Name the stream** In the **General** section, enter a **Name**. For example: `requests-to-siem`. 4. **Select the Applications data source** In the **Input** section, select *Applications* in **Data Source**. 5. **Turn off Sampling** In the **Transform** section, turn off **Sampling** while **Option** is still *All Current and Future Workloads*. 6. **Choose the workload** Set **Option** to *Filter Workloads*. In **Available Workload**, select the workload and move it to **Chosen Workload**. If your Console names domains where this guide names workloads, the same control reads *Filter Domains*. The pick list then reads **Available Domain** and **Chosen Domain**. 7. **Select the template** In the **Render Template** section, select *Applications + WAF Event Collector* in **Template**. The **Data Set** field shows the keys of each log line. 8. **Select the endpoint** In the **Output** section, select *Apache Kafka* in **Connector**. Enter `kafka1.example.com:9092,kafka2.example.com:9092` in **Bootstrap Servers** and `azion.requests` in **Kafka Topic**, and turn on **Enable Transport Layer Security (TLS)**. 9. **Keep the stream active** In the **Status** section, keep **Active** turned on. 10. **Select Save** The Console shows `Your data stream has been created`. The stream appears in the **Data Stream** list with `workloads` in the **Source** column. ### API To create the stream with the API, send the `workloads` data source, which is *Applications* in the Console, with template `184`: ```bash curl --request POST \ --url https://api.azion.com/v4/workspace/stream/streams \ --header 'Authorization: Token ' \ --header 'Content-Type: application/json' \ --data '{ "name": "requests-to-siem", "active": true, "inputs": [{ "type": "raw_logs", "attributes": { "data_source": "workloads" } }], "transform": [ { "type": "filter_workloads", "attributes": { "workloads": [] } }, { "type": "render_template", "attributes": { "template": 184 } } ], "outputs": [{ "type": "kafka", "attributes": { "bootstrap_servers": "kafka1.example.com:9092,kafka2.example.com:9092", "kafka_topic": "azion.requests", "use_tls": true } }] }' ``` The API answers `201` with `state` set to `executed` and the stored stream under `data`. Keep its `id`, which every later request on the stream uses. A stream with both `sampling` and `filter_workloads` is refused with `32007`, and one with neither with `32002`. The stream starts sending one to two minutes after it is saved. Saving checks the format of each field and does not contact the endpoint, so a wrong address or topic shows only when the stream sends. For a SIEM with its own connector, select it in **Connector**, or send its `type` in `outputs[0]`: *Splunk* takes the HTTP Event Collector URL and token, *IBM QRadar* a URL, and *Elasticsearch* a URL and an encoded API key. A stream keeps one endpoint, and the API drops any second entry in `outputs` with no error. For every field, refer to [Endpoints](/en/documentation/platform/data-stream/endpoints/). In the [Prepare regulated web applications for security audits](/en/documentation/use-cases/secure-applications-and-networks/prepare-regulated-web-applications-for-security-audits/) use case, this section takes these values: | Setting | Value | | -------------------------- | ------------------------------------------------------------------------------------------------------------------------ | | Name | `audit-events` | | Workload filter | The application's workload | | Endpoint | *Apache Kafka*, with `kafka1.example.com:9092` in **Bootstrap Servers**, `azion.audit` in **Kafka Topic**, and TLS on | | `outputs` entry in the API | `"type": "kafka"`, with `bootstrap_servers` `kafka1.example.com:9092`, `kafka_topic` `azion.audit`, and `use_tls` `true` | The stream activates one to two minutes after it is saved, and sends a batch every 60 seconds, or sooner at 2,000 log lines. --- ## Confirm the delivery Real-Time Events records every send of a stream, delivered or not, with the status code the endpoint returned. A stream sends a batch every 60 seconds, or sooner when it reaches 2,000 log lines. A failed send does not stop the stream, and its lines are not sent again. Send a few requests to the workload, then wait about a minute. ### Console To find the sends in Azion Console: 1. **Open Real-Time Events** Access [Azion Console](https://console.azion.com/) > **Real-Time Events**. 2. **Select the Data Stream data source** 3. **Read the latest sends** Find the rows whose **Endpoint Type** matches your endpoint, and read their **Status Code** and **Streamed Lines**. A **Status Code** of `200` means the endpoint accepted the batch. ### API To read the sends, query the `dataStreamedEvents` dataset, with a range that covers the activation of the stream: ```bash curl --request POST \ --url https://api.azion.com/v4/events/graphql \ --header 'Authorization: Token ' \ --header 'Content-Type: application/json' \ --data '{"query":"query { dataStreamedEvents(limit: 20, filter: {tsRange: {begin: \"2030-01-01T10:00:00\", end: \"2030-01-01T11:00:00\"}}, orderBy: [ts_DESC]) { ts endpointType statusCode streamedLines dataStreamed } }"}' ``` The API answers `200` with one record per send, the latest first. Each record carries the type of your endpoint in `endpointType`: ```json {"data":{"dataStreamedEvents":[{"ts":"