at_surface.py

#!/usr/bin/env python3
"""
Cartographie la surface AT d'un modem joignable en SSH.

Deux sources qui se completent:
  - statique: quelles commandes le service hote utilise vraiment, avec
    fichier:ligne. Le resultat cartographie du code tiers, donc il reste en
    local, on ne publie que la partie live.
  - live: quelles commandes le modem accepte, teste en forme "test"
    (AT+XXX=?) qui est en LECTURE SEULE et change aucun etat

Le live passe par SSH sur l'hote auquel le modem est rattache: c'est du
monitoring, pas un vecteur d'exploit.

Usage:
    python3 at_surface.py --rootfs /mnt/rootfs                  # statique
    python3 at_surface.py --ssh root@<hote>                  # live
    AT_SSH_PASS=xxx python3 at_surface.py --ssh root@<hote>  # live, par mdp
    python3 at_surface.py --rootfs /mnt/rootfs --ssh root@<hote> --json surface.json
"""

import argparse
import json
import os
import re
import subprocess
import sys

# at_core vit a cote des annotateurs, pas ici. Les deux copies sont
# identiques, on prend celle qu'on trouve.
HERE = os.path.dirname(os.path.abspath(__file__))
for scripts in (os.path.join(HERE, 'Ghidra', 'scripts'),
                os.path.join(HERE, 'IDA', 'scripts')):
    if os.path.isfile(os.path.join(scripts, 'at_core.py')):
        sys.path.insert(0, scripts)
        break
import at_core

# La table des familles sensibles vit dans at_core (partagee avec les
# annotateurs IDA/Ghidra).
RISKY = at_core.RISKY

# Commandes sondees en live, en plus de celles trouvees dans le rootfs.
PROBE_EXTRA = [
    'QLINUXCMD', 'QFLST', 'QFDWL', 'QFUPL', 'QFDEL', 'QFOTADL', 'QFIRMWARE',
    'QNVR', 'QNVW', 'QPRTPARA', 'QADBKEY', 'QDMEE', 'QSCLK', 'QTEMP',
    'QGMR', 'QCCID', 'QSIMSTAT', 'CSIM', 'CRSM', 'EGMR', 'CFUN', 'QCFG',
    'CGDCONT', 'CMGS', 'CMGL', 'CUSD', 'CGSN', 'CIMI', 'CSQ', 'COPS',
]

CMD_RE = re.compile(r'"\+([A-Z]{2,10})')


def scan_rootfs(rootfs):
    """Quelles commandes AT le code hote utilise, et ou."""
    base = os.path.join(rootfs, 'opt/SolidSense/modem_gps')
    found = {}
    if not os.path.isdir(base):
        print('[!] %s introuvable' % base)
        return found
    for name in sorted(os.listdir(base)):
        if not name.endswith('.py'):
            continue
        path = os.path.join(base, name)
        with open(path, 'r', errors='replace') as f:
            for num, line in enumerate(f, 1):
                for m in CMD_RE.finditer(line):
                    cmd = m.group(1)
                    found.setdefault(cmd, []).append('%s:%d' % (name, num))
    return found


REMOTE_PROBE = r'''
import serial, time, sys, json
cmds = json.loads(sys.stdin.readline())
s = serial.Serial("/dev/ttyUSB2", 115200, timeout=1)

def send(cmd, wait=0.6):
    s.reset_input_buffer()
    s.write((cmd + "\r\n").encode())
    time.sleep(wait)
    data = b""
    idle = 0
    while idle < 6:
        n = s.in_waiting
        if n:
            data += s.read(n)
            idle = 0
        else:
            idle += 1
            time.sleep(0.1)
    return data.decode("utf-8", "replace")

send("ATE0")
out = {}
for c in cmds:
    # forme test uniquement : lecture seule, ne change aucun etat
    r = send("AT+%s=?" % c)
    if "OK" in r:
        out[c] = "supporte"
    elif "CME ERROR" in r or "ERROR" in r:
        out[c] = "refuse"
    else:
        out[c] = "sans reponse"
print("###JSON###")
print(json.dumps(out))
'''


def probe_live(target, cmds):
    """Sonde le modem via SSH. Forme test seulement."""
    payload = json.dumps(cmds)
    remote = 'python3 -c %s' % shell_quote(REMOTE_PROBE)
    # StrictHostKeyChecking off: une cible reflashee souvent change de cle
    # a chaque fois.
    cmd = ['ssh', '-o', 'StrictHostKeyChecking=no',
           '-o', 'ConnectTimeout=10', target, remote]
    # par defaut ssh se debrouille avec une cle ou l'agent. AT_SSH_PASS si la
    # cible veut un mot de passe, ca evite de le trainer dans le fichier.
    passwd = os.environ.get('AT_SSH_PASS')
    if passwd:
        cmd = ['sshpass', '-p', passwd] + cmd
    proc = subprocess.run(
        cmd, input=payload + '\n', capture_output=True, text=True, timeout=600)
    out = proc.stdout
    if '###JSON###' not in out:
        print('[!] sonde live KO :')
        print(proc.stderr.strip()[-500:] or out[-500:])
        return {}
    return json.loads(out.split('###JSON###', 1)[1].strip())


def shell_quote(text):
    return "'" + text.replace("'", "'\\''") + "'"


def report(static, live):
    cmds = sorted(set(list(static.keys()) + list(live.keys())))
    print('\n=== surface AT ===')
    print('%-12s %-12s %-9s %s' % ('commande', 'live', 'hote', 'note'))
    supported = 0
    risky_ok = []
    for cmd in cmds:
        state = live.get(cmd, '-')
        used = 'oui' if cmd in static else '-'
        cat, note = RISKY.get(cmd, ('', ''))
        if state == 'supporte':
            supported += 1
            if cat:
                risky_ok.append((cmd, cat, note))
        label = ('[%s] %s' % (cat, note)) if cat else ''
        print('%-12s %-12s %-9s %s' % ('+' + cmd, state, used, label))

    print('\n%d commandes supportees par le modem sur %d sondees'
          % (supported, len([c for c in cmds if c in live])))
    print('%d familles utilisees par le service hote' % len(static))
    if risky_ok:
        print('\nsupportees ET sensibles')
        for cmd, cat, note in risky_ok:
            print('   +%-10s %-9s %s' % (cmd, cat, note))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--rootfs', help='rootfs monte de la cible')
    ap.add_argument('--ssh', help='cible ssh, ex root@hote')
    ap.add_argument('--json', help='fichier de sortie')
    args = ap.parse_args()

    static = scan_rootfs(args.rootfs) if args.rootfs else {}
    if static:
        print('[+] %d familles AT trouvees dans le code hote' % len(static))

    live = {}
    if args.ssh:
        cmds = sorted(set(list(static.keys()) + PROBE_EXTRA))
        print('[*] sondage live de %d commandes (forme test, lecture seule)...'
              % len(cmds))
        live = probe_live(args.ssh, cmds)

    report(static, live)

    if args.json:
        with open(args.json, 'w') as f:
            json.dump({'statique': static, 'live': live}, f, indent=1)
        print('\n[+] ecrit dans %s' % args.json)
    return 0


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