are MATH D'ING

Walkthrough

Challenge package

Players receive only chall.py and the remote endpoint. The public description and every message printed by the service are intentionally sarcastic and reveal nothing about the underlying construction.

The flag is 98 characters long, exactly twice the 49-character flag used by the inspiration challenge. The challenge begins at 999 points.

Intended solution

The source builds a deterministic Huffman tree for the flag and returns the encoded bit string without the tree. Its second action repeatedly chooses a uniformly random flag position and returns

P(ord(flag[position]))

where P is a degree-five polynomial with six unknown positive integer coefficients.

1. Request the second action many times. The relative occurrence count of each returned integer reveals the corresponding character frequency. Fit positive integer frequencies whose sum is the known flag length, 98.
2. All polynomial coefficients are positive. On positive inputs, P is strictly increasing, so sorting its outputs gives the same ordering as sorting the original characters.
3. Assign an ordered proxy alphabet to the sorted outputs. Combine the recovered frequencies with the source's frequency-and-symbol tie break to rebuild the exact Huffman tree, then decode the bit string into proxy characters.
4. The first six flag characters are the distinct, known characters Securi. Their proxy characters identify six polynomial outputs. Solve the resulting six-by-six Vandermonde system to recover all six coefficients of P.
5. Evaluate P over printable ASCII, map each output back to its proxy symbol, and substitute through the decoded proxy text.

The supplied solver uses a constrained maximum-likelihood fit for the integer frequencies, exact rational Gaussian elimination for the polynomial, and checks that the recovered flag reproduces the original bit string.

Reference solver

The reference solver supplied with the writeup:

#!/usr/bin/env python3

import argparse
from collections import Counter
from fractions import Fraction
import math
import os
from pathlib import Path
import re
import socket
import subprocess
import sys


FLAG_LENGTH = 98
KNOWN = "Securi"
PROMPT = b"> "


class Node:
    def __init__(self, freq, symbol, left=None, right=None):
        self.freq = freq
        self.symbol = symbol
        self.left = left
        self.right = right

    def __lt__(self, other):
        if self.freq == other.freq:
            return self.symbol < other.symbol
        return self.freq < other.freq


class Channel:
    def __init__(self, host=None, port=None):
        self.buffer = bytearray()
        self.process = None
        self.socket = None

        if host is None:
            root = Path(__file__).resolve().parent
            self.process = subprocess.Popen(
                [sys.executable, "-u", "chall.py"],
                cwd=root,
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.DEVNULL,
            )
            self.read_fd = self.process.stdout.fileno()
            self.write_fd = self.process.stdin.fileno()
        else:
            self.socket = socket.create_connection((host, port), timeout=30)
            self.socket.settimeout(30)

    def send(self, data):
        if self.socket is not None:
            self.socket.sendall(data)
            return

        view = memoryview(data)
        while view:
            written = os.write(self.write_fd, view)
            view = view[written:]

    def recv(self):
        if self.socket is not None:
            data = self.socket.recv(65536)
        else:
            data = os.read(self.read_fd, 65536)
        if not data:
            raise EOFError("service closed the connection")
        return data

    def read_until(self, marker):
        while True:
            position = self.buffer.find(marker)
            if position != -1:
                end = position + len(marker)
                result = bytes(self.buffer[:end])
                del self.buffer[:end]
                return result
            self.buffer.extend(self.recv())

    def read_prompts(self, count):
        found = 0
        search_from = 0
        while found < count:
            position = self.buffer.find(PROMPT, search_from)
            if position == -1:
                search_from = max(0, len(self.buffer) - len(PROMPT) + 1)
                self.buffer.extend(self.recv())
                continue
            found += 1
            search_from = position + len(PROMPT)

        result = bytes(self.buffer[:search_from])
        del self.buffer[:search_from]
        return result

    def close(self):
        try:
            self.send(b"3\n")
        except (BrokenPipeError, EOFError, OSError):
            pass
        if self.socket is not None:
            self.socket.close()
        elif self.process is not None:
            try:
                self.process.wait(timeout=2)
            except subprocess.TimeoutExpired:
                self.process.terminate()


def recover_frequencies(observations, total_length):
    values = sorted(observations)
    samples = sum(observations.values())
    count = len(values)
    negative_infinity = float("-inf")

    scores = [negative_infinity] * (total_length + 1)
    scores[0] = 0.0
    parents = []

    for index, value in enumerate(values):
        next_scores = [negative_infinity] * (total_length + 1)
        next_parents = [None] * (total_length + 1)
        remaining_symbols = count - index - 1
        observed = observations[value]

        for used, previous in enumerate(scores):
            if previous == negative_infinity:
                continue
            maximum = total_length - used - remaining_symbols
            for frequency in range(1, maximum + 1):
                target = used + frequency
                score = previous + observed * math.log(frequency)
                if score > next_scores[target]:
                    next_scores[target] = score
                    next_parents[target] = (used, frequency)

        scores = next_scores
        parents.append(next_parents)

    if scores[total_length] == negative_infinity:
        raise ValueError("could not fit the observed values to the known length")

    result = {}
    used = total_length
    for index in range(count - 1, -1, -1):
        previous, frequency = parents[index][used]
        result[values[index]] = frequency
        used = previous

    if used != 0 or sum(result.values()) != total_length:
        raise AssertionError("frequency recovery failed")
    if samples == 0:
        raise ValueError("no observations")
    return result


