histsdk/scripts/Run-DebianHistorianRelayCapture.ps1

param(
    [string]$SshUser = "dohertj2",
    [string]$SshHost = "10.100.0.35",
    [string]$RelayListenHost = "0.0.0.0",
    [int]$RelayListenPort = 32568,
    [string]$TargetHost = "10.100.0.48",
    [int]$TargetPort = 32568,
    [switch]$RewriteEndpointHost,
    [ValidateSet("history", "event")]
    [string]$Scenario = "history",
    [string]$TagName = "OtOpcUaParityTest_001.Counter",
    [string]$RetrievalMode = "Full",
    [int]$LookbackMinutes = 1440,
    [int]$MaxRows = 1,
    [UInt64]$ResolutionTicks = 0,
    [string[]]$HarnessExtraArgs = @(),
    [string]$OutputPath = $null,
    [string]$HarnessOutputPath = $null
)

$ErrorActionPreference = "Stop"

$repoRoot = Split-Path -Parent $PSScriptRoot
$stamp = Get-Date -Format "yyyyMMdd-HHmmss"
if ([string]::IsNullOrWhiteSpace($OutputPath)) {
    $OutputPath = Join-Path $repoRoot "docs\reverse-engineering\debian-relay-$Scenario-$stamp.ndjson"
}

if ([string]::IsNullOrWhiteSpace($HarnessOutputPath)) {
    $HarnessOutputPath = Join-Path $repoRoot "docs\reverse-engineering\native-trace-harness-via-debian-relay-$Scenario-$stamp.json"
}

$localPy = Join-Path $env:TEMP "histsdk-relay-$stamp.py"
$remotePy = "/tmp/histsdk-relay-$stamp.py"
$remoteLog = "/tmp/histsdk-relay-$stamp.ndjson"
$sshTarget = "$SshUser@$SshHost"

$relaySource = @'
import argparse, json, selectors, socket, time
parser = argparse.ArgumentParser()
parser.add_argument('--listen', default='0.0.0.0')
parser.add_argument('--listen-port', type=int, default=32568)
parser.add_argument('--target', required=True)
parser.add_argument('--target-port', type=int, default=32568)
parser.add_argument('--prefix', type=int, default=16)
parser.add_argument('--rewrite-from', default='')
parser.add_argument('--rewrite-to', default='')
args = parser.parse_args()
sel = selectors.DefaultSelector()
peers = {}
conn_id = 0

