Changeset 11130 in ntrip for trunk/BNC/tests/tools/mkcrafted.py

Timestamp:

Oct 28, 2026, 9:32:30 AM (5 days ago)

Author:

stuerze

Message:

Some fixes from security point of view

File:

• trunk/BNC/tests/tools/mkcrafted.py (modified) (6 diffs)

trunk/BNC/tests/tools/mkcrafted.py

@@ -6 +6 @@
   CRAFT0IGS0  IGS-SSR VTEC (4076/201), decoded by SsrCorrIgs, followed by
               RTCM 1300/1301/1302 with 31-character names (BNC's RTCM scan
-              lists them only for streams that also decode successfully)
+              lists them only for streams that also decode successfully);
+              in the first epoch also two RTCM 1033 (antenna and receiver
+              descriptors): a complete one and one whose receiver descriptor
+              length (200) runs beyond the end of the message
   CRAFT0RTC0  RTCM-SSR VTEC (1264), decoded by SsrCorrRtcm
   CRAFT0NEW0  new RTCM-SSR metadata (60), not dispatched by BNC at present;
@@ -12 +15 @@
                 - 7 entries (maximum), additional parameter of 40 and 0 bits
                 - 31 entries (5-bit count, more than META_MAXNUMENTRIES=7)
+              and satellite antenna messages (DF+010-DF+021):
+                - BDS (83): C01, C17, C33, C49, C64 (mask bits in all four
+                  16-bit quarters of DF394), individual blocks
+                - GPS (80): G01, G02, G32 sharing one block and one frequency
+                  block; nadir angle corrections incl. the invalid value
 
 VTEC epochs (every 5 s):
@@ -22 +30 @@
 by eye; the heights are 100 km * layer number.
 
+With --rawfile, a raw file with hostile chunk headers is written instead
+(test of the raw file reader, data/crafted_rawfile.raw):
+  ../ESCAPE0  station ID with a path: RTCM-SSR VTEC whose .ssr file must be
+              written inside the output directory
+  CRAFT0RTC0  RTCM-SSR VTEC, 5 s later
+  then a chunk header announcing 2000000000 bytes, which must be rejected
+
 Usage: mkcrafted.py <out.raw>
+       mkcrafted.py --rawfile <out.raw>
 """
 
@@ -140 +156 @@
 
 
+def descriptors(strings, truncate=False):
+    """RTCM 1033: antenna, setup ID, serial, receiver, firmware, serial."""
+    w = BitWriter()
+    w.u(1033, 12)
+    w.u(0, 12)               # reference station ID
+    ant, antser, rec, firm, recser = strings
+    w.text(ant, 8)
+    w.u(0, 8)                # antenna setup ID
+    w.text(antser, 8)
+    if truncate:
+        w.u(200, 8)          # receiver descriptor length beyond the message
+        for c in rec.encode("ascii"):
+            w.u(c, 8)
+    else:
+        w.text(rec, 8)
+        w.text(firm, 8)
+        w.text(recser, 8)
+    return frame(w.payload())
+
+
 def metadata(entries):
     """entries: list of None (default model) or number of additional bits."""
@@ -163 +199 @@
 
 
+def antenna(msg, sats, shared, nadir):
+    """Satellite antenna message; sats: PRNs (1..64) of the mask."""
+    w = BitWriter()
+    w.u(msg, 12)
+    w.u(4711, 16)            # provider ID
+    w.u(5, 6)                # satellite antenna IOD
+    w.u(1, 1)                # phase center information
+    w.u(0, 1)                # group delay information
+    w.u(1, 1)                # nadir angle dependent corrections follow
+    w.u(3, 5)                # maximum off-nadir angle: corrections 0..3 deg
+    w.u(1, 4)                # range extension: 3 + 1 = 4 bits
+    w.u(shared, 1)           # satellite set indicator
+    mask = 0
+    for prn in sats:
+        mask |= 1 << (64 - prn)
+    w.u(mask, 64)
+    for n in range(1 if shared else len(sats)):
+        w.u(shared, 1)       # frequency set indicator
+        w.u(0b101000, 6)     # frequencies 0 and 2
+        for f in range(1 if shared else 2):
+            w.u(1, 1)        # nadir correction follows
+            w.s(100 * n + 10 * f + 1, 12)    # mm
+            for deg in range(4):
+                w.s(nadir[(n + f + deg) % len(nadir)], 4)   # mm, -8: invalid
+    return frame(w.payload())
+
+
+def rawfile(path):
+    chunks = []
+    for i, sta in enumerate(["../ESCAPE0", "CRAFT0RTC0"]):
+        t = T0 + datetime.timedelta(seconds=5 * i)
+        chunks.append((t.isoformat(), sta, "RTCM_3", vtec_rtcm(t, [(2, 2)])))
+    write_chunks(path, chunks)
+    t = T0 + datetime.timedelta(seconds=10)
+    with open(path, "ab") as o:
+        o.write(("%s CRAFT0RTC0 RTCM_3 2000000000\n" % t.isoformat()).encode())
+        o.write(vtec_rtcm(t, [(2, 2)]) + b"\n")
+
+
 def main():
+    if len(sys.argv) == 3 and sys.argv[1] == "--rawfile":
+        rawfile(sys.argv[2])
+        return
     if len(sys.argv) != 2:
         sys.exit(__doc__)
@@ -175 +253 @@
         t = T0 + datetime.timedelta(seconds=5 * i)
         stamp = t.isoformat()
-        chunks.append((stamp, "CRAFT0IGS0", "RTCM_3", vtec_igs(t, layers)
-                       + service_crs() + helmert() + rtcm_crs()))
+        data = vtec_igs(t, layers) + service_crs() + helmert() + rtcm_crs()
+        if i == 0:
+            data += descriptors(["LEIAR25.R4      LEIT", "12345", "SEPT POLARX5",
+                                 "5.5.0", "3001234"])
+            data += descriptors(["TRUNCATED ANTENNA", "9", "RECEIVER", "", ""],
+                                truncate=True)
+        chunks.append((stamp, "CRAFT0IGS0", "RTCM_3", data))
         chunks.append((stamp, "CRAFT0RTC0", "RTCM_3", vtec_rtcm(t, layers)))
     stamp = (T0 + datetime.timedelta(seconds=15)).isoformat()
     chunks.append((stamp, "CRAFT0NEW0", "RTCM_3",
-                   metadata([40, 0] + [None] * 5) + metadata([None] * 31)))
+                   metadata([40, 0] + [None] * 5) + metadata([None] * 31)
+                   + antenna(83, [1, 17, 33, 49, 64], 0, [-7, 0, 7, 3])
+                   + antenna(80, [1, 2, 32], 1, [5, -8, -1, 2])))
     write_chunks(sys.argv[1], chunks)