Test tool for passing dummy json payloads between a client and server

cspayload.py

#!/usr/bin/env python3
"""
Simple HTTP Client/Server for JSON payload exchange

Basic Usage:
    Server mode: python cspayload.py server [options]
    Client mode: python cspayload.py client [options]

Server Options:
    --host HOST              Host to bind to (default: 0.0.0.0)
    --port PORT              Port to listen on (default: 8088)
    --payload FILE           Custom JSON payload file for responses
    --no-json                Respond with no JSON payload (empty body)
    --malformed              Respond with malformed JSON (random example)
    --malformed-custom STR   Respond with specific malformed JSON string

Client Options:
    --host HOST              Server host to connect to (default: localhost)
    --port PORT              Server port (default: 8088)
    --payload FILE           Custom JSON payload file to send
    --method METHOD          HTTP method: POST or GET (default: POST)
    --no-json                Send request with no JSON payload (empty body)
    --malformed              Send malformed JSON (random example)
    --malformed-custom STR   Send specific malformed JSON string
    --verbose, -v            Show debug information about what is being sent

Examples:
    # Start server with default settings
    python cspayload.py server
    
    # Start server that responds with malformed JSON
    python cspayload.py server --malformed
    
    # Send normal client request
    python cspayload.py client --host 10.0.2.50 --port 80
    
    # Send malformed JSON (use single quotes!)
    python cspayload.py client --malformed-custom '{"trailing": "comma",}'
    
    # Send request with no JSON payload
    python cspayload.py client --no-json
    
    # Debug what's being sent
    python cspayload.py client --malformed-custom '{"test": "value"}' --verbose

Note: When using --malformed-custom with JSON containing quotes, use single quotes
      on the outside: --malformed-custom '{"key": "value"}'
"""

import argparse
import json
import sys
import random
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.request import Request, urlopen
from urllib.error import URLError

# Default JSON payloads
DEFAULT_CLIENT_PAYLOAD = {
  "my_string": "Hello World",
  "my_number": 42,
  "my_boolean": True,
  "my_null": None,
  "my_array": [1, 2, 3],
  "my_object": {
    "nested_string": "I'm nested",
    "nested_array": ["a", "b", "c"]
  }
}

DEFAULT_SERVER_PAYLOAD = {
    "message": "Hello from server",
    "type": "response",
    "status": "success",
    "data": {
        "processed": True,
        "timestamp": "2026-01-29"
    }
}

# Malformed JSON examples
MALFORMED_JSON_EXAMPLES = [
    '{"incomplete": "missing closing brace"',
    '{"trailing": "comma",}',
    '{invalid: "no quotes on key"}',
    '{"bad_value": undefined}',
    '{"single": \'quotes\'}',
    'not json at all',
    '{"nested": {"missing": }}',
    '{"number": 01234}',  # Leading zeros
    '{"duplicate": 1, "duplicate": 2}',  # Duplicate keys (valid JSON but worth testing)
]


class JSONRequestHandler(BaseHTTPRequestHandler):
    """HTTP request handler for JSON payloads"""
    
    custom_response = None
    malformed_response = None
    no_json_response = False
    
    def do_POST(self):
        """Handle POST requests with JSON payload"""
        content_length = int(self.headers.get('Content-Length', 0))
        
        if content_length > 0:
            body = self.rfile.read(content_length)
            try:
                received_data = json.loads(body.decode('utf-8'))
                print("\n[Server] Received JSON payload:")
                print(json.dumps(received_data, indent=2))
            except json.JSONDecodeError as e:
                print(f"\n[Server] Received malformed JSON:")
                print(f"Raw data: {body.decode('utf-8', errors='replace')}")
                print(f"Error: {e}")
        else:
            print("\n[Server] Received empty POST request (no JSON payload)")
        
        # Send response
        self._send_response()
    
    def do_GET(self):
        """Handle GET requests"""
        print(f"\n[Server] Received GET request for path: {self.path}")
        self._send_response()
    
    def _send_response(self):
        """Send the configured response (normal, malformed, or no JSON)"""
        if self.no_json_response:
            # Send response with no JSON body
            print("[Server] Sending response with no JSON payload")
            self.send_response(200)
            self.send_header('Content-Length', '0')
            self.end_headers()
            
        elif self.malformed_response:
            # Send malformed JSON
            response_data = self.malformed_response
            print("[Server] Sending malformed JSON response:")
            print(f"Raw: {response_data}")
            
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(response_data)))
            self.end_headers()
            self.wfile.write(response_data.encode('utf-8'))
            
        else:
            # Send normal JSON response
            response_payload = self.custom_response if self.custom_response else DEFAULT_SERVER_PAYLOAD
            response_json = json.dumps(response_payload, indent=2)
            
            self.send_response(200)
            self.send_header('Content-Type', 'application/json')
            self.send_header('Content-Length', str(len(response_json)))
            self.end_headers()
            self.wfile.write(response_json.encode('utf-8'))
            
            print("[Server] Sent JSON response:")
            print(response_json)
    
    def log_message(self, format, *args):
        """Suppress default logging"""
        pass


def run_server(host, port, custom_payload=None, malformed=None, no_json=False):
    """Run HTTP server"""
    if custom_payload:
        JSONRequestHandler.custom_response = custom_payload
    
    if malformed:
        JSONRequestHandler.malformed_response = malformed
    
    if no_json:
        JSONRequestHandler.no_json_response = True
    
    server_address = (host, port)
    httpd = HTTPServer(server_address, JSONRequestHandler)
    
    print(f"[Server] Starting HTTP server on {host}:{port}")
    if no_json:
        print(f"[Server] Mode: NO JSON in responses")
    elif malformed:
        print(f"[Server] Mode: MALFORMED JSON in responses")
    else:
        print(f"[Server] Mode: Normal JSON responses")
    print(f"[Server] Press Ctrl+C to stop")
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n[Server] Shutting down...")
        httpd.shutdown()


