Our initial system for detecting steganography in user-uploaded images was a monolithic Python service running on a set of virtual machines. It polled a storage bucket, processed images using OpenCV, and logged the results. This architecture failed catastrophically within three months of production. The core issues were threefold: unpredictable, bursty traffic led to either massive over-provisioning or unacceptable processing delays; a security breach was a constant worry, as all processing logic ran within a single, overly-permissive network boundary; and the raw numerical output was useless for our security analysts, causing significant operational friction. The decision was made to rebuild from the ground up on Kubernetes, but simply containerizing the monolith was not an option. This is the breakdown of how we engineered its replacement, a decoupled, event-driven, and verifiably secure pipeline.
Initial Concept: Moving from Monolith to Event-Driven Functions
The fundamental flaw in the old system was its tightly coupled, synchronous nature. The new design had to be built on a few core principles: compute resources should scale directly with inbound request volume, including scaling to zero to control costs; components must be isolated and communicate over a secure, zero-trust network; and the system’s output must be immediately actionable intelligence, not raw data.
This led to a specific technology stack selection:
Knative: For the compute layer. We needed something more abstract than raw Kubernetes Deployments and Services. Knative Serving’s scale-to-zero capability was the primary draw for cost management. Its eventing model was the key to decoupling our processing stages. Instead of a single service doing everything, we could have small, single-purpose functions that react to events.
OpenCV: This was a non-negotiable carry-over. Its Python bindings provide the comprehensive toolset for the low-level image analysis required for steganography detection, specifically Least Significant Bit (LSB) analysis, which was our baseline detection method.
Linkerd: Our security posture demanded mutual TLS (mTLS) for all internal traffic. While options like Istio were considered, Linkerd’s simplicity, low operational overhead, and focus on providing mTLS and observability “out of the box” made it the pragmatic choice for our team. We needed to secure the system, not become full-time service mesh experts.
Seaborn: The security analysts needed to see the anomalies, not read log files of statistical deviations. We envisioned a separate service that would consume the analysis results and generate graphical reports. Seaborn, built on Matplotlib, is excellent for producing high-quality, server-side statistical visualizations without needing a GUI.
The proposed architecture was an asynchronous pipeline. An image upload would trigger an event. A Knative service running an OpenCV-based processor would consume this event, perform the analysis, and emit a new event with the results. A second Knative service, our reporting engine, would then consume the analysis event and use Seaborn to generate a visual report. Linkerd would be transparently draped over all of it, ensuring every network call between services was encrypted and authenticated.
flowchart TD
subgraph User Interaction
A[User uploads image to S3 Bucket]
end
subgraph Kubernetes Cluster with Linkerd Mesh
B[S3 Event Source] --> C{Knative Eventing: Broker};
C -->|Trigger: image.uploaded| D[Knative Service: Steganography-Analyzer];
subgraph D
direction LR
D1[Container: Python/Flask]
D2[Library: OpenCV]
end
D -- Emits analysis.completed event --> C;
C -->|Trigger: analysis.completed| E[Knative Service: Report-Generator];
subgraph E
direction LR
E1[Container: Python/Flask]
E2[Library: Seaborn/Matplotlib]
end
E --> F[Saves PNG Report to S3];
end
style D fill:#f9f,stroke:#333,stroke-width:2px
style E fill:#f9f,stroke:#333,stroke-width:2px
Part 1: The OpenCV Analyzer as a Knative Service
The first component was the core analyzer. It needed to be a containerized application that could be managed by Knative Serving. The primary challenge here was packaging OpenCV, a library with numerous system dependencies, into a slim, efficient container.
A common mistake is to start with a generic python:3.9 base image and install dependencies, leading to bloated images over 1GB. We opted for a multi-stage build to keep the final image leaner.
# Dockerfile for steganography-analyzer
# Stage 1: Builder with full build environment
FROM python:3.9-slim-buster AS builder
WORKDIR /app
# Install OS dependencies for OpenCV
RUN apt-get update && apt-get install -y --no-install-recommends \
libgl1-mesa-glx \
libglib2.0-0 \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies into a virtual environment
COPY requirements.txt .
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
RUN pip install --no-cache-dir -r requirements.txt
# Stage 2: Final image
FROM python:3.9-slim-buster
WORKDIR /app
# Copy only the necessary OS libraries from the builder stage
COPY /usr/lib/x86_64-linux-gnu/ /usr/lib/x86_64-linux-gnu/
COPY /lib/x86_64-linux-gnu/ /lib/x86_64-linux-gnu/
# Copy the virtual environment
COPY /opt/venv /opt/venv
# Copy application code
COPY ./src /app/src
# Set path and run command
ENV PATH="/opt/venv/bin:$PATH"
CMD ["python", "./src/app.py"]The application itself is a simple Flask server. Knative expects an HTTP server to be running on the port specified by the $PORT environment variable. The server listens for CloudEvents, which is the standard format used by Knative Eventing.
Here is the core processing logic. It’s not just “hello world”; it includes logging, error handling for invalid image data, and a structured JSON output.
# src/app.py for steganography-analyzer
import os
import cv2
import numpy as np
import base64
import logging
from flask import Flask, request, jsonify
from cloudevents.http import from_http
# Configure structured logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
app = Flask(__name__)
def analyze_lsb(image_array):
"""
Performs a basic LSB analysis on the image.
Extracts the LSB from each color channel and calculates variance.
High variance can indicate non-random data, a potential sign of steganography.
"""
try:
rows, cols, channels = image_array.shape
variances = []
for i in range(channels):
# Isolate one channel and extract the LSB plane
channel = image_array[:, :, i]
lsb_plane = channel & 1
# Calculate the variance. A truly random plane should have variance near 0.25
variance = np.var(lsb_plane.flatten())
variances.append(variance)
return variances
except Exception as e:
logging.error(f"Error during LSB analysis: {e}")
return None
@app.route('/', methods=['POST'])
def process_image_event():
try:
event = from_http(request.headers, request.get_data())
event_data = event.get_data()
image_id = event_data.get('imageId')
image_b64 = event_data.get('imageDataB64')
if not image_id or not image_b64:
logging.error("Received event with missing imageId or imageDataB64")
return jsonify({"status": "error", "message": "Missing required fields"}), 400
logging.info(f"Processing event for imageId: {image_id}")
# Decode image from base64
image_bytes = base64.b64decode(image_b64)
image_np_array = np.frombuffer(image_bytes, np.uint8)
image = cv2.imdecode(image_np_array, cv2.IMREAD_COLOR)
if image is None:
logging.error(f"Failed to decode image for imageId: {image_id}")
return jsonify({"status": "error", "message": "Invalid image format"}), 400
# Run the analysis
variances = analyze_lsb(image)
if variances is None:
return jsonify({"status": "error", "message": "Analysis failed internally"}), 500
result = {
"imageId": image_id,
"analysisType": "LSB_VARIANCE",
"results": {
"blue_channel_variance": variances[0],
"green_channel_variance": variances[1],
"red_channel_variance": variances[2]
}
}
# In a real system, this would emit a new CloudEvent.
# For this example, we just log and return the result.
logging.info(f"Analysis complete for {image_id}: {result}")
return jsonify(result), 200
except Exception as e:
logging.error(f"Unhandled exception in event handler: {e}")
return jsonify({"status": "error", "message": "Internal server error"}), 500
if __name__ == '__main__':
port = int(os.environ.get("PORT", 8080))
app.run(host='0.0.0.0', port=port)Finally, the Knative Service definition. A key lesson learned was managing cold starts. For this security-sensitive workload, a 5-10 second cold start was unacceptable. The trade-off was to set autoscaling.knative.dev/min-scale: "1" to keep at least one pod warm, sacrificing some cost-saving for predictable performance.
# knative/analyzer-service.yaml
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
name: steganography-analyzer
namespace: default
spec:
template:
metadata:
annotations:
# Keep one pod warm to mitigate cold start latency for critical path
autoscaling.knative.dev/min-scale: "1"
# Each pod can handle up to 10 concurrent requests before scaling up
autoscaling.knative.dev/target: "10"
spec:
containers:
- image: your-registry/steganography-analyzer:v1.0.0
ports:
- containerPort: 8080
env:
- name: PORT
value: "8080"
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "1"
memory: "1Gi"Part 2: Orchestration with Knative Eventing
With the processing unit built, we needed to feed it events. Knative Eventing’s broker/trigger model provides a robust pub/sub system on Kubernetes.
First, we install the Eventing components and a Broker. The Broker acts as a central hub for events.
# A default broker in the 'default' namespace
apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
name: default
namespace: defaultNext, we create a Trigger that subscribes to events of a specific type (type: dev.example.image.uploaded) and delivers them to our analyzer service. A critical production lesson was the need for a dead-letter sink (DLS). If our analyzer service repeatedly fails to process an event (e.g., due to a malformed image that crashes the parser), the event would be retried indefinitely, clogging the system. The DLS provides an escape hatch, sending poison pills to a separate service or storage for later inspection.
# knative/analyzer-trigger.yaml
apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
name: analyzer-trigger
namespace: default
spec:
broker: default
filter:
attributes:
type: dev.example.image.uploaded
subscriber:
ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: steganography-analyzer
uri: /
# Production readiness: Define a Dead Letter Sink for failed events
delivery:
deadLetterSink:
ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: event-dlq-handler # A separate service to log/store failed eventsPart 3: Zero-Trust Security with Linkerd
Deploying the services was not enough; we had to secure them. The requirement was strict: no unencrypted traffic inside the cluster. Manually managing TLS certificates for every service is an operational nightmare. This is where Linkerd proved its value.
After installing the Linkerd CLI and the control plane onto the cluster, we “meshed” our services. This involves injecting the Linkerd sidecar proxy into our pods. For Knative services, this is done by annotating the namespace or the service template itself. We chose to annotate the namespace.
kubectl annotate namespace default linkerd.io/inject=enabled
Any new pod in this namespace, including those scaled up by Knative, will automatically get the Linkerd proxy. The moment the proxy is present, it transparently intercepts all TCP traffic, automatically upgrading it to mTLS with any other meshed pod.
We verified this immediately. The linkerd viz stat command shows which services are talking to each other and confirms if the traffic has mTLS.
# Example output from linkerd viz stat
# > linkerd viz stat deployments -n default
NAME MESHED SUCCESS RPS LATENCY_P50 LATENCY_P95 LATENCY_P99 TCP_CONN
steganography-analyzer 1/1 100.00% 0.5rps 15ms 45ms 60ms 5
...The “MESHED” column confirms injection. Using linkerd viz tap allows deep inspection of live requests, showing headers and proving the traffic path through the proxies.
A significant pitfall emerged when the analyzer needed to fetch metadata from a legacy, non-meshed API outside the cluster. The calls failed because the Linkerd proxy, by default, doesn’t know how to handle this external traffic. The solution was to define a ServiceProfile, a custom Linkerd resource that provides the proxy with per-route configuration. This told Linkerd that traffic to legacy-api.internal was HTTP, could be retried on failure, and should not expect an mTLS handshake. This is a critical configuration step for integrating a service mesh into a brownfield environment.
# linkerd/legacy-api-profile.yaml
apiVersion: linkerd.io/v1alpha2
kind: ServiceProfile
metadata:
name: legacy-api.internal.svc.cluster.local
namespace: default
spec:
routes:
- name: 'GET /metadata/{id}'
isRetryable: true
timeout: 200msPart 4: The Seaborn Reporting Service
The final piece was turning data into insight. The analyzer service, upon completion, would emit its own event (type: dev.example.analysis.completed) containing the JSON results. A second Knative service, report-generator, subscribed to these events.
Its job was to take the variance data and generate a heatmap. A uniform image should have LSB variance near 0.25 across all its blocks. Steganographic content often creates localized pockets of higher or lower variance. A heatmap makes these patterns immediately obvious to a human analyst.
The key challenge here was running a plotting library like Seaborn/Matplotlib in a headless, server environment. The matplotlib.use('Agg') directive is essential; it selects a non-interactive backend that can render plots to files without needing a display server.
# src/reporter.py for report-generator
import os
import seaborn as sns
import matplotlib.pyplot as plt
import numpy as np
import logging
from flask import Flask, request, jsonify
from cloudevents.http import from_http
# ... (imports for S3 client)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Set Matplotlib to use a non-GUI backend
import matplotlib
matplotlib.use('Agg')
app = Flask(__name__)
def generate_variance_heatmap(image_id, variance_data, block_size=32):
"""
Generates a heatmap visualization of LSB variance across image blocks.
This is a simplified example; a real implementation would receive block-by-block variance.
Here, we simulate it to demonstrate the plotting.
"""
try:
logging.info(f"Generating heatmap for {image_id}")
# Simulate block data for demonstration
# A real system would pass this from the analyzer
simulated_blocks_rows = 16
simulated_blocks_cols = 16
# Create a heatmap where most values are around 0.25 but with an anomalous region
heatmap_data = np.full((simulated_blocks_rows, simulated_blocks_cols), 0.25)
heatmap_data[4:8, 4:8] = variance_data['red_channel_variance'] # Use one of the real variances
heatmap_data = heatmap_data + np.random.normal(0, 0.02, heatmap_data.shape)
fig, ax = plt.subplots(figsize=(10, 8))
sns.heatmap(heatmap_data, ax=ax, cmap="viridis", vmin=0, vmax=0.5)
ax.set_title(f'LSB Variance Heatmap for Image: {image_id}')
ax.set_xlabel('Image Block X-coordinate')
ax.set_ylabel('Image Block Y-coordinate')
filepath = f"/tmp/{image_id}_heatmap.png"
plt.savefig(filepath)
plt.close(fig) # Important to release memory
logging.info(f"Heatmap saved to {filepath}")
return filepath
except Exception as e:
logging.error(f"Failed to generate heatmap for {image_id}: {e}")
return None
@app.route('/', methods=['POST'])
def generate_report_event():
event = from_http(request.headers, request.get_data())
data = event.get_data()['results']
image_id = event.get_data()['imageId']
filepath = generate_variance_heatmap(image_id, data)
if filepath:
# Code to upload filepath to S3 bucket would go here
# s3_client.upload_file(filepath, 'report-bucket', f"{image_id}.png")
os.remove(filepath) # Clean up local file
return jsonify({"status": "success", "report_generated": image_id}), 200
else:
return jsonify({"status": "error", "message": "Failed to generate report"}), 500
if __name__ == '__main__':
port = int(os.environ.get("PORT", 8080))
app.run(host='0.0.0.0', port=port)This service was also deployed via a Knative service.yaml. A performance issue quickly became apparent: report generation was CPU-intensive. If multiple events arrived concurrently, a single pod would thrash, slowing all reports down. We solved this by setting autoscaling.knative.dev/container-concurrency: "1". This tells Knative to scale up a new pod for every concurrent request, ensuring each report generation process gets a full pod’s resources. This is another example of a critical, non-obvious tuning parameter required for production workloads.
The limitations of this system are clear. The LSB analysis is a naive first pass and easily defeated by more advanced steganographic algorithms. The next iteration of the analyzer needs a pluggable architecture to incorporate different detection modules, perhaps even a machine learning model, without requiring a full redeployment. Furthermore, the reliance on a warm pod (min-scale: 1) to manage latency is a temporary fix; exploring Knative’s activator behavior and custom scaling metrics could yield a more cost-effective solution for handling extreme traffic spikes. The reporting is also asynchronous. For a true Security Operations Center (SOC) dashboard, these results would need to be streamed via WebSockets, introducing a new set of stateful components into our otherwise stateless, event-driven world.
