source: ntrip/trunk/BNC/tests/tools/net_check.py

#!/usr/bin/env python3
"""Network checks of BNC on the loopback interface (no root needed):

  UDP sender filter: a fake server sends datagrams marked GOOD, a second
  sender on another loopback address datagrams marked SPOOFED to the same
  BNC port. BNC's raw output (rawOutFile) must contain only GOOD data:
    'U'   Ntrip 2 UDP mode (server 127.0.0.3)
    'R'   RTSP/RTP mode (RTSP server 127.0.0.3)
    'UN'  UDP without Ntrip, stream URL with sender 127.0.0.5
  and with the URL host 127.0.0.1 ('UN' as written by the 'Add Stream -
  Coming from UDP Port' dialog) data of any sender (both GOOD and SPOOFED).

  Ntrip 1 response header: an endless header is cut off ("Caster response
  header too long"), a normal Ntrip 1 stream still works.

  Upload status: a caster answering 'HTTP/1.1 401 Unauthorized' with a
  header that contains '200' must not be taken as success.

    tests/tools/net_check.py [--bnc PATH]      exit code 1 if a check fails
"""

import argparse
import glob
import os
import re
import shutil
import socket
import subprocess
import sys
import threading
import time

TOOLS = os.path.dirname(os.path.abspath(__file__))
TESTS = os.path.dirname(TOOLS)
ROOT = os.path.dirname(TESTS)
WORK = os.path.join(TESTS, "work", "net_check")
UPLOAD_RAW = os.path.join(TESTS, "data", "upload_rtnet_5min.raw")
SERVER, SPOOFER, UN_SENDER, UN_OTHER = "127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6"
DURATION = 8.0  # seconds of data
RTP_HEADER = bytes([128, 96]) + bytes(10)


def udp_socket(addr):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.bind((addr, 0))
    return s


def send_both(good, spoof, dest, header=b""):
    """GOOD from 'good', SPOOFED from 'spoof' to dest, alternating."""
    t0 = time.time()
    while time.time() - t0 < DURATION:
        good.sendto(header + b"GOOD" * 8, dest)
        spoof.sendto(header + b"SPOOFED!" * 4, dest)
        time.sleep(0.2)


def fake_udp_server(port):
    """Ntrip 2 UDP mode: request datagram -> reply with session -> data."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    srv.bind((SERVER, port))
    srv.settimeout(20)
    try:
        req, client = srv.recvfrom(4096)
    except socket.timeout:
        return
    srv.sendto(RTP_HEADER + b"HTTP/1.1 200 OK\r\nSession: 4711\r\n\r\n", client)
    send_both(srv, udp_socket(SPOOFER), client, RTP_HEADER)


def fake_rtsp_server(port):
    """RTSP: SETUP -> PLAY on TCP, then RTP data to the client port."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((SERVER, port))
    srv.listen(1)
    srv.settimeout(20)
    try:
        conn, peer = srv.accept()
    except socket.timeout:
        return
    conn.settimeout(10)
    rtp = udp_socket(SERVER)

    def request():
        buf = b""
        while b"\r\n\r\n" not in buf:
            d = conn.recv(4096)
            if not d:
                return None
            buf += d
        return buf.decode("latin-1")

    setup = request()
    cport = int(re.search(r"client_port=(\d+)", setup).group(1))
    conn.sendall(("RTSP/1.0 200 OK\r\nCSeq: 1\r\nSession: 4711\r\n"
                  "Transport: RTP/GNSS;unicast;client_port=%d;server_port=%d\r\n\r\n"
                  % (cport, rtp.getsockname()[1])).encode())
    request()  # PLAY
    conn.sendall(b"RTSP/1.0 200 OK\r\nCSeq: 2\r\nSession: 4711\r\n\r\n")
    send_both(rtp, udp_socket(SPOOFER), (peer[0], cport), RTP_HEADER)
    conn.close()


def fake_ntrip1_caster(port, endless):
    """Ntrip 1: endless response header, or ICY 200 OK followed by data."""
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((SERVER, port))
    srv.listen(5)
    srv.settimeout(1)
    t0 = time.time()
    while time.time() - t0 < DURATION:
        try:
            conn, _ = srv.accept()
        except socket.timeout:
            continue
        conn.settimeout(5)
        try:
            conn.recv(4096)
            if endless:
                t1 = time.time()
                while time.time() - t1 < 3:
                    conn.sendall(b"X-Padding: " + b"x" * 1000 + b"\r\n")
            else:
                conn.sendall(b"ICY 200 OK\r\n\r\n")
                while time.time() - t0 < DURATION:
                    conn.sendall(b"GOOD" * 8)
                    time.sleep(0.2)
        except OSError:
            pass
        conn.close()