def run_client(host, port, custom_payload=None, method='POST', no_json=False, 
               malformed=None, verbose=False):
    """Run HTTP client"""
    url = f"http://{host}:{port}/"
    
    print(f"[Client] Connecting to {url}")
    
    if verbose and malformed:
        print(f"[Client] DEBUG - Malformed string received by Python:")
        print(f"  Length: {len(malformed)} bytes")
        print(f"  Repr: {repr(malformed)}")
        has_quotes = '"' in malformed
        print(f"  Has quotes: {has_quotes}")
    
    try:
        if no_json:
            # Send request with no JSON payload
            print(f"[Client] Sending {method} request with NO JSON payload")
            
            if method == 'POST':
                request = Request(url, data=b'', headers={'Content-Length': '0'})
            else:  # GET
                request = Request(url)
                
        elif malformed:
            # Send malformed JSON
            print(f"[Client] Sending {method} request with MALFORMED JSON:")
            print(f"Raw: {malformed}")
            
            if method == 'POST':
                request = Request(url, data=malformed.encode('utf-8'), 
                                headers={'Content-Type': 'application/json'})
            else:
                print("[Client] Warning: GET requests don't typically have a body, sending as POST")
                method = 'POST'
                request = Request(url, data=malformed.encode('utf-8'),
                                headers={'Content-Type': 'application/json'})
                
        else:
            # Send normal JSON payload
            payload = custom_payload if custom_payload else DEFAULT_CLIENT_PAYLOAD
            print(f"[Client] Sending JSON payload ({method}):")
            print(json.dumps(payload, indent=2))
            
            if method == 'POST':
                json_data = json.dumps(payload).encode('utf-8')
                request = Request(url, data=json_data, headers={'Content-Type': 'application/json'})
            else:  # GET
                request = Request(url)
        
        with urlopen(request, timeout=10) as response:
            response_data = response.read().decode('utf-8')
            print(f"\n[Client] Received response (Status: {response.status}):")
            
            if not response_data:
                print("(empty response)")
            else:
                try:
                    response_json = json.loads(response_data)
                    print(json.dumps(response_json, indent=2))
                except json.JSONDecodeError as e:
                    print("Received malformed JSON:")
                    print(f"Raw: {response_data}")
                    print(f"Error: {e}")
    
    except URLError as e:
        print(f"\n[Client] Error: {e}")
        sys.exit(1)
    except Exception as e:
        print(f"\n[Client] Unexpected error: {e}")
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(description='Simple HTTP Client/Server for JSON payloads')
    subparsers = parser.add_subparsers(dest='mode', help='Mode: client or server')
    
    # Server arguments
    server_parser = subparsers.add_parser('server', help='Run as HTTP server')
    server_parser.add_argument('--host', default='0.0.0.0', help='Host to bind to (default: 0.0.0.0)')
    server_parser.add_argument('--port', type=int, default=8088, help='Port to listen on (default: 8088)')
    server_parser.add_argument('--payload', help='Custom JSON payload file for responses')
    server_parser.add_argument('--no-json', action='store_true', 
                              help='Respond with no JSON payload')
    server_parser.add_argument('--malformed', action='store_true',
                              help='Respond with malformed JSON (random example)')
    server_parser.add_argument('--malformed-custom', metavar='STRING',
                              help='Respond with specific malformed JSON string')
    
    # Client arguments
    client_parser = subparsers.add_parser('client', help='Run as HTTP client')
    client_parser.add_argument('--host', default='localhost', help='Server host (default: localhost)')
    client_parser.add_argument('--port', type=int, default=8088, help='Server port (default: 8088)')
    client_parser.add_argument('--payload', help='Custom JSON payload file to send')
    client_parser.add_argument('--method', choices=['POST', 'GET'], default='POST', 
                              help='HTTP method (default: POST)')
    client_parser.add_argument('--no-json', action='store_true',
                              help='Send request with no JSON payload')
    client_parser.add_argument('--malformed', action='store_true',
                              help='Send malformed JSON (random example)')
    client_parser.add_argument('--malformed-custom', metavar='STRING',
                              help='Send specific malformed JSON string')
    client_parser.add_argument('--verbose', '-v', action='store_true',
                              help='Show debug information about what is being sent')
    
    args = parser.parse_args()
    
    if not args.mode:
        parser.print_help()
        sys.exit(1)
    
    # Load custom payload if specified
    custom_payload = None
    if hasattr(args, 'payload') and args.payload:
        try:
            with open(args.payload, 'r') as f:
                custom_payload = json.load(f)
        except Exception as e:
            print(f"Error loading custom payload: {e}")
            sys.exit(1)
    
    # Determine malformed JSON to use
    malformed_json = None
    if hasattr(args, 'malformed_custom') and args.malformed_custom:
        malformed_json = args.malformed_custom
    elif hasattr(args, 'malformed') and args.malformed:
        malformed_json = random.choice(MALFORMED_JSON_EXAMPLES)
    
    # Check for conflicting options
    if hasattr(args, 'no_json') and args.no_json and malformed_json:
        print("Error: Cannot use both --no-json and --malformed/--malformed-custom")
        sys.exit(1)
    
    # Run in appropriate mode
    if args.mode == 'server':
        run_server(args.host, args.port, custom_payload, 
                  malformed=malformed_json, no_json=args.no_json)
    elif args.mode == 'client':
        verbose = args.verbose if hasattr(args, 'verbose') else False
        run_client(args.host, args.port, custom_payload, args.method,
                  no_json=args.no_json, malformed=malformed_json, verbose=verbose)


if __name__ == '__main__':
    main()