Merge pull request #5 from obsrvbl/version-3.1

Release 3.1

.gitignore

@@ -19,6 +19,9 @@ index.html
 images/iso/*.iso
 packaging/output/
 packaging/root/opt/obsrvbl-ona/netflow/
+packaging/root/opt/obsrvbl-ona/ipfix/
 packaging/root/opt/obsrvbl-ona/ona_service/
 packaging/root/opt/obsrvbl-ona/pna/
 packaging/root/opt/obsrvbl-ona/version
+packaging/root/opt/obsrvbl-ona/system/python-packages/*
+!packaging/root/opt/obsrvbl-ona/system/python-packages/LICENSE*

images/iso/ona/rules.v4

@@ -7,12 +7,11 @@
 -A INPUT -p tcp --dport 22 -m state --state NEW,ESTABLISHED -j ACCEPT
 -A OUTPUT -p tcp --sport 22 -m state --state ESTABLISHED -j ACCEPT
 
-# Inbound NetFlow
+# Inbound NetFlow - add other lines here
+-A INPUT -p udp --dport 2055 -m state --state NEW,ESTABLISHED -j ACCEPT
+-A INPUT -p udp --dport 4739 -m state --state NEW,ESTABLISHED -j ACCEPT
 -A INPUT -p udp --dport 9995 -m state --state NEW,ESTABLISHED -j ACCEPT
 
-# Inbound SFlow
-# -A INPUT -p udp --dport 6343 -m state --state NEW,ESTABLISHED -j ACCEPT
-
 # Outbound ping
 -A OUTPUT -p icmp --icmp-type echo-request -j ACCEPT
 -A INPUT -p icmp --icmp-type echo-reply -j ACCEPT

packaging/root/opt/obsrvbl-ona/config

@@ -141,6 +141,27 @@ OBSRVBL_CHECK_POINT_PUSHER="false"
 OBSRVBL_CHECK_POINT_LOGDIR="/opt/obsrvbl-ona/logs/check_point"
 OBSRVBL_CHECK_POINT_PATH="/var/log/check-point-fw.log"
 
+##
+# eta-capturer
+##
+OBSRVBL_ETA_CAPTURER="false"
+OBSRVBL_ETA_PCAP_DIR="/opt/obsrvbl-ona/logs/eta"
+OBSRVBL_ETA_CAPTURE_IFACE="any"
+OBSRVBL_ETA_CAPTURE_SECONDS="600"
+OBSRVBL_ETA_CAPTURE_MBITS="32"
+OBSRVBL_ETA_UDP_PORT="2055"
+
+##
+# nvzflow-monitor
+##
+OBSRVBL_NVZFLOW_CAPTURER="false"
+OBSRVBL_NVZFLOW_LOG_DIR="/opt/obsrvbl-ona/logs/nvzflow"
+
+##
+# kubernetes-watcher
+##
+OBSRVBL_KUBERNETES_WATCHER="false"
+
 
 # load up auto and local config
 [ -f $OBSRVBL_CONFIG_AUTO ] && . $OBSRVBL_CONFIG_AUTO

packaging/root/opt/obsrvbl-ona/system/supervisord/ona-eta-monitor.sh

@@ -0,0 +1,36 @@
+#!/bin/sh
+
+#  Copyright 2018 Observable Networks
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+. /opt/obsrvbl-ona/config
+cd /opt/obsrvbl-ona/
+
+# Ensure log directory exists
+mkdir -p $OBSRVBL_ETA_PCAP_DIR
+
+# Wait until the next interval
+sleep `expr $OBSRVBL_ETA_CAPTURE_SECONDS - \`date +%s\` % $OBSRVBL_ETA_CAPTURE_SECONDS`
+
+# Run the monitor
+exec /usr/bin/sudo \
+    /usr/sbin/tcpdump \
+        -w "$OBSRVBL_ETA_PCAP_DIR/logs_%s.pcap" \
+        -i "$OBSRVBL_ETA_CAPTURE_IFACE" \
+        -s 0 \
+        -C "$OBSRVBL_ETA_CAPTURE_MBITS" \
+        -G "$OBSRVBL_ETA_CAPTURE_SECONDS" \
+        -W "1" \
+        -U \
+        -Z "obsrvbl_ona" \
+        "(udp dst port $OBSRVBL_ETA_UDP_PORT) and (udp[8:2] == 9)"

packaging/root/opt/obsrvbl-ona/system/supervisord/ona-pdns-monitor.sh

@@ -32,4 +32,4 @@ exec /usr/bin/sudo \
         -G "$OBSRVBL_PDNS_CAPTURE_SECONDS" \
         -U \
         -Z "obsrvbl_ona" \
-        "ip and udp src port 53"
+        "udp src port 53"

src/scripts/ona_service/eta_pusher.py

@@ -0,0 +1,47 @@
+#  Copyright 2018 Observable Networks
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from __future__ import division, print_function, unicode_literals
+
+# python builtins
+from os import getenv
+
+# local
+from tcpdump_pusher import TcpdumpPusher
+
+ENV_ETA_PCAP_DIR = 'OBSRVBL_ETA_PCAP_DIR'
+DEFAULT_ETA_PCAP_DIR = './logs'
+
+
+class EtaPusher(TcpdumpPusher):
+    def __init__(self, *args, **kwargs):
+        init_kwargs = {
+            'data_type': 'logs',
+            'poll_seconds': 60,
+            'pcap_dir': getenv(ENV_ETA_PCAP_DIR, DEFAULT_ETA_PCAP_DIR),
+        }
+        kwargs.update(init_kwargs)
+        super(EtaPusher, self).__init__(*args, **kwargs)
+
+    def execute(self, now=None):
+        all_remote_paths = super(EtaPusher, self).execute(now=now)
+
+        for remote_path in all_remote_paths:
+            self.api.send_signal(
+                self.data_type,
+                data={'path': remote_path, 'log_type': 'eta-pcap'}
+            )
+
+
+if __name__ == '__main__':
+    EtaPusher().run()

src/scripts/ona_service/kubernetes_watcher.py

@@ -0,0 +1,165 @@
+#  Copyright 2018 Observable Networks
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from __future__ import print_function, unicode_literals
+
+# python builtins
+import io
+import json
+import logging
+import os
+from tempfile import NamedTemporaryFile
+
+# local
+from api import requests
+from service import Service
+
+FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+logging.basicConfig(level=logging.INFO, format=FORMAT)
+
+K8S_CA_CERT_PATH = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
+K8S_TOKEN_PATH = '/var/run/secrets/kubernetes.io/serviceaccount/token'
+
+DATA_TYPE = 'hostnames'
+POLL_SECONDS = 3600
+
+
+class KubernetesWatcher(Service):
+    def __init__(self, *args, **kwargs):
+        # Kubernetes API endpoint
+        self.k8s_host = os.environ.get('KUBERNETES_SERVICE_HOST')
+        self.k8s_port = os.environ.get('KUBERNETES_SERVICE_PORT')
+
+        # Kubernetes CA certificate
+        self.k8s_ca_cert_path = os.environ.get(
+            'K8S_CA_CERT_PATH', K8S_CA_CERT_PATH
+        )
+
+        # Kubernetes authentication token
+        k8s_token_path = os.environ.get(
+            'KUBERNETES_TOKEN_PATH', K8S_TOKEN_PATH
+        )
+        self.k8s_token = self._read_if_exists(k8s_token_path)
+
+        self.match_hostname = (
+            requests.packages.urllib3.connection.match_hostname
+        )
+
+        kwargs.update({
+            'poll_seconds': POLL_SECONDS,
+        })
+        super(KubernetesWatcher, self).__init__(*args, **kwargs)
+
+    def _read_if_exists(self, *args, **kwargs):
+        # Pass *args and **kwargs to io.open. Read the file handle and return
+        # its data. If the file doesn't exist, return None.
+        try:
+            with io.open(*args, **kwargs) as infile:
+                return infile.read()
+        except (IOError, OSError):
+            return None
+
+    def _send_update(self, resolved, now):
+        with NamedTemporaryFile() as f:
+            f.write(json.dumps(resolved))
+            f.seek(0)
+            path = self.api.send_file(DATA_TYPE, f.name, now, suffix='hosts')
+            if path is not None:
+                data = {'path': path}
+                self.api.send_signal(DATA_TYPE, data)
+
+    def _get_pods(self):
+        # Hit the k8s API server for pods in all namespaces
+        url = 'https://{}:{}/api/v1/pods/'.format(self.k8s_host, self.k8s_port)
+        headers = {
+            'Authorization': 'Bearer {}'.format(self.k8s_token),
+            'Accept': 'application/json',
+        }
+        resp = requests.get(url, headers=headers, verify=self.k8s_ca_cert_path)
+        resp.raise_for_status()
+        pod_data = resp.json()
+
+        pod_ip_map = {}
+        cluster_ip_map = {}
+        for item in pod_data.get('items', []):
+            metadata = item.get('metadata', {})
+            pod_name = metadata.get('name')
+            pod_namespace = metadata.get('namespace')
+
+            status = item.get('status', {})
+            pod_ip = status.get('podIP')
+
+            spec = item.get('spec', {})
+            host_network = spec.get('hostNetwork', False)
+            node_name = spec.get('nodeName')
+
+            # Skip incomplete entries
+            if (not pod_name) or (not pod_namespace) or (not pod_ip):
+                continue
+
+            # For pods that share the node's address, report that address
+            if host_network:
+                if node_name:
+                    cluster_ip_map[pod_ip] = node_name
+            # Otherwise, report pod-name.pod-namespace
+            else:
+                pod_ip_map[pod_ip] = '{}.{}'.format(pod_name, pod_namespace)
+
+        pod_ip_map.update(cluster_ip_map)
+        return pod_ip_map
+
+    def _set_hostname_match(self):
+        # Ensure that the hostname validates. This is required for certain
+        # versions of Python 2.7 and requests/urllib3.
+        def _match_hostname(cert, hostname):
+            try:
+                self.match_hostname(cert, hostname)
+            except Exception:
+                if hostname != self.k8s_host:
+                    raise
+
+        requests.packages.urllib3.connection.match_hostname = _match_hostname
+
+    def _check_access(self):
+        # All the access variables must be set, and the CA cert path
+        # must exist
+        check_vars = (
+            self.k8s_host, self.k8s_port, self.k8s_ca_cert_path, self.k8s_token
+        )
+        if any(x is None for x in check_vars):
+            return False
+
+        return os.path.exists(self.k8s_ca_cert_path)
+
+    def execute(self, now=None):
+        if not self._check_access():
+            logging.error('Missing Kubernetes connection parameters')
+            return
+
+        self._set_hostname_match()
+        try:
+            resolved = self._get_pods()
+        except Exception:
+            logging.exception('Error getting pod data')
+            return
+
+        if not resolved:
+            logging.error('No mappings were found')
+            return
+
+        self._send_update(resolved, now)
+
+
+if __name__ == '__main__':
+    watcher = KubernetesWatcher()
+    watcher.run()