oc-python

oc-python

#!/usr/bin/env python3


# Ported via vibe coding

import argparse
import subprocess
import sys
import os
import base64
import json
import time
import shlex

def main():
    # --- Configuration & Argument Parsing ---
    parser = argparse.ArgumentParser(
        description="Run Python scripts in an OpenShift Pod.",
        usage="%(prog)s [OPTIONS] <script.py> [args...]"
    )
    parser.add_argument("--namespace", "-n", help="Specify the OpenShift namespace")
    parser.add_argument("--destroy", action="store_true", help="Delete the resident runner pod and its services")
    parser.add_argument("--runner", default=None, help="Specific name for the pod, ServiceAccount, and labels. Defaults to 'oc-python-runner' with a timestamped pod if omitted.")
    parser.add_argument("--serviceaccount", help="Specify a custom ServiceAccount name to use (defaults to the runner name)")
    parser.add_argument("--expose", type=int, help="Expose a specific port on the pod via a Kubernetes Service")
    parser.add_argument("script", nargs="?", help="The local Python script to execute")

    # Use parse_known_args to capture native flags and leave script arguments in `remaining`
    args, remaining = parser.parse_known_args()

    # Determine if a custom runner name was explicitly given
    is_custom_runner = args.runner is not None
    runner_base = args.runner if is_custom_runner else "oc-python-runner"
    app_label = f"app={runner_base}"
    ns_args = ["-n", args.namespace] if args.namespace else []

    # --- Handle Destroy Flag ---
    if args.destroy:
        print(f"[STAGE] Initialization: Cleaning up all '{runner_base}' pods and services...")
        # Clean up both pods and services mapping to this app label
        cmd = ["oc", "delete", "pod,svc", "-l", app_label] + ns_args
        subprocess.run(cmd)
        print("[DONE] Cleanup complete.")
        sys.exit(0)

    # --- Validate Input ---
    if not args.script:
        print("[ERROR] No script specified.")
        parser.print_help()
        sys.exit(1)

    if not os.path.isfile(args.script):
        print(f"[ERROR] Script '{args.script}' not found.")
        sys.exit(1)

    # Use the explicitly provided service account, or default to the runner base name
    sa_name = args.serviceaccount if args.serviceaccount else runner_base
    image = "registry.redhat.io/rhel9/python-312"

    # --- Ensure ServiceAccount Exists ---
    print(f"[STAGE] Check: Verifying ServiceAccount '{sa_name}'...")
    res = subprocess.run(["oc", "get", "sa", sa_name] + ns_args, capture_output=True)
    if res.returncode != 0:
        print(f"[STAGE] Setup: Creating ServiceAccount '{sa_name}'...")
        subprocess.run(["oc", "create", "sa", sa_name] + ns_args, check=True)
    else:
        print("[STAGE] Check: ServiceAccount already exists.")

    # --- Ensure Runner Pod Exists ---
    needs_creation = False
    pod_name = None

    if is_custom_runner:
        pod_name = runner_base
        print(f"[STAGE] Check: Searching for specific runner pod '{pod_name}'...")
        check_cmd = ["oc", "get", "pod", pod_name] + ns_args + ["-o", "jsonpath={.status.phase}"]
        res = subprocess.run(check_cmd, capture_output=True, text=True)
        
        if res.returncode != 0:
            needs_creation = True
        else:
            status = res.stdout.strip()
            if status != "Running":
                print(f"[STAGE] Wait: Pod '{pod_name}' exists but is in '{status}' state. Waiting for 'Ready'...")
                subprocess.run(["oc", "wait"] + ns_args + ["--for=condition=Ready", f"pod/{pod_name}", "--timeout=60s"], check=True)
            else:
                print(f"[STAGE] Check: Found resident runner '{pod_name}'. Reusing container...")
    else:
        print("[STAGE] Check: Searching for existing resident runner...")
        cmd = ["oc", "get", "pods"] + ns_args + ["-l", app_label, "-o", 'jsonpath={.items[?(@.status.phase=="Running")].metadata.name}']
        res = subprocess.run(cmd, capture_output=True, text=True)
        
        pod_names = res.stdout.strip().split()
        pod_name = pod_names[0] if pod_names else None

        if not pod_name:
            needs_creation = True
            pod_name = f"{runner_base}-{int(time.time())}"
        else:
            print(f"[STAGE] Check: Found resident runner '{pod_name}'. Reusing container...")

    # --- Provisioning logic (if needed) ---
    if needs_creation:
        print(f"[STAGE] Deployment: Provisioning new Pod '{pod_name}'...")
        
        # Build Container Spec
        container_spec = {
            "name": "python-runner",
            "image": image,
            "command": ["/bin/bash", "-c", "trap : TERM INT; sleep infinity & wait"],
            "securityContext": {
                "allowPrivilegeEscalation": False,
                "capabilities": {"drop": ["ALL"]},
                "runAsNonRoot": True,
                "seccompProfile": {"type": "RuntimeDefault"}
            }
        }

        # Add ports to container spec if requested
        if args.expose:
            container_spec["ports"] = [{"containerPort": args.expose}]

        overrides = {
            "spec": {
                "serviceAccountName": sa_name,
                "containers": [container_spec]
            }
        }

        run_cmd = ["oc", "run", pod_name] + ns_args + [
            f"--image={image}",
            f"--labels={app_label}",
            "--restart=Never",
            f"--overrides={json.dumps(overrides)}"
        ]
        subprocess.run(run_cmd, check=True)

        print(f"[STAGE] Wait: Waiting for Pod/{pod_name} to reach 'Ready' state...")
        subprocess.run(["oc", "wait"] + ns_args + ["--for=condition=Ready", f"pod/{pod_name}", "--timeout=60s"], check=True)

        # Create a matching service if expose flag is provided
        if args.expose:
            print(f"[STAGE] Expose: Creating Service for port {args.expose}...")
            subprocess.run(["oc", "expose", "pod", pod_name, f"--port={args.expose}"] + ns_args)

    # --- Execute Script via oc exec ---
    print(f"[STAGE] Encoding: Base64 encoding local script '{args.script}'...")
    with open(args.script, "rb") as f:
        b64_script = base64.b64encode(f.read()).decode('utf-8')

    print("[STAGE] Execute: Transferring payload and triggering python3...")
    print("-" * 60)

    # Safely quote remaining arguments for bash evaluation
    safe_args = shlex.join(remaining)
    exec_payload = f"echo '{b64_script}' | base64 -d > /tmp/run.py && python3 /tmp/run.py {safe_args}"

    exec_cmd = ["oc", "exec", "-i", pod_name] + ns_args + ["--", "/bin/bash", "-c", exec_payload]
    
    # Subprocess.run directly pipes to standard out/err, mirroring bash behavior
    subprocess.run(exec_cmd)

    print("-" * 60)
    print("[STAGE] Finish: Execution complete.")

