Ingress Traffic Capture with Python

This guide shows you how to use Qtap to capture ingress (incoming) HTTP traffic to your applications. We'll use a simple Python FastAPI web server as an example, but the same principles apply to any HTTP server.

What You'll Learn

Configure Qtap to capture incoming HTTP requests
See request headers, bodies, and metadata from client requests
Understand the difference between ingress and egress capture
Monitor API endpoints your application exposes

Use Cases

Why capture ingress traffic?

Debug API requests from clients
Monitor what data clients are sending to your endpoints
Audit incoming traffic for security analysis
Understand client behavior and usage patterns
Troubleshoot issues with incoming requests

Prerequisites

Linux system with kernel 5.10+ and eBPF support
Python 3.7+ installed
Qtap installed (see Installation)
Root/sudo access

Part 1: The Python Web Server

First, let's create a simple FastAPI web server that we'll monitor.

Install FastAPI and Uvicorn

pip install fastapi uvicorn

Create the Web Server

Save this as api_server.py:

from fastapi import FastAPI, Header
from typing import Optional
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    """Simple health check endpoint"""
    return {"message": "API server is running", "status": "healthy"}

@app.get("/api/users/{user_id}")
async def get_user(user_id: int, x_request_id: Optional[str] = Header(None)):
    """Get user by ID with optional request tracking header"""
    return {
        "user_id": user_id,
        "name": f"User {user_id}",
        "email": f"user{user_id}@example.com",
        "request_id": x_request_id
    }

@app.post("/api/users")
async def create_user(user_data: dict):
    """Create a new user"""
    return {
        "message": "User created successfully",
        "data": user_data,
        "id": 12345
    }

@app.get("/api/slow")
async def slow_endpoint():
    """Simulate a slow endpoint"""
    import time
    time.sleep(2)
    return {"message": "This took 2 seconds"}

if __name__ == "__main__":
    # Listen on all interfaces so other containers/machines can reach it
    # This is required for true ingress capture from network sources
    uvicorn.run(app, host="0.0.0.0", port=8000)

Start the Server

In one terminal, start the Python server:

python api_server.py

You should see:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

The server is now listening on all network interfaces, making it accessible from other containers or machines on your network.

Part 2: Qtap Configuration for Ingress

Now let's configure Qtap to capture incoming requests to our Python server.

Create the Qtap Config

Save this as qtap-ingress.yaml:

version: 2

# Storage Configuration - Output to console
services:
  # Event metadata (connection info, timing)
  event_stores:
    - type: stdout

  # HTTP request/response data
  object_stores:
    - type: stdout

# Processing Stack - Capture incoming HTTP traffic
stacks:
  ingress_stack:
    plugins:
      # HTTP Capture plugin
      - type: http_capture
        config:
          level: full      # (none|summary|details|full) - Capture everything including request/response bodies
          format: text     # (json|text) - Human-readable format

# Traffic Capture Settings - INGRESS configuration
tap:
  direction: ingress         # (egress|ingress|all) - Only capture INCOMING traffic
  ignore_loopback: true      # (true|false) - True since we're capturing network traffic (not localhost)
  audit_include_dns: false   # (true|false) - Skip DNS for cleaner output

  http:
    stack: ingress_stack     # Use our ingress processing stack

Key Configuration Points:

direction: ingress - Only captures incoming HTTP requests from the network
ignore_loopback: true - We're capturing real network traffic, not localhost
level: full - Captures complete request/response including bodies

Part 3: Running Qtap and Testing

Step 1: Start Qtap (Before Making Requests!)

In a second terminal, start Qtap with the ingress config:

# Using Docker
docker run -d \
  --name qtap-ingress \
  --user 0:0 --privileged \
  --cap-add CAP_BPF --cap-add CAP_SYS_ADMIN \
  --pid=host --network=host \
  -v /sys:/sys \
  -v /var/run/docker.sock:/var/run/docker.sock \
  -v "$(pwd)/qtap-ingress.yaml:/app/config/qtap.yaml" \
  -e TINI_SUBREAPER=1 \
  --ulimit=memlock=-1 \
  us-docker.pkg.dev/qpoint-edge/public/qtap:v0 \
  --log-level=warn \
  --log-encoding=console \
  --config="/app/config/qtap.yaml"

# Wait for Qtap to initialize (CRITICAL - must wait before traffic!)
sleep 6

Or using the Linux binary:

sudo qtap \
  --config=qtap-ingress.yaml \
  --log-level=warn \
  --log-encoding=console

Step 2: Find Your Host's Network IP Address

Find the IP address of the machine running your Python server:

# On Linux - find your primary network interface
ip addr show | grep "inet " | grep -v 127.0.0.1

# Example output:
# inet 192.168.1.100/24 brd 192.168.1.255 scope global eth0

The IP address before the / is what you'll use (e.g., 192.168.1.100).

Note: You can test from the same machine using its network IP (not localhost), or from a different machine on the same network. Both demonstrate true ingress traffic.

