Merge pull request #1004 from snyk/feat/sysdig-integration

RUN-2052 Feat/sysdig integration

Author: KatieArmstr

snyk-monitor/README.md

@@ -25,13 +25,13 @@ abcd1234-abcd-1234-abcd-1234abcd1234
 ```
 The Snyk Integration ID is used in the `--from-literal=integrationId=` parameter in the next step.
 
-2. If you are not using any private registries, create a Kubernetes secret called `snyk-monitor` containing the Snyk Integration ID from the previous step running the following command:
+2. (Optional) If you are not using any private registries, create a Kubernetes secret called `snyk-monitor` containing the Snyk Integration ID from the previous step running the following command:
  ```shell
  kubectl create secret generic snyk-monitor -n snyk-monitor --from-literal=dockercfg.json={} --from-literal=integrationId=abcd1234-abcd-1234-abcd-1234abcd1234
  ```
  Continue to Helm installation instructions below.
 
-3. If you're using a private registry, you should create a `dockercfg.json` file. The `dockercfg` file is necessary to allow the monitor to look up images in private registries. Usually your credentials can be found in `$HOME/.docker/config.json`. These must also be added to the `dockercfg.json` file.
+3. (Optional) If you're using a private registry, you should create a `dockercfg.json` file. The `dockercfg` file is necessary to allow the monitor to look up images in private registries. Usually your credentials can be found in `$HOME/.docker/config.json`. These must also be added to the `dockercfg.json` file.
 
 Create a file named `dockercfg.json`. Store your credentials in there; it should look like this:
 
@@ -77,12 +77,12 @@ Finally, create the secret in Kubernetes by running the following command:
 kubectl create secret generic snyk-monitor -n snyk-monitor --from-file=./dockercfg.json --from-literal=integrationId=abcd1234-abcd-1234-abcd-1234abcd1234
 ```
 
-4. If your private registry requires installing certificates (*.crt, *.cert, *.key only) please put them in a folder and create the following ConfigMap:
+4. (Optional) If your private registry requires installing certificates (*.crt, *.cert, *.key only) please put them in a folder and create the following ConfigMap:
 ```shell
 kubectl create configmap snyk-monitor-certs -n snyk-monitor --from-file=<path_to_certs_folder>
 ```
 
