Skip to main content

Universal Protocol Analysis & Simulation Framework

Project description

๐ŸŽฏ UPAS - Universal Protocol Analysis & Simulation

UPAS Logo

Version Python License Status CI/CD Pipeline

Advanced Protocol Reverse Engineering & Network Service Emulation Platform

๐Ÿš€ Quick Start โ€ข ๐Ÿ“– Documentation โ€ข ๐ŸŽฎ Examples โ€ข ๐Ÿ› ๏ธ Development


๐ŸŽฏ What is UPAS?

UPAS (Universal Protocol Analysis & Simulation) is a production-ready framework for protocol reverse engineering, network service emulation, and advanced protocol testing. Originally designed for industrial protocol analysis, UPAS enables security researchers, network engineers, and developers to:

  • ๐Ÿ” Reverse engineer complex network protocols with dynamic pattern matching
  • ๐ŸŽญ Emulate sophisticated network services with multi-packet response capabilities
  • ๐Ÿญ Simulate industrial protocols (Modbus, CAN, custom IoT systems)
  • ๐Ÿ›ก๏ธ Test security vulnerabilities in network stacks and devices
  • ๐Ÿ“ก Develop protocol analyzers with real-time variable extraction
  • ๐ŸŽช Create behavior simulations with intelligent state management

โœจ Key Features

๐ŸŽฏ Advanced Pattern Matching

  • Dynamic variable capture: Extract values with [CAPTURE:VAR:size]
  • Wildcard operations: Skip content with [SKIP:n] and [WILDCARD:n]
  • Binary pattern support: Hex patterns with mixed static/dynamic content
  • Real-time analysis: Immediate pattern matching during packet processing

๐Ÿš€ Multi-Packet Response System

  • Sequential responses: Ordered packets with ACK validation and retry logic
  • Burst responses: Parallel transmission for complex protocols
  • Delayed sequences: Custom timing control for realistic simulation
  • Response strategies: Single, sequence, burst, and delayed modes

๐Ÿ”„ Intelligent State Management

  • Behavior-driven transitions: State changes triggered by execution success
  • State-only behaviors: Transitions without packets (timeouts, delays)
  • Conditional filtering: Behaviors active only in specific protocol phases
  • Entry/exit actions: Automatic behavior execution on state changes

๐ŸŒ Service-Aware Transport

  • Automatic routing: TCP/UDP service selection via transport layer
  • Multi-service support: UDP unicast/multicast, TCP client/server
  • Advanced socket management: Robust connection handling with fallbacks
  • Interface control: Multi-homed systems with interface binding

๐Ÿš€ Quick Start

Installation

UPAS is designed for minimal dependencies with optional feature sets:

# ๐ŸŽฏ Minimal Installation (Recommended - Zero dependencies)
pip install upas

# ๐Ÿ”ฌ With Analysis Features (PCAP support)
pip install upas[analysis]

# ๐ŸŒ With IoT Features (MQTT, Serial)
pip install upas[iot]

# ๐Ÿš€ Full Installation (All features)
pip install upas[full]

๐Ÿ“‹ Installation Modes: See the WIKI for detailed information about each installation mode and their specific use cases.

Basic Usage

# Run protocol with clean output (default)
upas run examples/simple_beacon.json

# Verbose logging with execution details
upas run examples/heartbeat_monitor.json -v

# Full debug logging for development
upas run examples/iot_discovery.json -d

# Quiet mode (errors only)
upas run examples/simple_beacon.json -q

# Run with specific network interface
upas run examples/modbus_simulation.json --interface eth0

Simple Protocol Example

{
  "protocol": {
    "name": "UDP_Beacon",
    "version": "1.0",
    "description": "Basic UDP beacon example"
  },
  "variables": {
    "MESSAGE": "Hello UPAS",
    "COUNTER": 0
  },
  "functions": {
    "increment": "lambda x: (x + 1) % 0xFF"
  },
  "transports": {
    "ethernet": {
      "type": "ethernet",
      "services": {
        "udp_service": {
          "type": "udp_unicast",
          "bind": "0.0.0.0:12345"
        }
      }
    }
  },
  "behaviors": {
    "beacon": {
      "type": "periodic",
      "interval": 2000,
      "transport": "ethernet",
      "destination": "127.0.0.1:12346",
      "payload": ["BEACON", "[MESSAGE]", "[COUNTER:1:increment]"]
    }
  }
}

๐Ÿ Programmatic API

UPAS provides a powerful Python API for integrating protocol simulation into your applications.

Basic Usage

import asyncio
import upas

async def main():
    # Run protocol for 30 seconds
    manager = await upas.run_protocol('protocol.json', duration=30)

    # Check current state
    print(f"Current state: {manager.get_current_state()}")

    # Get protocol variables
    variables = manager.get_variables()
    print(f"Variables: {variables}")