Step 3: Allow Firewall Access (If Needed)

If you have a firewall enabled, allow traffic on port 8000:

# On Ubuntu/Debian with ufw
sudo ufw allow 8000/tcp
sudo ufw status

# Or temporarily disable for testing
# sudo ufw disable

Step 4: Generate Test Traffic from Another Host

From another machine on your network (or the same machine using its network IP):

# Replace 192.168.1.100 with your server's IP address

# Test 1: Simple GET request (TRUE INGRESS!)
curl http://192.168.1.100:8000/

# Test 2: GET with path parameter
curl http://192.168.1.100:8000/api/users/42

# Test 3: GET with custom header
curl http://192.168.1.100:8000/api/users/99 \
  -H "X-Request-ID: test-12345"

# Test 4: POST with JSON body
curl -X POST http://192.168.1.100:8000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice Smith", "email": "[email protected]", "role": "admin"}'

# Test 5: Slow endpoint (tests latency tracking)
curl http://192.168.1.100:8000/api/slow

Why not use localhost? Using the network IP (even from the same machine) creates true ingress traffic from the network perspective. The kernel treats this as actual incoming network traffic, not loopback.

Alternative: Test from the same machine

If you don't have another machine available, you can still test from the same host using its network IP:

# On the SAME machine where Python and Qtap are running
# Use your network IP, NOT 127.0.0.1

curl http://192.168.1.100:8000/  # Replace with your actual IP

# This creates network traffic (goes through the network stack)
# Even though source and destination are the same physical machine

Step 5: View Captured Traffic

Check the Qtap logs to see the captured ingress traffic:

# If using Docker
docker logs qtap-ingress

# Look for output showing the Python process and incoming requests

What you should see:

=== HTTP Transaction ===
Source Process: python (PID: 12345)
Direction: INGRESS ← (incoming from network)
Source IP: 192.168.1.50 (client machine or same host via network IP)
Destination IP: 192.168.1.100:8000 (your Python server)
Method: POST
URL: http://192.168.1.100:8000/api/users
Status: 200 OK
Duration: 15ms

