simple utility for testing a CTFd instance

CTFd_util.py

import requests
import re
import logging
import time


class CTFd_Client:
    nonce_re = re.compile(r'\'?csrf(Nonce|_nonce)\'? ?[:=] "(.*)"')
    _header = {
        'Accept-Language': 'en-US,en;q=0.5',
        'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.14; rv:69.0) Gecko/20100101 Firefox/69.0'
    }

    def __init__(self, url, username=None, password=None):
        self.url = url
        self.ses = requests.session()
        self.ses.headers.update(self._header)
        if username:
            self.login(username, password)

    def _extract_token(self, text):
        try:
            self.ses.headers.pop('CSRF-Token')
        except:
            pass
        nonce = self.nonce_re.findall(text)[0][1]
        logging.info('updated csrf token')
        self.ses.headers.setdefault('CSRF-Token', nonce)

    def login(self, username, passwd):
        self._extract_token(self.ses.get(self.url+'/login').text)
        self._extract_token(self.ses.post(self.url+'/login', data={
            'name': username,
            'password': passwd,
            'nonce': self.ses.headers.get('CSRF-Token'),
        }).text)

    def view(self, chall_id):
        x = self.ses.get(
            self.url+'/api/v1/challenges/'+str(chall_id)
        ).json()
        return x['data']['description']

    def submit(self, chall_id, answer):
        x = self.ses.post(self.url+'/api/v1/challenges/attempt', json={
            'challenge_id': chall_id,
            'submission': answer
        }).json()
        if 'data' in x.keys():
            print(x['data']['status'])
            while x['data']['status'] == 'ratelimited':
                time.sleep(10)
                x = self.ses.post(self.url+'/api/v1/challenges/attempt', json={
                    'challenge_id': chall_id,
                    'submission': answer
                }).json()
                print(x['data']['status'])
        return x
    
    @staticmethod
    def _parse_events(stream):
        last = chr(0)
        buf = ''
        for c in stream:
            if last == '\n' and c == '\n':
                yield buf.rstrip()
                buf = ''
            buf += c
            last = c

    def events(self):
        for event in self._parse_events(
            self.ses.get(
                self.url+'/events', stream=True
            ).iter_content(1, decode_unicode=True)
        ):
            yield event


cli = CTFd_Client(input('url: '), input('username: '), input('password: '))

for i in cli.events():
    print(i)