Service Mesh 安全 for AI Microservices

"""
Audit service mesh sidecar deployment in AI namespaces.
Identifies pods without sidecars, misconfigured injection,
and performance-related bypasses that create 安全 gaps.
"""
 
import subprocess
import json
from typing import Any
 
 
class ServiceMeshAIAuditor:
    """Audit service mesh configuration for AI workloads."""
 
    def __init__(self, namespace: str = "ai-推論"):
        self.namespace = namespace
        self.findings: list[dict] = []
 
    def _kubectl_json(self, *args: str) -> dict[str, Any]:
        """Execute kubectl and return parsed JSON."""
        cmd = ["kubectl", "-n", self.namespace] + list(args) + ["-o", "json"]
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return {}
        return json.loads(result.stdout)
 
    def check_sidecar_coverage(self) -> None:
        """
        Verify all AI pods have service mesh sidecars injected.
        Missing sidecars mean unencrypted, unauthorized communication.
        """
        pods = self._kubectl_json("get", "pods")
 
        for pod in pods.get("items", []):
            name = pod["metadata"]["name"]
            containers = [
                c["name"]
                for c in pod.get("spec", {}).get("containers", [])
            ]
 
            # Check for known sidecar container names
            sidecar_names = {"istio-proxy", "linkerd-proxy", "envoy-sidecar"}
            has_sidecar = bool(sidecar_names & set(containers))
 
            # Check for sidecar injection annotation
            annotations = pod.get("metadata", {}).get("annotations", {})
            injection_disabled = (
                annotations.get("sidecar.istio.io/inject") == "false"
                or annotations.get("linkerd.io/inject") == "disabled"
            )
 
            if not has_sidecar:
                severity = "HIGH"
                detail = (
                    f"Pod {name} has no service mesh sidecar. "
                    f"Containers: {containers}. "
                )
                if injection_disabled:
                    detail += (
                        "Injection is explicitly disabled via annotation. "
                        "This was likely intentional but creates a 安全 gap."
                    )
                    severity = "CRITICAL"
                else:
                    detail += (
                        "Injection may have failed or the namespace is not "
                        "labeled for injection."
                    )
 
                # Check if 這是 a GPU/model-serving pod
                for container in pod.get("spec", {}).get("containers", []):
                    limits = container.get("resources", {}).get("limits", {})
                    if any("gpu" in k.lower() for k in limits):
                        detail += " THIS IS A GPU POD — model 推論 "
                        detail += "traffic is unprotected."
                        severity = "CRITICAL"
 
                self.findings.append({
                    "severity": severity,
                    "title": f"Missing sidecar: {name}",
                    "detail": detail,
                })
 
    def check_mtls_policy(self) -> None:
        """Verify mTLS is enforced in STRICT mode."""
        # Check Istio PeerAuthentication
        result = subprocess.run(
            [
                "kubectl", "get", "peerauthentication",
                "-n", self.namespace, "-o", "json",
            ],
            capture_output=True, text=True, timeout=30,
        )
 
        if result.returncode == 0:
            policies = json.loads(result.stdout)
            has_strict = False
 
            for policy in policies.get("items", []):
                mode = (
                    policy.get("spec", {})
                    .get("mtls", {})
                    .get("mode", "UNSET")
                )
                name = policy["metadata"]["name"]
 
                if mode == "STRICT":
                    has_strict = True
                elif mode == "PERMISSIVE":
                    self.findings.append({
                        "severity": "HIGH",
                        "title": f"PERMISSIVE mTLS: {name}",
                        "detail": (
                            "PERMISSIVE mode allows plaintext connections. "
                            "攻擊者 in the network can intercept AI "
                            "推論 requests and responses without TLS."
                        ),
                    })
                elif mode == "DISABLE":
                    self.findings.append({
                        "severity": "CRITICAL",
                        "title": f"mTLS DISABLED: {name}",
                        "detail": (
                            "mTLS is explicitly disabled. All service-to-"
                            "service communication is unencrypted."
                        ),
                    })
 
                # Check for port-level exceptions
                port_mtls = (
                    policy.get("spec", {}).get("portLevelMtls", {})
                )
                for port, config in port_mtls.items():
                    if config.get("mode") != "STRICT":
                        self.findings.append({
                            "severity": "HIGH",
                            "title": (
                                f"mTLS exception on port {port}: {name}"
                            ),
                            "detail": (
                                f"Port {port} has mTLS mode "
                                f"{config.get('mode', 'UNSET')}. "
                                f"If 這是 a model serving port, "
                                f"推論 traffic is unprotected."
                            ),
                        })
 
            if not has_strict:
                self.findings.append({
                    "severity": "HIGH",
                    "title": "No STRICT mTLS policy in namespace",
                    "detail": (
                        "Without a STRICT PeerAuthentication policy, "
                        "services can accept plaintext connections."
                    ),
                })
 
    def check_authorization_policies(self) -> None:
        """
        Verify that AuthorizationPolicies restrict which services
        can communicate in the AI 推論 pipeline.
        """
        result = subprocess.run(
            [
                "kubectl", "get", "authorizationpolicies",
                "-n", self.namespace, "-o", "json",
            ],
            capture_output=True, text=True, timeout=30,
        )
 
        if result.returncode != 0:
            self.findings.append({
                "severity": "HIGH",
                "title": "No AuthorizationPolicies found",
                "detail": (
                    "Without 授權 policies, any service in the "
                    "mesh can call any other service. 這意味著 a "
                    "compromised preprocessor can directly access model "
                    "servers, bypassing the routing layer."
                ),
            })
            return
 
        policies = json.loads(result.stdout)
        if not policies.get("items"):
            self.findings.append({
                "severity": "HIGH",
                "title": "No AuthorizationPolicies in namespace",
                "detail": (
                    "No 授權 policies restrict service-to-service "
                    "communication. 實作 least-privilege policies."
                ),
            })
            return
 
        # Check for overly broad ALLOW policies
        for policy in policies.get("items", []):
            name = policy["metadata"]["name"]
            rules = policy.get("spec", {}).get("rules", [])
 
            for rule in rules:
                from_sources = rule.get("from", [])
                if not from_sources:
                    self.findings.append({
                        "severity": "MEDIUM",
                        "title": f"No source restriction: {name}",
                        "detail": (
                            "Rule has no 'from' clause, allowing any "
                            "source to match. 考慮 restricting to "
                            "specific service accounts."
                        ),
                    })
 
    def check_envoy_filter_injection(self) -> None:
        """
        Check for EnvoyFilter resources that modify proxy behavior.
        These can be used to bypass 安全 policies.
        """
        result = subprocess.run(
            [
                "kubectl", "get", "envoyfilters",
                "-n", self.namespace, "-o", "json",
            ],
            capture_output=True, text=True, timeout=30,
        )
 
        if result.returncode == 0:
            filters = json.loads(result.stdout)
            for ef in filters.get("items", []):
                name = ef["metadata"]["name"]
                patches = (
                    ef.get("spec", {}).get("configPatches", [])
                )
                for patch in patches:
                    context = patch.get("match", {}).get("context", "")
                    operation = patch.get("patch", {}).get("operation", "")
                    self.findings.append({
                        "severity": "MEDIUM",
                        "title": f"EnvoyFilter found: {name}",
                        "detail": (
                            f"Context: {context}, Operation: {operation}. "
                            f"EnvoyFilters can modify proxy behavior to "
                            f"bypass mTLS, 授權, or rate limiting. "
                            f"Verify this filter is authorized."
                        ),
                    })
 
    def run_audit(self) -> list[dict]:
        """Run complete service mesh audit for AI namespace."""
        self.findings = []
        self.check_sidecar_coverage()
        self.check_mtls_policy()
        self.check_authorization_policies()
        self.check_envoy_filter_injection()
        return self.findings
 
 