def emit(kind, **payload):
    payload['kind'] = kind
    payload['ts'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
    print(json.dumps(payload, separators=(',', ':')), flush=True)

def safe_ascii(data):
    out = []
    for b in data:
        if 32 <= b <= 126:
            out.append(chr(b))
        elif b in (9, 10, 13):
            out.append(' ')
        else:
            out.append('.')
    return ''.join(out)

def classify(data):
    if data.startswith(b'\x00\x01\x00\x01\x02\x02') and b'net.tcp:/' in data:
        return 'nettcp-preamble'
    if data == b'\x0a':
        return 'nettcp-preamble-ack'
    if data.startswith(b'\x16\x03'):
        return 'tls-record'
    if len(data) >= 9 and data[5:13] == b'NTLMSSP\x00':
        return 'ntlmssp'
    if len(data) >= 11 and data[7:15] == b'NTLMSSP\x00':
        return 'ntlmssp'
    if data.startswith(b'\x09\x15application/'):
        return 'nettcp-content-type'
    return 'unknown'

def safe_detail(data):
    kind = classify(data)
    if kind in ('nettcp-preamble', 'nettcp-content-type'):
        return safe_ascii(data)
    return None

rewrite_from = args.rewrite_from.encode('ascii') if args.rewrite_from else b''
rewrite_to = args.rewrite_to.encode('ascii') if args.rewrite_to else b''
if rewrite_from and len(rewrite_from) != len(rewrite_to):
    raise SystemExit('rewrite strings must have equal byte length')

def accept(sock):
    global conn_id
    client, addr = sock.accept()
    client.setblocking(False)
    upstream = socket.create_connection((args.target, args.target_port), timeout=8)
    upstream.setblocking(False)
    conn_id += 1
    cid = conn_id
    peers[client] = (upstream, cid, 'c2s')
    peers[upstream] = (client, cid, 's2c')
    sel.register(client, selectors.EVENT_READ, relay)
    sel.register(upstream, selectors.EVENT_READ, relay)
    emit('connect', id=cid, client=str(addr), target=f'{args.target}:{args.target_port}')

def close_pair(sock):
    peer_info = peers.pop(sock, None)
    if peer_info is None:
        return
    other, cid, direction = peer_info
    peers.pop(other, None)
    for s in (sock, other):
        try: sel.unregister(s)
        except Exception: pass
        try: s.close()
        except Exception: pass
    emit('close', id=cid, direction=direction)

def relay(sock):
    peer_info = peers.get(sock)
    if peer_info is None:
        return
    other, cid, direction = peer_info
    try:
        data = sock.recv(65536)
    except Exception as exc:
        emit('recv_error', id=cid, direction=direction, error=repr(exc))
        close_pair(sock)
        return
    if not data:
        close_pair(sock)
        return
    if direction == 'c2s' and rewrite_from and rewrite_from in data:
        data = data.replace(rewrite_from, rewrite_to)
        emit('rewrite', id=cid, direction=direction, from_value=args.rewrite_from, to_value=args.rewrite_to)
    detail = safe_detail(data)
    event = {
        'id': cid,
        'direction': direction,
        'bytes': len(data),
        'classification': classify(data),
        'prefix': data[:args.prefix].hex()
    }
    if detail is not None:
        event['safe_ascii'] = detail
    emit('data', **event)
    try:
        other.sendall(data)
    except Exception as exc:
        emit('send_error', id=cid, direction=direction, error=repr(exc))
        close_pair(sock)

listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind((args.listen, args.listen_port))
listener.listen(16)
listener.setblocking(False)
sel.register(listener, selectors.EVENT_READ, accept)
emit('listening', listen=f'{args.listen}:{args.listen_port}', target=f'{args.target}:{args.target_port}')
while True:
    for key, _ in sel.select(timeout=1):
        key.data(key.fileobj)
'@

Push-Location $repoRoot
try {
    New-Item -ItemType Directory -Force -Path (Split-Path -Parent $OutputPath) | Out-Null
    New-Item -ItemType Directory -Force -Path (Split-Path -Parent $HarnessOutputPath) | Out-Null
    Set-Content -LiteralPath $localPy -Value $relaySource -Encoding UTF8
    scp -q $localPy "${sshTarget}:$remotePy"

    $rewriteArgs = ""
    if ($RewriteEndpointHost) {
        if ($SshHost.Length -ne $TargetHost.Length) {
            throw "RewriteEndpointHost requires SshHost and TargetHost to have equal string length."
        }
        $rewriteArgs = " --rewrite-from $SshHost --rewrite-to $TargetHost"
    }

    $remoteCommand = "nohup python3 $remotePy --listen $RelayListenHost --listen-port $RelayListenPort --target $TargetHost --target-port $TargetPort$rewriteArgs > $remoteLog 2>&1 < /dev/null & echo `$!"
    $relayPid = (ssh -o BatchMode=yes -o ConnectTimeout=8 $sshTarget $remoteCommand).Trim()
    Start-Sleep -Seconds 1
    ssh -o BatchMode=yes $sshTarget "cat $remoteLog" | Tee-Object -FilePath $OutputPath | Out-Host

    $harnessArgs = @(
        "--scenario", $Scenario,
        "--server-name", $SshHost,
        "--tag", $TagName,
        "--retrieval-mode", $RetrievalMode,
        "--lookback-minutes", $LookbackMinutes.ToString(),
        "--max-rows", $MaxRows.ToString()
    )

    if ($ResolutionTicks -gt 0) {
        $harnessArgs += @("--resolution-ticks", $ResolutionTicks.ToString())
    }

    if ($HarnessExtraArgs.Count -gt 0) {
        $harnessArgs += $HarnessExtraArgs
    }

    & .\tools\AVEVA.Historian.NativeTraceHarness\bin\Debug\net481\AVEVA.Historian.NativeTraceHarness.exe @harnessArgs |
        Set-Content -Path $HarnessOutputPath -Encoding UTF8
    $harnessExit = $LASTEXITCODE

    ssh -o BatchMode=yes $sshTarget "cat $remoteLog" | Set-Content -Path $OutputPath -Encoding UTF8
    exit $harnessExit
}
finally {
    try {
        if ($relayPid) {
            ssh -o BatchMode=yes $sshTarget "kill $relayPid 2>/dev/null || true; rm -f $remotePy" | Out-Null
        }
    }
    catch {
        Write-Warning "Relay cleanup failed: $_"
    }

    Remove-Item -LiteralPath $localPy -ErrorAction SilentlyContinue
    Pop-Location
}