def build_tree(frequencies):
    import heapq

    heap = [Node(frequency, symbol) for symbol, frequency in frequencies.items()]
    heapq.heapify(heap)

    while len(heap) > 1:
        left = heapq.heappop(heap)
        right = heapq.heappop(heap)
        heapq.heappush(
            heap,
            Node(
                left.freq + right.freq,
                min(left.symbol, right.symbol),
                left,
                right,
            ),
        )
    return heap[0]


def decode(bits, root):
    result = []
    node = root
    for bit in bits:
        node = node.left if bit == "0" else node.right
        if node.left is None and node.right is None:
            result.append(node.symbol)
            node = root
    if node is not root:
        raise ValueError("trailing partial symbol")
    return "".join(result)


def make_codes(node, prefix="", result=None):
    if result is None:
        result = {}
    if node.left is None and node.right is None:
        result[node.symbol] = prefix
    else:
        make_codes(node.left, prefix + "0", result)
        make_codes(node.right, prefix + "1", result)
    return result


def solve_linear_system(matrix, vector):
    size = len(vector)
    rows = [
        [Fraction(value) for value in matrix[index]] + [Fraction(vector[index])]
        for index in range(size)
    ]

    for column in range(size):
        pivot = next(
            (row for row in range(column, size) if rows[row][column] != 0),
            None,
        )
        if pivot is None:
            raise ValueError("singular system")
        rows[column], rows[pivot] = rows[pivot], rows[column]

        scale = rows[column][column]
        rows[column] = [value / scale for value in rows[column]]

        for row in range(size):
            if row == column:
                continue
            scale = rows[row][column]
            if scale == 0:
                continue
            rows[row] = [
                current - scale * source
                for current, source in zip(rows[row], rows[column])
            ]

    return [rows[index][-1] for index in range(size)]


def evaluate_polynomial(coefficients, value):
    result = Fraction(0)
    for coefficient in coefficients:
        result = result * value + coefficient
    return result


def collect(channel, samples, batch_size):
    channel.read_until(PROMPT)
    channel.send(b"1\n")
    response = channel.read_until(PROMPT)
    match = re.search(rb"([01]{32,})", response)
    if match is None:
        raise ValueError("could not find the long binary string")
    bits = match.group(1).decode("ascii")

    observations = Counter()
    completed = 0
    while completed < samples:
        amount = min(batch_size, samples - completed)
        channel.send(b"2\n" * amount)
        block = channel.read_prompts(amount)
        values = re.findall(rb"Groundbreaking: ([0-9]+)", block)
        if len(values) != amount:
            raise ValueError(f"expected {amount} observations, received {len(values)}")
        observations.update(int(value) for value in values)
        completed += amount
        if completed == samples or completed % 50_000 == 0:
            print(f"collected {completed}/{samples}", file=sys.stderr)

    return bits, observations


def recover_flag(bits, observations):
    value_frequencies = recover_frequencies(observations, FLAG_LENGTH)
    ordered_values = sorted(value_frequencies)
    if len(ordered_values) > 90:
        raise ValueError("proxy alphabet is too large")

    proxies = [chr(33 + index) for index in range(len(ordered_values))]
    value_to_proxy = dict(zip(ordered_values, proxies))
    proxy_to_value = {proxy: value for value, proxy in value_to_proxy.items()}
    proxy_frequencies = {
        value_to_proxy[value]: frequency
        for value, frequency in value_frequencies.items()
    }

    proxy_root = build_tree(proxy_frequencies)
    proxy_text = decode(bits, proxy_root)
    if len(proxy_text) != FLAG_LENGTH:
        raise ValueError("decoded text has the wrong length")

    matrix = []
    vector = []
    for proxy, actual in zip(proxy_text[: len(KNOWN)], KNOWN):
        value = ord(actual)
        matrix.append([value**power for power in range(5, -1, -1)])
        vector.append(proxy_to_value[proxy])
    coefficients = solve_linear_system(matrix, vector)

    proxy_to_actual = {}
    for codepoint in range(32, 127):
        output = evaluate_polynomial(coefficients, codepoint)
        if output.denominator != 1:
            continue
        proxy = value_to_proxy.get(int(output))
        if proxy is not None:
            proxy_to_actual[proxy] = chr(codepoint)

    missing = set(proxy_text) - set(proxy_to_actual)
    if missing:
        raise ValueError(f"unmapped proxy symbols: {sorted(missing)!r}")

    flag = "".join(proxy_to_actual[proxy] for proxy in proxy_text)
    if not flag.startswith("Securinets{") or not flag.endswith("}"):
        raise ValueError("recovered text does not have the expected wrapper")

    actual_root = build_tree(Counter(flag))
    codes = make_codes(actual_root)
    if "".join(codes[symbol] for symbol in flag) != bits:
        raise ValueError("recovered text does not reproduce the binary string")
    return flag


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("host", nargs="?")
    parser.add_argument("port", nargs="?", type=int)
    parser.add_argument("--samples", type=int, default=220_000)
    parser.add_argument("--batch", type=int, default=4_000)
    args = parser.parse_args()
    if (args.host is None) != (args.port is None):
        parser.error("provide both host and port, or neither for a local solve")
    if args.samples < 1 or args.batch < 1:
        parser.error("sample and batch counts must be positive")
    return args


def main():
    args = parse_args()
    channel = Channel(args.host, args.port)
    try:
        bits, observations = collect(channel, args.samples, args.batch)
    finally:
        channel.close()
    print(recover_flag(bits, observations))


if __name__ == "__main__":
    main()

Flag

Securinets{you_sampled_the_same_box_forever_and_now_you_are_suddenly_a_mathematician_how_precious}