main.py

#!/usr/bin/env python3

import json
import yaml
import logging
import requests
import base64
from datetime import datetime
from flask import Flask, request, jsonify
from kubernetes import client, config
import os

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class KubeGuardController:
    def __init__(self):
        self.config_data = {}
        self.load_kubernetes_config()
        self.load_config()

    def load_kubernetes_config(self):
        """Load Kubernetes configuration"""
        try:
            config.load_incluster_config()
            logger.info("Loaded in-cluster Kubernetes config")
        except Exception:
            try:
                config.load_kube_config()
                logger.info("Loaded local Kubernetes config")
            except Exception as e:
                logger.error(f"Failed to load Kubernetes config: {e}")
                raise

        self.k8s_client = client.CoreV1Api()

    def load_config(self):
        """Load configuration from ConfigMap"""
        try:
            config_map_name = os.getenv('CONFIG_MAP_NAME', 'kube-guard-config')
            config_map_namespace = os.getenv('CONFIG_MAP_NAMESPACE', 'kube-guard')

            config_map = self.k8s_client.read_namespaced_config_map(
                name=config_map_name,
                namespace=config_map_namespace
            )

            config_yaml = config_map.data.get('config.yaml', '{}')
            self.config_data = yaml.safe_load(config_yaml)
            logger.info("Configuration loaded successfully")

        except Exception as e:
            logger.error(f"Failed to load config from ConfigMap: {e}")
            self.config_data = {
                'mattermost': {
                    'webhook_url': os.getenv('MATTERMOST_WEBHOOK_URL', ''),
                    'channel': os.getenv('MATTERMOST_CHANNEL', 'alerts')
                },
                'monitored_namespaces': ['*'],
                'ignored_users': [],
                'notifications': {
                    'shell_access': True,
                    'port_forward': True,
                    'pvc_mount': True,
                    'privileged_pod': True,
                    'host_access': True,
                    'rbac_change': True,
                    'namespace_delete': True,
                }
            }

    def is_namespace_monitored(self, namespace):
        """Check if a namespace is in the monitored list. '*' means all."""
        namespaces = self.config_data.get('monitored_namespaces', ['*'])
        if '*' in namespaces:
            return True
        return namespace in namespaces

    def is_user_ignored(self, username):
        """Check if a user is in the ignored list (no notifications sent)."""
        ignored = self.config_data.get('ignored_users', [])
        return username in ignored

    def send_mattermost_notification(self, message, username='KubeGuard'):
        """Send notification to Mattermost"""
        webhook_url = self.config_data.get('mattermost', {}).get('webhook_url')
        channel = self.config_data.get('mattermost', {}).get('channel', 'alerts')

        if not webhook_url:
            logger.warning("Mattermost webhook URL not configured")
            return

        payload = {
            'channel': f"#{channel}",
            'username': username,
            'text': message,
            'icon_emoji': ':warning:'
        }

        try:
            response = requests.post(webhook_url, json=payload, timeout=10)
            response.raise_for_status()
            logger.info("Notification sent to Mattermost successfully")
        except Exception as e:
            logger.error(f"Failed to send Mattermost notification: {e}")

    def get_user_info(self, admission_request):
        """Extract user information from the admission request"""
        user_info = admission_request.get('userInfo', {})
        username = user_info.get('username', 'unknown')
        groups = user_info.get('groups', [])
        return username, groups

    def _format_time(self):
        return datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')

    def _cluster_name(self):
        return self.config_data.get('cluster_name', 'in-cluster')

    # --- Detection methods ---

    def check_shell_access(self, admission_request):
        """Detect kubectl exec into a pod"""
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'PodExecOptions':
            return None
        namespace = admission_request.get('namespace', '')
        if not self.is_namespace_monitored(namespace):
            return None
        username, _ = self.get_user_info(admission_request)
        pod_name = admission_request.get('name', 'unknown')
        return (f":warning: **Shell Access Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` opened a shell to pod `{namespace}/{pod_name}`\n"
                f"Time: {self._format_time()}")

    def check_port_forward(self, admission_request):
        """Detect kubectl port-forward to a pod"""
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'PodPortForwardOptions':
            return None
        namespace = admission_request.get('namespace', '')
        if not self.is_namespace_monitored(namespace):
            return None
        username, _ = self.get_user_info(admission_request)
        pod_name = admission_request.get('name', 'unknown')
        return (f":warning: **Port Forward Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` created a port-forward to pod `{namespace}/{pod_name}`\n"
                f"Time: {self._format_time()}")

    def check_pvc_mount(self, admission_request):
        """Detect ad-hoc pods that mount PersistentVolumeClaims"""
        if admission_request.get('operation') != 'CREATE':
            return None
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'Pod':
            return None
        namespace = admission_request.get('namespace', '')
        if not self.is_namespace_monitored(namespace):
            return None

        obj = admission_request.get('object', {})
        spec = obj.get('spec', {})
        volumes = spec.get('volumes', [])

        pvc_names = []
        for vol in volumes:
            pvc = vol.get('persistentVolumeClaim')
            if pvc:
                pvc_names.append(pvc.get('claimName', 'unknown'))

        if not pvc_names:
            return None

        # Check if the pod is unmanaged (no ownerReferences = ad-hoc)
        owner_refs = obj.get('metadata', {}).get('ownerReferences', [])
        is_adhoc = len(owner_refs) == 0
        adhoc_tag = " **(ad-hoc / unmanaged)**" if is_adhoc else ""

        username, _ = self.get_user_info(admission_request)
        pod_name = obj.get('metadata', {}).get('name', '') or obj.get('metadata', {}).get('generateName', 'unknown')
        return (f":rotating_light: **PVC Mount Alert**{adhoc_tag} on cluster `{self._cluster_name()}`\n"
                f"User `{username}` created pod `{namespace}/{pod_name}` mounting PVC(s): `{', '.join(pvc_names)}`\n"
                f"Time: {self._format_time()}")

    def check_privileged_pod(self, admission_request):
        """Detect creation of privileged containers"""
        if admission_request.get('operation') != 'CREATE':
            return None
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'Pod':
            return None
        namespace = admission_request.get('namespace', '')
        if not self.is_namespace_monitored(namespace):
            return None

        obj = admission_request.get('object', {})
        spec = obj.get('spec', {})
        containers = spec.get('containers', []) + spec.get('initContainers', [])

        privileged_containers = []
        for c in containers:
            sc = c.get('securityContext', {})
            if sc.get('privileged', False):
                privileged_containers.append(c.get('name', 'unknown'))

        if not privileged_containers:
            return None

        username, _ = self.get_user_info(admission_request)
        pod_name = obj.get('metadata', {}).get('name', '') or obj.get('metadata', {}).get('generateName', 'unknown')
        return (f":rotating_light: **Privileged Container Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` created pod `{namespace}/{pod_name}` with privileged container(s): "
                f"`{', '.join(privileged_containers)}`\n"
                f"Time: {self._format_time()}")

    def check_host_access(self, admission_request):
        """Detect pods with hostPath, hostNetwork, hostPID, or hostIPC"""
        if admission_request.get('operation') != 'CREATE':
            return None
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'Pod':
            return None
        namespace = admission_request.get('namespace', '')
        if not self.is_namespace_monitored(namespace):
            return None

        obj = admission_request.get('object', {})
        spec = obj.get('spec', {})

        flags = []
        if spec.get('hostNetwork', False):
            flags.append('hostNetwork')
        if spec.get('hostPID', False):
            flags.append('hostPID')
        if spec.get('hostIPC', False):
            flags.append('hostIPC')

        host_paths = []
        for vol in spec.get('volumes', []):
            hp = vol.get('hostPath')
            if hp:
                host_paths.append(hp.get('path', 'unknown'))

        if not flags and not host_paths:
            return None

        details = []
        if flags:
            details.append(f"flags: `{', '.join(flags)}`")
        if host_paths:
            details.append(f"hostPath volumes: `{', '.join(host_paths)}`")

        username, _ = self.get_user_info(admission_request)
        pod_name = obj.get('metadata', {}).get('name', '') or obj.get('metadata', {}).get('generateName', 'unknown')
        return (f":rotating_light: **Host Access Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` created pod `{namespace}/{pod_name}` with {', '.join(details)}\n"
                f"Time: {self._format_time()}")

    def check_rbac_change(self, admission_request):
        """Detect creation or modification of ClusterRoles and ClusterRoleBindings"""
        operation = admission_request.get('operation', '')
        if operation not in ('CREATE', 'UPDATE'):
            return None
        resource = admission_request.get('kind', {})
        kind = resource.get('kind', '')
        if kind not in ('ClusterRole', 'ClusterRoleBinding', 'Role', 'RoleBinding'):
            return None

        username, _ = self.get_user_info(admission_request)
        obj = admission_request.get('object', {})
        name = obj.get('metadata', {}).get('name', 'unknown')
        namespace = admission_request.get('namespace', '')
        ns_info = f" in namespace `{namespace}`" if namespace else ""
        return (f":lock: **RBAC Change Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` {operation.lower()}d `{kind}/{name}`{ns_info}\n"
                f"Time: {self._format_time()}")

    def check_namespace_delete(self, admission_request):
        """Detect namespace deletion"""
        if admission_request.get('operation') != 'DELETE':
            return None
        resource = admission_request.get('kind', {})
        if resource.get('kind') != 'Namespace':
            return None

        username, _ = self.get_user_info(admission_request)
        name = admission_request.get('name', 'unknown')
        return (f":rotating_light: **Namespace Deletion Alert** on cluster `{self._cluster_name()}`\n"
                f"User `{username}` deleted namespace `{name}`\n"
                f"Time: {self._format_time()}")

    # --- Main processing ---

    def process_admission_request(self, admission_request):
        """Process the admission request and send notifications if needed"""
        try:
            username, _ = self.get_user_info(admission_request)
            if self.is_user_ignored(username):
                return

            notifications = self.config_data.get('notifications', {})

            checks = [
                ('shell_access', self.check_shell_access),
                ('port_forward', self.check_port_forward),
                ('pvc_mount', self.check_pvc_mount),
                ('privileged_pod', self.check_privileged_pod),
                ('host_access', self.check_host_access),
                ('rbac_change', self.check_rbac_change),
                ('namespace_delete', self.check_namespace_delete),
            ]

            for config_key, check_fn in checks:
                if notifications.get(config_key, True):
                    message = check_fn(admission_request)
                    if message:
                        self.send_mattermost_notification(message)
                        return  # Only one alert per request

        except Exception as e:
            logger.error(f"Error processing admission request: {e}")


