9a3e3a0501
- cgcap.ko: cpif CP-frame capture via kprobe/kallsyms, stock-GKI build - ratlock.sh: hold modem to LTE+NR(5G), block 2G/3G downgrade - cgdetect.sh: always-on behavioral cell-site-simulator detection - cgctl.sh + WebUI: block/detect toggles, status + alerts - build-stock-gki.sh: reproducible stock-GKI module build pipeline
130 lines
5.7 KiB
Python
130 lines
5.7 KiB
Python
#!/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()
|