CellGuard/common/tools/cg_decode.py

#!/usr/bin/env python3
"""
cg_decode.py — decode Shannon CP->AP frames captured by cgcap.ko, and flag
IMSI-catcher / cell-site-simulator (CSS) signatures.

Input: text from /proc/cgcap/frames (RX lines). Each RX line ends with the raw
hex of skb->data captured at io_dev_recv_skb_single_from_link_dev.

Frame layering (RE'd + validated against 234 live frames):
  0x00 u16 magic  = 0xABCD        0x08 u8  channel (0xf6 = umts_ipc1 SIT)
  0x02 u16 counter                0x09 u8  counter2
  0x04 u16 0xC000                 0x0a u16 0x0000
  0x06 u16 length                 --- SIT message starts at frame byte 0x0c ---
  SIT: 0x0c type | 0x0e id(hi=grp,lo=cmd) | 0x10 len | 0x12 rsvd | params @0x14 (indication)

Cell-info list (SIT id 0x070F, NET) params (relative to params @0x14):
  +0x00 u16 sit_len  +0x04 u32 count  +0x08 record[0]
  record = u8 rat | u8 registered | u8 conn_status | struct  (stride incl. header)
  rat: 0=GSM 1=CDMA 2=LTE 3=WCDMA 4=TDSCDMA 5=NR ; stride LTE=0x13C NR=0x18C GSM=0xA4
All struct offsets proven from vendor.radio.protocol.sit.stream.so fill fns.
"""
import sys, re

def u16(b, o): return b[o] | (b[o+1] << 8) if o+1 < len(b) else 0
def u32(b, o): return int.from_bytes(b[o:o+4], 'little') if o+4 <= len(b) else 0
def u64(b, o): return int.from_bytes(b[o:o+8], 'little') if o+8 <= len(b) else 0
def s32(b, o):
    v = u32(b, o); return v - (1 << 32) if v & 0x80000000 else v
def astr(b, o, n):
    return b[o:o+n].split(b'\x00')[0].decode('ascii', 'replace').replace('#', '')

RAT   = {0: 'GSM', 1: 'CDMA', 2: 'LTE', 3: 'WCDMA', 4: 'TDSCDMA', 5: 'NR'}
STRIDE = {0: 0xA4, 1: 0x2B, 2: 0x13C, 3: 0x114, 4: 0x110, 5: 0x18C}
WEAK_RAT = {'GSM', 'WCDMA', 'TDSCDMA', 'CDMA'}   # 2G/3G = downgrade risk

SIT_NAMES = {
    0x070C: "NET/GetCellInfoList", 0x070E: "NET/RegState", 0x070F: "NET/CellInfoList",
    0x0742: "MISC/0x42", 0x0906: "PS/0x06", 0x0702: "MISC/0x02",
}

def parse_lte(s):
    return dict(rat='LTE', mcc=astr(s,0,3), mnc=astr(s,3,3), ci=u32(s,6), pci=u32(s,0xa),
                tac=u32(s,0xe), earfcn=u32(s,0x12), op=astr(s,0x1a,32),
                rsrp=-s32(s,0x121) if len(s)>0x124 else None,
                rsrq=s32(s,0x125) if len(s)>0x128 else None)

def parse_nr(s):
    return dict(rat='NR', mcc=astr(s,0,3), mnc=astr(s,3,3), nci=u64(s,6), pci=u32(s,0xe),
                tac=u32(s,0x12), arfcn=u32(s,0x16), op=astr(s,0x1a,32),
                rsrp=-s32(s,0x119) if len(s)>0x11c else None,
                sinr=s32(s,0x121) if len(s)>0x124 else None)

def parse_gsm(s):
    return dict(rat='GSM', mcc=astr(s,0,3), mnc=astr(s,3,3), lac=u32(s,6), cid=u32(s,0xa),
                arfcn=u32(s,0xe), bsic=s[0x12] if len(s)>0x12 else None, op=astr(s,0x13,32),
                rssi=s32(s,0x95) if len(s)>0x98 else None)

PARSER = {0: parse_gsm, 2: parse_lte, 5: parse_nr}

def decode_cellinfo(params):
    """params = SIT params (frame[0x14:]); list = len@0, count@4, records@8."""
    count = u32(params, 4)
    cells, off = [], 8
    while off + 3 < len(params) and len(cells) < count and count < 64:
        rat = params[off]; reg = params[off+1]
        stride = STRIDE.get(rat, 0)
        if not stride:
            break
        struct = params[off+3: off+stride]
        c = PARSER.get(rat, lambda s: {'rat': RAT.get(rat, rat)})(struct)
        c['registered'] = bool(reg)
        cells.append(c)
        off += stride
    return cells

def load_rx(text):
    out = []
    for ln in text.splitlines():
        m = re.match(r'^(\d+)\s+RX\s+ch=0x([0-9a-f]+).*?\s([0-9a-f]{8,})', ln)
        if m:
            h = m.group(3); out.append((int(m.group(1)), bytes.fromhex(h[:len(h)-len(h)%2])))
    return out

def heuristics(cells, serving_plmn):
    """Return CSS/IMSI-catcher flags for a cell-info snapshot."""
    alerts = []
    serving = [c for c in cells if c.get('registered')]
    for c in serving:
        if c.get('rat') in WEAK_RAT:
            alerts.append(f"DOWNGRADE: serving on {c['rat']} (2G/3G) — CSS downgrade signature")
        pl = (c.get('mcc', '') + c.get('mnc', ''))
        if serving_plmn and pl and pl != serving_plmn:
            alerts.append(f"PLMN MISMATCH: serving cell PLMN {pl} != expected {serving_plmn}")
    if serving and not [c for c in cells if not c.get('registered')]:
        alerts.append("NO NEIGHBORS: empty neighbor list — CSS often suppresses neighbors")
    return alerts

def main():
    text = open(sys.argv[1]).read() if len(sys.argv) > 1 else sys.stdin.read()
    frames = load_rx(text)
    ids, plmns, cellframes = {}, set(), 0
    print(f"RX frames: {len(frames)}")
    for seq, b in frames:
        if len(b) < 16 or u16(b, 0) != 0xABCD:
            continue
        sid = u16(b, 0x0e)
        ids[sid] = ids.get(sid, 0) + 1
        params = b[0x14:]
        for p in re.findall(rb'(\d{6})', params):
            plmns.add(p.decode())
        if sid == 0x070F:
            cellframes += 1
            cells = decode_cellinfo(params)
            sp = next(iter(plmns), None)
            print(f"\n[seq {seq}] CellInfoList: {len(cells)} cells")
            for c in cells:
                tag = "SERVING" if c.get('registered') else "neighbor"
                print(f"  {tag} {c.get('rat'):6} PLMN={c.get('mcc','')}{c.get('mnc','')} "
                      f"pci={c.get('pci')} tac={c.get('tac')} earfcn={c.get('earfcn') or c.get('arfcn')} "
                      f"rsrp={c.get('rsrp')} op={c.get('op','')!r}")
            for a in heuristics(cells, sp):
                print(f"  !! {a}")
    print(f"\nPLMNs: {sorted(plmns)}   0x070F cell frames: {cellframes}")
    print("SIT id counts:", {f"0x{k:04x}": v for k, v in sorted(ids.items(), key=lambda x: -x[1])[:12]})
    if cellframes == 0:
        print("\n(no 0x070F cell-list frames captured yet — trigger getCellInfo to exercise it)")

if __name__ == '__main__':
    main()