asyncio.run(main())

Advanced Protocol Control

import upas

async def advanced_example():
    # Create protocol manager
    manager = upas.ProtocolManager('protocol.json')

    # Register state change callbacks
    def on_connected():
        print("Protocol connected! Switching to monitoring mode...")
        manager.set_variable("MODE", "MONITORING")

    manager.on_state_change("CONNECTED", on_connected)

    # Start protocol
    await manager.start_async(duration=60)

    # Dynamic state transition
    success = manager.transition_to_state("AUTHENTICATED")
    if success:
        print("Successfully transitioned to AUTHENTICATED state")

    # Stop when done
    manager.stop()

Dynamic Protocol Switching

import upas

async def protocol_switching():
    # Start with discovery protocol
    manager = await upas.run_protocol('discovery.json')

    # Register protocol change callback
    def on_protocol_change(new_protocol):
        print(f"Switched to: {new_protocol['protocol']['name']}")

    manager.on_protocol_change("switch", on_protocol_change)

    # Wait for discovery completion
    await asyncio.sleep(10)

    # Switch to operational protocol
    await upas.change_protocol(manager, 'operational.json')

    # Continue with new protocol
    await asyncio.sleep(30)
    manager.stop()

Engine-Level Control

import upas

async def engine_control():
    # Create and configure engine directly
    engine = await upas.create_engine('protocol.json')

    # Start engine
    await engine.start()

    # Monitor execution
    while engine.is_running():
        stats = engine.get_statistics()
        print(f"State: {stats['current_state']}, Behaviors: {stats['behaviors']}")
        await asyncio.sleep(1)

    # Stop engine
    await engine.stop()

API Reference

Function Description
run_protocol() High-level protocol execution
ProtocolManager() Advanced protocol control class
load_protocol() Load protocol from file
create_engine() Create engine instance
transition_to_state() Force state transition
change_protocol() Dynamic protocol switching

๐ŸŽฎ Examples

๐Ÿ”ง Protocol Testing & Simulation

# Modbus RTU simulation
upas run examples/modbus_simulation.json -v

# Custom IoT protocol with state machine
upas run examples/iot_discovery.json -d

๐ŸŒ Network Service Emulation

# IoT device discovery protocol
upas run examples/iot_discovery.json

# Sensor network simulation
upas run examples/sensor_network.json

๐Ÿ“ก Advanced Protocol Analysis

# Multi-phase protocol testing
upas run examples/advanced_protocol_example.json

# Protocol composition chains
upas run protocols/compositions/hybrid_discovery.json

๐Ÿ“– Documentation

๐Ÿ“š Core Documentation

๐ŸŽฏ Key Concepts

Pattern Matching Syntax

"payload_pattern": "[PREFIX:8][CAPTURE:SESSION_ID:4][SKIP:8]DATA[SUFFIX:4]"

Multi-Packet Responses

"response": {
  "mode": "sequence",
  "packets": [
    {"id": "ack", "payload": ["ACK[SESSION_ID:4]"], "timeout": 5.0},
    {"id": "data", "payload": ["DATA[RESPONSE:*]"], "delay": 0.1}
  ]
}

State Machine Control

"active_states": ["DISCOVERING", "CONNECTED"],
"transition": "AUTHENTICATED"

๐Ÿ—๏ธ Protocol Development

Quick Start Template

{
  "protocol": {
    "name": "Your_Protocol",
    "version": "1.0",
    "description": "Protocol description"
  },
  "variables": {
    "PORT": 8080,
    "MESSAGE": "Hello World"
  },
  "transports": {
    "ethernet": {
      "type": "ethernet",
      "services": {
        "udp_service": {
          "type": "udp_unicast",
          "bind": "0.0.0.0:[PORT]"
        }
      }
    }
  },
  "behaviors": {
    "periodic_beacon": {
      "type": "periodic",
      "interval": 1000,
      "transport": "ethernet",
      "destination": "127.0.0.1:8081",
      "payload": ["[MESSAGE]"]
    }
  }
}

Protocol Validation

# Validate single protocol
upas validate examples/simple_beacon.json

# Validate all protocols in directory
upas validate protocols/

# Validate with verbose output
upas validate examples/ -v

๐Ÿ› ๏ธ Development

Project Structure