if __name__ == "__main__":
    import sys
    ns = sys.argv[1] if len(sys.argv) > 1 else "ai-推論"
    auditor = ServiceMeshAIAuditor(namespace=ns)
    findings = auditor.run_audit()
    for f in findings:
        print(f"[{f['severity']}] {f['title']}")
        print(f"  {f['detail']}\n")

"""
Detect service mesh sidecar bypass opportunities in AI pods.
"""
 
import subprocess
import json
from typing import Any
 
 
def detect_sidecar_bypasses(namespace: str = "ai-推論") -> list[dict]:
    """
    識別 pods and configurations that allow bypassing
    the service mesh sidecar proxy.
    """
    findings = []
 
    result = subprocess.run(
        ["kubectl", "get", "pods", "-n", namespace, "-o", "json"],
        capture_output=True, text=True, timeout=30,
    )
    if result.returncode != 0:
        return findings
 
    pods = json.loads(result.stdout)
 
    for pod in pods.get("items", []):
        name = pod["metadata"]["name"]
        annotations = pod.get("metadata", {}).get("annotations", {})
        spec = pod.get("spec", {})
 
        # Check for excluded ports
        excluded_out = annotations.get(
            "traffic.sidecar.istio.io/excludeOutboundPorts", ""
        )
        excluded_in = annotations.get(
            "traffic.sidecar.istio.io/excludeInboundPorts", ""
        )
 
        if excluded_out:
            findings.append({
                "severity": "HIGH",
                "title": f"Outbound port exclusion: {name}",
                "detail": (
                    f"Excluded outbound ports: {excluded_out}. "
                    f"Traffic on these ports bypasses mTLS and "
                    f"授權 policies."
                ),
            })
 
        if excluded_in:
            findings.append({
                "severity": "HIGH",
                "title": f"Inbound port exclusion: {name}",
                "detail": (
                    f"Excluded inbound ports: {excluded_in}. "
                    f"Inbound traffic on these ports is not authenticated."
                ),
            })
 
        # Check for excluded IP ranges
        excluded_ips = annotations.get(
            "traffic.sidecar.istio.io/excludeOutboundIPRanges", ""
        )
        if excluded_ips:
            findings.append({
                "severity": "MEDIUM",
                "title": f"Excluded IP ranges: {name}",
                "detail": (
                    f"Excluded ranges: {excluded_ips}. Traffic to these "
                    f"IPs bypasses the mesh."
                ),
            })
 
        # Check for host networking
        if spec.get("hostNetwork", False):
            findings.append({
                "severity": "CRITICAL",
                "title": f"Host networking: {name}",
                "detail": (
                    "Pod uses host networking, completely bypassing the "
                    "service mesh. All traffic is unencrypted and "
                    "unauthenticated."
                ),
            })
 
        # Check for containers running as istio-proxy UID (1337)
        for container in spec.get("containers", []):
            sec_ctx = container.get("securityContext", {})
            run_as_user = sec_ctx.get("runAsUser")
            if run_as_user == 1337:
                findings.append({
                    "severity": "CRITICAL",
                    "title": (
                        f"Container runs as mesh UID: "
                        f"{name}/{container['name']}"
                    ),
                    "detail": (
                        "Container runs as UID 1337 (Istio proxy UID). "
                        "Traffic from this container bypasses iptables "
                        "interception and the mesh entirely."
                    ),
                })
 
        # Check for privileged init containers that could modify iptables
        for init in spec.get("initContainers", []):
            sec = init.get("securityContext", {})
            caps = sec.get("capabilities", {}).get("add", [])
            if "NET_ADMIN" in caps or sec.get("privileged", False):
                # 這是 expected for istio-init, check if it's not
                if init["name"] not in ("istio-init", "linkerd-init"):
                    findings.append({
                        "severity": "HIGH",
                        "title": (
                            f"NET_ADMIN init container: "
                            f"{name}/{init['name']}"
                        ),
                        "detail": (
                            "Non-mesh init container with NET_ADMIN can "
                            "modify iptables rules to bypass sidecar "
                            "interception."
                        ),
                    })
 
    return findings

