TWCTF 2017 - Liar's Trap

0server.rb

#!/usr/bin/env ruby
require 'securerandom'
## Parameters
P = 115792089237316195423570985008687907853269984665640564039457584007913129639747
N = 100
K = 25
L = 38 # The number of liars

def apply_polynomial(coeffs, x)
    r = 0
    coeffs.inject(0) do |r, c|
        r = (r * x + c) % P
    end
end

def create_polynomial
    coeff = SecureRandom.hex(64 * K)
    coeff.chars.each_slice(128).map{|a|a.join.to_i(16) % P}
end

secret = SecureRandom.hex(64).to_i(16) % P
polynomial = create_polynomial
polynomial[-1] = secret

# Distribute Secret
user = Array.new(N) {|i| apply_polynomial(polynomial, i + 1)  }

# Some user are liars
[*0...N].shuffle[0, L].each{|i| user[i] = rand(P) }

STDOUT.puts "--- Distributed Secret ---"
N.times do |i|
    STDOUT.puts "User #{i}: #{user[i]}"
end
STDOUT.flush
STDOUT.puts

STDOUT.puts "What's secret? "
STDOUT.flush
input_secret = STDIN.gets.to_i
if input_secret == secret
    STDOUT.puts "OK. I'll give you the flag"
    STDOUT.puts File.read("flag").chomp
else
    STDOUT.puts "Wrong"
end
STDOUT.flush
STDOUT.close

1worker.py

#!/usr/bin/pypy

"""
In the challenge we have the Shamir's Secret Sharing scheme.
The secret is splitted into 100 shares with threshold 25.
However, 38 shares are trashed (the "liars").

This is related to Reed-Solomon codes decoding.
But since this is a PPC challenge, we can go bruteforce!

Indeed, the fraction of 25-element subsets without liars is quite high:

sage: math.log(  binomial(100-38, 26) / binomial(100, 26)  , 2)
-21.669242478193773

To verify that interpolation is good, we can simply wait until the decoded value is repeated twice.
For wrong decodings this is highly unlikely.

That is, we need to make around 2 x 2^21 = 4 millions interpolations.
The challenge server gives us 30 seconds. Maybe we won't fit, but trying a few times is not a big deal.

Additionally, let's optimize the interpolation. Instead of choosing subsets randomly each time,
we can modify a previous subset e.g. by replacing a single point in it.
This can improve the time from quadratic to linear per single interpolation.

In a few tries the script succeeds and gets the flag:

TWCTF{Error_correction_to_Shamir's_Secret_Sharing}
"""

from random import randint, choice, shuffle
from libnum import invmod

P = 115792089237316195423570985008687907853269984665640564039457584007913129639747
N = 100
K = 25
L = 38 # The number of liars


class Interpol(object):
    """
    Lagrange Interpolation for the constant term only.
    Allows fast replacement of a single point (in linear instead of quadratic time)
    """
    def __init__(self, lst):
        self.xset = {v[0] for v in lst}
        self.xs = [v[0] for v in lst]
        self.ys = [v[1] for v in lst]
        self.vecs = []
        self.n = len(lst)
        for j in xrange(self.n):
            top = 1
            bot = 1
            for i in xrange(self.n):
                if i == j:
                    continue
                top *= -self.xs[i]
                top %= P
                bot *= (self.xs[j] - self.xs[i])
                bot %= P
            self.vecs.append(top * invmod(bot, P) % P)
            self.vecs[-1] %= P

    def ans(self):
        return sum(y * v % P for y, v in zip(self.ys, self.vecs)) % P

    def replace(self, jj, xy):
        xnew, ynew = xy
        assert (self.xs[jj] == xnew) ^ (self.ys[jj] == ynew) ^ True
        if self.xs[jj] == xnew:
            return False
        if xnew in self.xset:
            return False
        self.xset.remove(self.xs[jj])
        self.xset.add(xnew)

        xold = self.xs[jj]
        yold = self.ys[jj]

        for j in xrange(self.n):
            if j == jj:
                continue
            # self.vecs[j] /= (-xold) / (self.xs[j] - xold)
            # self.vecs[j] *= (-xnew) / (self.xs[j] - xnew)
            top = (-xnew) * (self.xs[j] - xold) % P
            bot = (-xold) * (self.xs[j] - xnew) % P
            self.vecs[j] *= top * invmod(bot, P) % P
            self.vecs[j] %= P

        self.xs[jj] = xnew
        self.ys[jj] = ynew

        # vecs[jj] rebuild
        top = 1
        bot = 1
        for i in xrange(self.n):
            if i == jj:
                continue
            top *= (-self.xs[i])
            top %= P
            bot *= (self.xs[jj] - self.xs[i])
            bot %= P
        self.vecs[jj] = top * invmod(bot, P) % P
        return True

import ast, os

vals = ast.literal_eval(open("challenge").read())

shuffle(vals)
I = Interpol(vals[:25])

open("/tmp/tags/%d" % sum(I.ys), "w").close()
open("/tmp/ans/%d" % I.ans(), "w").close()

ntried = 0
while True:
    # random walk on subsets until a decoded value is repeated
    # tag is hash of the subset, needed to avoid false positives
    pos = randint(0, 24)
    valnew = choice(vals)
    if not I.replace(pos, valnew):
        continue
    ntried += 1

    ans = I.ans()
    tag = sum(I.ys)
    if os.path.isfile("/tmp/ans/%d" % ans) and not os.path.isfile("/tmp/tags/%d" % tag):
        print "Decoded!", "itr", ntried, "ans", ans
        break
    open("/tmp/tags/%d" % sum(I.ys), "w").close()
    open("/tmp/ans/%d" % I.ans(), "w").close()

2manage.py

#-*- coding:utf-8 -*-

"""
This script gets a challenge from the server and runs workers.
Can be run repeatedly to succeed.
"""

import gevent, os
from gevent.pool import Pool
from gevent.subprocess import check_output

from sock import *

print "cleaning ans, tags"

# shared storage for the poor
os.system("mkdir /tmp/{ans,tags}")
os.system("find /tmp/ans/ -type f -delete")
os.system("find /tmp/tags/ -type f -delete")

print "getting challenge"
f = Sock("ppc2.chal.ctf.westerns.tokyo 42453", timeout=1000)
f.read_line()
vals = []
for i in xrange(1, 101):
    parts = f.read_line().split()
    assert parts[0] == "User"
    assert parts[1] == str(i-1)+":"
    vals.append((i, int(parts[2])))
f.read_until("secret?")

open("challenge", "w").write(`vals`)


# parallel computing is hard!!

def worker(i):
    out = check_output(["./worker.py"])
    print "got solution:", "worker", i, "outout", out
    ans = int(out.split()[-1])
    f.send_line(str(ans))
    print "Answer:"
    print `f.read_all()`
    p.kill()
    quit()

print "running workers"
p = Pool(8)
p.map_async(worker, range(8))

gevent.sleep(30)
print "no luck.."
os.system("pkill worker.py")
p.kill()