if __name__ == "__main__":
    main()

oc-python.sh

#!/bin/bash

# Configuration
APP_LABEL="app=oc-python-runner"
IMAGE="registry.redhat.io/rhel9/python-312"
SA_NAME="oc-python-runner"
NS_ARG=""

show_help() {
    echo "Usage: $0 [OPTIONS] <script.py> [args...]"
    echo ""
    echo "Options:"
    echo "  --namespace <ns>    Specify the OpenShift namespace"
    echo "  --destroy           Delete the resident runner pod"
    echo "  --help              Show this help message"
    echo ""
    echo "Example (Istio Bookinfo):"
    echo "  # Run a 10-iteration traffic test matching reviews versions"
    echo "  ./oc-python.sh purl.py http://productpage:9080/productpage -i 10 -m 'reviews-v[0-9]+'"
    echo ""
    echo "Note: The first run starts a persistent Pod. Subsequent runs use 'oc exec' for speed."
}

# --- Handle Flags ---
if [ "$#" -eq 0 ] || [ "$1" == "--help" ]; then
    show_help
    exit 0
fi

if [ "$1" == "--namespace" ]; then
    NS_ARG="-n $2"
    echo "[STAGE] Namespace context set to: $2"
    shift 2
fi

if [ "$1" == "--destroy" ]; then
    echo "[STAGE] Initialization: Cleaning up all oc-python-runner pods..."
    oc delete pod -l $APP_LABEL $NS_ARG
    echo "[DONE] Cleanup complete."
    exit 0