"""
Demonstrate traffic manipulation 偵測 in AI 推論 pipelines.
This 監控 tool compares model server responses with what
downstream services report to detect in-transit manipulation.
"""
 
import hashlib
import json
import time
from dataclasses import dataclass
from typing import Optional
 
 
@dataclass
class InferenceAuditRecord:
    """Records model 輸出 at 模型 server for later comparison."""
    request_id: str
    model_name: str
    output_hash: str
    timestamp: float
    output_size: int
 
 
class InferenceIntegrityMonitor:
    """
    Detect manipulation of 推論 results in transit
    by comparing outputs at different points in the pipeline.
    """
 
    def __init__(self):
        self.model_records: dict[str, InferenceAuditRecord] = {}
        self.discrepancies: list[dict] = []
 
    def record_model_output(
        self,
        request_id: str,
        model_name: str,
        output_data: bytes,
    ) -> InferenceAuditRecord:
        """
        Record 模型's actual 輸出 at 模型 server.
        This runs as a sidecar or interceptor at 模型 pod.
        """
        output_hash = hashlib.sha256(output_data).hexdigest()
        record = InferenceAuditRecord(
            request_id=request_id,
            model_name=model_name,
            output_hash=output_hash,
            timestamp=time.time(),
            output_size=len(output_data),
        )
        self.model_records[request_id] = record
        return record
 
    def verify_downstream_output(
        self,
        request_id: str,
        received_data: bytes,
        verification_point: str = "postprocessor",
    ) -> Optional[dict]:
        """
        Verify that the 輸出 received by a downstream service
        matches what 模型 actually produced.
        """
        record = self.model_records.get(request_id)
        if record is None:
            return {
                "severity": "MEDIUM",
                "request_id": request_id,
                "detail": (
                    f"No model 輸出 record for request {request_id}. "
                    f"Cannot verify integrity at {verification_point}."
                ),
            }
 
        received_hash = hashlib.sha256(received_data).hexdigest()
        if received_hash != record.output_hash:
            discrepancy = {
                "severity": "CRITICAL",
                "request_id": request_id,
                "model": record.model_name,
                "verification_point": verification_point,
                "expected_hash": record.output_hash[:16],
                "received_hash": received_hash[:16],
                "expected_size": record.output_size,
                "received_size": len(received_data),
                "detail": (
                    "Model 輸出 was modified in transit between "
                    f"模型 server and {verification_point}. "
                    "Possible man-in-the-middle attack on 推論 pipeline."
                ),
            }
            self.discrepancies.append(discrepancy)
            return discrepancy
 
        return None  # 輸出 matches