-5. If you are using an insecure registry or your registry is using unqualified images, you can provide a `registries.conf` file. See [the documentation](https://github.com/containers/image/blob/master/docs/containers-registries.conf.5.md) for information on the format and examples.
+5. (Optional) If you are using an insecure registry or your registry is using unqualified images, you can provide a `registries.conf` file. See [the documentation](https://github.com/containers/image/blob/master/docs/containers-registries.conf.5.md) for information on the format and examples.
 
 Create a file named `registries.conf`, see example adding an insecure registry: 
 
@@ -128,6 +128,31 @@ If '--reset-values' is specified, this is ignored.
 
 If running with Operator Lifecycle Manager (OLM) then OLM will handle upgrades for you when you request to install the latest version. This applies to OpenShift (OCP) and regular installations of OLM.
 
+## Sysdig Integration ##
+
+We have partnered with Sysdig to enrich the issues detected by Snyk for workloads with runtime data provided by Sysdig.
+
+In order for the integration with Sysdig to work, the Snyk monitor requires an extra Secret in the `snyk-monitor` namespace. The Secret name is `sysdig-eve-secret`.
+
+Please refer to the [Sysdig Secret installation guide](https://docs.sysdig.com/en/docs/sysdig-secure/integrate-effective-vulnerability-exposure-with-snyk/#copy-the-sysdig-secret) to install the Secret. Once the Sysdig Secret is installed, you need to copy it over to the snyk-monitor namespace:
+
+```bash
+kubectl get secret sysdig-eve-secret -n sysdig-agent -o yaml | grep -v '^\s*namespace:\s' | kubectl apply -n snyk-monitor  -f -
+```
+
+To enable Snyk to integrate with Sysdig and collect information about packages executed at runtime, use `--set sysdig.enabled=true` when installing the snyk-monitor:
+
+```bash
+helm upgrade --install snyk-monitor snyk-charts/snyk-monitor \
+  --namespace snyk-monitor \
+  --set clusterName="Production cluster" \
+  --set sysdig.enabled=true
+```
+
+> NOTE: The above command should be executed right after installing Sysdig. This will upgrade or install the snyk monitor, to allow the detection of Sysdig in the cluster.
+
+The snyk-monitor will now collect data from Sysdig every 4 hours.
+
 ## Setting up proxying ##
 
 Proxying traffic through a forwarding proxy can be achieved by setting the following values in the Helm chart:

snyk-monitor/templates/deployment.yaml

@@ -132,6 +132,20 @@ spec:
             value: {{ quote .Values.skopeo.compression.level }}
           - name: SNYK_WORKERS_COUNT
             value: {{ quote .Values.workers.count }}
+          {{- if .Values.sysdig.enabled }}
+          - name: SNYK_SYSDIG_ENDPOINT
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.sysdig.secretName }}
+                key: endpoint
+                optional: true
+          - name: SNYK_SYSDIG_TOKEN
+            valueFrom:
+              secretKeyRef:
+                name: {{ .Values.sysdig.secretName }}
+                key: token
+                optional: true
+          {{- end }}
           {{- with .Values.envs }}
           {{- toYaml . | trim | nindent 10 -}}
           {{- end }}

snyk-monitor/values.yaml

@@ -143,3 +143,7 @@ skopeo:
 
 workers:
   count: 10
+sysdig:
+  enabled: false
+  namespace: sysdig-agent
+  secretName: sysdig-eve-secret

src/common/config.ts

@@ -45,6 +45,12 @@ config.EXCLUDED_NAMESPACES = loadExcludedNamespaces();
 config.WORKERS_COUNT = Number(config.WORKERS_COUNT) || 10;
 config.SKOPEO_COMPRESSION_LEVEL = Number(config.SKOPEO_COMPRESSION_LEVEL) || 6;
 
+// return Sysdig endpoint information
+if (config.SYSDIG_ENDPOINT && config.SYSDIG_TOKEN) {
+  config.SYSDIG_ENDPOINT = config.SYSDIG_ENDPOINT.trim();
+  config.SYSDIG_TOKEN = config.SYSDIG_TOKEN.trim();
+}
+
 /**
  * Important: we delete the following env vars because we don't want to proxy requests to the Kubernetes API server.
  * The Kubernetes client library would honor the NO/HTTP/HTTPS_PROXY env vars.

src/data-scraper/index.ts

@@ -0,0 +1,75 @@
+import { logger } from '../common/logger';
+import { config } from '../common/config';
+import { sendRuntimeData } from '../transmitter';
+import { constructRuntimeData } from '../transmitter/payload';
+import { retryRequest } from '../transmitter';
+import { IRuntimeImagesResponse } from '../transmitter/types';
+import { NeedleOptions } from 'needle';
+import { Agent as HttpsAgent } from 'https';
+
+function isSuccessStatusCode(statusCode: number | undefined): boolean {
+  return statusCode !== undefined && statusCode >= 200 && statusCode < 300;
+}
+
+export async function scrapeData(): Promise<void> {
+  const base: string = config.SYSDIG_ENDPOINT;
+  const header: string = `Bearer ${config.SYSDIG_TOKEN}`;
+
+  const url: string = base + '/v1/runtimeimages';
+  // limit: min 1, max 500, default 250
+  const limit: number = 10;
+  const reqOptions: NeedleOptions = {
+    agent: new HttpsAgent({
+      keepAlive: true,
+      // We agreed with Sysdig to skip TLS certificates validation for HTTPS connection.
+      rejectUnauthorized: false,
+    }),
+    headers: {
+      authorization: header,
+    },
+  };
+
+  let cursor: string = '';
+  while (true) {
+    try {
+      logger.info({ cursor }, 'attempting to get runtime images');
+
+      const { response, attempt } = await retryRequest(
+        'get',
+        `${url}?limit=${limit}&cursor=${cursor}`,
+        {},
+        reqOptions,
+      );
+      if (!isSuccessStatusCode(response.statusCode)) {
+        throw new Error(`${response.statusCode} ${response.statusMessage}`);
+      }
+
+      logger.info(
+        {
+          attempt,
+          cursor,
+        },
+        'runtime images received successfully',
+      );
+
+      const responseBody: IRuntimeImagesResponse = response.body;
+      const runtimeDataPayload = constructRuntimeData(responseBody.data);
+      logger.info({}, 'sending runtime data upstream');
+      await sendRuntimeData(runtimeDataPayload);
+
+      cursor = responseBody.page.next || '';
+      if (!cursor) {
+        break;
+      }
+    } catch (error) {
+      logger.error(
+        {
+          error,
+          cursor,
+        },
+        'could not get runtime images',
+      );
+      break;
+    }
+  }
+}

src/index.ts

@@ -10,6 +10,7 @@ import { beginWatchingWorkloads } from './supervisor/watchers';
 import { loadAndSendWorkloadEventsPolicy } from './common/policy';
 import { sendClusterMetadata } from './transmitter';
 import { setSnykMonitorAgentId } from './supervisor/agent';
+import { scrapeData } from './data-scraper';
 
 process.on('uncaughtException', (err) => {
   if (state.shutdownInProgress) {
@@ -68,4 +69,20 @@ setImmediate(async function setUpAndMonitor(): Promise<void> {
   await sendClusterMetadata();
   await loadAndSendWorkloadEventsPolicy();
   await monitor();
+
+  const interval: number = 4 * 60 * 60 * 1000; // 4 hours in milliseconds
+  if (config.SYSDIG_ENDPOINT && config.SYSDIG_TOKEN) {
+    setInterval(async () => {
+      try {
+        await scrapeData();
+      } catch (error) {
+        logger.error(
+          { error },
+          'an error occurred while scraping runtime data',
+        );
+      }
+    }, interval).unref();
+  } else {
+    logger.info({}, 'Sysdig integration not detected');
+  }
 });

src/transmitter/index.ts

@@ -1,4 +1,3 @@
-import { parse } from 'url';
 import { queue } from 'async';
 import needle from 'needle';
 import sleep from 'sleep-promise';
@@ -17,6 +16,7 @@ import {
   IDependencyGraphPayload,
   IWorkloadEventsPolicyPayload,
   IClusterMetadataPayload,
+  IRuntimeDataPayload,
 } from './types';
 import { getProxyAgent } from './proxy';
 
@@ -27,24 +27,29 @@ interface KubernetesUpstreamRequest {
     | IDependencyGraphPayload
     | ScanResultsPayload
     | IWorkloadMetadataPayload
-    | IDeleteWorkloadPayload;
+    | IDeleteWorkloadPayload
+    | IClusterMetadataPayload
+    | IRuntimeDataPayload;
 }
 
 const upstreamUrl =
   config.INTEGRATION_API || config.DEFAULT_KUBERNETES_UPSTREAM_URL;
 
-let agent = new HttpAgent({
+let httpAgent = new HttpAgent({
   keepAlive: true,
 });
 
-if (parse(upstreamUrl).protocol?.startsWith('https')) {
-  agent = new HttpsAgent({
-    keepAlive: true,
-  });
+let httpsAgent = new HttpsAgent({
+  keepAlive: true,
+});
+
+function getAgent(u: string): HttpAgent {
+  const url = new URL(u);
+  return url.protocol === 'https:' ? httpsAgent : httpAgent;
 }
 
 // Async queue wraps around the call to retryRequest in order to limit
-// the number of requests in flight to Homebase at any one time.
+// the number of requests in flight to kubernetes upstream at any one time.
 const reqQueue = queue(async function (req: KubernetesUpstreamRequest) {
   return await retryRequest(req.method, req.url, req.payload);
 }, config.REQUEST_QUEUE_LENGTH);
@@ -60,7 +65,7 @@ export async function sendDepGraph(
       const request: KubernetesUpstreamRequest = {
         method: 'post',
         url: `${upstreamUrl}/api/v1/dependency-graph`,
-        payload: payload,
+        payload,
       };
 
       const { response, attempt } = await reqQueue.pushAsync(request);
@@ -91,7 +96,7 @@ export async function sendScanResults(
       const request: KubernetesUpstreamRequest = {
         method: 'post',
         url: `${upstreamUrl}/api/v1/scan-results`,
-        payload: payload,
+        payload,
       };
 
       const { response, attempt } = await reqQueue.pushAsync(request);
@@ -127,7 +132,7 @@ export async function sendWorkloadMetadata(
     const request: KubernetesUpstreamRequest = {
       method: 'post',
       url: `${upstreamUrl}/api/v1/workload`,
-      payload: payload,
+      payload,
     };
 
     const { response, attempt } = await reqQueue.pushAsync(request);
@@ -198,7 +203,7 @@ export async function deleteWorkload(
     const request: KubernetesUpstreamRequest = {
       method: 'delete',
       url: `${upstreamUrl}/api/v1/workload`,
-      payload: payload,
+      payload,
     };
 
     const { response, attempt } = await reqQueue.pushAsync(request);
@@ -229,10 +234,11 @@ function isSuccessStatusCode(statusCode: number | undefined): boolean {
   return statusCode !== undefined && statusCode > 100 && statusCode < 400;
 }
 
-async function retryRequest(
+export async function retryRequest(
   verb: NeedleHttpVerbs,
   url: string,
   payload: object,
+  reqOptions: NeedleOptions = {},
 ): Promise<IResponseWithAttempts> {
   const retry = {
     attempts: 3,
@@ -241,7 +247,8 @@ async function retryRequest(
   const options: NeedleOptions = {
     json: true,
     compressed: true,
-    agent,
+    agent: getAgent(url),
+    ...reqOptions,
   };
 
   if (config.HTTP_PROXY || config.HTTPS_PROXY) {
@@ -317,11 +324,13 @@ export async function sendClusterMetadata(): Promise<void> {
       'attempting to send cluster metadata',
     );
 
-    const { response, attempt } = await retryRequest(
-      'post',
-      `${upstreamUrl}/api/v1/cluster`,
+    const request: KubernetesUpstreamRequest = {
+      method: 'post',
+      url: `${upstreamUrl}/api/v1/cluster`,
       payload,
-    );
+    };
+
+    const { response, attempt } = await reqQueue.pushAsync(request);
     if (!isSuccessStatusCode(response.statusCode)) {
       throw new Error(`${response.statusCode} ${response.statusMessage}`);
     }
@@ -347,3 +356,46 @@ export async function sendClusterMetadata(): Promise<void> {
     );
   }
 }
+
+export async function sendRuntimeData(
+  payload: IRuntimeDataPayload,
+): Promise<void> {
+  const logContext = {
+    userLocator: payload.target.userLocator,
+    cluster: payload.target.cluster,
+    agentId: payload.target.agentId,
+    identity: payload.identity,
+  };
+
+  try {
+    logger.info(logContext, 'attempting to send runtime data');
+
+    const request: KubernetesUpstreamRequest = {
+      method: 'post',
+      url: `${upstreamUrl}/api/v1/runtime-results`,
+      payload,
+    };
+
+    const { response, attempt } = await reqQueue.pushAsync(request);
+
+    if (!isSuccessStatusCode(response.statusCode)) {
+      throw new Error(`${response.statusCode} ${response.statusMessage}`);
+    }
+
+    logger.info(
+      {
+        attempt,
+        ...logContext,
+      },
+      'runtime data sent upstream successfully',
+    );
+  } catch (error) {
+    logger.error(
+      {
+        error,
+        ...logContext,
+      },
+      'could not send runtime data',
+    );
+  }
+}