fi

# --- Validate Input ---
SCRIPT_PATH=$1
shift # Remaining args in $@
if [ ! -f "$SCRIPT_PATH" ]; then
    echo "[ERROR] Script '$SCRIPT_PATH' not found."
    show_help
    exit 1
fi


# --- Ensure ServiceAccount Exists ---
echo "[STAGE] Check: Verifying ServiceAccount '$SA_NAME'..."
if ! oc get sa $SA_NAME $NS_ARG &> /dev/null; then
    echo "[STAGE] Setup: Creating ServiceAccount '$SA_NAME'..."
    oc create sa $SA_NAME $NS_ARG
else
    echo "[STAGE] Check: ServiceAccount already exists."
fi

# --- Ensure Runner Pod Exists ---
echo "[STAGE] Check: Searching for existing resident runner..."
POD_NAME=$(oc get pods $NS_ARG -l $APP_LABEL -o jsonpath='{.items[?(@.status.phase=="Running")].metadata.name}' | awk '{print $1}')

if [ -z "$POD_NAME" ]; then
    echo "[STAGE] Deployment: No active runner found. Provisioning new Pod..."
    POD_NAME="oc-python-$(date +%s)"
    
    OVERRIDES='{
      "spec": {
        "serviceAccountName": "'$SA_NAME'",
        "containers": [{
          "name": "python-runner",
          "image": "'$IMAGE'",
          "command": ["/bin/bash", "-c", "trap : TERM INT; sleep infinity & wait"],
          "securityContext": {
            "allowPrivilegeEscalation": false,
            "capabilities": { "drop": ["ALL"] },
            "runAsNonRoot": true,
            "seccompProfile": { "type": "RuntimeDefault" }
          }
        }]
      }
    }'

    oc run $POD_NAME $NS_ARG --image=$IMAGE --labels=$APP_LABEL --restart=Never --overrides="$OVERRIDES"
    
    echo "[STAGE] Wait: Waiting for Pod/$POD_NAME to reach 'Ready' state..."
    oc wait $NS_ARG --for=condition=Ready pod/$POD_NAME --timeout=60s
else
    echo "[STAGE] Check: Found resident runner $POD_NAME. Reusing container..."
fi

# --- Execute Script via oc exec ---
echo "[STAGE] Encoding: Base64 encoding local script '$SCRIPT_PATH'..."
B64_SCRIPT=$(base64 -w 0 "$SCRIPT_PATH")
REMAINING_ARGS="$@"

echo "[STAGE] Execute: Transferring payload and triggering python3..."
echo "------------------------------------------------------------"

oc exec -i $POD_NAME $NS_ARG -- /bin/bash -c \
    "echo '$B64_SCRIPT' | base64 -d > /tmp/run.py && python3 /tmp/run.py $REMAINING_ARGS"

echo "------------------------------------------------------------"
echo "[STAGE] Finish: Execution complete."

purl.py

import urllib.request
import urllib.error
import time
import re
import argparse
import ssl