def bnc_live(bnc, name, mountpoint, server=None):
    """Run BNC with one stream for DURATION + a few seconds; returns its raw
    output and log."""
    d = os.path.join(WORK, name)
    os.makedirs(os.path.join(d, "Output"))
    env = dict(os.environ, QT_QPA_PLATFORM="offscreen", LC_ALL="C", TZ="UTC")
    cmd = [bnc, "--nw", "--platform", "offscreen",
           "--conf", os.path.join(ROOT, "Example_Configs", "22_Empty.bnc"),
           "-key", "mountPoints", mountpoint,
           "-key", "rawOutFile", "Output/raw",
           "-key", "logFile", "Output/bnc.log"]
    with open(os.path.join(d, "bnc.out"), "wb") as out:
        p = subprocess.Popen(cmd, cwd=d, stdout=out, stderr=subprocess.STDOUT, env=env)
        if server:
            server()
        else:
            time.sleep(DURATION + 2)
        time.sleep(1)
        p.send_signal(2)
        p.wait(30)
    raw = b"".join(open(f, "rb").read() for f in glob.glob(os.path.join(d, "Output", "raw*")))
    log = "".join(open(f, errors="replace").read()
                  for f in glob.glob(os.path.join(d, "Output", "bnc.log*")))
    return raw, log


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--bnc", default=os.path.join(ROOT, "bnc"))
    args = ap.parse_args()
    bnc = os.path.abspath(args.bnc)
    shutil.rmtree(WORK, ignore_errors=True)
    failed = 0

    def check(name, ok, detail):
        nonlocal failed
        print("%-6s %-44s %s" % ("PASS" if ok else "FAIL", name, detail), flush=True)
        failed += not ok

    def counts(raw):
        return raw.count(b"GOOD" * 8), raw.count(b"SPOOFED!" * 4)

    def detail(raw, log):
        good, spoof = counts(raw)
        reconnects = log.count("reconnecting")
        return "%d GOOD, %d SPOOFED datagrams, %d reconnect(s)" % (good, spoof, reconnects)

    # 'U': Ntrip 2 in UDP mode
    port = 21211
    raw, log = bnc_live(bnc, "udp_ntrip", "//user:pass@%s:%d/TEST RTCM_3 DEU 50 8 no U" % (SERVER, port),
                        lambda: fake_udp_server(port))
    good, spoof = counts(raw)
    check("U: only datagrams of the caster", good > 0 and spoof == 0 and "reconnecting" not in log,
          detail(raw, log))

    # 'R': RTSP/RTP
    port = 21212
    raw, log = bnc_live(bnc, "rtsp", "//user:pass@%s:%d/TEST RTCM_3 DEU 50 8 no R" % (SERVER, port),
                        lambda: fake_rtsp_server(port))
    good, spoof = counts(raw)
    check("R: only RTP datagrams of the RTSP server", good > 0 and spoof == 0 and "reconnecting" not in log,
          detail(raw, log))

    # 'UN' with sender address: only that sender
    port = 21213
    sender = lambda: (time.sleep(2), send_both(udp_socket(UN_SENDER), udp_socket(UN_OTHER),
                                               ("127.0.0.1", port)))
    raw, log = bnc_live(bnc, "udp_raw_sender", "//%s:%d/TEST RTCM_3 DEU 50 8 no UN" % (UN_SENDER, port),
                        sender)
    good, spoof = counts(raw)
    check("UN with sender address: only that sender", good > 0 and spoof == 0, detail(raw, log))

    # 'UN' as written by the dialog (127.0.0.1): any sender, as before
    port = 21214
    sender = lambda: (time.sleep(2), send_both(udp_socket(UN_SENDER), udp_socket(UN_OTHER),
                                               ("127.0.0.1", port)))
    raw, log = bnc_live(bnc, "udp_raw_any", "//127.0.0.1:%d/TEST RTCM_3 DEU 50 8 no UN" % port, sender)
    good, spoof = counts(raw)
    check("UN with 127.0.0.1: any sender", good > 0 and spoof > 0, detail(raw, log))

    # Ntrip 1: endless response header / normal stream
    port = 21216
    raw, log = bnc_live(bnc, "ntrip1_endless", "//user:pass@%s:%d/TEST RTCM_3 DEU 50 8 no 1" % (SERVER, port),
                        lambda: fake_ntrip1_caster(port, True))
    check("Ntrip 1: endless response header cut off", "response header too long" in log and not raw,
          "%d bytes stored" % len(raw))
    port = 21217
    raw, log = bnc_live(bnc, "ntrip1_stream", "//user:pass@%s:%d/TEST RTCM_3 DEU 50 8 no 1" % (SERVER, port),
                        lambda: fake_ntrip1_caster(port, False))
    good, spoof = counts(raw)
    check("Ntrip 1: normal stream", good > 0, "%d GOOD blocks" % good)

    # Upload: rejection whose header contains "200"
    d = os.path.join(WORK, "upload_401")
    os.makedirs(os.path.join(d, "Output"))
    os.symlink(os.path.join(ROOT, "Example_Configs", "Input"), os.path.join(d, "Input"))
    fc = subprocess.Popen(
        [sys.executable, "-I", os.path.join(TOOLS, "fakecaster.py"), "--port", "21215",
         "--out", os.path.join(d, "Output", "upload.raw"), "--time", "2026-10-01T09:33:00",
         "--timeout", "8",
         "--reply", "HTTP/1.1 401 Unauthorized\\r\\nContent-Length: 200\\r\\n\\r\\n"],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    fc.stdout.readline()
    env = dict(os.environ, QT_QPA_PLATFORM="offscreen", LC_ALL="C", TZ="UTC")
    with open(os.path.join(d, "bnc.out"), "wb") as out:
        subprocess.run([bnc, "--nw", "--platform", "offscreen",
                        "--conf", os.path.join(ROOT, "Example_Configs", "16_Upload.bnc"),
                        "--file", UPLOAD_RAW, "-key", "uploadMountpointsOut",
                        "127.0.0.1,21215,CLOCK,2,user,pass,IGS20,IGS-SSR,0,,,,,,,0 byte(s),"],
                       cwd=d, stdout=out, stderr=subprocess.STDOUT, env=env, timeout=300)
    caster = fc.communicate(timeout=120)[0]
    log = "".join(open(f, errors="replace").read() for f in glob.glob(os.path.join(d, "Output", "*.log*")))
    opened, broken = log.count("Connection opened"), log.count("Connection broken")
    check("upload: 401 with '200' in header rejected", opened == 0 and broken > 0,
          "%d 'opened', %d 'broken' (%s)" % (opened, broken, caster.strip().splitlines()[-1]))

    print("%d check(s) failed, details in %s" % (failed, os.path.relpath(WORK, ROOT)))
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())