247 lines
9.1 KiB
Python
247 lines
9.1 KiB
Python
"""
|
|
Contec CMS7000PLUS / CMS9200PLUS Discovery Tool
|
|
==================================
|
|
This script listens on multiple TCP ports simultaneously to discover
|
|
what data the CMS7000PLUS / CMS9200PLUS patient monitor sends over the network.
|
|
|
|
Usage:
|
|
python contec_discovery.py
|
|
|
|
It will listen on ports: 511, 515, 516, 517, 518, 519, 520, 6060
|
|
and display all received data in hex + ASCII format for protocol analysis.
|
|
|
|
IMPORTANT: Run as Administrator (required for port 511 on Windows)
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
import os
|
|
from datetime import datetime
|
|
|
|
# --- Configuration ---
|
|
LISTEN_HOST = "0.0.0.0"
|
|
DISCOVERY_PORTS = [511, 515, 516, 517, 518, 519, 520, 6060]
|
|
LOG_DIR = "captures"
|
|
|
|
# --- Setup Logging ---
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s [%(levelname)s] %(message)s',
|
|
handlers=[logging.StreamHandler(sys.stdout)]
|
|
)
|
|
logger = logging.getLogger("contec_discovery")
|
|
|
|
|
|
def hex_dump(data: bytes, width: int = 16) -> str:
|
|
"""Format bytes as a hex dump with ASCII representation."""
|
|
lines = []
|
|
for i in range(0, len(data), width):
|
|
chunk = data[i:i + width]
|
|
hex_part = ' '.join(f'{b:02x}' for b in chunk)
|
|
ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
|
|
lines.append(f" {i:04x} {hex_part:<{width * 3}} |{ascii_part}|")
|
|
return '\n'.join(lines)
|
|
|
|
|
|
def try_decode_hl7(data: bytes) -> str | None:
|
|
"""Attempt to decode the data as an HL7 message."""
|
|
try:
|
|
# Strip MLLP framing if present
|
|
cleaned = data
|
|
if cleaned.startswith(b'\x0b'):
|
|
cleaned = cleaned[1:]
|
|
if b'\x1c\x0d' in cleaned:
|
|
cleaned = cleaned[:cleaned.find(b'\x1c\x0d')]
|
|
|
|
text = cleaned.decode('utf-8', errors='ignore').strip()
|
|
if 'MSH|' in text or 'MSH' in text[:20]:
|
|
return text
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
async def handle_discovery_client(
|
|
reader: asyncio.StreamReader,
|
|
writer: asyncio.StreamWriter,
|
|
port: int,
|
|
capture_file
|
|
):
|
|
"""Handle a single incoming connection and dump all data."""
|
|
client_ip, client_port = writer.get_extra_info('peername')
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
|
|
|
|
print("\n" + "=" * 80)
|
|
print(f" NEW CONNECTION on port {port}")
|
|
print(f" From: {client_ip}:{client_port}")
|
|
print(f" Time: {timestamp}")
|
|
print("=" * 80)
|
|
|
|
capture_file.write(f"\n{'='*80}\n")
|
|
capture_file.write(f"Connection on port {port} from {client_ip}:{client_port} at {timestamp}\n")
|
|
capture_file.write(f"{'='*80}\n")
|
|
capture_file.flush()
|
|
|
|
packet_count = 0
|
|
|
|
try:
|
|
while True:
|
|
data = await asyncio.wait_for(reader.read(8192), timeout=30.0)
|
|
if not data:
|
|
break
|
|
|
|
packet_count += 1
|
|
recv_time = datetime.now().strftime("%H:%M:%S.%f")
|
|
|
|
print(f"\n--- Packet #{packet_count} on port {port} [{recv_time}] ({len(data)} bytes) ---")
|
|
print(hex_dump(data))
|
|
|
|
# Try to decode as HL7
|
|
hl7_text = try_decode_hl7(data)
|
|
if hl7_text:
|
|
print(f"\n >>> HL7 MESSAGE DETECTED <<<")
|
|
segments = hl7_text.replace('\n', '\r').split('\r')
|
|
for seg in segments:
|
|
seg = seg.strip()
|
|
if seg:
|
|
print(f" {seg}")
|
|
|
|
# Extract vital signs from OBX segments
|
|
if seg.startswith("OBX|"):
|
|
fields = seg.split('|')
|
|
if len(fields) > 5:
|
|
obs_id = fields[3] if len(fields) > 3 else "?"
|
|
obs_value = fields[5] if len(fields) > 5 else "?"
|
|
obs_unit = fields[6] if len(fields) > 6 else ""
|
|
print(f" -> Vital: {obs_id} = {obs_value} {obs_unit}")
|
|
|
|
# Also try plain text decode
|
|
try:
|
|
text = data.decode('utf-8', errors='ignore')
|
|
printable_ratio = sum(1 for c in text if c.isprintable() or c in '\r\n\t') / max(len(text), 1)
|
|
if printable_ratio > 0.7 and not hl7_text:
|
|
print(f"\n [Text representation]:")
|
|
for line in text.split('\r'):
|
|
line = line.strip()
|
|
if line:
|
|
print(f" {line}")
|
|
except Exception:
|
|
pass
|
|
|
|
# Write to capture file
|
|
capture_file.write(f"\nPacket #{packet_count} [{recv_time}] ({len(data)} bytes)\n")
|
|
capture_file.write(hex_dump(data) + "\n")
|
|
if hl7_text:
|
|
capture_file.write(f"HL7: {hl7_text}\n")
|
|
capture_file.write(f"Raw bytes: {data.hex()}\n")
|
|
capture_file.flush()
|
|
|
|
except asyncio.TimeoutError:
|
|
print(f" [Timeout - no data for 30s on port {port}]")
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except ConnectionResetError:
|
|
print(f" [Connection reset by {client_ip} on port {port}]")
|
|
except Exception as e:
|
|
print(f" [Error on port {port}: {e}]")
|
|
finally:
|
|
print(f"\n Connection closed on port {port} from {client_ip} ({packet_count} packets received)")
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def start_port_listener(port: int, capture_file):
|
|
"""Start a TCP server on a specific port."""
|
|
async def client_handler(reader, writer):
|
|
await handle_discovery_client(reader, writer, port, capture_file)
|
|
|
|
try:
|
|
server = await asyncio.start_server(client_handler, LISTEN_HOST, port)
|
|
logger.info(f" ✓ Listening on port {port}")
|
|
return server
|
|
except PermissionError:
|
|
logger.error(f" ✗ Port {port} - Permission denied (try running as Administrator)")
|
|
return None
|
|
except OSError as e:
|
|
if "already in use" in str(e).lower() or e.errno == 10048:
|
|
logger.error(f" ✗ Port {port} - Already in use by another application")
|
|
else:
|
|
logger.error(f" ✗ Port {port} - {e}")
|
|
return None
|
|
|
|
|
|
async def main():
|
|
print(r"""
|
|
╔══════════════════════════════════════════════════════════════╗
|
|
║ Contec CMS7000PLUS / CMS9200PLUS Discovery Tool ║
|
|
║ ║
|
|
║ This tool listens for incoming data from your patient ║
|
|
║ monitor to discover its communication protocol. ║
|
|
║ ║
|
|
║ SETUP REQUIRED: ║
|
|
║ 1. Connect monitor to your PC via Ethernet ║
|
|
║ 2. Set your PC's Ethernet IP to: 192.168.1.50 ║
|
|
║ (or whatever subnet your monitor uses) ║
|
|
║ 3. On the monitor, set CMS/Server IP to YOUR PC's IP ║
|
|
║ 4. Run this script as Administrator ║
|
|
║ ║
|
|
║ Press Ctrl+C to stop ║
|
|
╚══════════════════════════════════════════════════════════════╝
|
|
""")
|
|
|
|
# Create captures directory
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
|
capture_filename = os.path.join(
|
|
LOG_DIR,
|
|
f"capture_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
|
|
)
|
|
capture_file = open(capture_filename, 'w', encoding='utf-8')
|
|
logger.info(f"Capture log: {capture_filename}")
|
|
|
|
# Show current network config hint
|
|
print("\nStarting listeners on all discovery ports...")
|
|
print("(If port 511 fails, run PowerShell as Administrator)\n")
|
|
|
|
servers = []
|
|
for port in DISCOVERY_PORTS:
|
|
server = await start_port_listener(port, capture_file)
|
|
if server:
|
|
servers.append(server)
|
|
|
|
if not servers:
|
|
logger.error("No ports could be opened! Check permissions and existing services.")
|
|
capture_file.close()
|
|
return
|
|
|
|
active_ports = [s.sockets[0].getsockname()[1] for s in servers if s.sockets]
|
|
print(f"\n{'─' * 60}")
|
|
print(f" Active on ports: {active_ports}")
|
|
print(f" Waiting for data from CMS7000PLUS / CMS9200PLUS...")
|
|
print(f" Make sure the monitor's CMS/Server IP points to this PC")
|
|
print(f"{'─' * 60}\n")
|
|
|
|
try:
|
|
# Keep running until Ctrl+C
|
|
await asyncio.Event().wait()
|
|
except asyncio.CancelledError:
|
|
pass
|
|
finally:
|
|
print("\nShutting down...")
|
|
for server in servers:
|
|
server.close()
|
|
for server in servers:
|
|
await server.wait_closed()
|
|
capture_file.close()
|
|
print(f"Capture saved to: {capture_filename}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
print("\n\nDiscovery stopped by user.")
|