#!/usr/bin/env bash
# Apply hardened service mesh configuration for AI 推論 namespace
# This script configures Istio 安全 policies
 
NAMESPACE="${1:-ai-推論}"
 
echo "=== Applying Service Mesh Hardening for $NAMESPACE ==="
 
# 1. Enforce STRICT mTLS
kubectl apply -f - <<EOF
apiVersion: 安全.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: strict-mtls
  namespace: $NAMESPACE
spec:
  mtls:
    mode: STRICT
EOF
echo "[+] STRICT mTLS PeerAuthentication applied"
 
# 2. Default deny all traffic
kubectl apply -f - <<EOF
apiVersion: 安全.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: deny-all
  namespace: $NAMESPACE
spec:
  {}
EOF
echo "[+] Default deny AuthorizationPolicy applied"
 
# 3. Allow only specific 推論 pipeline traffic
kubectl apply -f - <<EOF
apiVersion: 安全.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: allow-推論-pipeline
  namespace: $NAMESPACE
spec:
  selector:
    matchLabels:
      app: model-server
  rules:
  - from:
    - source:
        principals:
        - "cluster.local/ns/$NAMESPACE/sa/model-router"
    to:
    - operation:
        methods: ["POST"]
        paths: ["/v1/completions", "/v2/models/*/infer"]
EOF
echo "[+] Model server access restricted to model-router only"
 
# 4. Allow router access from preprocessor only
kubectl apply -f - <<EOF
apiVersion: 安全.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: allow-router-access
  namespace: $NAMESPACE
spec:
  selector:
    matchLabels:
      app: model-router
  rules:
  - from:
    - source:
        principals:
        - "cluster.local/ns/$NAMESPACE/sa/preprocessor"
        - "cluster.local/ns/$NAMESPACE/sa/api-gateway"
EOF
echo "[+] Model router access restricted to preprocessor and gateway"
 
echo ""
echo "=== Configuration Complete ==="
echo "Verify with: istioctl analyze -n $NAMESPACE"