--- Request Headers ---
Host: 192.168.1.100:8000
User-Agent: curl/7.81.0
Accept: */*
Content-Type: application/json
Content-Length: 67

--- Request Body ---
{"name": "Alice Smith", "email": "[email protected]", "role": "admin"}

--- Response Headers ---
Content-Type: application/json
Content-Length: 89

--- Response Body ---
{"message":"User created successfully","data":{"name":"Alice Smith","email":"[email protected]","role":"admin"},"id":12345}
========================

Key indicators that it's working:

✅ "exe": "/usr/bin/python3" - Python process identified
✅ Direction: INGRESS - True incoming traffic from network
✅ Source IP: 192.168.1.50 - NOT 127.0.0.1 - This is real network traffic!
✅ Destination IP: 192.168.1.100:8000 - Your server's network IP
✅ "protocol": "http1" - HTTP protocol parsed correctly
✅ Full request/response bodies visible
✅ Custom headers captured (X-Request-ID)
✅ Latency/duration tracked

Part 4: Configuration Variations

Variation 1: Capture Both Ingress and Egress

To capture both incoming requests AND outgoing requests your Python app makes:

tap:
  direction: all             # Capture everything
  ignore_loopback: false     # Include localhost
  http:
    stack: ingress_stack

Variation 2: Headers Only (No Bodies)

To reduce output size, capture only headers:

stacks:
  headers_only:
    plugins:
      - type: http_capture
        config:
          level: details     # (summary|details|full) - Details includes headers but not bodies
          format: text

Variation 3: Filter Specific Endpoints with Rules

Capture only POST requests or specific paths using Rulekit:

stacks:
  filtered_stack:
    plugins:
      - type: http_capture
        config:
          level: none        # Don't capture by default
          format: json
          rules:
            # Only capture POST requests
            - name: "POST requests only"
              expr: http.req.method == "POST"
              level: full

            # Capture slow requests (> 1 second)
            - name: "Slow endpoints"
              expr: http.res.duration_ms > 1000
              level: full

Variation 4: Production Setup with S3 Storage

For production, send sensitive request/response data to your own S3 bucket:

services:
  event_stores:
    - type: stdout          # Metadata to console

  object_stores:
    - type: s3              # Bodies to S3 (never leaves your environment)
      config:
        endpoint: https://s3.amazonaws.com
        region: us-east-1
        bucket: my-company-qtap-ingress
        access_key_id: ${AWS_ACCESS_KEY_ID}
        secret_access_key: ${AWS_SECRET_ACCESS_KEY}

See Level 4 of the Complete Guide for full S3 setup.

Understanding the Output

What Qtap Captures for Ingress

Event Metadata (anonymized):

Source IP and port (client)
Destination IP and port (your server)
Process information (python, PID, container)
Timing (request start, duration)
Status code

Object Data (sensitive):

HTTP method, path, query parameters
Request headers (including custom headers)
Request body (JSON, form data, etc.)
Response headers
Response body

Direction Field Explained

INGRESS (incoming): Traffic coming INTO your application
- Client → Your Python server
- Shows what clients are requesting
EGRESS (outgoing): Traffic going OUT from your application
- Your Python server → External API
- Shows what your server is calling

Troubleshooting

Not Seeing Any Traffic?

Check 1: Is your Python server listening on 0.0.0.0?

# In api_server.py - must bind to 0.0.0.0, not 127.0.0.1
uvicorn.run(app, host="0.0.0.0", port=8000)

Check 2: Is Qtap running BEFORE you made requests?

# Qtap must be running first, then generate traffic
docker logs qtap-ingress  # Check if Qtap started successfully

Check 3: Is your Python server actually running?

# Test with your network IP (NOT localhost!)
curl http://192.168.1.100:8000/  # Replace with your actual IP
# Should return JSON response

Check 4: Can clients reach your server?

# Check firewall status
sudo ufw status

# If active, ensure port 8000 is allowed
sudo ufw allow 8000/tcp

# Test network connectivity from client
ping 192.168.1.100  # Replace with your server's IP

Check 5: Check Qtap is hooking into Python correctly

docker logs qtap-ingress 2>&1 | grep -i python
# Should see logs about attaching to Python process

Seeing `"l7Protocol": "other"` instead of `"http1"`?

This means Qtap captured the connection but couldn't parse HTTP. Possible causes:

Python server using HTTPS (TLS) - Qtap should still see it via SSL hooks
Traffic is not HTTP
Qtap not fully initialized when request was made (wait 6+ seconds)

Too Much Noise from System Traffic?

Add process filters to exclude specific executables:

tap:
  direction: ingress
  ignore_loopback: true

  filters:
    custom:
      - exe: /usr/sbin/sshd     # Exclude SSH daemon
        strategy: exact
      - exe: /lib/systemd/      # Exclude systemd processes
        strategy: prefix

Real-World Example: API Gateway Monitoring

Here's a practical example for monitoring an API gateway:

version: 2

services:
  event_stores:
    - type: stdout
  object_stores:
    - type: s3  # Store sensitive request/response data securely
      config:
        endpoint: https://minio.internal.company.com
        bucket: api-gateway-audit
        # credentials from environment

stacks:
  # Stack for all ingress traffic
  api_monitoring:
    plugins:
      - type: http_capture
        config:
          level: none      # Default: don't capture
          format: json
          rules:
            # Capture all authentication requests
            - name: "Auth endpoints"
              expr: http.req.path matches /^\/api\/auth\//
              level: full

            # Capture all errors
            - name: "Error responses"
              expr: http.res.status >= 400
              level: full

            # Capture slow requests (for performance debugging)
            - name: "Slow requests"
              expr: http.res.duration_ms > 500
              level: details  # Headers only, no bodies

tap:
  direction: ingress
  ignore_loopback: true    # True for network traffic (production)
  http:
    stack: api_monitoring

Next Steps

Learn More About Ingress Capture:

Traffic Capture Settings - Complete tap section reference
Traffic Processing with Plugins - All plugin options
Complete Guide - Progressive tutorial covering all Qtap features

Production Deployment:

Storage Configuration - S3 setup for production
Kubernetes Manifest - Deploy Qtap in K8s
Helm Chart - Helm deployment

Alternative: Cloud Management:

Qplane - Manage Qtap with visual dashboards
POC Kick Off Guide - Quick start with cloud control plane

Cleanup

# Stop Qtap
docker rm -f qtap-ingress

# Stop Python server (CTRL+C in the terminal)

# Remove config file
rm qtap-ingress.yaml api_server.py

This guide uses validated configurations. All examples are tested and guaranteed to work.

PreviousStarter Configuration - Stdout Only NextHTTPS Header Capture Without Proxies

Last updated 17 hours ago

What You'll Learn

Use Cases

Prerequisites

Part 1: The Python Web Server

Install FastAPI and Uvicorn

Create the Web Server

Start the Server

Part 2: Qtap Configuration for Ingress

Create the Qtap Config

Part 3: Running Qtap and Testing

Step 1: Start Qtap (Before Making Requests!)

Step 2: Find Your Host's Network IP Address

Step 3: Allow Firewall Access (If Needed)

Step 4: Generate Test Traffic from Another Host

Step 5: View Captured Traffic

Part 4: Configuration Variations

Variation 1: Capture Both Ingress and Egress

Variation 2: Headers Only (No Bodies)

Variation 3: Filter Specific Endpoints with Rules

Variation 4: Production Setup with S3 Storage

Understanding the Output

What Qtap Captures for Ingress

Direction Field Explained

Troubleshooting

Not Seeing Any Traffic?

Seeing "l7Protocol": "other" instead of "http1"?

Too Much Noise from System Traffic?

Real-World Example: API Gateway Monitoring

Next Steps

Cleanup

Seeing `"l7Protocol": "other"` instead of `"http1"`?