# Global controller instance
try:
    controller = KubeGuardController()
except Exception as e:
    logger.warning(f"Failed to initialize KubeGuardController: {e}")
    controller = None

@app.route('/healthz', methods=['GET'])
def health_check():
    """Health check endpoint"""
    return jsonify({'status': 'healthy'}), 200

@app.route('/readyz', methods=['GET'])
def readiness_check():
    """Readiness check endpoint"""
    return jsonify({'status': 'ready'}), 200

@app.route('/validate', methods=['POST'])
def validate():
    """Validation webhook endpoint"""
    try:
        admission_review = request.get_json()

        if not admission_review or 'request' not in admission_review:
            return jsonify({'error': 'Invalid admission review'}), 400

        admission_request = admission_review['request']

        # Process the request for notifications (this doesn't block the request)
        controller.process_admission_request(admission_request)

        # Always allow the request (we're just monitoring, not blocking)
        admission_response = {
            'uid': admission_request.get('uid'),
            'allowed': True
        }

        return jsonify({
            'apiVersion': 'admission.k8s.io/v1',
            'kind': 'AdmissionReview',
            'response': admission_response
        })

    except Exception as e:
        logger.error(f"Error in validation webhook: {e}")
        return jsonify({'error': str(e)}), 500

@app.route('/mutate', methods=['POST'])
def mutate():
    """Mutation webhook endpoint (not used but required for completeness)"""
    try:
        admission_review = request.get_json()
        admission_request = admission_review['request']

        # No mutations, just allow (notification handled by /validate endpoint)
        admission_response = {
            'uid': admission_request.get('uid'),
            'allowed': True,
            'patchType': 'JSONPatch',
            'patch': base64.b64encode(json.dumps([]).encode()).decode()
        }

        return jsonify({
            'apiVersion': 'admission.k8s.io/v1',
            'kind': 'AdmissionReview',
            'response': admission_response
        })

    except Exception as e:
        logger.error(f"Error in mutation webhook: {e}")
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    port = int(os.getenv('PORT', 8443))

    # Check if TLS certificates are mounted
    tls_cert_path = '/etc/tls/tls.crt'
    tls_key_path = '/etc/tls/tls.key'

    if os.path.exists(tls_cert_path) and os.path.exists(tls_key_path):
        logger.info("Using mounted TLS certificates")
        ssl_context = (tls_cert_path, tls_key_path)
    else:
        logger.info("Using adhoc TLS certificates")
        ssl_context = 'adhoc'

    app.run(host='0.0.0.0', port=port, ssl_context=ssl_context)