upas-cli/
โ”œโ”€โ”€ src/upas/              # Core UPAS framework
โ”‚   โ”œโ”€โ”€ core/              # Engine, behaviors, transport
โ”‚   โ”‚   โ”œโ”€โ”€ behavior/      # Behavior system
โ”‚   โ”‚   โ”œโ”€โ”€ transport/     # Network transport layer
โ”‚   โ”‚   โ””โ”€โ”€ protocol/      # Protocol engine
โ”‚   โ”œโ”€โ”€ analysis/          # Protocol analysis tools
โ”‚   โ””โ”€โ”€ cli.py             # Command-line interface
โ”œโ”€โ”€ protocols/             # Standard protocol library
โ”‚   โ”œโ”€โ”€ behaviors/         # Reusable behavior components
โ”‚   โ”œโ”€โ”€ transports/        # Transport configurations
โ”‚   โ””โ”€โ”€ compositions/      # Protocol compositions
โ”œโ”€โ”€ examples/              # Usage examples
โ”œโ”€โ”€ docs/                  # Documentation
โ””โ”€โ”€ tests/                 # Test suite

Installation for Development

# Clone repository
git clone https://github.com/BitsDiver/upas-cli.git
cd upas-cli

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac
# or
venv\\Scripts\\activate   # Windows

# Install in development mode
pip install -e .
pip install -r requirements-dev.txt

# Run tests
python -m pytest tests/ -v

# Run with coverage
python -m pytest --cov=src/upas tests/

Contributing

  1. Fork the repository
  2. Create feature branch (git checkout -b feature/amazing-feature)
  3. Commit changes (git commit -m 'Add amazing feature')
  4. Push to branch (git push origin feature/amazing-feature)
  5. Open Pull Request

Testing

# Run unit tests
python -m pytest tests/

# Run integration tests
python -m pytest tests/integration/

# Run specific test module
python -m pytest tests/test_engine.py -v

# Generate coverage report
python -m pytest --cov=src/upas --cov-report=html tests/

๐ŸŽฏ Use Cases

๐Ÿ›ก๏ธ Security Research

  • IoT penetration testing - Emulate vulnerable devices for security assessment
  • Protocol fuzzing - Generate malformed packets for vulnerability discovery
  • Network reconnaissance - Service discovery and protocol fingerprinting

๐Ÿญ Industrial Automation

  • HMI simulation - Test SCADA interfaces and human-machine interactions
  • PLC emulation - Industrial protocol testing and commissioning
  • Network validation - Protocol compliance and interoperability testing

๐Ÿ”ฌ Research & Development

  • Protocol prototyping - Rapid development of custom network protocols
  • Network simulation - Large-scale network behavior modeling
  • Educational tools - Network protocol education and training

๐Ÿ“ž Support & Community

๐Ÿค Getting Help

๐ŸŽ“ Learning Resources


๐Ÿ“„ License

This project is licensed under the MIT License - see the LICENSE file for details.

MIT License

Copyright (c) 2025 BitsDiver

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

๐Ÿ™ Acknowledgments

UPAS was developed through extensive real-world testing with industrial protocols and IoT devices. Special thanks to the cybersecurity and industrial automation communities for their feedback and contributions.


๐ŸŽฏ UPAS v1.0.14 - Production Ready

โœจ Advanced Protocol Analysis & Simulation Platform โœจ

๐Ÿš€ Get Started โ€ข ๐Ÿ“– Documentation โ€ข ๐ŸŽฎ Examples โ€ข ๐Ÿค Contribute


Developed with โค๏ธ for the network protocol analysis community

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

upas-1.0.14.tar.gz (137.5 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

upas-1.0.14-py3-none-any.whl (99.8 kB view details)

Uploaded Python 3

File details

Details for the file upas-1.0.14.tar.gz.

File metadata

  • Download URL: upas-1.0.14.tar.gz
  • Upload date:
  • Size: 137.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.18

File hashes

Hashes for upas-1.0.14.tar.gz
Algorithm Hash digest
SHA256 e05a63ee8ec581a235b7b7420aaf503f2110a4587d45863e84f6e357f7f4ad6a
MD5 e07439468e7cc0891296b729c3fd98a7
BLAKE2b-256 b276e150cbec0580d0bd87258d6e1233f941bfc6b0abf5420e4163a726d6ec71

See more details on using hashes here.

File details

Details for the file upas-1.0.14-py3-none-any.whl.

File metadata

  • Download URL: upas-1.0.14-py3-none-any.whl
  • Upload date:
  • Size: 99.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.10.18

File hashes

Hashes for upas-1.0.14-py3-none-any.whl
Algorithm Hash digest
SHA256 9e2f75f115c1633e72f2aa65c9b27718ae4c0b328a29aba441633ac27c1698d4
MD5 a934ef30a4c0271785e235ad523d58a5
BLAKE2b-256 f6d8d14537e9aa521b9049fc29b5bd4e32da5a86fde0ab69683f5e0fda7a5c56

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page