⬢ DragonFlyBSD Kernel Audit
DF-5724 / drive.py
← back to finding ↓ download raw 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#!/usr/bin/env python3
"""DF-5724 — drive DDB over the socket serial console: unvalidated $tabstops.

Bug under test (sys/ddb/db_output.c + sys/ddb/db_variables.c):
  `set $tabstops=N` stores N into db_tab_stop_width with NO validation
  (db_variables.c:52,120-128).  db_force_whitespace() then computes
      next_tab = NEXT_TAB(last_print);                  // db_output.c:80
      #define NEXT_TAB(i) rounddown((i) + db_tab_stop_width, db_tab_stop_width)
  and rounddown(x,y) = ((x)/(y))*(y)  (sys/sys/param.h:400) has NO zero
  guard.

  variant A: tabstops=0  -> integer division by zero -> #DE inside DDB on
             the very next printing character that flushes whitespace
             (e.g. the 'd' of the next "db> " prompt, which has a pending
             trailing space).
  variant B: tabstops=-8 -> next_tab goes negative, `next_tab <= pos` holds,
             the inner `while (last_print < next_tab)` never advances and
             the outer `while (last_print < db_output_position)` spins
             forever: hard hang of the stopped-CPU debugger.
             Trigger: leading-space output at column 0, e.g. `print/d 5`
             -> db_printf("%11ld") emits 10 spaces then '5' at last_non=0.

usage: drive.py A|B
"""
import os, socket, sys, threading, time

SOCK  = os.environ.get("SOCK", "/tmp/opencode/df-5724-ser.sock")
DRAIN = os.environ.get("DRAIN", "/tmp/opencode/df-5724-drain-{}.log")

def connect():
    while True:
        try:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            s.connect(SOCK)
            return s
        except (FileNotFoundError, ConnectionRefusedError) as e:
            print(f"connect failed: {e}; retrying...", flush=True)
            time.sleep(2)

def send_raw(s, data, chunk=8, pace=0.15):
    for i in range(0, len(data), chunk):
        s.sendall(data[i:i+chunk])
        time.sleep(pace)

def send_line(s, line, wait=3):
    print(f"\n>>> {line!r}", flush=True)
    send_raw(s, line.encode() + b"\n")
    time.sleep(wait)

def main():
    variant = sys.argv[1].upper()
    s = connect()
    drain = DRAIN.format(variant.lower())
    def drainer(sock):
        with open(drain, "ab", buffering=0) as d:
            while True:
                try:
                    b = sock.recv(65536)
                except OSError:
                    return
                if not b:
                    return
                d.write(b)
                sys.stdout.write(b.decode("latin1")); sys.stdout.flush()
    threading.Thread(target=drainer, args=(s,), daemon=True).start()
    time.sleep(2)

    send_line(s, "x/x db_active", 3)        # control: output works, tab math live

    if variant == "A":
        send_line(s, "set $tabstops=0", 4)  # silent success; next prompt 'd' -> #DE
        send_line(s, "x/x db_active", 5)    # (probably never reached)
    else:
        send_line(s, "set $tabstops=-8", 4) # accepted
        send_line(s, "print/d 5", 8)        # leading spaces at col 0 -> hang
        send_line(s, "x/x db_active", 4)    # (never reached: CPU spinning)
    print("\n>>> session script complete", flush=True)
    time.sleep(2)

if __name__ == "__main__":
    main()