def purl(url, iterations, sleep_time, pattern, dump):
    stats = {}
    
    print(f"Targeting: {url}")
    if pattern:
        print(f"Pattern Tracking: {pattern}")
    print(f"{iterations} iterations, {sleep_time}s delay\n")
    
    if dump:
        print("Body Dumping: Enabled")
        
    print(f"{'#':<4} | {'Status':<6} | {'Latency':<10} | {'Result'}")
    print("-" * 60)
    
    context = ssl._create_unverified_context()
    
    for i in range(1, iterations + 1):
        tstart = time.perf_counter()
        try:
            with urllib.request.urlopen(url, timeout=5, context=context) as resp:
                content = resp.read().decode('utf-8')
                latency = (time.perf_counter() - tstart) * 1000
                status = resp.getcode()
                
                if pattern:
                    match = re.search(pattern, content)
                    val = match.group(0) if match else "No Match"
                    # Update dictionary statistics
                    stats[val] = stats.get(val, 0) + 1
                    result_text = val
                else:
                    result_text = f"{len(content)} bytes"
                
                print(f"{i:<4} | {status:<6} | {latency:>7.2f}ms | {result_text}")
                
                
                if dump:
                    print("-" * 20 + " BODY START " + "-" * 20)
                    print(content)
                    print("-" * 20 + "  BODY END  " + "-" * 20)
        except Exception as e:
            err_name = type(e).__name__
            stats[err_name] = stats.get(err_name, 0) + 1
            print(f"{i:<4} | Error  | {'N/A':>9} | {err_name}")

        if i < iterations:
            time.sleep(sleep_time)

    # Print Summary Table if matching was used
    if pattern and stats:
        print("\n" + "="*60)
        print(f"{'MATCHED VALUE':<40} | {'HITS':<6} | {'%'}")
        print("-" * 60)
        total = sum(stats.values())
        for key in sorted(stats.keys()):
            count = stats[key]
            percent = (count / total) * 100
            print(f"{key:<40} | {count:<6} | {percent:>5.1f}%")
        print("="*60)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("url")
    parser.add_argument("-i", "--iterations", type=int, default=1)
    parser.add_argument("-s", "--sleep", type=float, default=0.5)
    parser.add_argument("-m", "--match")
    parser.add_argument("-d", "--dump", action="store_true", help="Print response body to screen")
    
    args = parser.parse_args()
    purl(args.url, args.iterations, args.sleep, args.match, args.dump)

server.py

#!/usr/bin/env python3

"""
Header Dump Server for oc-python
--------------------------------
A lightweight HTTP server designed to be deployed via the oc-python wrapper.
It responds to GET and POST requests by echoing the pod's hostname and 
dumping all incoming HTTP request headers to both the client and the pod console.

Usage with oc-python:
1. Deploy and expose the server on port 8080:
   ./oc-python.py --runner header-dump-svc --expose 8080 server.py
   
2. Test the endpoint locally via port-forwarding:
   oc port-forward svc/header-dump-svc 8080:8080
   curl -H "X-Custom-Header: Test" http://localhost:8080

3. Or test internally from another pod in the same namespace:
   curl http://header-dump-svc:8080
"""

import os
import socket
import argparse
from http.server import HTTPServer, BaseHTTPRequestHandler

class DumpHeadersHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Fetch the pod name (hostname of the container)
        pod_name = socket.gethostname()

        # --- Dump request to the pod console ---
        print(f"\n--- Incoming {self.command} Request ---", flush=True)
        print(f"Client IP: {self.client_address[0]}", flush=True)
        print(f"Path: {self.path}", flush=True)
        print("=== Request Headers ===", flush=True)
        print(self.headers, flush=True)
        print("-------------------------------", flush=True)

        # Build the response text for the client
        response_text = f"Hello from {pod_name}\n\n"
        response_text += "=== Request Headers ===\n"
        response_text += str(self.headers)

        # Send a 200 OK HTTP response
        self.send_response(200)
        self.send_header('Content-type', 'text/plain; charset=utf-8')
        self.end_headers()

        # Write the payload back to the client
        self.wfile.write(response_text.encode('utf-8'))
        
    # Also support POST requests for easier debugging
    def do_POST(self):
        self.do_GET()

def run_server():
    parser = argparse.ArgumentParser(description="Simple header dumping server")
    parser.add_argument('--port', type=int, default=8080, help='Port to listen on (default: 8080)')
    args = parser.parse_args()

    # Bind to 0.0.0.0 to accept connections from outside the pod network
    server_address = ('0.0.0.0', args.port)
    httpd = HTTPServer(server_address, DumpHeadersHandler)
    
    print(f"Starting server on port {args.port}...", flush=True)
    print(f"Running inside pod/hostname: {socket.gethostname()}", flush=True)
    print("Waiting for requests...\n", flush=True)
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nShutting down server...", flush=True)
    finally:
        httpd.server_close()

if __name__ == '__main__':
    run_server()